Complete Belgian building corpus v2 workflow

scripts/assemble_belgium_building_corpus.py

@@ -169,7 +169,8 @@ def main() -> int:
                     "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"),
+                    "bbox_epsg4326": (raster.source_metadata or {}).get("bbox_epsg4326")
+                    or sample.get("bbox_epsg4326"),
                 }
             )
     manifest = {

scripts/audit_belgium_building_corpus.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Audit a frozen Belgian corpus and emit a deterministic review queue."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+REQUIRED_SPLITS = ("train", "val", "calibration", "test", "background-test")
+REGIONS = ("flanders", "wallonia", "brussels")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--corpus-dir", type=Path, required=True)
+    parser.add_argument("--output-dir", type=Path, required=True)
+    parser.add_argument("--review-decisions", type=Path)
+    args = parser.parse_args()
+    manifest = json.loads((args.corpus_dir / "operator_samples_manifest.json").read_text(encoding="utf-8"))
+    leakage = json.loads((args.corpus_dir / "spatial-leakage-audit.json").read_text(encoding="utf-8"))
+    samples = manifest["samples"]
+    split_counts = Counter((sample["region"], sample["split"]) for sample in samples)
+    decision_counts: Counter[str] = Counter()
+    review_queue: list[dict[str, Any]] = []
+    total_input = 0
+    total_accepted = 0
+    temporal_unknown = 0
+    failures: list[str] = []
+    for region in REGIONS:
+        for split in REQUIRED_SPLITS:
+            minimum = 4 if split == "train" else 2
+            if split_counts[(region, split)] < minimum:
+                failures.append(f"{region}/{split} has {split_counts[(region, split)]}, requires {minimum}")
+    for sample in samples:
+        audit_path = args.corpus_dir / "pairs" / sample["sample_slug"] / "label-audit.json"
+        audit = json.loads(audit_path.read_text(encoding="utf-8"))
+        total_input += int(audit["input_feature_count"])
+        total_accepted += int(audit["accepted_feature_count"])
+        decision_counts.update(audit["decision_counts"])
+        if audit.get("temporal_alignment_status") == "unknown":
+            temporal_unknown += 1
+        expected_empty = bool(sample.get("sample_role") == "background_candidate" and sample.get("require_empty"))
+        if expected_empty and audit["accepted_feature_count"] != 0:
+            failures.append(f"{sample['sample_slug']} is not pure empty after normalization")
+        priority = "high" if audit["accepted_feature_count"] >= 200 or sample["split"] in {"test", "background-test"} else "normal"
+        review_queue.append(
+            {
+                "sample_slug": sample["sample_slug"],
+                "region": sample["region"],
+                "context": sample.get("context"),
+                "split": sample["split"],
+                "accepted_feature_count": audit["accepted_feature_count"],
+                "temporal_alignment_status": audit.get("temporal_alignment_status"),
+                "priority": priority,
+                "decision": "pending_human_review",
+            }
+        )
+    if leakage.get("status") != "ok":
+        failures.append("spatial leakage audit failed")
+    reviewed = 0
+    review_complete = False
+    if args.review_decisions and args.review_decisions.is_file():
+        decisions = json.loads(args.review_decisions.read_text(encoding="utf-8"))
+        by_slug = {item["sample_slug"]: item for item in decisions.get("decisions", [])}
+        for item in review_queue:
+            decision = by_slug.get(item["sample_slug"])
+            if decision:
+                item["decision"] = decision.get("decision")
+                item["reviewer"] = decision.get("reviewer")
+                item["notes"] = decision.get("notes")
+                if item["decision"] in {"accepted", "rejected"} and item.get("reviewer"):
+                    reviewed += 1
+        review_complete = reviewed == len(review_queue) and all(item["decision"] == "accepted" for item in review_queue)
+    status = "failed" if failures else ("ok" if review_complete else "needs_human_review")
+    report = {
+        "status": status,
+        "dataset_version": manifest["dataset_version"],
+        "manifest_immutable": manifest["immutable"],
+        "sample_count": len(samples),
+        "split_counts": {f"{region}/{split}": split_counts[(region, split)] for region in REGIONS for split in REQUIRED_SPLITS},
+        "input_feature_count": total_input,
+        "accepted_feature_count": total_accepted,
+        "decision_counts": dict(sorted(decision_counts.items())),
+        "temporal_unknown_sample_count": temporal_unknown,
+        "spatial_leakage_status": leakage.get("status"),
+        "reviewed_sample_count": reviewed,
+        "review_complete": review_complete,
+        "failures": failures,
+        "review_queue": review_queue,
+    }
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    (args.output_dir / "belgium-building-corpus-audit.json").write_text(
+        json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
+    )
+    lines = [
+        f"# Belgian building corpus audit: {manifest['dataset_version']}",
+        "",
+        f"Status: `{status}`",
+        f"Samples: {len(samples)}; accepted labels: {total_accepted}/{total_input}.",
+        f"Spatial leakage: `{leakage.get('status')}`; human reviewed: {reviewed}/{len(samples)}.",
+        "",
+        "## Review queue",
+        "",
+        "| Sample | Region | Context | Split | Labels | Priority | Decision |",
+        "| --- | --- | --- | --- | ---: | --- | --- |",
+    ]
+    lines.extend(
+        f"| {item['sample_slug']} | {item['region']} | {item['context']} | {item['split']} | "
+        f"{item['accepted_feature_count']} | {item['priority']} | {item['decision']} |"
+        for item in review_queue
+    )
+    (args.output_dir / "belgium-building-corpus-audit.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
+    print(json.dumps({key: value for key, value in report.items() if key != "review_queue"}, ensure_ascii=False, indent=2))
+    return 1 if failures else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/provision_belgium_building_training_portfolio.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 
 import argparse
 import json
-from dataclasses import asdict, dataclass
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
@@ -31,6 +31,8 @@ AOIS = (
     Aoi("ghent-core-train", "flanders", "dense-urban", "train", 3.725, 51.052),
     Aoi("genk-industry-train", "flanders", "industrial", "train", 5.500, 50.965),
     Aoi("flanders-farms-train", "flanders", "rural-farms", "train", 4.850, 50.900),
+    Aoi("kalmthout-heath-train-bg", "flanders", "heath-negative", "train", 4.450, 51.390, "background_candidate"),
+    Aoi("limburg-forest-train-bg", "flanders", "forest-negative", "train", 5.550, 51.050, "background_candidate"),
     Aoi("bruges-val", "flanders", "historic-urban", "val", 3.224, 51.209),
     Aoi("turnhout-val", "flanders", "suburban", "val", 4.944, 51.322),
     Aoi("hasselt-cal", "flanders", "suburban", "calibration", 5.340, 50.930),
@@ -44,6 +46,8 @@ AOIS = (
     Aoi("charleroi-core-train", "wallonia", "dense-urban", "train", 4.440, 50.410),
     Aoi("seraing-industry-train", "wallonia", "industrial-valley", "train", 5.500, 50.600),
     Aoi("namur-residential-train", "wallonia", "residential", "train", 4.870, 50.470),
+    Aoi("ardennes-forest-train-bg", "wallonia", "forest-negative", "train", 5.700, 50.200, "background_candidate"),
+    Aoi("wallonia-quarry-train-hard", "wallonia", "quarry-hard-negative", "train", 5.130, 50.530, "background_candidate"),
     Aoi("tournai-val", "wallonia", "historic-urban", "val", 3.389, 50.606),
     Aoi("arlon-val", "wallonia", "small-city", "val", 5.817, 49.683),
     Aoi("verviers-cal", "wallonia", "suburban", "calibration", 5.860, 50.590),
@@ -57,13 +61,15 @@ AOIS = (
     Aoi("anderlecht-industry-train", "brussels", "industrial", "train", 4.320, 50.880),
     Aoi("uccle-residential-train", "brussels", "detached-residential", "train", 4.350, 50.795),
     Aoi("schaerbeek-train", "brussels", "dense-residential", "train", 4.380, 50.865),
+    Aoi("brussels-rail-train-hard", "brussels", "rail-hard-negative", "train", 4.345, 50.875, "background_candidate"),
+    Aoi("brussels-park-train-hard", "brussels", "park-hard-negative", "train", 4.400, 50.820, "background_candidate"),
     Aoi("woluwe-val", "brussels", "suburban", "val", 4.430, 50.845),
     Aoi("molenbeek-val", "brussels", "mixed-urban", "val", 4.325, 50.855),
     Aoi("brussels-park-cal", "brussels", "park-edge", "calibration", 4.380, 50.820),
     Aoi("brussels-canal-cal", "brussels", "canal-industry", "calibration", 4.340, 50.870),
     Aoi("brussels-rail-test", "brussels", "rail-context", "test", 4.330, 50.840),
     Aoi("jette-test", "brussels", "residential-park", "test", 4.325, 50.880),
-    Aoi("sonian-forest-hard", "brussels", "forest-hard-negative", "background-test", 4.410, 50.770, "background_candidate"),
+    Aoi("sonian-forest-hard", "brussels", "forest-hard-negative", "background-test", 4.420, 50.790, "background_candidate"),
     Aoi("bois-cambre-hard", "brussels", "park-hard-negative", "background-test", 4.375, 50.795, "background_candidate"),
 )
 
@@ -100,7 +106,8 @@ def bbox_for_center(lon: float, lat: float, side_m: float) -> dict[str, Any]:
 
 def post(session: requests.Session, url: str, payload: dict[str, Any]) -> dict[str, Any]:
     response = session.post(url, json=payload, timeout=180)
-    response.raise_for_status()
+    if not response.ok:
+        raise RuntimeError(f"Provider workflow HTTP {response.status_code}: {response.text[:1000]}")
     body = response.json()
     if body.get("error"):
         raise RuntimeError(f"{body['error']}: {body.get('message')}")
@@ -121,7 +128,17 @@ def main() -> int:
     args = parser.parse_args()
     session = requests.Session()
     samples: list[dict[str, Any]] = []
+    if args.output_spec.is_file():
+        existing = json.loads(args.output_spec.read_text(encoding="utf-8-sig"))
+        samples = list(existing.get("samples") or [])
+        for sample in samples:
+            if "sample_slug" not in sample and sample.get("slug"):
+                sample["sample_slug"] = sample.pop("slug")
+    completed = {str(sample.get("sample_slug") or sample.get("slug")) for sample in samples}
     for aoi in AOIS:
+        if aoi.slug in completed:
+            print(f"{aoi.slug}: checkpoint reused", flush=True)
+            continue
         contract = REGION_CONTRACT[aoi.region]
         bbox = bbox_for_center(aoi.lon, aoi.lat, args.side_m)
         common = {"bbox": bbox, "area_id": contract["area_id"], "force_refresh": args.force_refresh}
@@ -140,18 +157,33 @@ def main() -> int:
             raise RuntimeError(f"Pure-background AOI {aoi.slug} contains {feature_count} reference buildings")
         samples.append(
             {
-                **asdict(aoi),
+                "sample_slug": aoi.slug,
+                "region": aoi.region,
+                "context": aoi.context,
+                "split": aoi.split,
+                "sample_role": aoi.sample_role,
+                "require_empty": aoi.require_empty,
                 "bbox_epsg4326": [bbox["min_x"], bbox["min_y"], bbox["max_x"], bbox["max_y"]],
                 "raster_dataset_id": image_job["output_dataset_id"],
                 "reference_dataset_id": reference_job["output_dataset_id"],
                 "provider_reference_feature_count": feature_count,
             }
         )
+        checkpoint = {
+            "schema_version": 1,
+            "status": "in_progress",
+            "side_m": args.side_m,
+            "resolution_m": args.resolution_m,
+            "samples": samples,
+        }
+        args.output_spec.parent.mkdir(parents=True, exist_ok=True)
+        args.output_spec.write_text(json.dumps(checkpoint, ensure_ascii=False, indent=2), encoding="utf-8")
         print(f"{aoi.slug}: {feature_count} reference buildings", flush=True)
     payload = {
         "schema_version": 1,
         "side_m": args.side_m,
         "resolution_m": args.resolution_m,
+        "status": "complete",
         "samples": samples,
     }
     args.output_spec.parent.mkdir(parents=True, exist_ok=True)