#!/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 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]) 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 containment_overlap( 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) left_area = max(0.0, left[2] - left[0]) * max(0.0, left[3] - left[1]) right_area = max(0.0, right[2] - right[0]) * max(0.0, right[3] - right[1]) denominator = min(left_area, right_area) return intersection / denominator if denominator > 0 else 0.0 def suppress_contained_predictions( predictions: list[tuple[tuple[float, float, float, float], float]], threshold: float ) -> list[tuple[tuple[float, float, float, float], float]]: kept: list[tuple[tuple[float, float, float, float], float]] = [] for candidate in sorted(predictions, key=lambda item: -item[1]): if any(containment_overlap(candidate[0], existing[0]) >= threshold for existing in kept): continue kept.append(candidate) return kept def ensemble_predictions( primary: list[tuple[tuple[float, float, float, float], float]], secondary: list[tuple[tuple[float, float, float, float], float]], *, match_iou: float, mode: str, ) -> list[tuple[tuple[float, float, float, float], float]]: unmatched = set(range(len(secondary))) combined = [] unmatched_primary = [] for box, score in primary: candidates = [(iou(box, secondary[index][0]), index) for index in unmatched] overlap, index = max(candidates, default=(0.0, -1)) if overlap < match_iou: unmatched_primary.append((box, score)) continue other_box, other_score = secondary[index] unmatched.remove(index) combined.append((tuple((left + right) / 2 for left, right in zip(box, other_box)), (score + other_score) / 2)) if mode == "union": combined.extend(unmatched_primary) combined.extend(secondary[index] for index in unmatched) return combined def scale_box( box: tuple[float, float, float, float], factor: float, offset_x: float = 0.0, offset_y: float = 0.0 ) -> tuple[float, float, float, float]: """Scale a detector box around its center for calibration-only geometry correction.""" x1, y1, x2, y2 = box cx, cy = (x1 + x2) / 2, (y1 + y2) / 2 half_width, half_height = (x2 - x1) * factor / 2, (y2 - y1) * factor / 2 return ( cx - half_width + offset_x, cy - half_height + offset_y, cx + half_width + offset_x, cy + half_height + offset_y, ) 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") parser.add_argument("--augment", action="store_true", help="Enable deterministic YOLO test-time augmentation.") parser.add_argument("--imgsz", type=int, default=640) parser.add_argument( "--batch", type=int, default=16, help="Inference batch size; lower this for high-resolution CUDA evaluation.", ) parser.add_argument( "--max-det", type=int, default=1000, help="Maximum detections retained per tile; dense Belgian urban tiles exceed YOLO's default 300.", ) parser.add_argument( "--nms-iou", type=float, default=0.7, help="Inference NMS IoU; freeze calibration-selected values before protected test evaluation.", ) parser.add_argument( "--containment-nms", type=float, default=1.0, help="Suppress lower-score nested boxes at this intersection-over-minimum-area threshold.", ) parser.add_argument("--box-scale", type=float, default=1.0) 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) parser.add_argument("--proposal-classifier-threshold", type=float, default=0.5) parser.add_argument("--proposal-crop-scale", type=float, default=1.4) parser.add_argument("--proposal-classifier-batch", type=int, default=64) args = parser.parse_args() if args.batch < 1: parser.error("--batch must be positive") if args.proposal_classifier_batch < 1: parser.error("--proposal-classifier-batch must be positive") if not 0.0 < args.nms_iou < 1.0: 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 proposal_transform = None if args.proposal_classifier: import torch from torchvision.models import ResNet18_Weights proposal_classifier = torch.jit.load(str(args.proposal_classifier), map_location=args.device).eval() proposal_transform = ResNet18_Weights.DEFAULT.transforms() 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")) or ( item.get("sample_role") == "background_candidate" and int(item.get("reference_feature_count") or 0) == 0 ) } 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] 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(selected_paths), args.batch): bounded_results.extend( model.predict( selected_paths[start : start + args.batch], conf=min(args.thresholds), device=args.device, augment=args.augment, imgsz=args.imgsz, batch=args.batch, max_det=args.max_det, iou=args.nms_iou, verbose=False, ) ) 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))) observations: list[dict[str, Any]] = [] for result_index, (tile, result) in enumerate(zip(tiles, results, strict=True)): height, width = result.orig_shape predictions = [ ( scale_box( tuple(map(float, box)), args.box_scale, args.box_offset_x, args.box_offset_y ), float(score), ) for box, score in zip(result.boxes.xyxy.cpu().tolist(), result.boxes.conf.cpu().tolist(), strict=True) ] predictions = suppress_contained_predictions(predictions, args.containment_nms) if additional_results is not None: secondary = [ ( scale_box(tuple(map(float, box)), args.box_scale, args.box_offset_x, args.box_offset_y), float(score), ) for box, score in zip( additional_results[result_index].boxes.xyxy.cpu().tolist(), additional_results[result_index].boxes.conf.cpu().tolist(), strict=True, ) ] secondary = suppress_contained_predictions(secondary, args.containment_nms) predictions = ensemble_predictions( predictions, secondary, match_iou=args.ensemble_match_iou, mode=args.ensemble_mode ) if proposal_classifier is not None: from PIL import Image import torch with Image.open(tile["image_path"]) as opened: source = opened.convert("RGB") crops = [] for box, _score in predictions: x1, y1, x2, y2 = box cx, cy = (x1 + x2) / 2, (y1 + y2) / 2 side = max(x2 - x1, y2 - y1, 8.0) * args.proposal_crop_scale left = max(0, min(source.width - 1, int(cx - side / 2))) top = max(0, min(source.height - 1, int(cy - side / 2))) right = max(left + 1, min(source.width, int(cx + side / 2))) bottom = max(top + 1, min(source.height, int(cy + side / 2))) crop = source.crop((left, top, right, bottom)) crops.append(proposal_transform(crop.convert("RGB"))) if crops: probabilities = [] with torch.inference_mode(): for start in range(0, len(crops), args.proposal_classifier_batch): batch = torch.stack(crops[start : start + args.proposal_classifier_batch]).to(args.device) probabilities.extend( torch.sigmoid(proposal_classifier(batch).flatten()).cpu().tolist() ) predictions = [item for item, probability in zip(predictions, probabilities, strict=True) if probability >= args.proposal_classifier_threshold] 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, "test_time_augmentation": args.augment, "inference_imgsz": args.imgsz, "inference_batch": args.batch, "max_detections_per_tile": args.max_det, "nms_iou": args.nms_iou, "containment_nms": args.containment_nms, "box_scale": args.box_scale, "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, "proposal_classifier_threshold": args.proposal_classifier_threshold if args.proposal_classifier else None, "proposal_crop_scale": args.proposal_crop_scale if args.proposal_classifier else None, "proposal_classifier_batch": args.proposal_classifier_batch if args.proposal_classifier else None, "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())