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