from __future__ import annotations from dataclasses import dataclass from math import isfinite from typing import Any from uuid import UUID from pyproj import CRS, Transformer from shapely.geometry import GeometryCollection, MultiPolygon, Polygon, box from shapely.geometry.base import BaseGeometry from shapely.ops import transform as shapely_transform from shapely.ops import unary_union from shapely.validation import make_valid from app.core.errors import AppError from app.services.qa_service import QaMatchEvidence @dataclass(frozen=True) class DetectionQaCoverage: geometry: BaseGeometry manifest_path: str tile_count: int source_crs_values: tuple[str, ...] @dataclass(frozen=True) class CoveragePopulation: geometries: list[tuple[dict[str, Any], BaseGeometry]] raw_count: int evaluated_count: int excluded_outside_count: int clipped_boundary_count: int class DetectionQaService: @staticmethod def tile_manifest_path(parameters: Any) -> str | None: if not isinstance(parameters, dict): return None value = parameters.get("tile_manifest_path") if isinstance(value, str) and value.strip(): return value.strip() nested = parameters.get("parameters_json") if isinstance(nested, dict): value = nested.get("tile_manifest_path") if isinstance(value, str) and value.strip(): return value.strip() return None @staticmethod def build_tile_coverage( manifest: dict[str, Any], *, manifest_path: str, expected_dataset_id: UUID | None, ) -> DetectionQaCoverage: manifest_dataset_id = manifest.get("source_dataset_id") or manifest.get("source_raster_id") if expected_dataset_id is not None and manifest_dataset_id and str(manifest_dataset_id) != str(expected_dataset_id): raise AppError( code="DETECTION_QA_COVERAGE_MISMATCH", message="Detection tile manifest belongs to a different raster dataset", details={ "analysis_dataset_id": str(expected_dataset_id), "manifest_dataset_id": str(manifest_dataset_id), }, status_code=422, ) tiles = manifest.get("tiles") if not isinstance(tiles, list) or not tiles: raise AppError( code="DETECTION_QA_COVERAGE_INVALID", message="Detection tile manifest has no usable tile coverage", status_code=422, ) default_crs = manifest.get("crs") or manifest.get("source_crs") or manifest.get("dataset_crs") coverage_parts: list[BaseGeometry] = [] source_crs_values: set[str] = set() target_crs = CRS.from_epsg(4326) for tile_index, tile in enumerate(tiles): if not isinstance(tile, dict): raise DetectionQaService._coverage_error("Tile manifest entries must be objects", tile_index) raw_bounds = tile.get("bounds") if not isinstance(raw_bounds, (list, tuple)) or len(raw_bounds) != 4: raise DetectionQaService._coverage_error("Tile manifest entries require four bounds values", tile_index) try: left, bottom, right, top = (float(value) for value in raw_bounds) except (TypeError, ValueError) as exc: raise DetectionQaService._coverage_error("Tile bounds must be numeric", tile_index) from exc if not all(isfinite(value) for value in (left, bottom, right, top)) or left >= right or bottom >= top: raise DetectionQaService._coverage_error("Tile bounds must define a finite non-empty extent", tile_index) raw_crs = tile.get("crs") or default_crs if not isinstance(raw_crs, str) or not raw_crs.strip(): raise DetectionQaService._coverage_error("Tile coverage requires explicit CRS metadata", tile_index) try: source_crs = CRS.from_user_input(raw_crs) except Exception as exc: raise DetectionQaService._coverage_error("Tile coverage CRS is invalid", tile_index) from exc source_crs_values.add(source_crs.to_string()) tile_geometry: BaseGeometry = box(left, bottom, right, top) if source_crs != target_crs: transformer = Transformer.from_crs(source_crs, target_crs, always_xy=True) tile_geometry = shapely_transform(transformer.transform, tile_geometry) tile_geometry = DetectionQaService._valid_geometry(tile_geometry, tile_index=tile_index) coverage_parts.append(tile_geometry) coverage_geometry = DetectionQaService._valid_geometry(unary_union(coverage_parts)) min_x, min_y, max_x, max_y = coverage_geometry.bounds if min_x < -180 or max_x > 180 or min_y < -90 or max_y > 90: raise AppError( code="DETECTION_QA_COVERAGE_INVALID", message="Transformed tile coverage falls outside EPSG:4326 bounds", details={"bounds": [min_x, min_y, max_x, max_y]}, status_code=422, ) return DetectionQaCoverage( geometry=coverage_geometry, manifest_path=manifest_path, tile_count=len(tiles), source_crs_values=tuple(sorted(source_crs_values)), ) @staticmethod def filter_population( geometries: list[tuple[dict[str, Any], BaseGeometry]], coverage: DetectionQaCoverage, ) -> CoveragePopulation: evaluated: list[tuple[dict[str, Any], BaseGeometry]] = [] excluded_outside_count = 0 clipped_boundary_count = 0 for feature, geometry in geometries: if geometry.is_empty or not geometry.intersects(coverage.geometry): excluded_outside_count += 1 continue try: clipped = geometry.intersection(coverage.geometry) except Exception as exc: raise AppError( code="GEOMETRY_OPERATION_UNSUPPORTED", message="Unable to clip QA geometry to persisted tile coverage", details={"reason": str(exc)}, status_code=422, ) from exc if clipped.is_empty or (geometry.geom_type in {"Polygon", "MultiPolygon"} and clipped.area <= 0): excluded_outside_count += 1 continue clipped = DetectionQaService._valid_geometry(clipped) if not coverage.geometry.covers(geometry): clipped_boundary_count += 1 evaluated.append((feature, clipped)) return CoveragePopulation( geometries=evaluated, raw_count=len(geometries), evaluated_count=len(evaluated), excluded_outside_count=excluded_outside_count, clipped_boundary_count=clipped_boundary_count, ) @staticmethod def box_to_footprint_diagnostics( strict_evidence: QaMatchEvidence, envelope_evidence: QaMatchEvidence, *, iou_threshold: float, ) -> dict[str, Any]: envelope_metrics = DetectionQaService._metrics(envelope_evidence) return { "diagnostic_only": True, "canonical_method": "candidate_polygon_vs_reference_footprint_iou", "diagnostic_method": "candidate_polygon_vs_reference_envelope_iou", "iou_threshold": iou_threshold, "strict_matches": strict_evidence.matches, "envelope_matches": envelope_evidence.matches, "possible_box_to_footprint_mismatch_count": max(0, envelope_evidence.matches - strict_evidence.matches), **envelope_metrics, } @staticmethod def _metrics(evidence: QaMatchEvidence) -> dict[str, Any]: precision = ( evidence.matches / (evidence.matches + evidence.false_positives) if evidence.matches + evidence.false_positives > 0 else None ) recall = ( evidence.matches / (evidence.matches + evidence.false_negatives) if evidence.matches + evidence.false_negatives > 0 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 mean_iou = ( sum(evidence.match_iou_values) / len(evidence.match_iou_values) if evidence.match_iou_values else None ) return { "envelope_false_positives": evidence.false_positives, "envelope_false_negatives": evidence.false_negatives, "envelope_precision": precision, "envelope_recall": recall, "envelope_f1_score": f1_score, "envelope_mean_iou": mean_iou, } @staticmethod def _valid_geometry(geometry: BaseGeometry, *, tile_index: int | None = None) -> BaseGeometry: if not geometry.is_valid: geometry = make_valid(geometry) if geometry.is_empty or not geometry.is_valid: raise DetectionQaService._coverage_error("Tile coverage geometry is empty or invalid", tile_index) if isinstance(geometry, GeometryCollection): polygonal_parts = [part for part in geometry.geoms if isinstance(part, (Polygon, MultiPolygon)) and not part.is_empty] if polygonal_parts: geometry = unary_union(polygonal_parts) return geometry @staticmethod def _coverage_error(message: str, tile_index: int | None = None) -> AppError: details = {"tile_index": tile_index} if tile_index is not None else None return AppError( code="DETECTION_QA_COVERAGE_INVALID", message=message, details=details, status_code=422, )