geointel/backend/app/services/detection_comparison_service.py

"""Placing two detection runs side by side, honestly.

The workbench ranks model variants by a stored F1, each measured at that
variant's own confidence threshold. That number says as much about the
threshold as about the model: a conservatively calibrated detector looks worse
than a liberal one without detecting anything differently. Average precision
describes the whole ranking the model produced and is the comparable figure.

Comparability comes first, though. Two runs over different rasters, scored
against different references, or covering different ground are not two answers
to one question, and no metric makes them so.
"""

from __future__ import annotations

from typing import Any

from app.core.errors import AppError


class DetectionComparisonService:
    # A run whose inference footprint was never established cannot be placed
    # beside one that was: their recalls have different denominators.
    PROVEN_COVERAGE_MODES = ("persisted_tile_manifest_union",)

    @staticmethod
    def assess_comparability(entries: list[dict[str, Any]]) -> dict[str, Any]:
        """Say whether these runs answer the same question, and why not if they don't."""

        if len(entries) < 2:
            raise AppError(
                code="DETECTION_COMPARISON_NEEDS_TWO_RUNS",
                message="Comparing detection models requires at least two runs",
                status_code=400,
            )

        reasons: list[str] = []
        source_rasters = {str(entry.get("dataset_id")) for entry in entries}
        if len(source_rasters) > 1:
            reasons.append("different_source_raster")

        references = {str(entry.get("reference_dataset_id")) for entry in entries}
        if len(references) > 1:
            reasons.append("different_reference_dataset")

        if any(
            str(entry.get("coverage_mode")) not in DetectionComparisonService.PROVEN_COVERAGE_MODES
            for entry in entries
        ):
            reasons.append("coverage_not_proven")

        populations = {int(entry.get("reference_evaluated_count") or 0) for entry in entries}
        if len(populations) > 1:
            reasons.append("different_evaluated_population")

        return {
            "comparable": not reasons,
            "blocking_reasons": reasons,
            "source_raster_count": len(source_rasters),
            "reference_dataset_count": len(references),
            "evaluated_population_counts": sorted(populations),
        }

    @staticmethod
    def rank(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
        """Order runs by average precision, stating the margin and any tie.

        Ranking on the F1 each run happened to be read at would order the
        thresholds, not the models.
        """

        ordered = sorted(
            rows,
            key=lambda row: (-(row.get("average_precision") or 0.0), str(row.get("model_asset_id") or "")),
        )
        if not ordered:
            return []

        leader = ordered[0].get("average_precision") or 0.0
        tied_count = sum(1 for row in ordered if (row.get("average_precision") or 0.0) == leader)
        runner_up = (ordered[1].get("average_precision") or 0.0) if len(ordered) > 1 else leader

        ranked: list[dict[str, Any]] = []
        for row in ordered:
            average_precision = row.get("average_precision") or 0.0
            is_leader = average_precision == leader
            ranked.append(
                {
                    **row,
                    "rank": 1 if is_leader else 1 + sum(
                        1 for other in ordered if (other.get("average_precision") or 0.0) > average_precision
                    ),
                    # Distance behind the best run; zero for the leader itself.
                    "average_precision_gap": leader - average_precision,
                    "tied": is_leader and tied_count > 1,
                    # Only the leader has a lead; stating it on every row would
                    # invite reading a follower's gap as an advantage.
                    "lead_over_next": (leader - runner_up) if is_leader and tied_count == 1 else None,
                }
            )
        return ranked


    @staticmethod
    def compare_runs(
        db,
        *,
        analysis_run_ids: list,
        reference_dataset_id,
        iou_threshold: float = 0.5,
    ) -> dict[str, Any]:
        """Score several runs against one reference and rank them on AP.

        Each run is scored through the same QA path the workbench uses, so the
        comparison and the persisted quality checks cannot drift apart.
        """

        # Lazy: detection_service imports this module's siblings at load.
        from app.models import AnalysisRun
        from app.services.detection_service import DetectionService

        if len(set(analysis_run_ids)) < 2:
            raise AppError(
                code="DETECTION_COMPARISON_NEEDS_TWO_RUNS",
                message="Comparing detection models requires at least two distinct runs",
                status_code=400,
            )

        entries: list[dict[str, Any]] = []
        rows: list[dict[str, Any]] = []
        for analysis_run_id in analysis_run_ids:
            result = DetectionService.compare_detections_with_reference(
                db,
                analysis_run_id=analysis_run_id,
                reference_dataset_id=reference_dataset_id,
                iou_threshold=iou_threshold,
            )
            run = db.get(AnalysisRun, analysis_run_id)
            parameters = (run.parameters_json if run and isinstance(run.parameters_json, dict) else {}) or {}
            coverage = result.get("coverage") if isinstance(result.get("coverage"), dict) else {}
            curve = result.get("precision_recall_curve") or {}

            entries.append(
                {
                    "analysis_run_id": analysis_run_id,
                    "dataset_id": getattr(run, "dataset_id", None),
                    "model_id": getattr(run, "model_name", None),
                    "model_asset_id": parameters.get("model_asset_id"),
                    "reference_dataset_id": reference_dataset_id,
                    "coverage_mode": coverage.get("mode"),
                    "reference_evaluated_count": coverage.get("reference_evaluated_count"),
                }
            )
            rows.append(
                {
                    "analysis_run_id": str(analysis_run_id),
                    "quality_check_id": result.get("quality_check_id"),
                    "model_id": getattr(run, "model_name", None),
                    "model_asset_id": parameters.get("model_asset_id"),
                    "run_confidence_threshold": parameters.get("confidence_threshold"),
                    "average_precision": curve.get("average_precision"),
                    "best_f1": curve.get("best_f1"),
                    "best_f1_threshold": curve.get("best_f1_threshold"),
                    # The figure the workbench used to rank on, kept visible so
                    # the difference between the two readings is auditable.
                    "f1_at_run_threshold": result.get("f1_score"),
                    "precision_at_run_threshold": result.get("precision"),
                    "recall_at_run_threshold": result.get("recall"),
                }
            )

        comparability = DetectionComparisonService.assess_comparability(entries)
        return {
            "reference_dataset_id": str(reference_dataset_id),
            "iou_threshold": iou_threshold,
            "comparability": comparability,
            "ranking_metric": "average_precision",
            "rows": DetectionComparisonService.rank(rows),
        }