#!/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 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 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 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("--output-dir", type=Path, required=True) 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")) 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"] = "building-be-v30-rotated-holdouts-r1" 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) 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), "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} ), } 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())