Add fail-closed regional detector routing

scripts/evaluate_belgium_building_candidate.py

@@ -10,6 +10,21 @@ from pathlib import Path
 from typing import Any
 
 
+def parse_regional_models(values: list[str]) -> dict[str, Path]:
+    """Parse explicit REGION=MODEL routing without silently accepting ambiguity."""
+    models: dict[str, Path] = {}
+    for value in values:
+        region, separator, model = value.partition("=")
+        region = region.strip().lower()
+        model = model.strip()
+        if not separator or not region or not model:
+            raise ValueError(f"Invalid regional model {value!r}; expected REGION=/path/to/model.pt")
+        if region in models:
+            raise ValueError(f"Duplicate regional model for {region!r}")
+        models[region] = Path(model)
+    return models
+
+
 def iou(left: tuple[float, float, float, float], right: tuple[float, float, float, float]) -> float:
     x1, y1 = max(left[0], right[0]), max(left[1], right[1])
     x2, y2 = min(left[2], right[2]), min(left[3], right[3])
@@ -173,6 +188,13 @@ def main() -> int:
     parser.add_argument("--box-offset-x", type=float, default=0.0)
     parser.add_argument("--box-offset-y", type=float, default=0.0)
     parser.add_argument("--additional-model", type=Path)
+    parser.add_argument(
+        "--regional-model",
+        action="append",
+        default=[],
+        metavar="REGION=MODEL",
+        help="Route tiles from one manifest region to an explicit expert model; repeat per region.",
+    )
     parser.add_argument("--ensemble-mode", choices=("consensus", "union"), default="consensus")
     parser.add_argument("--ensemble-match-iou", type=float, default=0.3)
     parser.add_argument("--proposal-classifier", type=Path)
@@ -188,6 +210,10 @@ def main() -> int:
         parser.error("--nms-iou must be between zero and one")
     if not 0.0 < args.containment_nms <= 1.0:
         parser.error("--containment-nms must be above zero and at most one")
+    try:
+        regional_models = parse_regional_models(args.regional_model)
+    except ValueError as exc:
+        parser.error(str(exc))
 
     from ultralytics import YOLO
     proposal_classifier = None
@@ -213,12 +239,17 @@ def main() -> int:
     }
     tiles = [item for item in summary["tiles"] if item.get("kept", True) and item["split"] == args.split]
     image_paths = [item["image_path"] for item in tiles]
-    def predict_bounded(model: YOLO) -> list[Any]:
+    unknown_regions = sorted(set(regional_models) - set(regions.values()))
+    if unknown_regions:
+        parser.error(f"Regional model keys absent from corpus manifest: {', '.join(unknown_regions)}")
+
+    def predict_bounded(model: YOLO, paths: list[str] | None = None) -> list[Any]:
+        selected_paths = image_paths if paths is None else paths
         bounded_results: list[Any] = []
-        for start in range(0, len(image_paths), args.batch):
+        for start in range(0, len(selected_paths), args.batch):
             bounded_results.extend(
                 model.predict(
-                    image_paths[start : start + args.batch],
+                    selected_paths[start : start + args.batch],
                     conf=min(args.thresholds),
                     device=args.device,
                     augment=args.augment,
@@ -232,6 +263,17 @@ def main() -> int:
         return bounded_results
 
     results = predict_bounded(YOLO(str(args.model)))
+    routed_tile_counts: dict[str, int] = {}
+    for region, model_path in regional_models.items():
+        indices = [index for index, tile in enumerate(tiles) if regions[tile["sample_slug"]] == region]
+        if not indices:
+            parser.error(f"Regional model {region!r} has no tiles in selected split")
+        routed_results = predict_bounded(YOLO(str(model_path)), [image_paths[index] for index in indices])
+        if len(routed_results) != len(indices):
+            raise RuntimeError(f"Regional model {region!r} returned an incomplete result set")
+        for index, result in zip(indices, routed_results, strict=True):
+            results[index] = result
+        routed_tile_counts[region] = len(indices)
     additional_results = None
     if args.additional_model:
         additional_results = predict_bounded(YOLO(str(args.additional_model)))
@@ -337,6 +379,8 @@ def main() -> int:
         "box_offset_x": args.box_offset_x,
         "box_offset_y": args.box_offset_y,
         "additional_model": str(args.additional_model) if args.additional_model else None,
+        "regional_models": {region: str(path) for region, path in sorted(regional_models.items())},
+        "regional_model_tile_counts": dict(sorted(routed_tile_counts.items())),
         "ensemble_mode": args.ensemble_mode if args.additional_model else None,
         "ensemble_match_iou": args.ensemble_match_iou if args.additional_model else None,
         "proposal_classifier": str(args.proposal_classifier) if args.proposal_classifier else None,

tests/test_evaluate_belgium_building_candidate.py

@@ -0,0 +1,23 @@
+from pathlib import Path
+
+import pytest
+
+from scripts.evaluate_belgium_building_candidate import parse_regional_models
+
+
+def test_parse_regional_models_accepts_explicit_unique_routes() -> None:
+    assert parse_regional_models(["flanders=/models/flanders.pt", "wallonia=/models/wallonia.pt"]) == {
+        "flanders": Path("/models/flanders.pt"),
+        "wallonia": Path("/models/wallonia.pt"),
+    }
+
+
+@pytest.mark.parametrize("value", ["flanders", "=/model.pt", "flanders="])
+def test_parse_regional_models_rejects_incomplete_routes(value: str) -> None:
+    with pytest.raises(ValueError, match="expected REGION"):
+        parse_regional_models([value])
+
+
+def test_parse_regional_models_rejects_duplicate_routes() -> None:
+    with pytest.raises(ValueError, match="Duplicate regional model"):
+        parse_regional_models(["flanders=/first.pt", "FLANDERS=/second.pt"])