#!/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 from pathlib import Path from typing import Any from uuid import UUID from app.db.session import SessionLocal from app.models import Dataset from normalize_belgium_building_labels import normalize REGION_SOURCES = { "flanders": ("digitaal_vlaanderen_orthophoto", "grb"), "wallonia": ("spw_orthophoto", "spw_picc"), "brussels": ("urbis_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_raster, expected_reference = REGION_SOURCES[region] if raster.source_name != expected_raster: 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 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("--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 else None, reference_observed_at=reference.observed_at.isoformat() if reference.observed_at else None, ) 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"), "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"), } ) 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") 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())