Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
158 lines
7.4 KiB
Python
158 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Render deterministic reference/prediction error overlays for protected calibration AOIs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import statistics
|
|
from pathlib import Path
|
|
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
from evaluate_belgium_building_candidate import iou, read_references
|
|
|
|
|
|
def match_details(predictions, references, match_iou: float):
|
|
candidates = sorted(predictions, key=lambda item: -item[1])
|
|
unmatched = set(range(len(references)))
|
|
matched_predictions: list[tuple[tuple[float, float, float, float], float]] = []
|
|
false_predictions: list[tuple[tuple[float, float, float, float], float]] = []
|
|
for prediction in candidates:
|
|
best = max(unmatched, key=lambda index: iou(prediction[0], references[index]), default=None)
|
|
if best is not None and iou(prediction[0], references[best]) >= match_iou:
|
|
unmatched.remove(best)
|
|
matched_predictions.append(prediction)
|
|
else:
|
|
false_predictions.append(prediction)
|
|
return matched_predictions, false_predictions, [references[index] for index in sorted(unmatched)]
|
|
|
|
|
|
def draw_boxes(draw, boxes, *, color, width=3, scores=False):
|
|
font = ImageFont.load_default()
|
|
for item in boxes:
|
|
box, score = item if scores else (item, None)
|
|
draw.rectangle(box, outline=color, width=width)
|
|
if score is not None:
|
|
draw.text((box[0] + 2, box[1] + 2), f"{score:.2f}", fill=color, font=font)
|
|
|
|
|
|
def geometry(box):
|
|
width = max(0.0, box[2] - box[0])
|
|
height = max(0.0, box[3] - box[1])
|
|
return width * height, max(width / height, height / width) if width and height else float("inf")
|
|
|
|
|
|
def distribution(values):
|
|
ordered = sorted(value for value in values if math.isfinite(value))
|
|
if not ordered:
|
|
return {"count": 0}
|
|
def percentile(fraction):
|
|
return ordered[round((len(ordered) - 1) * fraction)]
|
|
return {
|
|
"count": len(ordered), "median": statistics.median(ordered),
|
|
"p90": percentile(0.9), "p95": percentile(0.95), "p99": percentile(0.99),
|
|
"max": ordered[-1],
|
|
}
|
|
|
|
|
|
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("--output-dir", type=Path, required=True)
|
|
parser.add_argument("--sample", action="append", required=True)
|
|
parser.add_argument("--threshold", type=float, required=True)
|
|
parser.add_argument("--match-iou", type=float, default=0.25)
|
|
parser.add_argument("--device", default="cuda:0")
|
|
parser.add_argument("--imgsz", type=int, default=640)
|
|
parser.add_argument("--max-det", type=int, default=1000)
|
|
parser.add_argument("--nms-iou", type=float, default=0.7)
|
|
parser.add_argument("--max-tiles", type=int, default=24)
|
|
parser.add_argument("--columns", type=int, default=4)
|
|
args = parser.parse_args()
|
|
|
|
from ultralytics import YOLO
|
|
|
|
summary = json.loads(args.summary.read_text(encoding="utf-8"))
|
|
selected = [tile for tile in summary["tiles"] if tile["sample_slug"] in set(args.sample)]
|
|
image_paths = [Path(tile["image_path"]) for tile in selected]
|
|
results = YOLO(str(args.model)).predict(
|
|
image_paths, conf=args.threshold, device=args.device, imgsz=args.imgsz,
|
|
max_det=args.max_det, iou=args.nms_iou, verbose=False, stream=False,
|
|
)
|
|
evidence = []
|
|
cards = []
|
|
geometry_evidence = {role: {"area_share": [], "aspect_ratio": []} for role in ("reference", "matched", "false_positive", "false_negative")}
|
|
for tile, image_path, result in zip(selected, image_paths, results, strict=True):
|
|
image = Image.open(image_path).convert("RGB")
|
|
predictions = [
|
|
((float(box[0]), float(box[1]), float(box[2]), float(box[3])), float(score))
|
|
for box, score in zip(result.boxes.xyxy.cpu().tolist(), result.boxes.conf.cpu().tolist(), strict=True)
|
|
]
|
|
references = read_references(Path(tile["label_path"]), image.width, image.height)
|
|
matched, false_positive, false_negative = match_details(predictions, references, args.match_iou)
|
|
for role, boxes in (("reference", references), ("false_negative", false_negative)):
|
|
for box in boxes:
|
|
area, aspect = geometry(box)
|
|
geometry_evidence[role]["area_share"].append(area / (image.width * image.height))
|
|
geometry_evidence[role]["aspect_ratio"].append(aspect)
|
|
for role, boxes in (("matched", matched), ("false_positive", false_positive)):
|
|
for box, _score in boxes:
|
|
area, aspect = geometry(box)
|
|
geometry_evidence[role]["area_share"].append(area / (image.width * image.height))
|
|
geometry_evidence[role]["aspect_ratio"].append(aspect)
|
|
evidence.append({
|
|
"sample_slug": tile["sample_slug"], "image_path": str(image_path),
|
|
"label_count": len(references), "true_positive": len(matched),
|
|
"false_positive": len(false_positive), "false_negative": len(false_negative),
|
|
})
|
|
draw = ImageDraw.Draw(image)
|
|
draw_boxes(draw, references, color=(255, 215, 0), width=2)
|
|
draw_boxes(draw, matched, color=(0, 220, 80), scores=True)
|
|
draw_boxes(draw, false_positive, color=(255, 40, 40), scores=True)
|
|
draw_boxes(draw, false_negative, color=(255, 0, 220), width=3)
|
|
cards.append((len(false_positive) + len(false_negative), tile["sample_slug"], image_path.name, image))
|
|
|
|
cards.sort(key=lambda item: (-item[0], item[1], item[2]))
|
|
cards = cards[: args.max_tiles]
|
|
header = 34
|
|
rendered = []
|
|
font = ImageFont.load_default()
|
|
for errors, slug, name, image in cards:
|
|
card = Image.new("RGB", (image.width, image.height + header), (20, 31, 44))
|
|
card.paste(image, (0, header))
|
|
ImageDraw.Draw(card).text((6, 7), f"{slug} | {name} | errors={errors}", fill="white", font=font)
|
|
rendered.append(card)
|
|
gap = 10
|
|
rows = math.ceil(len(rendered) / args.columns) if rendered else 0
|
|
if rendered:
|
|
width = args.columns * rendered[0].width + (args.columns + 1) * gap
|
|
height = rows * rendered[0].height + (rows + 1) * gap
|
|
sheet = Image.new("RGB", (width, height), (226, 232, 240))
|
|
for index, card in enumerate(rendered):
|
|
x = gap + (index % args.columns) * (card.width + gap)
|
|
y = gap + (index // args.columns) * (card.height + gap)
|
|
sheet.paste(card, (x, y))
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
sheet.save(args.output_dir / "candidate_error_contact_sheet.png")
|
|
report = {
|
|
"schema_version": 1, "model": str(args.model), "summary": str(args.summary),
|
|
"threshold": args.threshold, "match_iou": args.match_iou, "samples": args.sample,
|
|
"legend": {"reference": "yellow", "matched_prediction": "green", "false_positive": "red", "false_negative": "magenta"},
|
|
"geometry": {
|
|
role: {name: distribution(values) for name, values in metrics.items()}
|
|
for role, metrics in geometry_evidence.items()
|
|
},
|
|
"tiles": evidence,
|
|
}
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
(args.output_dir / "candidate_error_contact_sheet.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
|
|
print(json.dumps({"status": "ok", "tile_count": len(evidence), "rendered": len(rendered)}, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|