JensandClaude Opus 5
2cd2c49389
Threshold calibration ran the model over every tile once per threshold — three GPU passes to compare 0.50, 0.25 and 0.15 on a hundred-tile raster. The answer is already in a single run at the lowest value: detections above a higher cut are a subset of it, and duplicate suppression walks candidates in descending confidence, so a lower-confidence box can never displace a higher-confidence one. The kept set above any cut is identical whichever threshold the run used, which is what makes one pass sufficient rather than merely cheaper. QA now takes calibration_thresholds and reads each operating point off the same precision/recall walk it already performs, marking the F1-optimal cut. The lab runs inference once and fills its table from the sweep. The contract test asserted the per-threshold loop by name, pinning the waste it was meant to describe. It now states what calibration owes an operator: a row per requested threshold, from one run. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
236 lines
10 KiB
Python
236 lines
10 KiB
Python
"""Threshold-independent detection metrics.
|
|
|
|
A single precision/recall/F1 triple describes one operating point. Which point
|
|
that is depends on the confidence threshold the operator typed, so two models
|
|
cannot be compared from it: a conservatively calibrated model looks worse at a
|
|
low cut and better at a high one without detecting anything differently.
|
|
|
|
This service produces the standard alternative — the full precision/recall
|
|
curve over every confidence value that occurs in the run, the average
|
|
precision derived from it, and the threshold where F1 actually peaks — using
|
|
the same greedy IoU matching rule as the rest of QA so the numbers stay
|
|
comparable with the persisted quality checks.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from shapely.geometry.base import BaseGeometry
|
|
from shapely.strtree import STRtree
|
|
|
|
from app.services.qa_service import QaService
|
|
|
|
|
|
class DetectionMetricsService:
|
|
SUPPORTED_GEOMETRY_TYPES = QaService.SUPPORTED_GEOMETRY_TYPES
|
|
|
|
@staticmethod
|
|
def _rank_candidates(
|
|
candidates: list[tuple[dict[str, Any], BaseGeometry]],
|
|
) -> list[tuple[float, str, dict[str, Any], BaseGeometry]]:
|
|
"""Order candidates by confidence, highest first, deterministically."""
|
|
|
|
ranked: list[tuple[float, str, dict[str, Any], BaseGeometry]] = []
|
|
for index, (feature, geometry) in enumerate(candidates):
|
|
if geometry.is_empty or geometry.geom_type not in DetectionMetricsService.SUPPORTED_GEOMETRY_TYPES:
|
|
continue
|
|
if geometry.area <= 0:
|
|
continue
|
|
confidence = QaService._feature_confidence(feature)
|
|
identifier = QaService._feature_identifier(feature, "candidate", index)
|
|
ranked.append((float(confidence if confidence is not None else 0.0), identifier, feature, geometry))
|
|
ranked.sort(key=lambda item: (-item[0], item[1]))
|
|
return ranked
|
|
|
|
@staticmethod
|
|
def _greedy_hits(
|
|
ranked: list[tuple[float, str, dict[str, Any], BaseGeometry]],
|
|
references: list[tuple[dict[str, Any], BaseGeometry]],
|
|
iou_threshold: float,
|
|
) -> tuple[list[bool], int]:
|
|
"""Mark each ranked candidate as a hit or a miss, best score first.
|
|
|
|
Walking the ranking once and consuming references as they are claimed
|
|
is exactly the COCO/PASCAL rule, and it is what makes the result
|
|
independent of the order rows came out of the database.
|
|
"""
|
|
|
|
supported = [
|
|
(index, geometry)
|
|
for index, (_, geometry) in enumerate(references)
|
|
if geometry.geom_type in DetectionMetricsService.SUPPORTED_GEOMETRY_TYPES
|
|
and not geometry.is_empty
|
|
and geometry.area > 0
|
|
]
|
|
reference_count = len(supported)
|
|
if not supported:
|
|
return [False] * len(ranked), 0
|
|
|
|
tree = STRtree([geometry for _, geometry in supported])
|
|
claimed: set[int] = set()
|
|
hits: list[bool] = []
|
|
|
|
for _, _, _, geometry in ranked:
|
|
best_iou = 0.0
|
|
best_index: int | None = None
|
|
for position in sorted(int(value) for value in tree.query(geometry)):
|
|
if position in claimed:
|
|
continue
|
|
_, reference_geometry = supported[position]
|
|
intersection_area = geometry.intersection(reference_geometry).area
|
|
if intersection_area <= 0:
|
|
continue
|
|
union_area = geometry.area + reference_geometry.area - intersection_area
|
|
if union_area <= 0:
|
|
continue
|
|
iou = intersection_area / union_area
|
|
if iou > best_iou:
|
|
best_iou = iou
|
|
best_index = position
|
|
if best_index is not None and best_iou >= iou_threshold:
|
|
claimed.add(best_index)
|
|
hits.append(True)
|
|
else:
|
|
hits.append(False)
|
|
|
|
return hits, reference_count
|
|
|
|
@staticmethod
|
|
def _average_precision(points: list[dict[str, Any]]) -> float:
|
|
"""Area under the precision/recall curve, with precision made monotone.
|
|
|
|
Interpolating precision to its running maximum from the right is the
|
|
VOC/COCO convention; without it the sawtooth from individual false
|
|
positives shows up as noise in the score.
|
|
"""
|
|
|
|
if not points:
|
|
return 0.0
|
|
|
|
recalls = [0.0] + [point["recall"] for point in points]
|
|
precisions = [1.0] + [point["precision"] for point in points]
|
|
for index in range(len(precisions) - 2, -1, -1):
|
|
precisions[index] = max(precisions[index], precisions[index + 1])
|
|
|
|
area = 0.0
|
|
for index in range(1, len(recalls)):
|
|
area += (recalls[index] - recalls[index - 1]) * precisions[index]
|
|
return area
|
|
|
|
@staticmethod
|
|
def operating_point(curve: dict[str, Any], *, min_confidence: float) -> dict[str, Any]:
|
|
"""The metrics that hold when detections below ``min_confidence`` are dropped.
|
|
|
|
Read off the curve rather than recomputed: the curve already walked the
|
|
ranking once, and every threshold is a prefix of that walk. Running the
|
|
model again per threshold would spend N GPU passes to reproduce numbers
|
|
that are already here.
|
|
"""
|
|
|
|
points = curve.get("points") or []
|
|
reference_count = int(curve.get("reference_count") or 0)
|
|
admitted = [point for point in points if point["confidence_threshold"] >= min_confidence]
|
|
# Points are cumulative down the ranking, so the last admitted one is
|
|
# the complete tally at this cut.
|
|
tally = admitted[-1] if admitted else None
|
|
|
|
true_positives = int(tally["true_positives"]) if tally else 0
|
|
false_positives = int(tally["false_positives"]) if tally else 0
|
|
false_negatives = max(0, reference_count - true_positives)
|
|
candidate_count = true_positives + false_positives
|
|
|
|
precision = true_positives / candidate_count if candidate_count else None
|
|
recall = true_positives / reference_count if reference_count else None
|
|
f1_score = None
|
|
if precision is not None and recall is not None:
|
|
f1_score = (2 * precision * recall / (precision + recall)) if precision + recall > 0 else 0.0
|
|
|
|
return {
|
|
"min_confidence": min_confidence,
|
|
"confidence_threshold": tally["confidence_threshold"] if tally else None,
|
|
"candidate_count": candidate_count,
|
|
"true_positives": true_positives,
|
|
"false_positives": false_positives,
|
|
"false_negatives": false_negatives,
|
|
"precision": precision,
|
|
"recall": recall,
|
|
"f1_score": f1_score,
|
|
}
|
|
|
|
@staticmethod
|
|
def calibration_sweep(curve: dict[str, Any], *, thresholds: list[float]) -> list[dict[str, Any]]:
|
|
"""Every requested operating point, strictest first, from one curve.
|
|
|
|
Recall cannot fall as the cut loosens — that monotonicity is exactly
|
|
why a single run answers the whole sweep.
|
|
"""
|
|
|
|
ordered = sorted({float(value) for value in thresholds}, reverse=True)
|
|
rows = [
|
|
DetectionMetricsService.operating_point(curve, min_confidence=value) for value in ordered
|
|
]
|
|
best_f1 = max((row["f1_score"] or 0.0) for row in rows) if rows else 0.0
|
|
marked = False
|
|
for row in rows:
|
|
is_best = not marked and (row["f1_score"] or 0.0) == best_f1
|
|
row["best_f1_in_sweep"] = is_best
|
|
marked = marked or is_best
|
|
return rows
|
|
|
|
@staticmethod
|
|
def precision_recall_curve(
|
|
candidates: list[tuple[dict[str, Any], BaseGeometry]],
|
|
references: list[tuple[dict[str, Any], BaseGeometry]],
|
|
*,
|
|
iou_threshold: float,
|
|
) -> dict[str, Any]:
|
|
"""Sweep every confidence value present and describe the whole curve."""
|
|
|
|
ranked = DetectionMetricsService._rank_candidates(candidates)
|
|
hits, reference_count = DetectionMetricsService._greedy_hits(ranked, references, iou_threshold)
|
|
|
|
points: list[dict[str, Any]] = []
|
|
true_positives = 0
|
|
for position, hit in enumerate(hits):
|
|
if hit:
|
|
true_positives += 1
|
|
false_positives = position + 1 - true_positives
|
|
precision = true_positives / (position + 1)
|
|
recall = true_positives / reference_count if reference_count else 0.0
|
|
f1 = (2 * precision * recall / (precision + recall)) if precision + recall > 0 else 0.0
|
|
points.append(
|
|
{
|
|
"confidence_threshold": ranked[position][0],
|
|
"candidate_count": position + 1,
|
|
"true_positives": true_positives,
|
|
"false_positives": false_positives,
|
|
"false_negatives": max(0, reference_count - true_positives),
|
|
"precision": precision,
|
|
"recall": recall,
|
|
"f1_score": f1,
|
|
}
|
|
)
|
|
|
|
# Keep one point per distinct threshold: the last one, which is the
|
|
# complete tally for everything at or above that confidence.
|
|
deduplicated: list[dict[str, Any]] = []
|
|
for point in points:
|
|
if deduplicated and deduplicated[-1]["confidence_threshold"] == point["confidence_threshold"]:
|
|
deduplicated[-1] = point
|
|
else:
|
|
deduplicated.append(point)
|
|
|
|
best = max(deduplicated, key=lambda point: (point["f1_score"], point["confidence_threshold"]), default=None)
|
|
return {
|
|
"iou_threshold": iou_threshold,
|
|
"reference_count": reference_count,
|
|
"candidate_count": len(ranked),
|
|
"average_precision": DetectionMetricsService._average_precision(points),
|
|
"best_f1": best["f1_score"] if best else 0.0,
|
|
"best_f1_threshold": best["confidence_threshold"] if best else None,
|
|
"best_f1_precision": best["precision"] if best else None,
|
|
"best_f1_recall": best["recall"] if best else None,
|
|
"points": deduplicated,
|
|
}
|