evaluate models on fresh regional calibration AOIs

scripts/README.md

@@ -53,6 +53,23 @@ marker before release verification or CUDA allocation. Non-overlap proves only
 that tiles do not share pixels; checkpoint reports explicitly avoid claiming
 statistical independence for adjacent tiles from the same AOI.
 
+`provision_belgium_building_training_portfolio.py --aoi-spec` provisions a
+pre-registered custom portfolio but deliberately accepts only `train`, `val`
+and `calibration`; protected test roles are rejected. For regional diagnostic
+work where the PICC/UrbIS harmonisation contract is still pending,
+`assemble_belgium_building_corpus.py
+--evaluation-only-pending-regional-contracts` preserves the exact authority
+failure, writes `NO_TRAINING.json` and makes no release claim.
+
+`export_yolo_diagnostic_evaluation_tiles.py` accepts only such a
+training-prohibited calibration manifest. It emits empty train directories and
+retains only full canonical non-overlapping grid cells; generic final-edge
+coverage windows are removed from the generated view and listed explicitly.
+`audit_evaluation_lineage_spatial_independence.py` then resolves every ancestral
+sample back to its georeferenced source raster and checks the frozen metric
+distance floor. These controls supplement, rather than replace, exact sample
+lineage checking in the checkpoint evaluator.
+
 Setup-, import-, demo- en maintenance-scripts voor GeoIntel.
 
 ## WALOUS source provisioning

scripts/assemble_belgium_building_corpus.py

@@ -62,6 +62,7 @@ def _validate_pair(
     reference: Dataset,
     *,
     fixture_mode: bool,
+    evaluation_only_pending_regional_contracts: bool,
 ) -> tuple[str, str, dict[str, Any]]:
     region = str(sample.get("region") or "").lower()
     if region not in REGION_SOURCES:
@@ -89,9 +90,22 @@ def _validate_pair(
                 for reason in eligibility[role]["reasons"]
             }
         )
-        raise SystemExit(
-            f"Dataset pair is not eligible for training for {sample['sample_slug']}: {', '.join(reasons)}"
+        allowed_pending = bool(
+            evaluation_only_pending_regional_contracts
+            and split == "calibration"
+            and region in {"wallonia", "brussels"}
+            and set(reasons) == {"reference_building_validation_not_primary"}
         )
+        if not allowed_pending:
+            raise SystemExit(
+                f"Dataset pair is not eligible for training for {sample['sample_slug']}: {', '.join(reasons)}"
+            )
+        eligibility["evaluation_only_exception"] = {
+            "allowed": True,
+            "reason": "regional_building_authority_contract_pending",
+            "training_allowed": False,
+            "release_claim_allowed": False,
+        }
     return region, expected_reference, eligibility
 
 
@@ -130,6 +144,15 @@ def main() -> int:
     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")
+    parser.add_argument(
+        "--evaluation-only-pending-regional-contracts",
+        action="store_true",
+        help=(
+            "Allow calibration-only PICC/UrbIS pairs whose sole training gate "
+            "failure is a pending regional authority contract. Writes an "
+            "explicit NO_TRAINING marker and never makes a release claim."
+        ),
+    )
     parser.add_argument(
         "--fixture-mode",
         action="store_true",
@@ -168,6 +191,9 @@ def main() -> int:
                 raster,
                 reference,
                 fixture_mode=args.fixture_mode,
+                evaluation_only_pending_regional_contracts=(
+                    args.evaluation_only_pending_regional_contracts
+                ),
             )
             raster_source = _dataset_path(raster)
             reference_source_path = _dataset_path(reference)
@@ -223,9 +249,18 @@ def main() -> int:
         "immutable": bool(args.freeze),
         "training_eligibility": {
             "policy_version": TRAINING_ELIGIBILITY_POLICY_VERSION,
-            "status": "eligible",
+            "status": (
+                "not_eligible_evaluation_only"
+                if args.evaluation_only_pending_regional_contracts
+                else "eligible"
+            ),
             "fixture_mode": bool(args.fixture_mode),
         },
+        "purpose": (
+            "non_protected_diagnostic_evaluation"
+            if args.evaluation_only_pending_regional_contracts
+            else "training_corpus"
+        ),
         "samples": manifest_samples,
     }
     manifest_path = output_dir / "operator_samples_manifest.json"
@@ -244,8 +279,21 @@ def main() -> int:
         "immutable": bool(args.freeze),
         "training_eligibility_policy": TRAINING_ELIGIBILITY_POLICY_VERSION,
         "fixture_mode": bool(args.fixture_mode),
+        "training_allowed": not args.evaluation_only_pending_regional_contracts,
+        "release_claim_allowed": False,
     }
     (output_dir / "corpus-freeze.json").write_text(json.dumps(freeze, indent=2), encoding="utf-8")
+    if args.evaluation_only_pending_regional_contracts:
+        marker = {
+            "schema_version": 1,
+            "reason": "evaluation_only_pending_regional_authority_contracts",
+            "training_allowed": False,
+            "release_claim_allowed": False,
+            "manifest_sha256": freeze["manifest_sha256"],
+        }
+        (output_dir / "NO_TRAINING.json").write_text(
+            json.dumps(marker, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+        )
     print(json.dumps(freeze))
     return 0
 

scripts/audit_evaluation_lineage_spatial_independence.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""Audit fresh evaluation AOIs against every raster in a model's corpus lineage."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+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_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def metric_box(bounds: list[float] | tuple[float, ...]) -> Any:
+    transformer = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True)
+    return shapely_transform(transformer.transform, box(*map(float, bounds)))
+
+
+def raster_bounds_epsg4326(path: Path) -> list[float]:
+    import rasterio
+    from rasterio.warp import transform_bounds
+
+    with rasterio.open(path) as dataset:
+        if dataset.crs is None:
+            raise ValueError(f"lineage raster has no CRS: {path}")
+        bounds = transform_bounds(dataset.crs, "EPSG:4326", *dataset.bounds)
+    return list(map(float, bounds))
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--evaluation-manifest", type=Path, required=True)
+    parser.add_argument("--lineage-summary", type=Path, action="append", required=True)
+    parser.add_argument("--manifest-search-root", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--minimum-distance-m", type=float, default=2000.0)
+    args = parser.parse_args()
+    if args.output.exists():
+        parser.error(f"output already exists: {args.output}")
+    evaluation_manifest = args.evaluation_manifest.resolve(strict=True)
+    evaluation = json.loads(evaluation_manifest.read_text(encoding="utf-8-sig"))
+    evaluation_rows = evaluation.get("samples")
+    if not isinstance(evaluation_rows, list) or not evaluation_rows:
+        raise SystemExit("evaluation manifest contains no samples")
+
+    lineage_slugs: set[str] = set()
+    summary_evidence: list[dict[str, Any]] = []
+    for raw in args.lineage_summary:
+        summary = raw.resolve(strict=True)
+        payload = json.loads(summary.read_text(encoding="utf-8"))
+        tiles = payload.get("tiles")
+        if not isinstance(tiles, list):
+            raise SystemExit(f"lineage summary has no tiles: {summary}")
+        slugs = {str(row.get("sample_slug") or "").strip() for row in tiles}
+        slugs.discard("")
+        lineage_slugs.update(slugs)
+        summary_evidence.append(
+            {"path": str(summary), "sha256": sha256_file(summary), "sample_slugs": sorted(slugs)}
+        )
+
+    candidates: dict[str, dict[str, dict[str, Any]]] = {
+        slug: {} for slug in lineage_slugs
+    }
+    for manifest_path in sorted(
+        args.manifest_search_root.rglob("operator_samples_manifest.json")
+    ):
+        try:
+            payload = json.loads(manifest_path.read_text(encoding="utf-8-sig"))
+        except (OSError, json.JSONDecodeError):
+            continue
+        for sample in payload.get("samples") or []:
+            if not isinstance(sample, dict):
+                continue
+            slug = str(sample.get("sample_slug") or "").strip()
+            raster_raw = sample.get("raster_path")
+            if slug not in candidates or not isinstance(raster_raw, str):
+                continue
+            raster = Path(raster_raw)
+            if not raster.is_file():
+                continue
+            key = str(raster.resolve())
+            candidates[slug].setdefault(
+                key,
+                {
+                    "sample_slug": slug,
+                    "raster_path": key,
+                    "declared_raster_sha256": sample.get("raster_sha256"),
+                    "source_manifest_path": str(manifest_path),
+                    "source_manifest_sha256": sha256_file(manifest_path),
+                },
+            )
+
+    missing = sorted(slug for slug, rows in candidates.items() if not rows)
+    lineage_geometries: list[tuple[dict[str, Any], Any]] = []
+    for rows in candidates.values():
+        for record in rows.values():
+            bounds = raster_bounds_epsg4326(Path(record["raster_path"]))
+            record["bbox_epsg4326"] = bounds
+            lineage_geometries.append((record, metric_box(bounds)))
+
+    results: list[dict[str, Any]] = []
+    for sample in evaluation_rows:
+        slug = str(sample.get("sample_slug") or "").strip()
+        bounds = sample.get("bbox_epsg4326")
+        if not isinstance(bounds, list) or len(bounds) != 4:
+            raise SystemExit(f"evaluation sample has no governed bbox: {slug}")
+        geometry = metric_box(bounds)
+        distances = [
+            (float(geometry.distance(other)), record)
+            for record, other in lineage_geometries
+        ]
+        distances.sort(key=lambda row: row[0])
+        minimum, nearest = distances[0]
+        results.append(
+            {
+                "sample_slug": slug,
+                "bbox_epsg4326": bounds,
+                "minimum_lineage_distance_m": minimum,
+                "nearest_lineage_sample_slug": nearest["sample_slug"],
+                "nearest_lineage_raster_path": nearest["raster_path"],
+                "passes_minimum_distance": minimum >= args.minimum_distance_m,
+            }
+        )
+
+    status = "independent" if not missing and all(
+        row["passes_minimum_distance"] for row in results
+    ) else "failed"
+    output = {
+        "schema_version": 1,
+        "status": status,
+        "minimum_distance_m": args.minimum_distance_m,
+        "evaluation_manifest_path": str(evaluation_manifest),
+        "evaluation_manifest_sha256": sha256_file(evaluation_manifest),
+        "lineage_summaries": summary_evidence,
+        "lineage_sample_count": len(lineage_slugs),
+        "resolved_lineage_sample_count": len(lineage_slugs) - len(missing),
+        "missing_lineage_sample_slugs": missing,
+        "evaluation_samples": results,
+    }
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(output, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    print(json.dumps(output, indent=2, sort_keys=True))
+    return 0 if status == "independent" else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/evaluate_yolo_checkpoint_matrix.py

@@ -270,12 +270,39 @@ def background_detection_count(
     return len(selected), count
 
 
+def validate_pure_background_prefixes(
+    images: list[Path], prefixes: tuple[str, ...]
+) -> list[Path]:
+    selected = [path for path in images if path.stem.casefold().startswith(prefixes)]
+    if not selected:
+        raise ValueError(
+            "no validation images match the declared pure-background prefixes"
+        )
+    nonempty: list[str] = []
+    for image in selected:
+        try:
+            relative = image.relative_to(image.parents[1])
+        except ValueError as exc:
+            raise ValueError(f"cannot resolve label path for {image}") from exc
+        label = image.parents[1].parent / "labels" / relative.with_suffix(".txt")
+        if not label.is_file():
+            raise ValueError(f"pure-background image has no label file: {image}")
+        if label.read_text(encoding="utf-8").strip():
+            nonempty.append(str(label))
+    if nonempty:
+        raise ValueError(
+            "pure-background prefixes include non-empty labels: "
+            + ", ".join(nonempty[:10])
+        )
+    return selected
+
+
 def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--dataset-yaml", type=Path, required=True)
     parser.add_argument("--model", type=Path, action="append", required=True)
     parser.add_argument("--output", type=Path, required=True)
-    parser.add_argument("--background-prefix", action="append", required=True)
+    parser.add_argument("--background-prefix", action="append", default=[])
     parser.add_argument("--background-confidence", type=float, default=0.15)
     parser.add_argument(
         "--lineage-summary",
@@ -321,6 +348,8 @@ def main() -> int:
         write_blocked_manifest(args.output, dataset_yaml, independence_evidence)
         return 3
     prefixes = tuple(value.casefold() for value in args.background_prefix)
+    if prefixes:
+        validate_pure_background_prefixes(images, prefixes)
 
     import torch
     from ultralytics import YOLO
@@ -359,14 +388,6 @@ def main() -> int:
                 save_json=False,
                 verbose=False,
             )
-            background_images, background_detections = background_detection_count(
-                model,
-                images,
-                prefixes=prefixes,
-                confidence=args.background_confidence,
-                image_size=args.imgsz,
-                device=args.device,
-            )
             row.update(
                 {
                     "status": "ok",
@@ -374,10 +395,23 @@ def main() -> int:
                     "recall": metric_value(metrics, "mr"),
                     "map50": metric_value(metrics, "map50"),
                     "map50_95": metric_value(metrics, "map"),
-                    "pure_background_image_count": background_images,
-                    "pure_background_detection_count": background_detections,
                 }
             )
+            if prefixes:
+                background_images, background_detections = background_detection_count(
+                    model,
+                    images,
+                    prefixes=prefixes,
+                    confidence=args.background_confidence,
+                    image_size=args.imgsz,
+                    device=args.device,
+                )
+                row.update(
+                    {
+                        "pure_background_image_count": background_images,
+                        "pure_background_detection_count": background_detections,
+                    }
+                )
         except Exception as exc:  # preserve the complete attempted matrix
             row.update(
                 {
@@ -396,7 +430,7 @@ def main() -> int:
     successful = [row for row in rows if row["status"] == "ok"]
     successful.sort(
         key=lambda row: (
-            int(row["pure_background_detection_count"] == 0),
+            int(row.get("pure_background_detection_count") == 0 and bool(prefixes)),
             row["map50_95"],
             row["map50"],
             row["precision"],

scripts/export_yolo_diagnostic_evaluation_tiles.py

@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+"""Tile an explicitly training-prohibited calibration corpus for diagnostics."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+try:
+    from scripts.export_operator_yolo_tile_dataset import (
+        ensure_dependencies,
+        ensure_yolo_directories,
+        export_sample_tiles,
+        file_sha256,
+        write_dataset_yaml,
+    )
+except ModuleNotFoundError:  # direct execution from /app/scripts
+    from export_operator_yolo_tile_dataset import (
+        ensure_dependencies,
+        ensure_yolo_directories,
+        export_sample_tiles,
+        file_sha256,
+        write_dataset_yaml,
+    )
+
+
+def validate_diagnostic_manifest(manifest_path: Path) -> dict[str, Any]:
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8-sig"))
+    if manifest.get("purpose") != "non_protected_diagnostic_evaluation":
+        raise ValueError("manifest is not a diagnostic-evaluation corpus")
+    eligibility = manifest.get("training_eligibility")
+    if not isinstance(eligibility, dict) or eligibility.get("status") != (
+        "not_eligible_evaluation_only"
+    ):
+        raise ValueError("diagnostic manifest must be explicitly training-ineligible")
+    samples = manifest.get("samples")
+    if not isinstance(samples, list) or not samples:
+        raise ValueError("diagnostic manifest contains no samples")
+    slugs: set[str] = set()
+    for sample in samples:
+        if not isinstance(sample, dict):
+            raise ValueError("diagnostic samples must be objects")
+        slug = str(sample.get("sample_slug") or "").strip()
+        if not slug or slug in slugs:
+            raise ValueError(f"missing or duplicate diagnostic sample slug: {slug}")
+        if sample.get("split") != "calibration":
+            raise ValueError(f"diagnostic sample is not calibration-only: {slug}")
+        if sample.get("sample_role") not in {"positive", "background_candidate"}:
+            raise ValueError(f"invalid diagnostic sample role: {slug}")
+        slugs.add(slug)
+    marker_path = manifest_path.parent / "NO_TRAINING.json"
+    if not marker_path.is_file():
+        raise ValueError("diagnostic corpus has no NO_TRAINING.json marker")
+    marker = json.loads(marker_path.read_text(encoding="utf-8"))
+    if marker.get("training_allowed") is not False:
+        raise ValueError("NO_TRAINING marker does not prohibit training")
+    if marker.get("manifest_sha256") != file_sha256(manifest_path):
+        raise ValueError("NO_TRAINING marker is not bound to the diagnostic manifest")
+    return manifest
+
+
+def is_canonical_evaluation_window(window: dict[str, Any], tile_size: int) -> bool:
+    return bool(
+        int(window["row_off"]) % tile_size == 0
+        and int(window["col_off"]) % tile_size == 0
+        and int(window["height"]) == tile_size
+        and int(window["width"]) == tile_size
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--manifest", type=Path, required=True)
+    parser.add_argument("--output-dir", type=Path, required=True)
+    parser.add_argument("--tile-size", type=int, default=512)
+    parser.add_argument("--stride", type=int, default=512)
+    parser.add_argument("--min-label-px", type=float, default=4.0)
+    parser.add_argument("--min-label-visible-ratio", type=float, default=0.35)
+    args = parser.parse_args()
+    if args.output_dir.exists():
+        raise SystemExit(f"refusing to overwrite evaluation tiles: {args.output_dir}")
+    if args.stride != args.tile_size:
+        raise SystemExit("diagnostic evaluation tiles must be non-overlapping")
+    manifest_path = args.manifest.expanduser().resolve(strict=True)
+    try:
+        manifest = validate_diagnostic_manifest(manifest_path)
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        raise SystemExit(str(exc)) from exc
+
+    ensure_dependencies()
+    ensure_yolo_directories(args.output_dir)
+    exported: list[dict[str, Any]] = []
+    for sample in manifest["samples"]:
+        exported.extend(
+            export_sample_tiles(
+                sample=sample,
+                manifest_path=manifest_path,
+                output_dir=args.output_dir,
+                val_slugs={str(sample["sample_slug"]).lower()},
+                tile_size=args.tile_size,
+                stride=args.stride,
+                negative_keep_ratio=1.0,
+                min_label_px=args.min_label_px,
+                min_label_visible_ratio=args.min_label_visible_ratio,
+                background_negative_repeat=1,
+                drop_low_variance_negatives=False,
+                blank_range_threshold=8.0,
+                reference_source=str(sample["reference_source"]),
+                reference_layer=str(sample["reference_layer"]),
+            )
+        )
+    kept: list[dict[str, Any]] = []
+    excluded_edge_cover: list[dict[str, Any]] = []
+    for row in exported:
+        window = row["window"]
+        is_canonical_grid = bool(
+            row["kept"]
+            and is_canonical_evaluation_window(window, args.tile_size)
+        )
+        if is_canonical_grid:
+            kept.append(row)
+            continue
+        if row["kept"]:
+            excluded_edge_cover.append(
+                {
+                    "sample_slug": row["sample_slug"],
+                    "tile_index": row["tile_index"],
+                    "window": window,
+                    "reason": "overlapping_edge_cover_tile",
+                }
+            )
+            Path(row["image_path"]).unlink(missing_ok=True)
+            Path(row["label_path"]).unlink(missing_ok=True)
+    if not kept or any(row["split"] != "val" for row in kept):
+        raise SystemExit("diagnostic exporter produced invalid split membership")
+    if any((args.output_dir / "images" / "train").iterdir()) or any(
+        (args.output_dir / "labels" / "train").iterdir()
+    ):
+        raise SystemExit("diagnostic exporter wrote training data")
+
+    dataset_yaml = write_dataset_yaml(args.output_dir, "building")
+    summary = {
+        "schema_version": 1,
+        "status": "ok_evaluation_only",
+        "purpose": "non_protected_diagnostic_evaluation",
+        "training_allowed": False,
+        "release_claim_allowed": False,
+        "dataset_yaml": str(dataset_yaml),
+        "dataset_yaml_sha256": file_sha256(dataset_yaml),
+        "source_manifest": str(manifest_path),
+        "source_manifest_sha256": file_sha256(manifest_path),
+        "tile_size": args.tile_size,
+        "stride": args.stride,
+        "tile_count": len(kept),
+        "label_count": sum(int(row["label_count"]) for row in kept),
+        "excluded_overlapping_edge_cover_tile_count": len(excluded_edge_cover),
+        "excluded_overlapping_edge_cover_tiles": excluded_edge_cover,
+        "selected_sample_slugs": sorted(
+            {str(row["sample_slug"]) for row in kept}
+        ),
+        "tiles": kept,
+    }
+    summary_path = args.output_dir / "yolo_tile_dataset_summary.json"
+    summary_path.write_text(
+        json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    marker = {
+        "schema_version": 1,
+        "reason": "non_protected_diagnostic_evaluation_only",
+        "training_allowed": False,
+        "release_claim_allowed": False,
+        "source_manifest_sha256": summary["source_manifest_sha256"],
+        "summary_sha256": file_sha256(summary_path),
+    }
+    (args.output_dir / "NO_TRAINING.json").write_text(
+        json.dumps(marker, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    print(json.dumps({key: value for key, value in summary.items() if key != "tiles"}))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/provision_belgium_building_training_portfolio.py

@@ -267,6 +267,62 @@ REGION_CONTRACT = {
     },
 }
 
+NON_PROTECTED_SPLITS = {"train", "val", "calibration"}
+
+
+def load_custom_aois(path: Path) -> tuple[Aoi, ...]:
+    payload = json.loads(path.read_text(encoding="utf-8-sig"))
+    rows = payload.get("aois")
+    if not isinstance(rows, list) or not rows:
+        raise SystemExit("--aoi-spec requires a non-empty aois list")
+    result: list[Aoi] = []
+    seen: set[str] = set()
+    for row in rows:
+        if not isinstance(row, dict):
+            raise SystemExit("--aoi-spec entries must be objects")
+        slug = str(row.get("slug") or "").strip()
+        region = str(row.get("region") or "").strip().lower()
+        split = str(row.get("split") or "").strip().lower()
+        context = str(row.get("context") or "").strip()
+        role = str(row.get("sample_role") or "positive").strip()
+        if not slug or slug in seen:
+            raise SystemExit(f"Custom AOI slug is missing or duplicated: {slug}")
+        if region not in REGION_CONTRACT:
+            raise SystemExit(f"Unsupported custom AOI region for {slug}: {region}")
+        if split not in NON_PROTECTED_SPLITS:
+            raise SystemExit(
+                f"Custom AOI {slug} uses protected/unsupported split {split!r}; "
+                "fresh provisioning is limited to train, val and calibration"
+            )
+        if not context:
+            raise SystemExit(f"Custom AOI context is missing: {slug}")
+        try:
+            lon = float(row["lon"])
+            lat = float(row["lat"])
+        except (KeyError, TypeError, ValueError) as exc:
+            raise SystemExit(f"Custom AOI coordinates are invalid: {slug}") from exc
+        if not 2.4 <= lon <= 6.5 or not 49.4 <= lat <= 51.7:
+            raise SystemExit(f"Custom AOI is outside the Belgium workbench: {slug}")
+        if role not in {"positive", "background_candidate"}:
+            raise SystemExit(f"Unsupported custom AOI sample_role for {slug}: {role}")
+        require_empty = bool(row.get("require_empty", False))
+        if require_empty and role != "background_candidate":
+            raise SystemExit(f"Only background candidates may require empty labels: {slug}")
+        seen.add(slug)
+        result.append(
+            Aoi(
+                slug,
+                region,
+                context,
+                split,
+                lon,
+                lat,
+                role,
+                require_empty,
+            )
+        )
+    return tuple(result)
+
 
 def bbox_for_center(lon: float, lat: float, side_m: float) -> dict[str, Any]:
     to_metric = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True)
@@ -298,6 +354,14 @@ def main() -> int:
     parser.add_argument("--side-m", type=float, default=256.0)
     parser.add_argument("--resolution-m", type=float, default=0.25)
     parser.add_argument("--force-refresh", action="store_true")
+    parser.add_argument(
+        "--aoi-spec",
+        type=Path,
+        help=(
+            "Optional JSON file containing fresh non-protected AOIs. Custom "
+            "specs cannot provision test or background-test roles."
+        ),
+    )
     parser.add_argument(
         "--only-slug",
         action="append",
@@ -310,11 +374,16 @@ def main() -> int:
         help="Read an operator session token from a local mode-0600 file instead of exposing it on the command line.",
     )
     args = parser.parse_args()
-    known_slugs = {aoi.slug for aoi in AOIS}
+    if args.aoi_spec and args.only_slug:
+        parser.error("--aoi-spec and --only-slug are mutually exclusive")
+    available_aois = load_custom_aois(args.aoi_spec) if args.aoi_spec else AOIS
+    known_slugs = {aoi.slug for aoi in available_aois}
     unknown_slugs = sorted(set(args.only_slug) - known_slugs)
     if unknown_slugs:
         raise SystemExit(f"Unknown --only-slug value(s): {', '.join(unknown_slugs)}")
-    selected_aois = tuple(aoi for aoi in AOIS if not args.only_slug or aoi.slug in args.only_slug)
+    selected_aois = tuple(
+        aoi for aoi in available_aois if not args.only_slug or aoi.slug in args.only_slug
+    )
     session = requests.Session()
     if args.session_token_file:
         token = args.session_token_file.read_text(encoding="utf-8").strip()