Add fail-closed Belgian training loop

scripts/evaluate_belgium_building_candidate.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+"""Evaluate one building detector without using protected data for tuning."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+
+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])
+    intersection = max(0.0, x2 - x1) * max(0.0, y2 - y1)
+    union = (left[2] - left[0]) * (left[3] - left[1]) + (right[2] - right[0]) * (
+        right[3] - right[1]
+    ) - intersection
+    return intersection / union if union > 0 else 0.0
+
+
+def match_boxes(
+    predictions: list[tuple[tuple[float, float, float, float], float]],
+    references: list[tuple[float, float, float, float]],
+    *,
+    confidence: float,
+    match_iou: float,
+) -> tuple[int, int, int]:
+    unmatched = set(range(len(references)))
+    true_positive = 0
+    considered = sorted((item for item in predictions if item[1] >= confidence), key=lambda item: -item[1])
+    for box, _score in considered:
+        candidates = [(iou(box, references[index]), index) for index in unmatched]
+        best_iou, best_index = max(candidates, default=(0.0, -1))
+        if best_iou >= match_iou:
+            unmatched.remove(best_index)
+            true_positive += 1
+    return true_positive, len(considered) - true_positive, len(unmatched)
+
+
+def metrics(tp: int, fp: int, fn: int) -> dict[str, float | int]:
+    precision = tp / (tp + fp) if tp + fp else 1.0
+    recall = tp / (tp + fn) if tp + fn else 1.0
+    return {
+        "true_positive": tp,
+        "false_positive": fp,
+        "false_negative": fn,
+        "precision": precision,
+        "recall": recall,
+        "f1": 2 * precision * recall / (precision + recall) if precision + recall else 0.0,
+    }
+
+
+def read_references(path: Path, width: int, height: int) -> list[tuple[float, float, float, float]]:
+    boxes = []
+    for line in path.read_text(encoding="utf-8").splitlines() if path.is_file() else []:
+        parts = line.split()
+        if len(parts) != 5:
+            raise ValueError(f"Invalid YOLO label row in {path}: {line}")
+        _class_id, cx, cy, box_width, box_height = map(float, parts)
+        boxes.append(
+            (
+                (cx - box_width / 2) * width,
+                (cy - box_height / 2) * height,
+                (cx + box_width / 2) * width,
+                (cy + box_height / 2) * height,
+            )
+        )
+    return boxes
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--model", type=Path, required=True)
+    parser.add_argument("--summary", type=Path, required=True)
+    parser.add_argument("--corpus-manifest", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--thresholds", type=float, nargs="+", default=[0.1, 0.15, 0.2, 0.25, 0.3, 0.4])
+    parser.add_argument("--match-iou", type=float, default=0.25)
+    parser.add_argument("--device", default="cuda:0")
+    parser.add_argument("--split", default="val")
+    args = parser.parse_args()
+
+    from ultralytics import YOLO
+
+    summary = json.loads(args.summary.read_text(encoding="utf-8"))
+    manifest = json.loads(args.corpus_manifest.read_text(encoding="utf-8"))
+    regions = {item["sample_slug"]: item["region"] for item in manifest["samples"]}
+    pure_empty_slugs = {
+        item["sample_slug"] for item in manifest["samples"] if bool(item.get("require_empty"))
+    }
+    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]
+    results = YOLO(str(args.model)).predict(image_paths, conf=min(args.thresholds), device=args.device, verbose=False)
+    observations: list[dict[str, Any]] = []
+    for tile, result in zip(tiles, results, strict=True):
+        height, width = result.orig_shape
+        predictions = [
+            (tuple(map(float, box)), float(score))
+            for box, score in zip(result.boxes.xyxy.cpu().tolist(), result.boxes.conf.cpu().tolist(), strict=True)
+        ]
+        observations.append(
+            {
+                "sample_slug": tile["sample_slug"],
+                "region": regions[tile["sample_slug"]],
+                "references": read_references(Path(tile["label_path"]), width, height),
+                "predictions": predictions,
+            }
+        )
+
+    sweeps = []
+    for threshold in args.thresholds:
+        totals: defaultdict[str, list[int]] = defaultdict(lambda: [0, 0, 0])
+        pure_empty_fp = 0
+        for item in observations:
+            tp, fp, fn = match_boxes(
+                item["predictions"], item["references"], confidence=threshold, match_iou=args.match_iou
+            )
+            for key in ("all", item["region"], item["sample_slug"]):
+                totals[key][0] += tp
+                totals[key][1] += fp
+                totals[key][2] += fn
+            if item["sample_slug"] in pure_empty_slugs:
+                pure_empty_fp += fp
+        sweeps.append(
+            {
+                "threshold": threshold,
+                "aggregate": metrics(*totals["all"]),
+                "regions": {region: metrics(*totals[region]) for region in sorted(set(regions.values()))},
+                "samples": {slug: metrics(*counts) for slug, counts in sorted(totals.items()) if slug not in {"all", *regions.values()}},
+                "pure_empty_false_positives": pure_empty_fp,
+            }
+        )
+    payload = {
+        "schema_version": 1,
+        "model": str(args.model),
+        "summary": str(args.summary),
+        "split": args.split,
+        "match_iou": args.match_iou,
+        "tile_count": len(tiles),
+        "sweeps": sweeps,
+    }
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+    print(json.dumps(payload, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())