Audit Belgian corpus splits and record CUDA evidence

scripts/assemble_belgium_building_corpus.py

@@ -7,14 +7,25 @@ import argparse
 import hashlib
 import json
 import shutil
+import sys
 from pathlib import Path
 from typing import Any
 from uuid import UUID
 
-from app.db.session import SessionLocal
-from app.models import Dataset
+from pyproj import Transformer
+from shapely.geometry import box
+from shapely.ops import transform as shapely_transform
 
-from normalize_belgium_building_labels import normalize
+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"),
@@ -58,6 +69,33 @@ def _validate_pair(sample: dict[str, Any], raster: Dataset, reference: Dataset)
     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)
@@ -137,6 +175,12 @@ def main() -> int:
     }
     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),