#!/usr/bin/env python3 """Build a leak-free regional YOLO view over an immutable tiled corpus.""" from __future__ import annotations import argparse import hashlib import json 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, ) PROTECTED_SPLITS = {"calibration", "test", "background-test", "challenge"} def 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 write_lines(path: Path, values: list[str]) -> None: path.write_text("".join(f"{value}\n" for value in values), encoding="utf-8") def build( *, summary: dict[str, Any], manifest: dict[str, Any], region: str, priority_contexts: set[str], priority_repeat: int, negative_repeat: int, ) -> tuple[list[str], list[str], dict[str, Any]]: samples = {item["sample_slug"]: item for item in manifest["samples"]} selected: list[tuple[dict[str, Any], dict[str, Any]]] = [] protected: list[str] = [] unknown: list[str] = [] for tile in summary["tiles"]: if not tile.get("kept", True): continue sample = samples.get(tile["sample_slug"]) if sample is None: unknown.append(tile["sample_slug"]) continue tile_split = str(tile.get("split") or "") manifest_split = str(sample.get("split") or "") if tile_split in PROTECTED_SPLITS or manifest_split in PROTECTED_SPLITS: protected.append(tile["sample_slug"]) continue if tile_split != manifest_split: raise ValueError( f"tile/manifest split mismatch for {tile['sample_slug']}: " f"{tile_split!r} != {manifest_split!r}" ) if sample.get("region") == region and tile_split in {"train", "val"}: selected.append((tile, sample)) if unknown: raise ValueError(f"summary references unknown samples: {sorted(set(unknown))}") if protected: raise ValueError(f"summary contains protected tiles: {sorted(set(protected))}") train: list[str] = [] val: list[str] = [] sample_counts: Counter[str] = Counter() context_counts: Counter[str] = Counter() negative_count = 0 for tile, sample in selected: image_path = str(tile["image_path"]) split = str(tile.get("split") or sample.get("split")) if split == "val": val.append(image_path) continue context = str(sample.get("context") or "unknown") is_negative = bool(tile.get("is_negative")) repeat = negative_repeat if is_negative else priority_repeat if context in priority_contexts else 1 train.extend([image_path] * repeat) sample_counts[sample["sample_slug"]] += repeat context_counts[context] += repeat negative_count += repeat if is_negative else 0 if not train or not val: raise ValueError(f"regional dataset requires non-empty train and val lists: {region}") if set(train) & set(val): raise ValueError("regional train/validation image leakage") evidence = { "schema_version": 1, "status": "ok", "region": region, "priority_contexts": sorted(priority_contexts), "priority_repeat": priority_repeat, "negative_repeat": negative_repeat, "train_entry_count": len(train), "train_unique_image_count": len(set(train)), "validation_image_count": len(val), "negative_train_entry_count": negative_count, "sample_entry_counts": dict(sorted(sample_counts.items())), "context_entry_counts": dict(sorted(context_counts.items())), "protected_samples_in_training": [], "train_validation_overlap": [], } return train, val, evidence def select_paths( summary: dict[str, Any], manifest: dict[str, Any], region: str ) -> tuple[list[str], list[str]]: """Backward-compatible unweighted regional selection API.""" samples = {item["sample_slug"]: item for item in manifest["samples"]} compatible_summary = { **summary, "tiles": [ tile for tile in summary["tiles"] if tile["sample_slug"] in samples and samples[tile["sample_slug"]].get("split") in {"train", "val"} ], } train, val, _evidence = build( summary=compatible_summary, manifest=manifest, region=region, priority_contexts=set(), priority_repeat=1, negative_repeat=1, ) return sorted(train), sorted(val) 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("--region", required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--priority-context", action="append", default=[]) parser.add_argument("--priority-repeat", type=int, default=2) parser.add_argument("--negative-repeat", type=int, default=2) parser.add_argument( "--review-audit", type=Path, help="Accepted corpus-audit evidence; mandatory for an operational training release.", ) parser.add_argument( "--fixture-mode", action="store_true", help="Accept only an explicitly fixture-only corpus manifest; never use for operational training data.", ) args = parser.parse_args() if args.priority_repeat < 1 or args.negative_repeat < 1: parser.error("repeat counts must be positive") summary = json.loads(args.summary.read_text(encoding="utf-8")) manifest = json.loads(args.corpus_manifest.read_text(encoding="utf-8")) try: assert_frozen_manifest_training_eligible( args.corpus_manifest, fixture_mode=args.fixture_mode, verify_live=True, ) 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 train, val, evidence = build( summary=summary, manifest=manifest, region=args.region, priority_contexts=set(args.priority_context), priority_repeat=args.priority_repeat, negative_repeat=args.negative_repeat, ) args.output_dir.mkdir(parents=True, exist_ok=False) train_path = args.output_dir / "train.txt" val_path = args.output_dir / "val.txt" write_lines(train_path, train) write_lines(val_path, val) yaml_path = args.output_dir / "dataset.yaml" yaml_path.write_text( f"path: /\ntrain: {train_path}\nval: {val_path}\nnames:\n 0: building\n", encoding="utf-8", ) try: release_paths = create_training_release_manifest( train_yaml=yaml_path, 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 evidence.update({ "source_summary": str(args.summary), "source_summary_sha256": sha256(args.summary), "corpus_manifest": str(args.corpus_manifest), "corpus_manifest_sha256": sha256(args.corpus_manifest), "train_sha256": sha256(train_path), "validation_sha256": sha256(val_path), "dataset_yaml": str(yaml_path), "training_release_manifest": str(release_paths["release_manifest"]), "training_release_manifest_sha256": 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), }) encoded_evidence = json.dumps(evidence, indent=2) (args.output_dir / "regional-dataset-evidence.json").write_text(encoded_evidence, encoding="utf-8") (args.output_dir / "regional-dataset.json").write_text(encoded_evidence, encoding="utf-8") print(json.dumps(evidence, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())