317 lines
13 KiB
Python
317 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a leak-free YOLO sampling manifest from failed release gates."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import re
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
PRECISION_NEGATIVE_CONTEXTS = {
|
|
"coastal-urban": {"port-hard-negative", "dunes-negative"},
|
|
"industrial": {"industrial-hard-negative", "rail-hard-negative", "port-hard-negative"},
|
|
"ribbon-development": {"farmland-hard-negative", "forest-hard-negative"},
|
|
"rural-town": {"farmland-hard-negative", "forest-hard-negative", "quarry-hard-negative"},
|
|
"regional-architecture": {"forest-hard-negative", "quarry-hard-negative"},
|
|
}
|
|
|
|
|
|
def file_sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def dataset_validation_source(source_yaml: Path) -> str:
|
|
"""Preserve the source dataset's validation contract verbatim."""
|
|
for raw_line in source_yaml.read_text(encoding="utf-8").splitlines():
|
|
key, separator, value = raw_line.partition(":")
|
|
if separator and key.strip() == "val" and value.strip():
|
|
return value.strip()
|
|
raise ValueError(f"Source dataset YAML has no validation source: {source_yaml}")
|
|
|
|
|
|
def build_sampling(
|
|
*,
|
|
summary: dict[str, Any],
|
|
manifest: dict[str, Any],
|
|
assessment: dict[str, Any],
|
|
positive_repeat: int = 3,
|
|
negative_repeat: int = 4,
|
|
precision_positive_repeat: int = 1,
|
|
context_positive_repeat: int = 5,
|
|
context_negative_repeat: int = 6,
|
|
max_region_share: float = 0.65,
|
|
sampling_round: int = 0,
|
|
precision_guard_band: float = 0.03,
|
|
recall_guard_band: float = 0.03,
|
|
) -> tuple[list[str], dict[str, Any]]:
|
|
if assessment.get("status") != "continue_training_loop":
|
|
raise ValueError("Failure-driven sampling requires a failed assessment")
|
|
if min(
|
|
positive_repeat,
|
|
negative_repeat,
|
|
precision_positive_repeat,
|
|
context_positive_repeat,
|
|
context_negative_repeat,
|
|
) < 1:
|
|
raise ValueError("Repeat factors must be positive")
|
|
if not 0 < max_region_share <= 1:
|
|
raise ValueError("max_region_share must be in (0, 1]")
|
|
if sampling_round < 0:
|
|
raise ValueError("sampling_round must be non-negative")
|
|
if precision_guard_band < 0 or recall_guard_band < 0:
|
|
raise ValueError("Guard bands must be non-negative")
|
|
|
|
samples = {item["sample_slug"]: item for item in manifest["samples"]}
|
|
gates = assessment["gates"]
|
|
evaluation = assessment.get("test") or assessment.get("calibration")
|
|
if not evaluation or "regions" not in evaluation:
|
|
raise ValueError("Assessment has no regional calibration or test evidence")
|
|
regions = evaluation["regions"]
|
|
weak_recall_regions = {
|
|
region
|
|
for region, metrics in regions.items()
|
|
if metrics["f1"] < gates["min_region_f1"]
|
|
or metrics["recall"] < gates["min_region_recall"] + recall_guard_band
|
|
}
|
|
weak_precision_regions = {
|
|
region
|
|
for region, metrics in regions.items()
|
|
if metrics["precision"] < gates["min_region_precision"] + precision_guard_band
|
|
}
|
|
background = assessment.get("background")
|
|
background_failed = bool(
|
|
background
|
|
and background["pure_empty_false_positives"]
|
|
> gates["max_pure_empty_false_positives"]
|
|
)
|
|
recall_dominant_regions = {
|
|
region
|
|
for region in weak_recall_regions & weak_precision_regions
|
|
if not background_failed
|
|
and (
|
|
regions[region]["recall"] / gates["min_region_recall"]
|
|
< regions[region]["precision"] / gates["min_region_precision"]
|
|
)
|
|
}
|
|
weak_recall_contexts: set[tuple[str, str]] = set()
|
|
weak_precision_contexts: set[tuple[str, str]] = set()
|
|
for sample_slug, metrics in evaluation.get("samples", {}).items():
|
|
sample = samples.get(sample_slug)
|
|
if not sample:
|
|
continue
|
|
key = (sample["region"], sample.get("context", "unknown"))
|
|
if sample["region"] in weak_recall_regions and (
|
|
metrics["f1"] < gates["min_region_f1"]
|
|
or metrics["recall"] < gates["min_region_recall"]
|
|
):
|
|
weak_recall_contexts.add(key)
|
|
if (
|
|
sample["region"] in weak_precision_regions
|
|
and metrics["precision"] < gates["min_region_precision"]
|
|
):
|
|
weak_precision_contexts.add(key)
|
|
targeted_negative_contexts = set(weak_precision_contexts)
|
|
for region, context in weak_precision_contexts:
|
|
targeted_negative_contexts.update(
|
|
(region, related)
|
|
for related in PRECISION_NEGATIVE_CONTEXTS.get(context, set())
|
|
)
|
|
|
|
base_paths_by_region: dict[str, list[str]] = {}
|
|
extra_paths_by_region: dict[str, list[str]] = {}
|
|
selected_samples: set[str] = set()
|
|
protected_samples: set[str] = set()
|
|
for tile in summary["tiles"]:
|
|
sample = samples[tile["sample_slug"]]
|
|
if sample["split"] != "train" or tile["split"] != "train":
|
|
protected_samples.add(tile["sample_slug"])
|
|
continue
|
|
region = sample["region"]
|
|
context_key = (region, sample.get("context", "unknown"))
|
|
repeat = 1
|
|
if tile["label_count"] > 0 and region in weak_recall_regions:
|
|
repeat = (
|
|
context_positive_repeat
|
|
if context_key in weak_recall_contexts
|
|
else positive_repeat
|
|
)
|
|
elif tile["label_count"] > 0 and region in weak_precision_regions:
|
|
# Precision-only correction still needs positive examples to avoid
|
|
# shifting the classifier toward background and sacrificing recall.
|
|
repeat = precision_positive_repeat
|
|
if (
|
|
tile["label_count"] == 0
|
|
and (background_failed or region in weak_precision_regions)
|
|
and region not in recall_dominant_regions
|
|
):
|
|
repeat = (
|
|
context_negative_repeat
|
|
if context_key in targeted_negative_contexts
|
|
else negative_repeat
|
|
)
|
|
path = str(Path(tile["image_path"]).resolve())
|
|
base_paths_by_region.setdefault(region, []).append(path)
|
|
extras = extra_paths_by_region.setdefault(region, [])
|
|
repeated = [path] * (repeat - 1)
|
|
if tile["label_count"] == 0 and context_key in targeted_negative_contexts:
|
|
# Preserve the most diagnostic hard-negative repeats when the
|
|
# regional cap has to remove lower-priority repetition.
|
|
extras[:0] = repeated
|
|
else:
|
|
extras.extend(repeated)
|
|
selected_samples.add(tile["sample_slug"])
|
|
|
|
pre_cap_counts = Counter({
|
|
region: len(paths) + len(extra_paths_by_region[region])
|
|
for region, paths in base_paths_by_region.items()
|
|
})
|
|
capped_counts = Counter(pre_cap_counts)
|
|
if max_region_share < 1:
|
|
while capped_counts:
|
|
region, count = max(capped_counts.items(), key=lambda item: (item[1], item[0]))
|
|
total = sum(capped_counts.values())
|
|
if count / total <= max_region_share:
|
|
break
|
|
others = total - count
|
|
limit = math.floor(max_region_share / (1 - max_region_share) * others)
|
|
limit = max(limit, len(base_paths_by_region[region]))
|
|
if limit >= count:
|
|
break
|
|
capped_counts[region] = limit
|
|
image_paths: list[str] = []
|
|
for region in sorted(base_paths_by_region):
|
|
base_paths = base_paths_by_region[region]
|
|
extra_limit = capped_counts[region] - len(base_paths)
|
|
image_paths.extend(base_paths)
|
|
extras = extra_paths_by_region[region]
|
|
if extras and extra_limit < len(extras):
|
|
# Rotate the capped repeat window between loop rounds so persistent
|
|
# failures cannot yield the exact same training list indefinitely.
|
|
offset = sampling_round % len(extras)
|
|
extras = extras[offset:] + extras[:offset]
|
|
image_paths.extend(extras[:extra_limit])
|
|
repeat_counts = Counter({region: capped_counts[region] for region in capped_counts})
|
|
if not image_paths:
|
|
raise ValueError("No train-only tiles selected")
|
|
metadata = {
|
|
"schema_version": 1,
|
|
"status": "ok",
|
|
"strategy": "failed-region-positive-and-hard-negative-repeat",
|
|
"failure_evidence_source": "test" if assessment.get("test") else "calibration",
|
|
"weak_recall_regions": sorted(weak_recall_regions),
|
|
"weak_precision_regions": sorted(weak_precision_regions),
|
|
"recall_dominant_regions": sorted(recall_dominant_regions),
|
|
"weak_recall_contexts": [f"{region}:{context}" for region, context in sorted(weak_recall_contexts)],
|
|
"weak_precision_contexts": [f"{region}:{context}" for region, context in sorted(weak_precision_contexts)],
|
|
"targeted_negative_contexts": [
|
|
f"{region}:{context}" for region, context in sorted(targeted_negative_contexts)
|
|
],
|
|
"background_gate_failed": background_failed,
|
|
"positive_repeat": positive_repeat,
|
|
"negative_repeat": negative_repeat,
|
|
"precision_positive_repeat": precision_positive_repeat,
|
|
"context_positive_repeat": context_positive_repeat,
|
|
"context_negative_repeat": context_negative_repeat,
|
|
"max_region_share": max_region_share,
|
|
"sampling_round": sampling_round,
|
|
"precision_guard_band": precision_guard_band,
|
|
"recall_guard_band": recall_guard_band,
|
|
"source_train_tile_count": sum(
|
|
1
|
|
for tile in summary["tiles"]
|
|
if samples[tile["sample_slug"]]["split"] == "train" and tile["split"] == "train"
|
|
),
|
|
"sampled_train_entry_count": len(image_paths),
|
|
"sampled_entries_by_region": dict(sorted(repeat_counts.items())),
|
|
"pre_cap_entries_by_region": dict(sorted(pre_cap_counts.items())),
|
|
"dropped_region_repeat_count": sum(pre_cap_counts.values()) - len(image_paths),
|
|
"selected_train_sample_count": len(selected_samples),
|
|
"protected_sample_count": len(protected_samples),
|
|
"protected_samples_in_training": [],
|
|
}
|
|
return image_paths, metadata
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--summary", type=Path, required=True)
|
|
parser.add_argument("--corpus-manifest", type=Path, required=True)
|
|
parser.add_argument("--assessment", type=Path, required=True)
|
|
parser.add_argument("--output-dir", type=Path, required=True)
|
|
parser.add_argument("--positive-repeat", type=int, default=3)
|
|
parser.add_argument("--negative-repeat", type=int, default=4)
|
|
parser.add_argument("--precision-positive-repeat", type=int, default=1)
|
|
parser.add_argument("--context-positive-repeat", type=int, default=5)
|
|
parser.add_argument("--context-negative-repeat", type=int, default=6)
|
|
parser.add_argument("--max-region-share", type=float, default=0.65)
|
|
parser.add_argument("--sampling-round", type=int)
|
|
parser.add_argument("--precision-guard-band", type=float, default=0.03)
|
|
parser.add_argument("--recall-guard-band", type=float, default=0.03)
|
|
args = parser.parse_args()
|
|
|
|
summary = json.loads(args.summary.read_text(encoding="utf-8"))
|
|
manifest = json.loads(args.corpus_manifest.read_text(encoding="utf-8"))
|
|
assessment = json.loads(args.assessment.read_text(encoding="utf-8"))
|
|
sampling_round = args.sampling_round
|
|
if sampling_round is None:
|
|
match = re.fullmatch(r"iteration-(\d+)", args.output_dir.parent.name)
|
|
sampling_round = int(match.group(1)) if match else 0
|
|
paths, metadata = build_sampling(
|
|
summary=summary,
|
|
manifest=manifest,
|
|
assessment=assessment,
|
|
positive_repeat=args.positive_repeat,
|
|
negative_repeat=args.negative_repeat,
|
|
precision_positive_repeat=args.precision_positive_repeat,
|
|
context_positive_repeat=args.context_positive_repeat,
|
|
context_negative_repeat=args.context_negative_repeat,
|
|
max_region_share=args.max_region_share,
|
|
sampling_round=sampling_round,
|
|
precision_guard_band=args.precision_guard_band,
|
|
recall_guard_band=args.recall_guard_band,
|
|
)
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
train_list = args.output_dir / "train-failure-driven.txt"
|
|
train_list.write_text("\n".join(paths) + "\n", encoding="utf-8")
|
|
source_yaml = args.summary.parent / "dataset.yaml"
|
|
val_source = dataset_validation_source(source_yaml)
|
|
dataset_yaml = args.output_dir / "dataset.yaml"
|
|
dataset_yaml.write_text(
|
|
f"path: {args.output_dir}\n"
|
|
f"train: {train_list}\n"
|
|
f"val: {val_source}\n"
|
|
"names:\n 0: building\n",
|
|
encoding="utf-8",
|
|
)
|
|
metadata.update(
|
|
{
|
|
"summary": str(args.summary),
|
|
"summary_sha256": file_sha256(args.summary),
|
|
"corpus_manifest": str(args.corpus_manifest),
|
|
"corpus_manifest_sha256": file_sha256(args.corpus_manifest),
|
|
"assessment": str(args.assessment),
|
|
"assessment_sha256": file_sha256(args.assessment),
|
|
"source_dataset_yaml": str(source_yaml),
|
|
"train_list": str(train_list),
|
|
"dataset_yaml": str(dataset_yaml),
|
|
}
|
|
)
|
|
output = args.output_dir / "failure-driven-sampling.json"
|
|
output.write_text(json.dumps(metadata, indent=2), encoding="utf-8")
|
|
print(json.dumps(metadata, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|