#!/usr/bin/env python3 """Assemble immutable detector input pairs from persisted governed Datasets.""" from __future__ import annotations import argparse import hashlib import json import shutil import sys from pathlib import Path from typing import Any from uuid import UUID from pyproj import Transformer from shapely.geometry import box from shapely.ops import transform as shapely_transform REPO_ROOT = Path(__file__).resolve().parents[1] APP_ROOT = REPO_ROOT if (REPO_ROOT / "app").is_dir() else REPO_ROOT / "backend" for import_root in (Path(__file__).resolve().parent, APP_ROOT): if str(import_root) not in sys.path: sys.path.insert(0, str(import_root)) from app.db.session import SessionLocal # noqa: E402 from app.models import Dataset # noqa: E402 from normalize_belgium_building_labels import normalize # noqa: E402 REGION_SOURCES = { "flanders": ({"digitaal_vlaanderen_orthophoto"}, "grb"), "wallonia": ({"spw_orthophoto"}, "spw_picc"), "brussels": ({"urbis_orthophoto", "digitaal_vlaanderen_orthophoto"}, "urbis"), } SPLITS = {"train", "val", "calibration", "test", "background-test"} 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 _dataset_path(dataset: Dataset) -> Path: if not dataset.storage_path: raise SystemExit(f"Dataset {dataset.id} has no persisted storage path") path = Path(dataset.storage_path) if not path.is_file(): raise SystemExit(f"Dataset {dataset.id} artifact is unreadable: {path}") return path def _validate_pair(sample: dict[str, Any], raster: Dataset, reference: Dataset) -> tuple[str, str]: region = str(sample.get("region") or "").lower() if region not in REGION_SOURCES: raise SystemExit(f"Unsupported region for {sample.get('sample_slug')}: {region}") expected_rasters, expected_reference = REGION_SOURCES[region] if raster.source_name not in expected_rasters: raise SystemExit(f"Raster provider mismatch for {sample['sample_slug']}: {raster.source_name}") if reference.source_name != expected_reference or reference.reference_layer_name != "buildings": raise SystemExit(f"Reference provider/layer mismatch for {sample['sample_slug']}") split = str(sample.get("split") or "") if split not in SPLITS: raise SystemExit(f"Unsupported split for {sample['sample_slug']}: {split}") if raster.status != "ready" or reference.status != "ready": raise SystemExit(f"Dataset pair is not ready for {sample['sample_slug']}") return region, expected_reference 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) findings: list[dict[str, Any]] = [] metric_boxes: list[tuple[dict[str, Any], Any]] = [] for sample in samples: bounds = sample.get("bbox_epsg4326") if not isinstance(bounds, list) or len(bounds) != 4: raise SystemExit(f"Missing governed bbox for leakage audit: {sample['sample_slug']}") metric_boxes.append((sample, shapely_transform(transformer.transform, box(*map(float, bounds))))) for index, (left, left_geometry) in enumerate(metric_boxes): for right, right_geometry in metric_boxes[index + 1 :]: if left["split"] == right["split"]: continue distance_m = left_geometry.distance(right_geometry) if distance_m < buffer_m: findings.append( { "left": left["sample_slug"], "left_split": left["split"], "right": right["sample_slug"], "right_split": right["split"], "distance_m": distance_m, } ) return {"status": "ok" if not findings else "failed", "buffer_m": buffer_m, "findings": findings} def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--spec", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--version", default="building-be-v1") parser.add_argument("--min-label-px", type=float, default=3.0) parser.add_argument("--merge-touching-roofs", action="store_true") parser.add_argument("--freeze", action="store_true") args = parser.parse_args() spec = json.loads(args.spec.read_text(encoding="utf-8-sig")) samples = spec.get("samples") if not isinstance(samples, list) or not samples: raise SystemExit("Corpus spec must contain at least one sample") output_dir = args.output_dir.resolve() if output_dir.exists() and any(output_dir.iterdir()): raise SystemExit(f"Refusing to overwrite non-empty corpus directory: {output_dir}") output_dir.mkdir(parents=True, exist_ok=True) pairs_dir = output_dir / "pairs" pairs_dir.mkdir() manifest_samples: list[dict[str, Any]] = [] seen_slugs: set[str] = set() with SessionLocal() as db: for sample in samples: slug = str(sample.get("sample_slug") or "").strip() if not slug or slug in seen_slugs: raise SystemExit(f"Missing or duplicate sample_slug: {slug}") seen_slugs.add(slug) raster = db.get(Dataset, UUID(str(sample["raster_dataset_id"]))) reference = db.get(Dataset, UUID(str(sample["reference_dataset_id"]))) if raster is None or reference is None: raise SystemExit(f"Persisted Dataset pair not found for {slug}") region, reference_source = _validate_pair(sample, raster, reference) raster_source = _dataset_path(raster) reference_source_path = _dataset_path(reference) sample_dir = pairs_dir / slug sample_dir.mkdir() raster_target = sample_dir / "image.tif" normalized_target = sample_dir / "buildings.normalized.geojson" audit_target = sample_dir / "label-audit.json" shutil.copyfile(raster_source, raster_target) normalized, audit = normalize( reference_path=reference_source_path, raster_path=raster_target, source_name=reference_source, min_label_px=args.min_label_px, imagery_observed_at=( raster.observed_at.isoformat() if raster.observed_at and (raster.source_metadata or {}).get("observation_time_precision") != "unknown_per_pixel" else None ), reference_observed_at=reference.observed_at.isoformat() if reference.observed_at else None, imagery_valid_to=raster.valid_to.isoformat() if raster.valid_to else None, merge_touching_roofs=args.merge_touching_roofs, ) normalized_target.write_text(json.dumps(normalized, ensure_ascii=False), encoding="utf-8") audit_target.write_text(json.dumps(audit, ensure_ascii=False, indent=2), encoding="utf-8") manifest_samples.append( { "sample_slug": slug, "sample_role": sample.get("sample_role", "positive"), "require_empty": bool(sample.get("require_empty", False)), "region": region, "context": sample.get("context"), "split": sample["split"], "raster_path": str(raster_target), "reference_path": str(normalized_target), "reference_source": reference_source, "reference_layer": "buildings", "reference_feature_count": audit["accepted_feature_count"], "raster_dataset_id": str(raster.id), "reference_dataset_id": str(reference.id), "raster_sha256": sha256(raster_target), "reference_sha256": sha256(normalized_target), "label_audit_sha256": sha256(audit_target), "bbox_epsg4326": (raster.source_metadata or {}).get("bbox_epsg4326") or sample.get("bbox_epsg4326"), } ) manifest = { "schema_version": 1, "dataset_version": args.version, "immutable": bool(args.freeze), "samples": manifest_samples, } manifest_path = output_dir / "operator_samples_manifest.json" manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") leakage_audit = audit_spatial_leakage(manifest_samples) (output_dir / "spatial-leakage-audit.json").write_text( json.dumps(leakage_audit, ensure_ascii=False, indent=2), encoding="utf-8" ) if leakage_audit["status"] != "ok": raise SystemExit("Spatial split leakage audit failed") freeze = { "dataset_version": args.version, "manifest_sha256": sha256(manifest_path), "sample_count": len(manifest_samples), "immutable": bool(args.freeze), } (output_dir / "corpus-freeze.json").write_text(json.dumps(freeze, indent=2), encoding="utf-8") print(json.dumps(freeze)) return 0 if __name__ == "__main__": raise SystemExit(main())