Add fail-closed failure-driven YOLO sampling

scripts/build_failure_driven_yolo_sampling.py

@@ -0,0 +1,152 @@
+#!/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
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+
+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 build_sampling(
+    *,
+    summary: dict[str, Any],
+    manifest: dict[str, Any],
+    assessment: dict[str, Any],
+    positive_repeat: int = 3,
+    negative_repeat: int = 4,
+) -> tuple[list[str], dict[str, Any]]:
+    if assessment.get("status") != "continue_training_loop":
+        raise ValueError("Failure-driven sampling requires a failed assessment")
+    if positive_repeat < 1 or negative_repeat < 1:
+        raise ValueError("Repeat factors must be positive")
+
+    samples = {item["sample_slug"]: item for item in manifest["samples"]}
+    gates = assessment["gates"]
+    regions = assessment["test"]["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"]
+    }
+    weak_precision_regions = {
+        region
+        for region, metrics in regions.items()
+        if metrics["precision"] < gates["min_region_precision"]
+    }
+    background_failed = (
+        assessment["background"]["pure_empty_false_positives"]
+        > gates["max_pure_empty_false_positives"]
+    )
+
+    image_paths: list[str] = []
+    repeat_counts: Counter[str] = Counter()
+    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"]
+        repeat = 1
+        if tile["label_count"] > 0 and region in weak_recall_regions:
+            repeat = positive_repeat
+        if tile["label_count"] == 0 and (background_failed or region in weak_precision_regions):
+            repeat = negative_repeat
+        path = str(Path(tile["image_path"]).resolve())
+        image_paths.extend([path] * repeat)
+        repeat_counts[region] += repeat
+        selected_samples.add(tile["sample_slug"])
+
+    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",
+        "weak_recall_regions": sorted(weak_recall_regions),
+        "weak_precision_regions": sorted(weak_precision_regions),
+        "background_gate_failed": background_failed,
+        "positive_repeat": positive_repeat,
+        "negative_repeat": negative_repeat,
+        "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())),
+        "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)
+    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"))
+    paths, metadata = build_sampling(
+        summary=summary,
+        manifest=manifest,
+        assessment=assessment,
+        positive_repeat=args.positive_repeat,
+        negative_repeat=args.negative_repeat,
+    )
+    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_dir = args.summary.parent / "images" / "val"
+    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_dir}\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())