240 lines
10 KiB
Python
240 lines
10 KiB
Python
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,
|
|
*,
|
|
raw_count: int | None = None,
|
|
) -> CoveragePopulation:
|
|
evaluated: list[tuple[dict[str, Any], BaseGeometry]] = []
|
|
resolved_raw_count = len(geometries) if raw_count is None else raw_count
|
|
if resolved_raw_count < len(geometries):
|
|
raise ValueError("raw_count cannot be smaller than the supplied geometry population")
|
|
excluded_outside_count = resolved_raw_count - len(geometries)
|
|
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=resolved_raw_count,
|
|
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,
|
|
)
|