#!/usr/bin/env python3 """Create a leak-free rotated release portfolio without copying immutable imagery.""" from __future__ import annotations import argparse import hashlib import json import shutil import sys from collections import Counter from pathlib import Path from typing import Any from pyproj import Transformer from shapely.geometry import box from shapely.ops import transform as shapely_transform 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, ) 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 parse_slugs(raw: str) -> set[str]: return {value.strip() for value in raw.split(",") if value.strip()} def validate_dataset_version(value: str) -> str: version = value.strip() if not version or any(character not in "abcdefghijklmnopqrstuvwxyz0123456789-" for character in version): raise ValueError("dataset version must be a lowercase canonical slug") return version def write_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2), encoding="utf-8") def audit_spatial_leakage(samples: list[dict[str, Any]], buffer_m: float = 64.0) -> dict[str, Any]: transformer = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True) metric = [] for sample in samples: bounds = sample.get("bbox_epsg4326") if not isinstance(bounds, list) or len(bounds) != 4: raise SystemExit(f"missing bbox for {sample['sample_slug']}") metric.append((sample, shapely_transform(transformer.transform, box(*map(float, bounds))))) findings = [] for index, (left, left_geometry) in enumerate(metric): for right, right_geometry in metric[index + 1:]: if left["split"] == right["split"]: continue distance = left_geometry.distance(right_geometry) if distance < buffer_m: findings.append({ "left": left["sample_slug"], "left_split": left["split"], "right": right["sample_slug"], "right_split": right["split"], "distance_m": distance, }) return {"status": "ok" if not findings else "failed", "buffer_m": buffer_m, "findings": findings} def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--corpus-manifest", type=Path, required=True) parser.add_argument("--source-summary", type=Path, action="append", required=True) parser.add_argument("--calibration-samples", required=True) parser.add_argument("--test-samples", required=True) parser.add_argument("--background-samples", required=True) parser.add_argument("--internal-val-samples", required=True) parser.add_argument("--version", required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument( "--fixture-mode", action="store_true", help="Accept only an explicitly fixture-only source corpus; never use for operational holdout rotation.", ) args = parser.parse_args() role_slugs = { "calibration": parse_slugs(args.calibration_samples), "test": parse_slugs(args.test_samples), "background-test": parse_slugs(args.background_samples), } internal_val_slugs = parse_slugs(args.internal_val_samples) all_holdouts = set().union(*role_slugs.values()) if sum(map(len, role_slugs.values())) != len(all_holdouts) or internal_val_slugs & all_holdouts: raise SystemExit("rotated holdout lists overlap") 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, ) for path in args.source_summary: assert_yolo_summary_bound_to_embedded_training_release( summary_path=path, corpus_manifest=args.corpus_manifest, fixture_mode=args.fixture_mode, ) except (TrainingEligibilityError, TrainingReleaseError) as exc: raise SystemExit(str(exc)) from exc source_samples = {item["sample_slug"]: item for item in manifest["samples"]} missing = sorted(all_holdouts - source_samples.keys()) if missing: raise SystemExit(f"unknown holdout samples: {missing}") nonfresh = sorted(slug for slug in all_holdouts if source_samples[slug]["split"] != "train") if nonfresh: raise SystemExit(f"rotated holdouts must come from the former train split: {nonfresh}") invalid_internal = sorted( slug for slug in internal_val_slugs if slug not in source_samples or source_samples[slug]["split"] != "train" ) if invalid_internal: raise SystemExit(f"internal validation samples must come from former train: {invalid_internal}") tiles_by_sample: dict[str, list[dict[str, Any]]] = {} summary_hashes = {} for path in args.source_summary: summary = json.loads(path.read_text(encoding="utf-8")) summary_hashes[str(path)] = sha256(path) for tile in summary["tiles"]: slug = tile["sample_slug"] tiles_by_sample.setdefault(slug, []).append({**tile, "source_split": tile["split"]}) missing_tiles = sorted(source_samples.keys() - tiles_by_sample.keys()) if missing_tiles: raise SystemExit(f"manifest samples have no source tiles: {missing_tiles}") rotated_manifest = json.loads(json.dumps(manifest)) rotated_manifest["dataset_version"] = validate_dataset_version(args.version) rotated_manifest["immutable"] = True assignment = {slug: role for role, slugs in role_slugs.items() for slug in slugs} for sample in rotated_manifest["samples"]: sample["previous_split"] = sample["split"] sample["split"] = ( "val" if sample["sample_slug"] in internal_val_slugs else assignment.get(sample["sample_slug"], "train") ) args.output_dir.mkdir(parents=True, exist_ok=True) manifest_path = args.output_dir / "operator_samples_manifest.json" write_json(manifest_path, rotated_manifest) write_json( args.output_dir / "corpus-freeze.json", { "schema_version": 2, "dataset_version": rotated_manifest["dataset_version"], "manifest_sha256": sha256(manifest_path), "sample_count": len(rotated_manifest["samples"]), "immutable": True, "training_eligibility_policy": rotated_manifest["training_eligibility"]["policy_version"], "fixture_mode": bool(args.fixture_mode), }, ) leakage = audit_spatial_leakage(rotated_manifest["samples"]) write_json(args.output_dir / "spatial-leakage-audit.json", leakage) if leakage["status"] != "ok": raise SystemExit(f"rotated spatial leakage audit failed: {leakage['findings']}") source_pairs = args.corpus_manifest.parent / "pairs" for sample in rotated_manifest["samples"]: source_audit = source_pairs / sample["sample_slug"] / "label-audit.json" target_audit = args.output_dir / "pairs" / sample["sample_slug"] / "label-audit.json" target_audit.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source_audit, target_audit) train_tiles: list[dict[str, Any]] = [] internal_val_tiles: list[dict[str, Any]] = [] role_counts: dict[str, Counter[str]] = {} for role in ("calibration", "test", "background-test"): tiles = [] role_counts[role] = Counter() for slug in sorted(role_slugs[role]): role_counts[role][source_samples[slug]["region"]] += 1 tiles.extend({**tile, "split": "val"} for tile in tiles_by_sample[slug]) role_dir = args.output_dir / role summary_path = role_dir / "yolo_tile_dataset_summary.json" write_json(summary_path, {"schema_version": 1, "output_dir": str(role_dir), "tiles": tiles}) for slug, tiles in tiles_by_sample.items(): if slug in all_holdouts: continue if slug in internal_val_slugs: internal_val_tiles.extend({**tile, "split": "val"} for tile in tiles) continue for tile in tiles: # Preserve the original train corpus's internal checkpoint split only. if source_samples[slug]["split"] == "train" and tile["source_split"] == "val": internal_val_tiles.append({**tile, "split": "val"}) else: train_tiles.append({**tile, "split": "train"}) train_dir = args.output_dir / "train" train_dir.mkdir(parents=True, exist_ok=True) train_list = train_dir / "train.txt" val_list = train_dir / "val.txt" train_list.write_text("\n".join(tile["image_path"] for tile in train_tiles) + "\n", encoding="utf-8") val_list.write_text("\n".join(tile["image_path"] for tile in internal_val_tiles) + "\n", encoding="utf-8") (train_dir / "dataset.yaml").write_text( f"path: {train_dir}\ntrain: {train_list}\nval: {val_list}\nnames:\n 0: building\n", encoding="utf-8", ) write_json( train_dir / "yolo_tile_dataset_summary.json", {"schema_version": 1, "output_dir": str(train_dir), "tiles": train_tiles + internal_val_tiles}, ) evidence = { "schema_version": 1, "status": "ok", "strategy": "fresh-former-train-holdout-rotation", "source_manifest": str(args.corpus_manifest), "source_manifest_sha256": sha256(args.corpus_manifest), "dataset_version": rotated_manifest["dataset_version"], "source_summary_sha256": summary_hashes, "rotated_manifest": str(manifest_path), "rotated_manifest_sha256": sha256(manifest_path), "spatial_leakage_status": leakage["status"], "assignments": {role: sorted(slugs) for role, slugs in role_slugs.items()}, "internal_val_samples": sorted(internal_val_slugs), "holdout_sample_counts_by_region": { role: dict(sorted(counts.items())) for role, counts in role_counts.items() }, "train_tile_count": len(train_tiles), "internal_val_tile_count": len(internal_val_tiles), "protected_samples_in_training": sorted(all_holdouts.intersection({tile["sample_slug"] for tile in train_tiles + internal_val_tiles})), "fit_samples_in_internal_validation": sorted( {tile["sample_slug"] for tile in train_tiles} & {tile["sample_slug"] for tile in internal_val_tiles} ), "fixture_mode": bool(args.fixture_mode), "training_eligible": False, "training_eligibility_reason": ( "A rotated split needs a new human review, label contract validation and immutable training release." ), } if evidence["protected_samples_in_training"]: raise SystemExit("protected samples leaked into rotated training lists") if evidence["fit_samples_in_internal_validation"] or not internal_val_tiles: raise SystemExit("internal validation must be non-empty and AOI-disjoint from fit training") write_json(args.output_dir / "holdout-rotation-evidence.json", evidence) print(json.dumps(evidence, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())