geointel/backend/app/services/detection_metrics_service.py

"""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 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,
        }