#!/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 import sys from collections import Counter from pathlib import Path from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from training_dataset_eligibility import ( # noqa: E402 TrainingEligibilityError, assert_frozen_manifest_training_eligible, ) from training_release_manifest import ( # noqa: E402 TrainingReleaseError, assert_yolo_summary_bound_to_embedded_training_release, create_training_release_manifest, ) 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 spread_repeats(paths: list[str], *, sampling_round: int, lane: str) -> list[str]: """Deterministically spread adjacent repeats before a regional cap.""" indexed = enumerate(paths) return [ path for _index, path in sorted( indexed, key=lambda item: hashlib.sha256( f"{sampling_round}:{lane}:{item[0]}:{item[1]}".encode() ).digest(), ) ] def interleave(left: list[str], right: list[str]) -> list[str]: combined: list[str] = [] for index in range(max(len(left), len(right))): if index < len(left): combined.append(left[index]) if index < len(right): combined.append(right[index]) return combined 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") protected_feedback = [ role for role in ("test", "background") if assessment.get(role) is not None ] if protected_feedback: raise ValueError( "Failure-driven sampling is prohibited after protected " + "/".join(protected_feedback) + " evidence was opened" ) 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("calibration") if not evaluation or "regions" not in evaluation: raise ValueError("Assessment has no regional calibration 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]] = {} priority_positive_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) priority_positives = priority_positive_paths_by_region.setdefault(region, []) extras = extra_paths_by_region.setdefault(region, []) repeated = [path] * (repeat - 1) if tile["label_count"] > 0 and context_key in weak_recall_contexts: # Preserve diagnostic positives before precision-oriented # negatives when the regional cap truncates repeat entries. priority_positives.extend(repeated) elif 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(priority_positive_paths_by_region[region]) + 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) priority_positives = spread_repeats( priority_positive_paths_by_region[region], sampling_round=sampling_round, lane=f"{region}:positive", ) corrections = spread_repeats( extra_paths_by_region[region], sampling_round=sampling_round, lane=f"{region}:correction", ) extras = interleave(priority_positives, corrections) 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": "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())), "priority_positive_repeat_count": sum( len(paths) for paths in priority_positive_paths_by_region.values() ), "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( "--review-audit", type=Path, help="Passed corpus audit containing accepted human-review evidence for the frozen corpus.", ) 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) parser.add_argument( "--fixture-mode", action="store_true", help="Accept only an explicitly fixture-only corpus manifest; never use for operational sampling.", ) args = parser.parse_args() try: assert_frozen_manifest_training_eligible( args.corpus_manifest, fixture_mode=args.fixture_mode, verify_live=True, ) source_release = assert_yolo_summary_bound_to_embedded_training_release( summary_path=args.summary, corpus_manifest=args.corpus_manifest, fixture_mode=args.fixture_mode, ) except (TrainingEligibilityError, TrainingReleaseError) as exc: raise SystemExit(str(exc)) from exc 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 = Path(str(source_release["dataset_yaml"]["path"])) 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", ) try: release_paths = create_training_release_manifest( train_yaml=dataset_yaml, corpus_manifest=args.corpus_manifest, review_audit_path=args.review_audit, fixture_mode=args.fixture_mode, ) except TrainingReleaseError as exc: raise SystemExit(str(exc)) from exc 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), "training_release_manifest": str(release_paths["release_manifest"]), "training_release_manifest_sha256": file_sha256(release_paths["release_manifest"]), "training_release_freeze": str(release_paths["release_freeze"]), "training_asset_manifest": str(release_paths["asset_manifest"]), "fixture_mode": bool(args.fixture_mode), } ) 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())