diff --git a/backend/app/api/routes/detection.py b/backend/app/api/routes/detection.py index 8afbffb5..324787af 100644 --- a/backend/app/api/routes/detection.py +++ b/backend/app/api/routes/detection.py @@ -18,6 +18,7 @@ from app.schemas import ( DetectionRunResponse, Envelope, GeoJsonFeatureCollection, + JobRead, ModelAssetListResponse, YoloPreflightResponse, ) @@ -71,6 +72,29 @@ def run_detection(payload: DetectionRunRequest, db: Session = Depends(get_db)) - return envelope(result.model_dump()) +@router.post("/run-async", response_model=Envelope[JobRead]) +def queue_detection(payload: DetectionRunRequest, db: Session = Depends(get_db)) -> dict: + """Queue a detection run for the background worker. + + Tiled GPU inference takes minutes; ``POST /detection/run`` performs it + inside the request and is only appropriate for a handful of tiles. Poll + ``GET /jobs/{id}`` for the queued run instead. + """ + + job = DetectionService.enqueue_detection( + db=db, + project_id=payload.project_id, + dataset_id=payload.dataset_id, + model_id=payload.model_id, + model_asset_id=payload.model_asset_id, + confidence_threshold=payload.confidence_threshold, + class_filter=payload.class_filter, + tile_manifest_path=payload.tile_manifest_path, + parameters_json=payload.parameters_json, + ) + return envelope(JobRead.model_validate(job).model_dump(mode="json")) + + @router.get("/runs", response_model=Envelope[DetectionRunListResponse]) def list_detection_runs( project_id: UUID | None = None, diff --git a/backend/app/api/routes/segmentation.py b/backend/app/api/routes/segmentation.py index baeb595f..6bed8eca 100644 --- a/backend/app/api/routes/segmentation.py +++ b/backend/app/api/routes/segmentation.py @@ -10,6 +10,7 @@ from app.schemas import ( AnalysisQaResponse, Envelope, GeoJsonFeatureCollection, + JobRead, SegmentationListResponse, SegmentationModelsResponse, SegmentationQaRequest, @@ -46,6 +47,27 @@ def run_segmentation(payload: SegmentationRunRequest, db: Session = Depends(get_ return envelope(result.model_dump()) +@router.post("/run-async", response_model=Envelope[JobRead]) +def queue_segmentation(payload: SegmentationRunRequest, db: Session = Depends(get_db)) -> dict: + """Queue a segmentation run for the background worker. + + Configured segmentation walks the same tile manifest as detection and is + just as unsuited to running inside the request. Poll ``GET /jobs/{id}``. + """ + + job = SegmentationService.enqueue_segmentation( + db=db, + project_id=payload.project_id, + dataset_id=payload.dataset_id, + model_id=payload.model_id, + confidence_threshold=payload.confidence_threshold, + class_filter=payload.class_filter, + tile_manifest_path=payload.tile_manifest_path, + parameters_json=payload.parameters_json, + ) + return envelope(JobRead.model_validate(job).model_dump(mode="json")) + + @router.get("/runs", response_model=Envelope[SegmentationRunListResponse]) def list_segmentation_runs( project_id: UUID | None = None, diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 0f716021..05d727b5 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -350,6 +350,12 @@ class Settings(BaseSettings): ) aoi_worker_enabled: bool = Field(default=False, validation_alias="GEOINTEL_AOI_WORKER_ENABLED") aoi_worker_poll_seconds: float = Field(default=2.0, ge=0.5, le=60.0, validation_alias="GEOINTEL_AOI_WORKER_POLL_SECONDS") + # Executes queued detection.run / segmentation.run jobs so tiled GPU + # inference never blocks an HTTP request. + analysis_worker_enabled: bool = Field(default=False, validation_alias="GEOINTEL_ANALYSIS_WORKER_ENABLED") + analysis_worker_poll_seconds: float = Field( + default=2.0, ge=0.5, le=60.0, validation_alias="GEOINTEL_ANALYSIS_WORKER_POLL_SECONDS" + ) database_statement_timeout_ms: int = Field(default=5_000, validation_alias="DATABASE_STATEMENT_TIMEOUT_MS") yolo_enabled: bool = Field(default=False, validation_alias="YOLO_ENABLED") yolo_models_dir: str = Field(default="/app/models", validation_alias="YOLO_MODELS_DIR") @@ -377,6 +383,9 @@ class Settings(BaseSettings): yolo_max_tiles: int = Field(default=100, validation_alias="YOLO_MAX_TILES") yolo_max_detections: int = Field(default=1000, validation_alias="YOLO_MAX_DETECTIONS") yolo_duplicate_iou_threshold: float = Field(default=0.5, ge=0.0, le=1.0, validation_alias="YOLO_DUPLICATE_IOU_THRESHOLD") + yolo_suppress_tile_edge_detections: bool = Field( + default=True, validation_alias="YOLO_SUPPRESS_TILE_EDGE_DETECTIONS" + ) yolo_batch_size: int = Field(default=1, validation_alias="YOLO_BATCH_SIZE") yolo_seg_enabled: bool = Field(default=False, validation_alias="YOLO_SEG_ENABLED") yolo_seg_model_path: str | None = Field(default=None, validation_alias="YOLO_SEG_MODEL_PATH") diff --git a/backend/app/main.py b/backend/app/main.py index 983135a4..0ce07d41 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -20,6 +20,7 @@ from app.core.request_context import reset_request_id, set_request_id from app.db.session import SessionLocal from app.services.runtime_reconciliation_service import RuntimeReconciliationService from app.services.auth_service import AuthService +from app.services.analysis_job_worker import AnalysisJobWorker from app.services.aoi_operation_worker import AoiOperationWorker @@ -50,6 +51,7 @@ def create_app() -> FastAPI: async def lifespan(_: FastAPI): worker_stop = asyncio.Event() worker_task = None + analysis_worker_task = None if settings.reconcile_interrupted_runs_on_startup: db = SessionLocal() try: @@ -69,12 +71,17 @@ def create_app() -> FastAPI: db.close() if settings.aoi_worker_enabled: worker_task = asyncio.create_task(AoiOperationWorker.run(worker_stop, settings.aoi_worker_poll_seconds)) + if settings.analysis_worker_enabled: + analysis_worker_task = asyncio.create_task( + AnalysisJobWorker.run(worker_stop, settings.analysis_worker_poll_seconds) + ) try: yield finally: worker_stop.set() - if worker_task is not None: - await worker_task + for task in (worker_task, analysis_worker_task): + if task is not None: + await task app = FastAPI( title="GeoIntel", diff --git a/backend/app/services/analysis_job_worker.py b/backend/app/services/analysis_job_worker.py new file mode 100644 index 00000000..1b9501f3 --- /dev/null +++ b/backend/app/services/analysis_job_worker.py @@ -0,0 +1,158 @@ +"""Background execution for queued analysis runs. + +Tiled GPU inference is minutes of work. Running it inside the HTTP request +holds a worker thread for the whole duration, times the client out and leaves +the operator without progress. Queued ``detection.run`` and +``segmentation.run`` jobs are picked up here instead, mirroring the polling +worker the AOI operations already use so the runtime keeps one job model. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any +from uuid import UUID + +from app.db.session import SessionLocal +from app.models import Job + +logger = logging.getLogger("geointel.analysis_worker") + + +class AnalysisJobWorker: + HANDLED_JOB_TYPES = ("detection.run", "segmentation.run") + BATCH_SIZE = 4 + + @staticmethod + def _uuid(value: Any) -> UUID | None: + if isinstance(value, UUID): + return value + try: + return UUID(str(value)) + except (TypeError, ValueError): + return None + + @staticmethod + def _dispatch(db, job: Job) -> Any: + # Imported lazily: both services import each other's helpers, and the + # worker must not add a third edge to that cycle at module load. + from app.services.detection_service import DetectionService + from app.services.segmentation_service import SegmentationService + + parameters = job.parameters_json if isinstance(job.parameters_json, dict) else {} + project_id = AnalysisJobWorker._uuid(parameters.get("project_id")) + dataset_id = AnalysisJobWorker._uuid(parameters.get("dataset_id")) + if project_id is None or dataset_id is None: + raise ValueError("Queued analysis job is missing project_id or dataset_id") + + common = { + "db": db, + "project_id": project_id, + "dataset_id": dataset_id, + "model_id": parameters.get("model_id"), + "confidence_threshold": float(parameters.get("confidence_threshold") or 0.0), + "class_filter": parameters.get("class_filter") or [], + "tile_manifest_path": parameters.get("tile_manifest_path"), + "parameters_json": parameters.get("parameters_json") or {}, + "existing_job": job, + } + if job.job_type == "detection.run": + return DetectionService.run_detection( + model_asset_id=parameters.get("model_asset_id"), + **common, + ) + return SegmentationService.run_segmentation(**common) + + @staticmethod + def _claim(db, job: Job) -> None: + """Take the job out of the queue before doing any work on it. + + Without this the next poll would pick the same row up again while the + first execution is still running on the GPU. + """ + + job.status = "running" + db.add(job) + db.commit() + + @staticmethod + def _finalize(db, job: Job, result: Any) -> None: + """Close a job the handler left open. + + The analysis services normally set the terminal status themselves. + If one returns without doing so, recording the outcome here is what + keeps the job from sitting in "running" for ever. + """ + + if job.status != "running": + return + status = getattr(result, "status", None) + if status == "success": + job.status = "success" + job.result_json = { + "detection_count": getattr(result, "detection_count", None), + "segmentation_count": getattr(result, "segmentation_count", None), + } + else: + job.status = "failed" + job.error_message = getattr(result, "message", None) or "Analysis run did not complete" + job.result_json = {"error_code": getattr(result, "error_code", None) or "ANALYSIS_JOB_INCOMPLETE"} + db.add(job) + db.commit() + + @staticmethod + def _mark_failed(db, job: Job, *, code: str, message: str) -> None: + try: + db.rollback() + except Exception: + pass + job.status = "failed" + job.error_message = message + job.result_json = {"error_code": code, "message": message} + db.add(job) + db.commit() + + @staticmethod + def run_once(db=None) -> int: + """Execute one batch of queued analysis jobs. Returns the batch size.""" + + owns_session = db is None + session = db if db is not None else SessionLocal() + try: + rows = [ + job + for job in ( + session.query(Job) + .filter(Job.status == "queued") + .filter(Job.job_type.in_(AnalysisJobWorker.HANDLED_JOB_TYPES)) + .order_by(Job.created_at) + .limit(AnalysisJobWorker.BATCH_SIZE) + .all() + ) + if job.job_type in AnalysisJobWorker.HANDLED_JOB_TYPES and job.status == "queued" + ] + for job in rows: + try: + AnalysisJobWorker._claim(session, job) + result = AnalysisJobWorker._dispatch(session, job) + AnalysisJobWorker._finalize(session, job, result) + except Exception as exc: + code = getattr(exc, "code", None) or "ANALYSIS_JOB_INTERNAL_ERROR" + message = getattr(exc, "message", None) or str(exc) or "Unexpected analysis job failure" + AnalysisJobWorker._mark_failed(session, job, code=str(code), message=str(message)) + logger.exception("Analysis job failed job_id=%s job_type=%s", job.id, job.job_type) + return len(rows) + finally: + if owns_session: + session.close() + + @staticmethod + async def run(stop_event: asyncio.Event, poll_seconds: float) -> None: + while not stop_event.is_set(): + processed = await asyncio.to_thread(AnalysisJobWorker.run_once) + if processed == 0: + try: + await asyncio.wait_for(stop_event.wait(), timeout=poll_seconds) + except TimeoutError: + pass diff --git a/backend/app/services/detection_georeferencing.py b/backend/app/services/detection_georeferencing.py index ee79a557..bde40bc2 100644 --- a/backend/app/services/detection_georeferencing.py +++ b/backend/app/services/detection_georeferencing.py @@ -8,6 +8,24 @@ from shapely.geometry import Polygon from app.core.errors import AppError +def _require_source_crs(crs: str | None, tile: dict[str, Any]) -> str: + """Resolve the CRS a pixel coordinate is measured in, or fail. + + Falling back to EPSG:4326 turned a missing manifest field into geometry + that sits in the wrong place while still looking like a valid polygon on + the map. A georeferenced result without a known CRS is not a result. + """ + + for candidate in (crs, tile.get("crs"), tile.get("source_crs")): + if isinstance(candidate, str) and candidate.strip(): + return candidate.strip() + raise AppError( + code="DETECTION_TILE_CRS_REQUIRED", + message="Georeferencing a tile requires explicit CRS metadata", + status_code=422, + ) + + def pixel_bbox_to_epsg4326_polygon(bbox: list[float], tile: dict[str, Any], crs: str | None = None) -> Polygon: if len(bbox) != 4: raise AppError(code="DETECTION_INVALID_BBOX", message="YOLO detection bbox must contain four pixel coordinates", status_code=422) @@ -28,7 +46,7 @@ def pixel_bbox_to_epsg4326_polygon(bbox: list[float], tile: dict[str, Any], crs: else: corners = _corners_from_bounds(bbox=[x_min, y_min, x_max, y_max], tile=tile) - source_crs = crs or tile.get("crs") or tile.get("source_crs") or "EPSG:4326" + source_crs = _require_source_crs(crs, tile) if str(source_crs).upper() not in {"EPSG:4326", "4326"}: transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True) corners = [transformer.transform(x, y) for x, y in corners] @@ -61,7 +79,7 @@ def pixel_points_to_epsg4326_polygon(points: list[list[float]], tile: dict[str, else: coordinates = [_project_pixel_with_bounds(tile, x, y) for x, y in pixel_points] - source_crs = crs or tile.get("crs") or tile.get("source_crs") or "EPSG:4326" + source_crs = _require_source_crs(crs, tile) if str(source_crs).upper() not in {"EPSG:4326", "4326"}: transformer = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True) coordinates = [transformer.transform(x, y) for x, y in coordinates] diff --git a/backend/app/services/detection_service.py b/backend/app/services/detection_service.py index 329eef3c..52ab48c4 100644 --- a/backend/app/services/detection_service.py +++ b/backend/app/services/detection_service.py @@ -9,7 +9,11 @@ from typing import Any from typing import Type from geoalchemy2.shape import from_shape, to_shape +from pyproj import Transformer +from shapely.geometry import box as shapely_box from shapely.geometry import mapping, shape +from shapely.ops import transform as shapely_transform +from shapely.strtree import STRtree from sqlalchemy import func from app.core.config import Settings, get_settings @@ -18,6 +22,7 @@ from app.core.request_context import get_request_id from app.models import AnalysisRun, Area, Dataset, Detection, Job, Project, VectorFeature from app.schemas.detection import DetectionListResponse, DetectionRead, DetectionRunListResponse, DetectionRunRead, DetectionRunResponse from app.services.detection_georeferencing import pixel_bbox_to_epsg4326_polygon +from app.services.detection_metrics_service import DetectionMetricsService from app.services.detection_qa_service import DetectionQaService from app.services.dataset_consumption_gate_service import DatasetConsumptionGate from app.services.model_asset_catalog_service import ModelAssetCatalogService @@ -51,22 +56,11 @@ class DetectionService: parameters_json: dict[str, Any] | None = None, settings: Settings | None = None, yolo_adapter_class: Type[YoloDetectionAdapter] = YoloDetectionAdapter, + existing_job: Job | None = None, ) -> DetectionRunResponse: parameters = dict(parameters_json or {}) resolved_settings = settings or get_settings() - project = db.get(Project, project_id) - if not project: - raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) - dataset = db.get(Dataset, dataset_id) - if not dataset or dataset.project_id != project_id: - raise AppError(code="DATASET_NOT_FOUND", message="Dataset not found", status_code=404) - if dataset.dataset_type != "raster": - raise AppError( - code="INVALID_DATASET_TYPE", - message="Detection requires a raster dataset", - details={"dataset_type": dataset.dataset_type}, - status_code=400, - ) + dataset = DetectionService._validate_run_request(db, project_id=project_id, dataset_id=dataset_id) TemporalCompatibilityService.ensure_detection_source_supported(dataset) selected_model_asset = None @@ -122,7 +116,7 @@ class DetectionService: "tile_manifest_path": tile_manifest_path, "parameters_json": parameters, } - job = DetectionService._create_job(db, project_id, dataset_id, run_parameters) + job = DetectionService._create_job(db, project_id, dataset_id, run_parameters, existing_job=existing_job) analysis_run = DetectionService._create_analysis_run(db, project_id, dataset_id, job.id, model, run_parameters) logger.info( "detection_started request_id=%s project_id=%s dataset_id=%s job_id=%s analysis_run_id=%s model_id=%s", @@ -419,7 +413,19 @@ class DetectionService: class_name=class_name, min_confidence=min_confidence, ) - raw_candidate_geometries = [({"id": str(row.id), "class_name": row.class_name}, to_shape(row.geometry)) for row in detections] + raw_candidate_geometries = [ + ( + { + "id": str(row.id), + "class_name": row.class_name, + # Confidence lets the matcher rank candidates the way + # detection benchmarks do instead of by row order. + "confidence": row.confidence, + }, + to_shape(row.geometry), + ) + for row in detections + ] candidate_geometries = raw_candidate_geometries coverage = None @@ -510,10 +516,31 @@ class DetectionService: reference_envelopes, iou_threshold, ) + candidate_geometry_mode = DetectionQaService.candidate_geometry_mode(candidate_geometries) box_to_footprint_diagnostics = DetectionQaService.box_to_footprint_diagnostics( evidence, envelope_evidence, iou_threshold=iou_threshold, + candidate_geometry_mode=candidate_geometry_mode, + ) + box_to_footprint_diagnostics["envelope_precision_recall_curve"] = ( + DetectionMetricsService.precision_recall_curve( + candidate_geometries, + reference_envelopes, + iou_threshold=iou_threshold, + ) + ) + if candidate_geometry_mode == "axis_aligned_boxes": + coverage_warnings.append( + "Candidates are axis-aligned detector boxes; strict footprint IoU cannot reach 1 for " + "rotated or non-rectangular buildings. See box_to_footprint_diagnostics." + ) + # Threshold-independent view of the same populations, so the run can be + # compared with another model instead of only with itself. + precision_recall_curve = DetectionMetricsService.precision_recall_curve( + candidate_geometries, + reference_geometries, + iou_threshold=iou_threshold, ) mean_iou = None if not evidence.match_iou_values else sum(evidence.match_iou_values) / len(evidence.match_iou_values) precision = evidence.matches / (evidence.matches + evidence.false_positives) if evidence.matches + evidence.false_positives > 0 else None @@ -549,6 +576,7 @@ class DetectionService: "coverage": coverage_summary, "temporal_compatibility": temporal_compatibility, "box_to_footprint_diagnostics": box_to_footprint_diagnostics, + "precision_recall_curve": precision_recall_curve, "match_evidence": evidence.match_evidence, "false_positive_evidence": evidence.false_positive_evidence, "false_negative_evidence": evidence.false_negative_evidence, @@ -560,6 +588,9 @@ class DetectionService: "mean_iou": mean_iou, "false_positive_count": evidence.false_positives, "false_negative_count": evidence.false_negatives, + "average_precision": precision_recall_curve["average_precision"], + "best_f1": precision_recall_curve["best_f1"], + "best_f1_threshold": precision_recall_curve["best_f1_threshold"], }, ) logger.info( @@ -594,13 +625,32 @@ class DetectionService: "coverage": coverage_summary, "temporal_compatibility": temporal_compatibility, "box_to_footprint_diagnostics": box_to_footprint_diagnostics, + "precision_recall_curve": precision_recall_curve, "match_evidence": evidence.match_evidence, "false_positive_evidence": evidence.false_positive_evidence, "false_negative_evidence": evidence.false_negative_evidence, } @staticmethod - def _create_job(db, project_id: uuid.UUID, dataset_id: uuid.UUID, parameters: dict[str, Any]) -> Job: + def _create_job( + db, + project_id: uuid.UUID, + dataset_id: uuid.UUID, + parameters: dict[str, Any], + existing_job: Job | None = None, + ) -> Job: + if existing_job is not None: + # A queued job already represents this run; reuse it so the client + # keeps polling one identifier from request to result. + existing_job.status = "running" + existing_job.dataset_id = dataset_id + existing_job.input_dataset_id = dataset_id + existing_job.parameters_json = {**(existing_job.parameters_json or {}), **parameters} + existing_job.started_at = DetectionService._now() + db.add(existing_job) + db.commit() + db.refresh(existing_job) + return existing_job job = Job( id=uuid.uuid4(), job_type="detection.run", @@ -616,6 +666,78 @@ class DetectionService: db.refresh(job) return job + @staticmethod + def enqueue_detection( + db, + project_id: uuid.UUID, + dataset_id: uuid.UUID, + model_id: str, + confidence_threshold: float, + model_asset_id: str | None = None, + class_filter: list[str] | None = None, + tile_manifest_path: str | None = None, + parameters_json: dict[str, Any] | None = None, + ) -> Job: + """Accept a detection run for background execution. + + Everything cheap enough to answer inside the request is checked here, + so an operator learns about a missing dataset or an unvalidated class + immediately rather than from a job that fails minutes later. + """ + + DetectionService._validate_run_request( + db, + project_id=project_id, + dataset_id=dataset_id, + ) + job = Job( + id=uuid.uuid4(), + job_type="detection.run", + status="queued", + project_id=project_id, + dataset_id=dataset_id, + input_dataset_id=dataset_id, + parameters_json={ + "project_id": str(project_id), + "dataset_id": str(dataset_id), + "model_id": model_id, + "model_asset_id": model_asset_id, + "confidence_threshold": confidence_threshold, + "class_filter": class_filter or [], + "tile_manifest_path": tile_manifest_path, + "parameters_json": dict(parameters_json or {}), + }, + ) + db.add(job) + db.commit() + db.refresh(job) + logger.info( + "detection_queued request_id=%s project_id=%s dataset_id=%s job_id=%s model_id=%s", + get_request_id(), + project_id, + dataset_id, + job.id, + model_id, + ) + return job + + @staticmethod + def _validate_run_request(db, *, project_id: uuid.UUID, dataset_id: uuid.UUID) -> Dataset: + project = db.get(Project, project_id) + if not project: + raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) + dataset = db.get(Dataset, dataset_id) + if not dataset or dataset.project_id != project_id: + raise AppError(code="DATASET_NOT_FOUND", message="Dataset not found", status_code=404) + if dataset.dataset_type != "raster": + raise AppError( + code="INVALID_DATASET_TYPE", + message="Detection requires a raster dataset", + details={"dataset_type": dataset.dataset_type}, + status_code=400, + ) + return dataset + @staticmethod def _query_detection_rows( db, @@ -634,7 +756,14 @@ class DetectionService: query = query.filter(Detection.class_name == class_name) if min_confidence is not None: query = query.filter(Detection.confidence >= min_confidence) - return query.order_by(Detection.created_at.desc()).all() + # ``created_at`` defaults to the transaction timestamp, so every + # detection in a run shares one value and ordering by it alone leaves + # the row order undefined. Confidence first, id as a stable tiebreak. + return query.order_by( + Detection.confidence.desc(), + Detection.created_at.desc(), + Detection.id.asc(), + ).all() @staticmethod def _detection_properties(detection: Detection) -> dict[str, Any]: @@ -792,10 +921,20 @@ class DetectionService: model = adapter.load_model(model_path) allowed_classes = {DetectionService._canonical_class_name(value) for value in class_filter if DetectionService._canonical_class_name(value)} candidates: list[dict[str, Any]] = [] - manifest_crs = manifest.get("crs") or manifest.get("source_crs") or manifest.get("dataset_crs") or "EPSG:4326" - for tile in manifest["tiles"]: - tile_path = DetectionService._resolve_tile_path(tile, Path(tile_manifest_path or "").expanduser()) - for raw in adapter.predict_tile(model, tile_path, confidence_threshold): + manifest_crs = DetectionService._require_manifest_crs(manifest) + raster_bounds = DetectionService._bounds_to_epsg4326(manifest.get("bounds"), manifest_crs) + tiles = list(manifest["tiles"]) + tile_paths = [ + DetectionService._resolve_tile_path(tile, Path(tile_manifest_path or "").expanduser()) for tile in tiles + ] + # Batched so the GPU is not idle between tiles; each tile keeps its own + # transform for georeferencing, so results stay per tile and in order. + detections_per_tile = adapter.predict_tiles(model, tile_paths, confidence_threshold) + for tile, tile_path, raw_detections in zip(tiles, tile_paths, detections_per_tile): + tile_crs = tile.get("crs") or manifest_crs + tile_bounds_4326 = DetectionService._bounds_to_epsg4326(tile.get("bounds"), tile_crs) + tile_edge_tolerance = DetectionService._tile_edge_tolerance(tile, tile_bounds_4326) + for raw in raw_detections: model_class_name = str(raw.get("class_name") or "").strip() class_name = DetectionService._canonical_class_name(model_class_name) confidence = float(raw.get("confidence", 0.0)) @@ -806,7 +945,7 @@ class DetectionService: bbox = raw.get("bbox") if not isinstance(bbox, list): raise AppError(code="DETECTION_INVALID_BBOX", message="YOLO adapter returned a detection without bbox", status_code=422) - geometry = pixel_bbox_to_epsg4326_polygon(bbox=bbox, tile=tile, crs=tile.get("crs") or manifest_crs) + geometry = pixel_bbox_to_epsg4326_polygon(bbox=bbox, tile=tile, crs=tile_crs) properties = dict(raw.get("properties") or {}) if model_class_name and model_class_name != class_name: properties.setdefault("model_class_name", model_class_name) @@ -818,10 +957,19 @@ class DetectionService: "bbox": bbox, "source_tile_path": str(tile_path), "properties": {**properties, "tile_index": tile.get("index")}, + "tile_bounds": tile_bounds_4326, + "tile_edge_tolerance": tile_edge_tolerance, } ) + edge_filtered_candidates = candidates + if settings.yolo_suppress_tile_edge_detections: + edge_filtered_candidates = DetectionService._drop_tile_edge_truncations( + candidates, + raster_bounds=raster_bounds, + tolerance=0.0, + ) filtered_candidates = DetectionService._suppress_duplicate_candidates( - candidates, + edge_filtered_candidates, iou_threshold=float(settings.yolo_duplicate_iou_threshold), ) persisted: list[Detection] = [] @@ -858,7 +1006,9 @@ class DetectionService: return persisted, { "raw_detection_count": len(candidates), "suppressed_detection_count": len(candidates) - len(filtered_candidates), + "tile_edge_truncated_count": len(candidates) - len(edge_filtered_candidates), "duplicate_iou_threshold": float(settings.yolo_duplicate_iou_threshold), + "containment_suppression_threshold": DetectionService.CONTAINMENT_SUPPRESSION_THRESHOLD, "runtime_model_provenance": runtime_model_provenance.as_dict(), } @@ -887,21 +1037,90 @@ class DetectionService: def _canonical_class_name(value: Any) -> str: return str(value or "").strip().casefold() + # An object wider than the tile overlap is truncated by both tiles, so the + # two halves barely intersect and IoU alone never suppresses them. Overlap + # measured against the smaller box catches that case; the threshold is + # deliberately strict so that terraced houses stay separate detections. + CONTAINMENT_SUPPRESSION_THRESHOLD = 0.85 + @staticmethod def _suppress_duplicate_candidates(candidates: list[dict[str, Any]], iou_threshold: float) -> list[dict[str, Any]]: if iou_threshold <= 0 or len(candidates) < 2: return candidates + + ordered = sorted( + candidates, + key=lambda item: (-float(item["confidence"]), str(item.get("source_tile_path") or "")), + ) kept: list[dict[str, Any]] = [] - for candidate in sorted(candidates, key=lambda item: float(item["confidence"]), reverse=True): + kept_geometries: list[Any] = [] + tree = None + + for candidate in ordered: + geometry = candidate["geometry"] duplicate = False - for kept_candidate in kept: + # Only geometries that actually touch this candidate can suppress + # it, so an index keeps a dense AOI from turning into an O(n^2) scan. + neighbour_indexes = range(len(kept)) if tree is None else (int(index) for index in tree.query(geometry)) + for index in neighbour_indexes: + kept_candidate = kept[index] if candidate["class_name"] != kept_candidate["class_name"]: continue - if DetectionService._geometry_iou(candidate["geometry"], kept_candidate["geometry"]) >= iou_threshold: + other = kept_geometries[index] + if DetectionService._geometry_iou(geometry, other) >= iou_threshold: + duplicate = True + break + if ( + DetectionService._geometry_containment(geometry, other) + >= DetectionService.CONTAINMENT_SUPPRESSION_THRESHOLD + ): duplicate = True break if not duplicate: kept.append(candidate) + kept_geometries.append(geometry) + tree = STRtree(kept_geometries) + return kept + + @staticmethod + def _drop_tile_edge_truncations( + candidates: list[dict[str, Any]], + *, + raster_bounds: tuple[float, float, float, float] | None, + tolerance: float, + ) -> list[dict[str, Any]]: + """Discard boxes cut off by an interior tile edge. + + Such a box describes only the part of the object that fell inside its + tile. Because tiles overlap, the neighbouring tile saw the object whole + and contributed the box worth keeping. A box against the outer raster + edge has no such neighbour and is kept. + """ + + if raster_bounds is None or tolerance <= 0: + return candidates + + raster_left, raster_bottom, raster_right, raster_top = raster_bounds + kept: list[dict[str, Any]] = [] + for candidate in candidates: + tile_bounds = candidate.get("tile_bounds") + if not tile_bounds or len(tuple(tile_bounds)) != 4: + kept.append(candidate) + continue + tile_left, tile_bottom, tile_right, tile_top = (float(value) for value in tile_bounds) + left, bottom, right, top = candidate["geometry"].bounds + # A pixel-sized tolerance per tile: a fixed degree value would be + # wrong for both a 10 cm orthophoto and a coarse thematic raster. + tolerance = float(candidate.get("tile_edge_tolerance") or 0.0) or tolerance + + touches_interior_edge = ( + (abs(left - tile_left) <= tolerance and abs(tile_left - raster_left) > tolerance) + or (abs(right - tile_right) <= tolerance and abs(tile_right - raster_right) > tolerance) + or (abs(bottom - tile_bottom) <= tolerance and abs(tile_bottom - raster_bottom) > tolerance) + or (abs(top - tile_top) <= tolerance and abs(tile_top - raster_top) > tolerance) + ) + if not touches_interior_edge: + kept.append(candidate) return kept @staticmethod @@ -916,6 +1135,78 @@ class DetectionService: return 0.0 return intersection_area / union_area + @staticmethod + def _geometry_containment(left, right) -> float: + """Intersection over the smaller of the two areas.""" + + if left.is_empty or right.is_empty: + return 0.0 + smaller_area = min(left.area, right.area) + if smaller_area <= 0: + return 0.0 + intersection_area = left.intersection(right).area + if intersection_area <= 0: + return 0.0 + return intersection_area / smaller_area + + @staticmethod + def _require_manifest_crs(manifest: dict[str, Any]) -> str: + """Refuse to georeference inference output against a guessed CRS. + + Detection QA already rejects a tile without explicit CRS metadata. + Silently assuming EPSG:4326 on the inference side produced geometry + that looks plausible on a map but sits in the wrong place. + """ + + raw_crs = manifest.get("crs") or manifest.get("source_crs") or manifest.get("dataset_crs") + if not isinstance(raw_crs, str) or not raw_crs.strip(): + raise AppError( + code="DETECTION_TILE_MANIFEST_INVALID", + message="Raster tile manifest requires explicit CRS metadata for georeferencing", + status_code=422, + ) + return raw_crs.strip() + + @staticmethod + def _bounds_to_epsg4326(bounds: Any, crs: str | None) -> tuple[float, float, float, float] | None: + if not isinstance(bounds, (list, tuple)) or len(bounds) != 4: + return None + try: + left, bottom, right, top = (float(value) for value in bounds) + except (TypeError, ValueError): + return None + if left >= right or bottom >= top: + return None + if not crs or str(crs).strip().upper() in {"EPSG:4326", "4326"}: + return (left, bottom, right, top) + try: + transformer = Transformer.from_crs(crs, "EPSG:4326", always_xy=True) + # Transform the whole rectangle, not just two corners: a projected + # box does not stay axis-aligned after reprojection. + projected = shapely_transform(transformer.transform, shapely_box(left, bottom, right, top)) + return projected.bounds + except Exception: + return None + + @staticmethod + def _tile_edge_tolerance(tile: dict[str, Any], tile_bounds_4326: tuple[float, float, float, float] | None) -> float: + """One and a half pixels, expressed in the degrees the boxes live in.""" + + if tile_bounds_4326 is None: + return 0.0 + pixel_window = tile.get("pixel_window") + if not (isinstance(pixel_window, (list, tuple)) and len(pixel_window) == 4): + return 0.0 + try: + width = float(pixel_window[2]) + height = float(pixel_window[3]) + except (TypeError, ValueError): + return 0.0 + if width <= 0 or height <= 0: + return 0.0 + left, bottom, right, top = tile_bounds_4326 + return 1.5 * max((right - left) / width, (top - bottom) / height) + @staticmethod def _load_tile_manifest(tile_manifest_path: str | None, max_tiles: int) -> dict[str, Any]: if not tile_manifest_path: diff --git a/backend/app/services/segmentation_service.py b/backend/app/services/segmentation_service.py index 95c8bf24..a606d65e 100644 --- a/backend/app/services/segmentation_service.py +++ b/backend/app/services/segmentation_service.py @@ -8,6 +8,7 @@ from typing import Any from geoalchemy2.shape import from_shape, to_shape from shapely.geometry import MultiPolygon, Polygon, mapping, shape from shapely.validation import make_valid +from sqlalchemy import func from app.core.config import Settings, get_settings from app.core.errors import AppError @@ -20,6 +21,7 @@ from app.schemas.segmentation import ( SegmentationRunResponse, ) from app.services.detection_georeferencing import pixel_points_to_epsg4326_polygon +from app.services.detection_qa_service import DetectionQaService from app.services.detection_service import DetectionService from app.services.dataset_consumption_gate_service import DatasetConsumptionGate from app.services.model_registry_service import ModelRegistryService @@ -51,22 +53,11 @@ class SegmentationService: settings: Settings | None = None, yolo_seg_adapter_class: type[YoloSegmentationAdapter] = YoloSegmentationAdapter, sam_adapter_class: type[SamSegmentationAdapter] = SamSegmentationAdapter, + existing_job: Job | None = None, ) -> SegmentationRunResponse: parameters = dict(parameters_json or {}) resolved_settings = settings or get_settings() - project = db.get(Project, project_id) - if not project: - raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) - dataset = db.get(Dataset, dataset_id) - if not dataset or dataset.project_id != project_id: - raise AppError(code="DATASET_NOT_FOUND", message="Dataset not found", status_code=404) - if dataset.dataset_type != "raster": - raise AppError( - code="INVALID_DATASET_TYPE", - message="Segmentation requires a raster dataset", - details={"dataset_type": dataset.dataset_type}, - status_code=400, - ) + dataset = SegmentationService._validate_run_request(db, project_id=project_id, dataset_id=dataset_id) model = ModelRegistryService.get_model_capability( model_id, @@ -110,7 +101,7 @@ class SegmentationService: "tile_manifest_path": tile_manifest_path, "parameters_json": parameters, } - job = SegmentationService._create_job(db, project_id, dataset_id, run_parameters) + job = SegmentationService._create_job(db, project_id, dataset_id, run_parameters, existing_job=existing_job) analysis_run = SegmentationService._create_analysis_run(db, project_id, dataset_id, job.id, model, run_parameters) if not model.configured: @@ -374,16 +365,98 @@ class SegmentationService: message="Segmentation run has no persisted geometries for QA", status_code=422, ) - references = db.query(VectorFeature).filter(VectorFeature.dataset_id == reference_dataset_id).all() - if not references: + # Score against the footprint the model actually saw. Without this the + # whole reference dataset is the denominator for recall, and every + # building outside the inferred tiles is counted as a miss. + manifest_path = DetectionQaService.tile_manifest_path(run_parameters) + coverage = None + if manifest_path: + manifest = DetectionService._load_tile_manifest(manifest_path, get_settings().yolo_max_tiles) + coverage = DetectionQaService.build_tile_coverage( + manifest, + manifest_path=manifest_path, + expected_dataset_id=run.dataset_id, + ) + + reference_query = db.query(VectorFeature).filter(VectorFeature.dataset_id == reference_dataset_id) + if coverage is not None and hasattr(reference_query, "count"): + reference_raw_count = reference_query.count() + references = reference_query.filter( + func.ST_Intersects(VectorFeature.geometry, from_shape(coverage.geometry, srid=4326)) + ).all() + else: + references = reference_query.all() + reference_raw_count = len(references) + if reference_raw_count == 0: raise AppError( code="REFERENCE_FEATURES_NOT_FOUND", message="Reference dataset has no persisted vector features for QA", status_code=422, ) - candidate_geometries = [({"id": str(row.id), "class_name": row.class_name}, to_shape(row.geometry)) for row in segmentations] - reference_geometries = [({"id": str(row.id), "feature_class": row.feature_class}, to_shape(row.geometry)) for row in references] + raw_candidate_geometries = [ + ( + {"id": str(row.id), "class_name": row.class_name, "confidence": row.confidence}, + to_shape(row.geometry), + ) + for row in segmentations + ] + raw_reference_geometries = [ + ({"id": str(row.id), "feature_class": row.feature_class}, to_shape(row.geometry)) for row in references + ] + candidate_geometries = raw_candidate_geometries + reference_geometries = raw_reference_geometries + + coverage_summary: dict[str, Any] = { + "applied": False, + "mode": "unbounded_no_manifest", + "manifest_path": None, + "tile_count": 0, + "source_crs_values": [], + "candidate_raw_count": len(raw_candidate_geometries), + "candidate_evaluated_count": len(raw_candidate_geometries), + "candidate_excluded_outside_count": 0, + "candidate_clipped_boundary_count": 0, + "reference_raw_count": reference_raw_count, + "reference_evaluated_count": len(raw_reference_geometries), + "reference_excluded_outside_count": 0, + "reference_clipped_boundary_count": 0, + } + coverage_warnings: list[str] = [] + if coverage is not None: + candidate_population = DetectionQaService.filter_population(raw_candidate_geometries, coverage) + reference_population = DetectionQaService.filter_population( + raw_reference_geometries, + coverage, + raw_count=reference_raw_count, + ) + candidate_geometries = candidate_population.geometries + reference_geometries = reference_population.geometries + if not reference_geometries: + raise AppError( + code="REFERENCE_FEATURES_OUTSIDE_COVERAGE", + message="Reference dataset has no polygon features inside persisted inference tile coverage", + status_code=422, + ) + coverage_summary = { + "applied": True, + "mode": "persisted_tile_manifest_union", + "manifest_path": coverage.manifest_path, + "tile_count": coverage.tile_count, + "source_crs_values": list(coverage.source_crs_values), + "candidate_raw_count": candidate_population.raw_count, + "candidate_evaluated_count": candidate_population.evaluated_count, + "candidate_excluded_outside_count": candidate_population.excluded_outside_count, + "candidate_clipped_boundary_count": candidate_population.clipped_boundary_count, + "reference_raw_count": reference_population.raw_count, + "reference_evaluated_count": reference_population.evaluated_count, + "reference_excluded_outside_count": reference_population.excluded_outside_count, + "reference_clipped_boundary_count": reference_population.clipped_boundary_count, + } + coverage_warnings.append( + "QA populations were clipped to the union of persisted inference tile footprints before matching." + ) + evidence = QaService._match_io_u_evidence( candidate_geometries, reference_geometries, @@ -411,13 +484,15 @@ class SegmentationService: "iou_threshold": iou_threshold, "class_name": class_name, "min_confidence": min_confidence, + "coverage_policy": coverage_summary["mode"], }, findings={ "matches": evidence.matches, "false_positives": evidence.false_positives, "false_negatives": evidence.false_negatives, - "warnings": evidence.warnings, + "warnings": coverage_warnings + evidence.warnings, "unsupported_geometry": evidence.unsupported, + "coverage": coverage_summary, "match_evidence": evidence.match_evidence, "false_positive_evidence": evidence.false_positive_evidence, "false_negative_evidence": evidence.false_negative_evidence, @@ -438,6 +513,8 @@ class SegmentationService: "reference_dataset_id": str(reference_dataset_id), "candidate_feature_count": len(candidate_geometries), "reference_feature_count": len(reference_geometries), + "candidate_feature_count_raw": len(raw_candidate_geometries), + "reference_feature_count_raw": reference_raw_count, "matches": evidence.matches, "false_positives": evidence.false_positives, "false_negatives": evidence.false_negatives, @@ -446,7 +523,8 @@ class SegmentationService: "f1_score": f1_score, "mean_iou": mean_iou, "iou_threshold": iou_threshold, - "warnings": evidence.warnings, + "warnings": coverage_warnings + evidence.warnings, + "coverage": coverage_summary, "match_evidence": evidence.match_evidence, "false_positive_evidence": evidence.false_positive_evidence, "false_negative_evidence": evidence.false_negative_evidence, @@ -458,7 +536,24 @@ class SegmentationService: return (Path(storage_root) / "masks" / str(project_id) / str(analysis_run_id) / tile_folder / f"mask_{segmentation_id}.png").as_posix() @staticmethod - def _create_job(db, project_id: uuid.UUID, dataset_id: uuid.UUID, parameters: dict[str, Any]) -> Job: + def _create_job( + db, + project_id: uuid.UUID, + dataset_id: uuid.UUID, + parameters: dict[str, Any], + existing_job: Job | None = None, + ) -> Job: + if existing_job is not None: + # Reuse the queued job so the operator polls one identifier. + existing_job.status = "running" + existing_job.dataset_id = dataset_id + existing_job.input_dataset_id = dataset_id + existing_job.parameters_json = {**(existing_job.parameters_json or {}), **parameters} + existing_job.started_at = SegmentationService._now() + db.add(existing_job) + db.commit() + db.refresh(existing_job) + return existing_job job = Job( id=uuid.uuid4(), job_type="segmentation.run", @@ -474,6 +569,59 @@ class SegmentationService: db.refresh(job) return job + @staticmethod + def _validate_run_request(db, *, project_id: uuid.UUID, dataset_id: uuid.UUID) -> Dataset: + project = db.get(Project, project_id) + if not project: + raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) + dataset = db.get(Dataset, dataset_id) + if not dataset or dataset.project_id != project_id: + raise AppError(code="DATASET_NOT_FOUND", message="Dataset not found", status_code=404) + if dataset.dataset_type != "raster": + raise AppError( + code="INVALID_DATASET_TYPE", + message="Segmentation requires a raster dataset", + details={"dataset_type": dataset.dataset_type}, + status_code=400, + ) + return dataset + + @staticmethod + def enqueue_segmentation( + db, + project_id: uuid.UUID, + dataset_id: uuid.UUID, + model_id: str, + confidence_threshold: float, + class_filter: list[str] | None = None, + tile_manifest_path: str | None = None, + parameters_json: dict[str, Any] | None = None, + ) -> Job: + """Accept a segmentation run for background execution.""" + + SegmentationService._validate_run_request(db, project_id=project_id, dataset_id=dataset_id) + job = Job( + id=uuid.uuid4(), + job_type="segmentation.run", + status="queued", + project_id=project_id, + dataset_id=dataset_id, + input_dataset_id=dataset_id, + parameters_json={ + "project_id": str(project_id), + "dataset_id": str(dataset_id), + "model_id": model_id, + "confidence_threshold": confidence_threshold, + "class_filter": class_filter or [], + "tile_manifest_path": tile_manifest_path, + "parameters_json": dict(parameters_json or {}), + }, + ) + db.add(job) + db.commit() + db.refresh(job) + return job + @staticmethod def _create_analysis_run(db, project_id, dataset_id, job_id, model, parameters: dict[str, Any]) -> AnalysisRun: analysis_run = AnalysisRun( @@ -568,7 +716,7 @@ class SegmentationService: model = adapter.load_model(model_path) allowed_classes = {DetectionService._canonical_class_name(value) for value in class_filter if DetectionService._canonical_class_name(value)} - manifest_crs = manifest.get("crs") or manifest.get("source_crs") or manifest.get("dataset_crs") or "EPSG:4326" + manifest_crs = DetectionService._require_manifest_crs(manifest) candidates: list[dict[str, Any]] = [] for tile in manifest["tiles"]: tile_path = DetectionService._resolve_tile_path(tile, Path(tile_manifest_path or "").expanduser()) @@ -787,7 +935,14 @@ class SegmentationService: query = query.filter(Segmentation.class_name == class_name) if min_confidence is not None: query = query.filter(Segmentation.confidence >= min_confidence) - return query.order_by(Segmentation.created_at.desc()).all() + # One transaction timestamp is shared by every row in a run, so + # ordering by it alone leaves the row order — and therefore the QA + # score — undefined. See DetectionService._query_detection_rows. + return query.order_by( + Segmentation.confidence.desc(), + Segmentation.created_at.desc(), + Segmentation.id.asc(), + ).all() @staticmethod def _segmentation_properties(segmentation: Segmentation) -> dict[str, Any]: diff --git a/backend/app/services/yolo_adapter.py b/backend/app/services/yolo_adapter.py index 45f9828f..a1acb030 100644 --- a/backend/app/services/yolo_adapter.py +++ b/backend/app/services/yolo_adapter.py @@ -1,6 +1,6 @@ from __future__ import annotations -from contextlib import contextmanager +from contextlib import ExitStack, contextmanager from pathlib import Path import tempfile from typing import Any @@ -86,53 +86,90 @@ class YoloDetectionAdapter: ) def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict[str, Any]]: - if not tile_path.exists() or not tile_path.is_file(): - raise AppError( - code="DETECTION_TILE_NOT_FOUND", - message="Tile referenced by manifest does not exist", - details={"tile_path": str(tile_path)}, - status_code=422, - ) - try: - with _prediction_source(tile_path) as prediction_source: - results = model.predict( - source=prediction_source, - conf=float(confidence_threshold), - imgsz=int(self.settings.yolo_image_size), - device=self.settings.yolo_device, - max_det=int(self.settings.yolo_max_detections), - verbose=False, - ) - except AppError: - raise - except Exception as exc: - raise AppError( - code="DETECTION_INFERENCE_FAILED", - message="Configured YOLO inference failed for a raster tile", - details={"tile_path": str(tile_path), "error": str(exc)}, - status_code=503, - ) from exc + return self.predict_tiles(model, [tile_path], confidence_threshold)[0] - detections: list[dict[str, Any]] = [] - for result in results: - names = getattr(result, "names", {}) or {} - boxes = getattr(result, "boxes", None) - if boxes is None: - continue - xyxy_values = _to_list(getattr(boxes, "xyxy", [])) - confidence_values = _to_list(getattr(boxes, "conf", [])) - class_values = _to_list(getattr(boxes, "cls", [])) - for index, bbox in enumerate(xyxy_values): - class_id = int(class_values[index]) if index < len(class_values) else -1 - detections.append( - { - "class_name": str(names.get(class_id, class_id)), - "confidence": float(confidence_values[index]) if index < len(confidence_values) else 0.0, - "bbox": [float(value) for value in bbox], - "properties": {"class_id": class_id}, - } + def predict_tiles( + self, + model, + tile_paths: list[Path], + confidence_threshold: float, + ) -> list[list[dict[str, Any]]]: + """Run inference over several tiles per GPU call. + + One ``predict`` call per tile leaves an RTX-class card mostly idle on a + run of a hundred tiles. Results are returned per tile, in the order the + tiles were given, so the caller can still georeference each detection + against its own tile transform. + """ + + for tile_path in tile_paths: + if not tile_path.exists() or not tile_path.is_file(): + raise AppError( + code="DETECTION_TILE_NOT_FOUND", + message="Tile referenced by manifest does not exist", + details={"tile_path": str(tile_path)}, + status_code=422, ) - return detections + + batch_size = max(1, int(self.settings.yolo_batch_size or 1)) + detections_per_tile: list[list[dict[str, Any]]] = [] + + for start in range(0, len(tile_paths), batch_size): + batch = tile_paths[start : start + batch_size] + with ExitStack() as stack: + sources = [stack.enter_context(_prediction_source(path)) for path in batch] + try: + results = model.predict( + source=sources, + conf=float(confidence_threshold), + imgsz=int(self.settings.yolo_image_size), + device=self.settings.yolo_device, + max_det=int(self.settings.yolo_max_detections), + verbose=False, + ) + except AppError: + raise + except Exception as exc: + raise AppError( + code="DETECTION_INFERENCE_FAILED", + message="Configured YOLO inference failed for a raster tile", + details={"tile_path": str(batch[0]), "error": str(exc)}, + status_code=503, + ) from exc + + results = list(results) + for offset in range(len(batch)): + result = results[offset] if offset < len(results) else None + detections_per_tile.append(_detections_from_result(result)) + + return detections_per_tile + + +def _detections_from_result(result: Any) -> list[dict[str, Any]]: + """Flatten one ultralytics result into the adapter's detection dicts.""" + + if result is None: + return [] + names = getattr(result, "names", {}) or {} + boxes = getattr(result, "boxes", None) + if boxes is None: + return [] + xyxy_values = _to_list(getattr(boxes, "xyxy", [])) + confidence_values = _to_list(getattr(boxes, "conf", [])) + class_values = _to_list(getattr(boxes, "cls", [])) + + detections: list[dict[str, Any]] = [] + for index, bbox in enumerate(xyxy_values): + class_id = int(class_values[index]) if index < len(class_values) else -1 + detections.append( + { + "class_name": str(names.get(class_id, class_id)), + "confidence": float(confidence_values[index]) if index < len(confidence_values) else 0.0, + "bbox": [float(value) for value in bbox], + "properties": {"class_id": class_id}, + } + ) + return detections def _to_list(value: Any) -> list[Any]: @@ -147,8 +184,95 @@ def _to_list(value: Any) -> list[Any]: return list(value) +def _rgb_band_indexes(dataset: Any) -> list[int]: + """Pick the three bands that carry visible colour, in R, G, B order. + + Belgian orthophoto tiles are commonly 4-band RGB + near-infrared. Taking + bands blindly would feed the detector an infrared channel as if it were + colour, so an explicit colour interpretation wins when the raster has one. + """ + + count = int(getattr(dataset, "count", 0) or 0) + if count <= 0: + raise ValueError("Raster tile has no bands") + if count == 1: + return [1, 1, 1] + + try: + from rasterio.enums import ColorInterp + + interpretations = list(getattr(dataset, "colorinterp", ()) or ()) + wanted = (ColorInterp.red, ColorInterp.green, ColorInterp.blue) + if all(interpretation in interpretations for interpretation in wanted): + return [interpretations.index(interpretation) + 1 for interpretation in wanted] + except Exception: + pass + + if count == 2: + return [1, 1, 1] + return [1, 2, 3] + + +def _stretch_to_uint8(data: Any, valid: Any) -> Any: + """Scale a (bands, H, W) array to 0-255 with one shared percentile stretch. + + ``uint8`` data is already display-ready and is passed through untouched; + inventing a stretch for it would change pixel values the model was trained + on. Anything wider (12-bit and 16-bit orthophotos, float reflectance) would + otherwise be truncated to near-black by a plain dtype cast. + + The stretch bounds are computed over all bands together, not per band. + A per-band stretch white-balances the tile and shifts every hue, while the + detector learned on ordinary RGB orthophotos. + """ + + import numpy as np + + if data.dtype == np.uint8: + return data + + if valid is not None and valid.any(): + sample = data[:, valid].reshape(-1) + else: + sample = data.reshape(-1) + if sample.size == 0: + return np.zeros(data.shape, dtype=np.uint8) + + low, high = (float(value) for value in np.percentile(sample.astype("float64"), (2.0, 98.0))) + if not high > low: + low, high = float(sample.min()), float(sample.max()) + if not high > low: + return np.full(data.shape, 0 if low == 0 else 255, dtype=np.uint8) + + scaled = (data.astype("float64") - low) * (255.0 / (high - low)) + return np.clip(scaled, 0.0, 255.0).astype(np.uint8) + + +def _read_tile_as_rgb(tile_path: Path) -> Any: + """Read a raster tile into an (H, W, 3) uint8 array fit for inference.""" + + import numpy as np + import rasterio + + with rasterio.open(tile_path) as dataset: + indexes = _rgb_band_indexes(dataset) + raw = dataset.read(indexes, masked=True) + + data = np.ma.getdata(raw) + mask = np.ma.getmaskarray(raw) + valid = ~mask.any(axis=0) + + rgb = np.moveaxis(_stretch_to_uint8(data, valid), 0, -1) + # Nodata collars stay black instead of dragging the stretch toward zero. + rgb = np.ascontiguousarray(rgb) + rgb[~valid] = 0 + return rgb + + @contextmanager def _prediction_source(tile_path: Path) -> Iterator[str]: + """Yield a path to an 8-bit RGB rendering of ``tile_path`` for the model.""" + temp_path: Path | None = None try: try: @@ -157,6 +281,20 @@ def _prediction_source(tile_path: Path) -> Iterator[str]: yield str(tile_path) return + try: + rgb = _read_tile_as_rgb(tile_path) + except Exception: + rgb = None + + if rgb is not None: + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as handle: + temp_path = Path(handle.name) + Image.fromarray(rgb).save(temp_path) + yield str(temp_path) + return + + # rasterio is unavailable or cannot read this file (a plain PNG/JPEG + # fixture, for instance). Fall back to the previous PIL handling. try: with Image.open(tile_path) as image: if image.mode == "RGB" and len(image.getbands()) == 3: diff --git a/backend/tests/test_analysis_job_queue.py b/backend/tests/test_analysis_job_queue.py new file mode 100644 index 00000000..2421d68b --- /dev/null +++ b/backend/tests/test_analysis_job_queue.py @@ -0,0 +1,176 @@ +"""Tiled GPU inference must not run inside an HTTP request. + +A configured YOLO run walks up to ``YOLO_MAX_TILES`` tiles through the GPU. +Doing that in the request handler holds a worker thread for minutes, gives the +operator no progress, and times the client out before the result exists. The +run is queued as a Job instead and executed by a background worker, which is +the same pattern the AOI operations already use. +""" + +from __future__ import annotations + +from uuid import uuid4 + +import pytest + +from app.core.errors import AppError +from app.models import AnalysisRun, Detection, Job +from app.services.analysis_job_worker import AnalysisJobWorker +from app.services.detection_service import DetectionService + + +class FakeQuery: + def __init__(self, rows): + self.rows = list(rows) + + def filter(self, *criteria): + return self + + def order_by(self, *_args): + return self + + def limit(self, count): + self.rows = self.rows[:count] + return self + + def all(self): + return list(self.rows) + + +class FakeSession: + def __init__(self, objects=None, query_rows=None): + self.objects = dict(objects or {}) + self.query_rows = query_rows or {} + self.added = [] + + def get(self, model, item_id): + return self.objects.get((model, item_id)) + + def query(self, model): + return FakeQuery(self.query_rows.get(model, [])) + + def add(self, item): + self.added.append(item) + if getattr(item, "id", None) is not None: + self.objects[(item.__class__, item.id)] = item + + def commit(self): + return None + + def rollback(self): + return None + + def refresh(self, _item): + return None + + def close(self): + return None + + +def _queued_job(**parameters) -> Job: + payload = { + "project_id": str(uuid4()), + "dataset_id": str(uuid4()), + "model_id": "yolo-configured", + "confidence_threshold": 0.4, + "class_filter": ["building"], + "tile_manifest_path": "/tiles/manifest.json", + "parameters_json": {}, + } + payload.update(parameters) + return Job( + id=uuid4(), + job_type="detection.run", + status="queued", + project_id=uuid4(), + parameters_json=payload, + ) + + +def test_queued_detection_job_is_dispatched_to_the_detection_service(monkeypatch) -> None: + job = _queued_job() + db = FakeSession(objects={(Job, job.id): job}, query_rows={Job: [job]}) + calls: list[dict] = [] + + def fake_run(**kwargs): + calls.append(kwargs) + return type( + "Result", + (), + { + "status": "success", + "detection_count": 3, + "analysis_run_id": uuid4(), + "job_id": kwargs["existing_job"].id, + "model_dump": lambda self, **_: {"status": "success", "detection_count": 3}, + }, + )() + + monkeypatch.setattr(DetectionService, "run_detection", staticmethod(fake_run)) + + processed = AnalysisJobWorker.run_once(db=db) + + assert processed == 1 + assert calls[0]["model_id"] == "yolo-configured" + assert calls[0]["confidence_threshold"] == 0.4 + assert calls[0]["tile_manifest_path"] == "/tiles/manifest.json" + assert calls[0]["existing_job"] is job + assert job.status == "success" + + +def test_a_failing_run_marks_the_job_failed_instead_of_leaving_it_running(monkeypatch) -> None: + job = _queued_job() + db = FakeSession(objects={(Job, job.id): job}, query_rows={Job: [job]}) + + def exploding(**_kwargs): + raise AppError(code="DETECTION_TILE_NOT_FOUND", message="missing tile", status_code=422) + + monkeypatch.setattr(DetectionService, "run_detection", staticmethod(exploding)) + + processed = AnalysisJobWorker.run_once(db=db) + + assert processed == 1 + assert job.status == "failed" + assert job.error_message == "missing tile" + assert job.result_json["error_code"] == "DETECTION_TILE_NOT_FOUND" + + +def test_an_unexpected_error_still_closes_the_job(monkeypatch) -> None: + job = _queued_job() + db = FakeSession(objects={(Job, job.id): job}, query_rows={Job: [job]}) + + def exploding(**_kwargs): + raise RuntimeError("CUDA out of memory") + + monkeypatch.setattr(DetectionService, "run_detection", staticmethod(exploding)) + + AnalysisJobWorker.run_once(db=db) + + assert job.status == "failed" + assert job.result_json["error_code"] == "ANALYSIS_JOB_INTERNAL_ERROR" + + +def test_job_types_the_worker_does_not_own_are_left_alone() -> None: + job = _queued_job() + job.job_type = "raster.clip" + db = FakeSession(objects={(Job, job.id): job}, query_rows={Job: [job]}) + + assert AnalysisJobWorker.run_once(db=db) == 0 + assert job.status == "queued" + + +def test_enqueue_validates_before_accepting_the_job() -> None: + """A bad request is rejected up front, not minutes later in the worker.""" + + db = FakeSession() + + with pytest.raises(AppError) as exc_info: + DetectionService.enqueue_detection( + db=db, + project_id=uuid4(), + dataset_id=uuid4(), + model_id="yolo-configured", + confidence_threshold=0.4, + ) + + assert exc_info.value.code == "PROJECT_NOT_FOUND" diff --git a/backend/tests/test_detection_georeferencing_crs_strictness.py b/backend/tests/test_detection_georeferencing_crs_strictness.py new file mode 100644 index 00000000..1139fe60 --- /dev/null +++ b/backend/tests/test_detection_georeferencing_crs_strictness.py @@ -0,0 +1,73 @@ +"""Inference must refuse to guess a CRS. + +Detection QA rejects a tile without explicit CRS metadata, but the inference +side silently assumed EPSG:4326. That produced geometry that renders as a +plausible polygon in the wrong place, which is worse than a clear failure: +"fail closed" is the stated rule for the runtime. +""" + +from __future__ import annotations + +import pytest + +from app.core.errors import AppError +from app.services.detection_georeferencing import ( + pixel_bbox_to_epsg4326_polygon, + pixel_points_to_epsg4326_polygon, +) +from app.services.detection_service import DetectionService + + +TILE_WITHOUT_CRS = { + "bounds": [4.0, 51.0, 5.0, 52.0], + "pixel_window": [0, 0, 100, 100], +} + + +def test_manifest_without_crs_is_rejected() -> None: + with pytest.raises(AppError) as exc_info: + DetectionService._require_manifest_crs({"tiles": [{"bounds": [0, 0, 1, 1]}]}) + + assert exc_info.value.code == "DETECTION_TILE_MANIFEST_INVALID" + + +def test_manifest_crs_is_read_from_any_of_the_documented_keys() -> None: + assert DetectionService._require_manifest_crs({"crs": "EPSG:31370"}) == "EPSG:31370" + assert DetectionService._require_manifest_crs({"source_crs": "EPSG:31370"}) == "EPSG:31370" + assert DetectionService._require_manifest_crs({"dataset_crs": "EPSG:3812"}) == "EPSG:3812" + + +def test_bbox_georeferencing_requires_an_explicit_crs() -> None: + with pytest.raises(AppError) as exc_info: + pixel_bbox_to_epsg4326_polygon(bbox=[0.0, 0.0, 10.0, 10.0], tile=TILE_WITHOUT_CRS) + + assert exc_info.value.code == "DETECTION_TILE_CRS_REQUIRED" + + +def test_mask_georeferencing_requires_an_explicit_crs() -> None: + with pytest.raises(AppError) as exc_info: + pixel_points_to_epsg4326_polygon( + points=[[0.0, 0.0], [10.0, 0.0], [10.0, 10.0]], tile=TILE_WITHOUT_CRS + ) + + assert exc_info.value.code == "DETECTION_TILE_CRS_REQUIRED" + + +def test_explicit_crs_on_the_tile_is_used() -> None: + tile = {**TILE_WITHOUT_CRS, "crs": "EPSG:4326"} + + polygon = pixel_bbox_to_epsg4326_polygon(bbox=[0.0, 0.0, 50.0, 50.0], tile=tile) + + assert polygon.bounds == pytest.approx((4.0, 51.5, 4.5, 52.0)) + + +def test_projected_bounds_are_reprojected_as_a_whole_rectangle() -> None: + # Lambert 72 around Mol. All four corners must be transformed, otherwise a + # rotated footprint is understated. + bounds = DetectionService._bounds_to_epsg4326([200000.0, 200000.0, 201000.0, 201000.0], "EPSG:31370") + + assert bounds is not None + min_x, min_y, max_x, max_y = bounds + assert 4.0 < min_x < 6.0 + assert 50.0 < min_y < 52.0 + assert max_x > min_x and max_y > min_y diff --git a/backend/tests/test_detection_tile_seam_handling.py b/backend/tests/test_detection_tile_seam_handling.py new file mode 100644 index 00000000..d96a2df6 --- /dev/null +++ b/backend/tests/test_detection_tile_seam_handling.py @@ -0,0 +1,125 @@ +"""Detections that straddle a tile seam must not become two half buildings. + +Tiling uses a fixed overlap. An object wider than that overlap is truncated by +both tiles, so the two boxes barely intersect and plain IoU suppression keeps +them both: two false positives plus one missed footprint for every seam +building. The suppressor therefore also compares overlap against the smaller +box, and truncated boxes that sit against an interior tile edge are dropped in +favour of the neighbouring tile's complete view. +""" + +from __future__ import annotations + +from shapely.geometry import box + +from app.services.detection_service import DetectionService + + +def _candidate(name: str, geometry, confidence: float, *, tile_index: int = 0, tile_bounds=None): + return { + "class_name": "building", + "confidence": confidence, + "geometry": geometry, + "bbox": [0.0, 0.0, 1.0, 1.0], + "source_tile_path": f"/tiles/tile_{tile_index:04d}.tif", + "properties": {"tile_index": tile_index, "name": name}, + "tile_bounds": tile_bounds, + } + + +def test_identical_overlapping_predictions_are_still_suppressed() -> None: + kept = DetectionService._suppress_duplicate_candidates( + [ + _candidate("a", box(0.0, 0.0, 1.0, 1.0), 0.7), + _candidate("b", box(0.02, 0.02, 1.02, 1.02), 0.9), + ], + iou_threshold=0.5, + ) + + assert [item["properties"]["name"] for item in kept] == ["b"] + + +def test_a_box_contained_in_a_larger_one_is_suppressed() -> None: + """A truncated seam half sits inside the complete box from the next tile.""" + + complete = box(0.0, 0.0, 10.0, 10.0) + truncated_half = box(0.0, 0.0, 4.0, 10.0) # IoU with ``complete`` is 0.4 + + kept = DetectionService._suppress_duplicate_candidates( + [ + _candidate("complete", complete, 0.88), + _candidate("truncated", truncated_half, 0.61), + ], + iou_threshold=0.5, + ) + + assert [item["properties"]["name"] for item in kept] == ["complete"] + + +def test_genuinely_adjacent_buildings_are_both_kept() -> None: + """Terraced houses touch but do not contain one another.""" + + kept = DetectionService._suppress_duplicate_candidates( + [ + _candidate("left", box(0.0, 0.0, 10.0, 10.0), 0.9), + _candidate("right", box(10.0, 0.0, 20.0, 10.0), 0.85), + ], + iou_threshold=0.5, + ) + + assert sorted(item["properties"]["name"] for item in kept) == ["left", "right"] + + +def test_different_classes_are_never_merged() -> None: + first = _candidate("a", box(0.0, 0.0, 10.0, 10.0), 0.9) + second = _candidate("b", box(0.0, 0.0, 10.0, 10.0), 0.8) + second["class_name"] = "solar_panel" + + kept = DetectionService._suppress_duplicate_candidates([first, second], iou_threshold=0.5) + + assert len(kept) == 2 + + +def test_boxes_clipped_by_an_interior_tile_edge_are_dropped() -> None: + """The overlapping neighbour tile still sees the whole object.""" + + tile = box(0.0, 0.0, 10.0, 10.0) + raster = box(0.0, 0.0, 30.0, 10.0) + + candidates = [ + # Sits against the tile's right edge: truncated by the tile, not real. + _candidate("edge", box(9.0, 2.0, 10.0, 4.0), 0.8, tile_bounds=tile.bounds), + # Comfortably inside the tile. + _candidate("interior", box(2.0, 2.0, 4.0, 4.0), 0.8, tile_bounds=tile.bounds), + ] + + kept = DetectionService._drop_tile_edge_truncations( + candidates, raster_bounds=raster.bounds, tolerance=0.001 + ) + + assert [item["properties"]["name"] for item in kept] == ["interior"] + + +def test_boxes_against_the_raster_edge_are_kept() -> None: + """No neighbouring tile exists there, so the box is all the evidence there is.""" + + tile = box(0.0, 0.0, 10.0, 10.0) + raster = box(0.0, 0.0, 10.0, 10.0) + + candidates = [_candidate("edge", box(9.0, 2.0, 10.0, 4.0), 0.8, tile_bounds=tile.bounds)] + + kept = DetectionService._drop_tile_edge_truncations( + candidates, raster_bounds=raster.bounds, tolerance=0.001 + ) + + assert [item["properties"]["name"] for item in kept] == ["edge"] + + +def test_edge_filter_keeps_candidates_without_tile_bounds() -> None: + candidates = [_candidate("unknown", box(2.0, 2.0, 4.0, 4.0), 0.8, tile_bounds=None)] + + kept = DetectionService._drop_tile_edge_truncations( + candidates, raster_bounds=(0.0, 0.0, 30.0, 10.0), tolerance=0.001 + ) + + assert len(kept) == 1 diff --git a/backend/tests/test_model_asset_catalog.py b/backend/tests/test_model_asset_catalog.py index ef0c676a..2888b009 100644 --- a/backend/tests/test_model_asset_catalog.py +++ b/backend/tests/test_model_asset_catalog.py @@ -50,6 +50,10 @@ class MockYoloAdapter: def load_model(self, model_path: Path): return {"model_path": str(model_path)} + def predict_tiles(self, model, tile_paths, confidence_threshold: float) -> list[list[dict]]: + # The service batches tiles; this double still answers per tile. + return [self.predict_tile(model, tile_path, confidence_threshold) for tile_path in tile_paths] + def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]: assert model["model_path"].endswith("building-detector.pt") return [ @@ -119,12 +123,15 @@ def _manifest(tmp_path: Path) -> Path: { "tile_set_id": "tiles-fixture", "count": 1, + "crs": "EPSG:4326", + "bounds": [4.0, 51.0, 5.0, 52.0], "tiles": [ { "path": str(tile_path), "pixel_window": [0, 0, 100, 100], "bounds": [4.0, 51.0, 5.0, 52.0], "transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01], + "crs": "EPSG:4326", "index": 0, } ], diff --git a/backend/tests/test_segmentation_configured_models.py b/backend/tests/test_segmentation_configured_models.py index bc5c09ad..d2c1f8c7 100644 --- a/backend/tests/test_segmentation_configured_models.py +++ b/backend/tests/test_segmentation_configured_models.py @@ -257,6 +257,7 @@ def _manifest(tmp_path: Path, tile_count: int = 1) -> Path: "pixel_window": [0, 0, 100, 100], "bounds": [4.0, 51.0, 5.0, 52.0], "transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01], + "crs": "EPSG:4326", "index": index, } ) @@ -267,6 +268,8 @@ def _manifest(tmp_path: Path, tile_count: int = 1) -> Path: "tile_set_id": "tiles-fixture", "source_dataset_id": str(uuid4()), "source_raster_id": str(uuid4()), + "crs": "EPSG:4326", + "bounds": [4.0, 51.0, 5.0, 52.0], "tile_size": 100, "overlap": 0, "count": tile_count, diff --git a/backend/tests/test_segmentation_qa_coverage.py b/backend/tests/test_segmentation_qa_coverage.py new file mode 100644 index 00000000..de9b1bed --- /dev/null +++ b/backend/tests/test_segmentation_qa_coverage.py @@ -0,0 +1,168 @@ +"""Segmentation QA must score against the area it actually inferred. + +Detection QA already clips both populations to the union of the persisted +inference tiles. Segmentation QA compared candidates against every reference +feature in the dataset, so every building outside the inferred tiles counted +as a false negative and recall collapsed for no modelling reason. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from uuid import uuid4 + +import pytest +from geoalchemy2.shape import from_shape +from shapely.geometry import MultiPolygon, box + +from app.core.errors import AppError +from app.models import AnalysisRun, Dataset, Segmentation, VectorFeature +from app.services.segmentation_service import SegmentationService + +from tests.test_sprint9_segmentation_foundation import ( # noqa: F401 + FakeSession, + _authoritative_reference, +) + + +def _manifest(tmp_path: Path, dataset_id, bounds: list[float]) -> str: + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text( + json.dumps( + { + "source_dataset_id": str(dataset_id), + "crs": "EPSG:4326", + "tiles": [{"path": "tile_0000.tif", "bounds": bounds, "crs": "EPSG:4326"}], + } + ), + encoding="utf-8", + ) + return str(manifest_path) + + +def _segmentation(project_id, dataset_id, analysis_run_id, geom): + return Segmentation( + id=uuid4(), + project_id=project_id, + dataset_id=dataset_id, + analysis_run_id=analysis_run_id, + job_id=uuid4(), + model_name="fixture-segmenter", + model_version="fixture-v1", + class_name="building", + confidence=0.9, + geometry=from_shape(MultiPolygon([geom]), srid=4326), + ) + + +def _reference(dataset_id, geom) -> VectorFeature: + return VectorFeature( + id=uuid4(), + dataset_id=dataset_id, + feature_class="building", + geometry=from_shape(geom, srid=4326), + ) + + +def _session(tmp_path: Path, *, with_manifest: bool): + project_id = uuid4() + dataset_id = uuid4() + reference_dataset_id = uuid4() + analysis_run_id = uuid4() + + parameters = {} + if with_manifest: + parameters = {"tile_manifest_path": _manifest(tmp_path, dataset_id, [0.0, 0.0, 1.0, 1.0])} + + reference_dataset = _authoritative_reference( + Dataset( + id=reference_dataset_id, + project_id=project_id, + name="reference.geojson", + dataset_type="vector", + source="test", + dataset_role="reference", + ) + ) + db = FakeSession( + objects={ + (AnalysisRun, analysis_run_id): AnalysisRun( + id=analysis_run_id, + project_id=project_id, + dataset_id=dataset_id, + analysis_type="segmentation", + status="success", + parameters_json=parameters, + ), + (Dataset, dataset_id): Dataset( + id=dataset_id, project_id=project_id, name="fixture.tif", dataset_type="raster", source="test" + ), + (Dataset, reference_dataset_id): reference_dataset, + }, + query_rows={ + Segmentation: [_segmentation(project_id, dataset_id, analysis_run_id, box(0.1, 0.1, 0.2, 0.2))], + VectorFeature: [ + # Inside the inferred tile: a genuine match. + _reference(reference_dataset_id, box(0.1, 0.1, 0.2, 0.2)), + # Far outside it: never looked at by the model. + _reference(reference_dataset_id, box(8.0, 8.0, 8.1, 8.1)), + _reference(reference_dataset_id, box(9.0, 9.0, 9.1, 9.1)), + ], + }, + ) + return db, analysis_run_id, reference_dataset_id + + +def test_segmentation_qa_scores_only_inside_persisted_tile_coverage(tmp_path: Path) -> None: + db, analysis_run_id, reference_dataset_id = _session(tmp_path, with_manifest=True) + + result = SegmentationService.compare_segmentations_with_reference( + db=db, + analysis_run_id=analysis_run_id, + reference_dataset_id=reference_dataset_id, + iou_threshold=0.5, + ) + + assert result["matches"] == 1 + assert result["false_negatives"] == 0 + assert result["recall"] == 1.0 + assert result["coverage"]["applied"] is True + assert result["coverage"]["reference_raw_count"] == 3 + assert result["coverage"]["reference_evaluated_count"] == 1 + assert result["coverage"]["reference_excluded_outside_count"] == 2 + assert any("tile" in warning for warning in result["warnings"]) + + +def test_segmentation_qa_without_manifest_reports_unbounded_coverage(tmp_path: Path) -> None: + db, analysis_run_id, reference_dataset_id = _session(tmp_path, with_manifest=False) + + result = SegmentationService.compare_segmentations_with_reference( + db=db, + analysis_run_id=analysis_run_id, + reference_dataset_id=reference_dataset_id, + iou_threshold=0.5, + ) + + # Unchanged behaviour, but the response now says the score was not bounded + # by an inference footprint so the recall can be read correctly. + assert result["false_negatives"] == 2 + assert result["coverage"]["applied"] is False + assert result["coverage"]["mode"] == "unbounded_no_manifest" + + +def test_segmentation_qa_rejects_reference_entirely_outside_coverage(tmp_path: Path) -> None: + db, analysis_run_id, reference_dataset_id = _session(tmp_path, with_manifest=True) + db.query_rows[VectorFeature] = [ + _reference(reference_dataset_id, box(8.0, 8.0, 8.1, 8.1)), + ] + + with pytest.raises(AppError) as exc_info: + SegmentationService.compare_segmentations_with_reference( + db=db, + analysis_run_id=analysis_run_id, + reference_dataset_id=reference_dataset_id, + iou_threshold=0.5, + ) + + assert exc_info.value.code == "REFERENCE_FEATURES_OUTSIDE_COVERAGE" diff --git a/backend/tests/test_sprint8b_yolo_foundation.py b/backend/tests/test_sprint8b_yolo_foundation.py index eafa1b23..c383215b 100644 --- a/backend/tests/test_sprint8b_yolo_foundation.py +++ b/backend/tests/test_sprint8b_yolo_foundation.py @@ -71,6 +71,10 @@ class MockYoloAdapter: self.loaded_model_path = model_path return object() + def predict_tiles(self, model, tile_paths, confidence_threshold: float) -> list[list[dict]]: + # The service batches tiles; this double still answers per tile. + return [self.predict_tile(model, tile_path, confidence_threshold) for tile_path in tile_paths] + def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]: assert tile_path.name == "tile_0000.tif" assert confidence_threshold == 0.5 @@ -130,19 +134,21 @@ class RecordingPredictModel: def predict(self, *, source, conf, imgsz, device, verbose, max_det): from PIL import Image - with Image.open(source) as image: - self.seen_sources.append( - { - "path": str(source), - "mode": image.mode, - "bands": len(image.getbands()), - "conf": conf, - "imgsz": imgsz, - "device": device, - "verbose": verbose, - "max_det": max_det, - } - ) + # Tiles are handed to the model in batches, so ``source`` is a list. + for item in source if isinstance(source, list) else [source]: + with Image.open(item) as image: + self.seen_sources.append( + { + "path": str(item), + "mode": image.mode, + "bands": len(image.getbands()), + "conf": conf, + "imgsz": imgsz, + "device": device, + "verbose": verbose, + "max_det": max_det, + } + ) return [] @@ -311,6 +317,7 @@ def _manifest(tmp_path: Path, tile_count: int = 1) -> Path: "pixel_window": [0, 0, 100, 100], "bounds": [4.0, 51.0, 5.0, 52.0], "transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01], + "crs": "EPSG:4326", "index": index, } ) @@ -321,6 +328,8 @@ def _manifest(tmp_path: Path, tile_count: int = 1) -> Path: "tile_set_id": "tiles-fixture", "source_dataset_id": str(uuid4()), "source_raster_id": str(uuid4()), + "crs": "EPSG:4326", + "bounds": [4.0, 51.0, 5.0, 52.0], "tile_size": 100, "overlap": 0, "count": tile_count, diff --git a/backend/tests/test_sprint8c_detection_visualization_qa.py b/backend/tests/test_sprint8c_detection_visualization_qa.py index 2fbb7f1e..6c3ea5ac 100644 --- a/backend/tests/test_sprint8c_detection_visualization_qa.py +++ b/backend/tests/test_sprint8c_detection_visualization_qa.py @@ -290,6 +290,11 @@ def test_detection_qa_persists_quality_check_and_metrics() -> None: "mean_iou", "false_positive_count", "false_negative_count", + # Threshold-independent metrics, so two models can be compared without + # both having to be read at the same confidence cut. + "average_precision", + "best_f1", + "best_f1_threshold", ] diff --git a/backend/tests/test_yolo_batched_inference.py b/backend/tests/test_yolo_batched_inference.py new file mode 100644 index 00000000..f4558b34 --- /dev/null +++ b/backend/tests/test_yolo_batched_inference.py @@ -0,0 +1,104 @@ +"""Tiles must reach the GPU in batches. + +``YOLO_BATCH_SIZE`` existed in the settings but nothing read it: every tile was +a separate ``model.predict`` call plus a separate temporary PNG. On an RTX-class +card that leaves most of the throughput unused for a run of a hundred tiles. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from app.core.errors import AppError +from app.services.yolo_adapter import YoloDetectionAdapter + + +class RecordingModel: + def __init__(self) -> None: + self.batches: list[list[str]] = [] + + def predict(self, *, source, conf, imgsz, device, verbose, max_det): + self.batches.append(list(source) if isinstance(source, list) else [source]) + return [] + + +def _settings(tmp_path: Path, **overrides): + from app.core.config import Settings + + values = { + "yolo_model_path": str(tmp_path / "model.pt"), + "yolo_device": "cpu", + "yolo_image_size": 64, + "yolo_max_detections": 1000, + "yolo_require_cuda": False, + "yolo_batch_size": 4, + } + values.update(overrides) + return Settings(**values) + + +def _tiles(tmp_path: Path, count: int) -> list[Path]: + Image = pytest.importorskip("PIL.Image") + paths = [] + for index in range(count): + path = tmp_path / f"tile_{index:04d}.png" + Image.new("RGB", (16, 16), (index, 20, 30)).save(path) + paths.append(path) + return paths + + +def test_tiles_are_predicted_in_configured_batches(tmp_path: Path) -> None: + tiles = _tiles(tmp_path, 9) + model = RecordingModel() + + YoloDetectionAdapter(_settings(tmp_path, yolo_batch_size=4)).predict_tiles(model, tiles, 0.25) + + assert [len(batch) for batch in model.batches] == [4, 4, 1] + + +def test_batch_size_one_still_works(tmp_path: Path) -> None: + tiles = _tiles(tmp_path, 3) + model = RecordingModel() + + YoloDetectionAdapter(_settings(tmp_path, yolo_batch_size=1)).predict_tiles(model, tiles, 0.25) + + assert [len(batch) for batch in model.batches] == [1, 1, 1] + + +def test_results_are_returned_per_tile_in_order(tmp_path: Path) -> None: + tiles = _tiles(tmp_path, 3) + + class PerTileModel: + def predict(self, *, source, conf, imgsz, device, verbose, max_det): + sources = list(source) if isinstance(source, list) else [source] + return [_FakeResult(index) for index, _ in enumerate(sources)] + + results = YoloDetectionAdapter(_settings(tmp_path)).predict_tiles(PerTileModel(), tiles, 0.25) + + assert len(results) == 3 + assert [len(detections) for detections in results] == [1, 1, 1] + + +def test_a_missing_tile_is_reported_before_the_batch_runs(tmp_path: Path) -> None: + tiles = _tiles(tmp_path, 2) + [tmp_path / "absent.png"] + + with pytest.raises(AppError) as exc_info: + YoloDetectionAdapter(_settings(tmp_path)).predict_tiles(RecordingModel(), tiles, 0.25) + + assert exc_info.value.code == "DETECTION_TILE_NOT_FOUND" + + +class _FakeBoxes: + xyxy = [[0.0, 0.0, 4.0, 4.0]] + conf = [0.9] + cls = [0] + + +class _FakeResult: + names = {0: "building"} + boxes = _FakeBoxes() + + def __init__(self, _index: int) -> None: + pass diff --git a/backend/tests/test_yolo_tile_image_preparation.py b/backend/tests/test_yolo_tile_image_preparation.py new file mode 100644 index 00000000..0a62e5bb --- /dev/null +++ b/backend/tests/test_yolo_tile_image_preparation.py @@ -0,0 +1,123 @@ +"""Orthophoto tiles must reach the model as a faithful 8-bit RGB image. + +Belgian orthophoto products are routinely 16-bit and/or 4-band (RGB + NIR). +Handing those to ``PIL.Image.convert("RGB")`` truncates the high byte, so a +bright roof arrives as a near-black pixel and the detector sees nothing that +resembles its training data. The tile is read with rasterio instead, the RGB +bands are selected explicitly and the values are percentile-stretched. +""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pytest + +from app.services.yolo_adapter import _prediction_source + +rasterio = pytest.importorskip("rasterio") +Image = pytest.importorskip("PIL.Image") + + +def _write_tile(path: Path, array: np.ndarray, dtype: str) -> None: + count, height, width = array.shape + with rasterio.open( + path, + "w", + driver="GTiff", + width=width, + height=height, + count=count, + dtype=dtype, + ) as dataset: + dataset.write(array.astype(dtype)) + + +def _prepared(tile_path: Path) -> np.ndarray: + with _prediction_source(tile_path) as source: + with Image.open(source) as image: + assert image.mode == "RGB" + return np.array(image) + + +def test_uint16_tile_keeps_its_contrast_instead_of_going_black(tmp_path: Path) -> None: + # A typical 12-bit-in-16-bit orthophoto: values well below 65535. + array = np.zeros((3, 32, 32), dtype=np.uint16) + array[0] = 800 + array[1] = 1600 + array[2] = 3200 + array[:, 0, 0] = 40 # a dark corner so the stretch has a low anchor + tile_path = tmp_path / "uint16.tif" + _write_tile(tile_path, array, "uint16") + + prepared = _prepared(tile_path) + + assert prepared.shape == (32, 32, 3) + # Naive 16->8 bit truncation would map 800/1600/3200 to near zero. + assert prepared.max() > 200 + # The three bands stay distinguishable rather than collapsing together. + assert prepared[16, 16, 0] < prepared[16, 16, 1] < prepared[16, 16, 2] + + +def test_four_band_rgbi_tile_drops_the_infrared_band(tmp_path: Path) -> None: + array = np.zeros((4, 16, 16), dtype=np.uint8) + array[0] = 10 + array[1] = 120 + array[2] = 240 + array[3] = 255 # near-infrared must not be treated as an alpha or a colour + tile_path = tmp_path / "rgbi.tif" + _write_tile(tile_path, array, "uint8") + + prepared = _prepared(tile_path) + + assert prepared.shape == (16, 16, 3) + assert prepared[8, 8, 0] < prepared[8, 8, 1] < prepared[8, 8, 2] + + +def test_single_band_tile_is_replicated_across_rgb(tmp_path: Path) -> None: + array = np.full((1, 16, 16), 128, dtype=np.uint8) + array[0, 0, 0] = 0 + array[0, 15, 15] = 255 + tile_path = tmp_path / "gray.tif" + _write_tile(tile_path, array, "uint8") + + prepared = _prepared(tile_path) + + assert prepared.shape == (16, 16, 3) + assert prepared[8, 8, 0] == prepared[8, 8, 1] == prepared[8, 8, 2] + + +def test_eight_bit_rgb_tile_is_passed_through_unchanged(tmp_path: Path) -> None: + array = np.zeros((3, 16, 16), dtype=np.uint8) + array[0] = 10 + array[1] = 120 + array[2] = 240 + tile_path = tmp_path / "rgb.tif" + _write_tile(tile_path, array, "uint8") + + prepared = _prepared(tile_path) + + # Already display-ready: no stretch should be invented for it. + assert prepared[8, 8].tolist() == [10, 120, 240] + + +def test_nodata_pixels_do_not_drive_the_stretch(tmp_path: Path) -> None: + array = np.full((3, 32, 32), 2000, dtype=np.uint16) + array[:, :4, :] = 0 # nodata collar from a clipped orthophoto + tile_path = tmp_path / "nodata.tif" + with rasterio.open( + tile_path, + "w", + driver="GTiff", + width=32, + height=32, + count=3, + dtype="uint16", + nodata=0, + ) as dataset: + dataset.write(array) + + prepared = _prepared(tile_path) + + assert prepared[16, 16].tolist() != [0, 0, 0]