#!/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 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, ) -> 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 or precision_positive_repeat < 1: raise ValueError("Repeat factors must be positive") 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"] } weak_precision_regions = { region for region, metrics in regions.items() if metrics["precision"] < gates["min_region_precision"] } background = assessment.get("background") background_failed = bool( background and 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 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): 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", "failure_evidence_source": "test" if assessment.get("test") else "calibration", "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, "precision_positive_repeat": precision_positive_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) parser.add_argument("--precision-positive-repeat", type=int, default=1) 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, precision_positive_repeat=args.precision_positive_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_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())