from __future__ import annotations import json from pathlib import Path import sys from types import SimpleNamespace from uuid import uuid4 import pytest from app.core.config import Settings from app.core.errors import AppError from app.models import AnalysisRun, Area, Dataset, Detection, Job, Project from app.services.detection_georeferencing import pixel_bbox_to_epsg4326_polygon from app.services.detection_service import DetectionService from app.services.model_registry_service import ModelRegistryService from app.services.yolo_adapter import YoloDetectionAdapter ROOT = Path(__file__).resolve().parents[2] class FakeSession: def __init__(self, objects=None) -> None: self.objects = objects or {} self.added = [] self.commits = 0 self.refreshes = [] def get(self, model, item_id): return self.objects.get((model, item_id)) def add(self, item) -> None: self.added.append(item) if getattr(item, "id", None) is not None: self.objects[(item.__class__, item.id)] = item def commit(self) -> None: self.commits += 1 def refresh(self, item) -> None: self.refreshes.append(item) class AvailableAdapter: @staticmethod def dependencies_available() -> bool: return True class MissingDependencyAdapter: @staticmethod def dependencies_available() -> bool: return False class MockYoloAdapter: def __init__(self, settings: Settings) -> None: self.settings = settings self.loaded_model_path: Path | None = None @staticmethod def dependencies_available() -> bool: return True def load_model(self, model_path: Path): self.loaded_model_path = model_path return object() 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 return [ { "class_name": "building", "confidence": 0.91, "bbox": [10.0, 20.0, 30.0, 40.0], "properties": {"adapter": "mock"}, } ] class MixedCaseYoloAdapter(MockYoloAdapter): def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]: return [ { "class_name": "Building", "confidence": 0.91, "bbox": [10.0, 20.0, 30.0, 40.0], "properties": {"adapter": "mock"}, } ] class OverlappingTileYoloAdapter(MockYoloAdapter): def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]: tile_index = int(tile_path.stem.split("_")[-1]) if tile_index == 0: bbox = [10.0, 20.0, 30.0, 40.0] confidence = 0.82 else: bbox = [11.0, 21.0, 31.0, 41.0] confidence = 0.91 return [ { "class_name": "building", "confidence": confidence, "bbox": bbox, "properties": {"adapter": "overlap"}, } ] class RecordingPredictModel: def __init__(self) -> None: self.seen_sources: list[dict] = [] 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, } ) return [] class ExplodingPredictModel: def predict(self, *, source, conf, imgsz, device, verbose): raise RuntimeError("expected input[1, 1, 480, 640] to have 3 channels, but got 1 channels instead") def _project_and_dataset(dataset_type: str = "raster"): project_id = uuid4() dataset_id = uuid4() project = Project(id=project_id, name="Geel") dataset = Dataset( id=dataset_id, project_id=project_id, name="source.tif", dataset_type=dataset_type, source="user_upload", storage_path="storage/uploads/source.tif", ) db = FakeSession(objects={(Project, project_id): project, (Dataset, dataset_id): dataset}) return db, project_id, dataset_id def _settings(tmp_path: Path, **overrides) -> Settings: model_path = tmp_path / "model.pt" values = { "yolo_enabled": True, "yolo_model_path": str(model_path), "yolo_max_tiles": 4, } values.update(overrides) return Settings(**values) def _manifest(tmp_path: Path, tile_count: int = 1) -> Path: tiles = [] for index in range(tile_count): tile_path = tmp_path / f"tile_{index:04d}.tif" tile_path.write_bytes(b"fixture") tiles.append( { "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], "index": index, } ) manifest_path = tmp_path / "manifest.json" manifest_path.write_text( json.dumps( { "tile_set_id": "tiles-fixture", "source_dataset_id": str(uuid4()), "source_raster_id": str(uuid4()), "tile_size": 100, "overlap": 0, "count": tile_count, "tiles": tiles, } ), encoding="utf-8", ) return manifest_path def test_yolo_configured_model_reports_not_configured_when_disabled(tmp_path: Path) -> None: settings = _settings(tmp_path, yolo_enabled=False) models = {model.model_id: model for model in ModelRegistryService.list_model_capabilities(settings=settings)} assert "yolo-configured" in models assert models["yolo-configured"].configured is False assert models["yolo-configured"].status == "not_configured" def test_yolo_configured_model_reports_dependency_unavailable(tmp_path: Path) -> None: model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) model = ModelRegistryService.get_model_capability( "yolo-configured", settings=settings, yolo_adapter_class=MissingDependencyAdapter, ) assert model is not None assert model.configured is False assert model.status == "dependency_unavailable" def test_yolo_configured_model_reports_configured_with_local_model_and_dependencies(tmp_path: Path) -> None: model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) model = ModelRegistryService.get_model_capability("yolo-configured", settings=settings, yolo_adapter_class=AvailableAdapter) assert model is not None assert model.configured is True assert model.status == "configured" assert model.version == settings.yolo_model_version assert model.nationally_validated is False assert model.operator_review_required is True assert model.validated_regions == ["flanders_mol_kempen"] assert model.supported_classes == ["building"] assert "Mol and the Kempen" in (model.validation_scope or "") def test_yolo_dependency_check_uses_real_imports_not_find_spec() -> None: source = (ROOT / "backend" / "app" / "services" / "yolo_adapter.py").read_text(encoding="utf-8") assert 'find_spec("ultralytics")' not in source assert "import ultralytics" in source assert "import torch" in source def test_yolo_runtime_fails_closed_when_cuda_is_required_but_unavailable(tmp_path: Path, monkeypatch) -> None: settings = _settings(tmp_path, yolo_device="cuda:0", yolo_require_cuda=True) monkeypatch.setitem(sys.modules, "torch", SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: False))) with pytest.raises(AppError) as exc_info: YoloDetectionAdapter(settings).validate_runtime() assert exc_info.value.code == "DETECTION_ACCELERATOR_UNAVAILABLE" def test_yolo_runtime_rejects_cpu_device_when_cuda_is_required(tmp_path: Path, monkeypatch) -> None: settings = _settings(tmp_path, yolo_device="cpu", yolo_require_cuda=True) monkeypatch.setitem(sys.modules, "torch", SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: True))) with pytest.raises(AppError) as exc_info: YoloDetectionAdapter(settings).validate_runtime() assert exc_info.value.code == "DETECTION_ACCELERATOR_MISCONFIGURED" def test_yolo_validation_scope_requires_persisted_validated_area(tmp_path: Path) -> None: dataset = Dataset(id=uuid4(), project_id=uuid4(), name="image.tif", dataset_type="raster", source="test", area_id=uuid4()) wrong_area = Area(id=dataset.area_id, project_id=dataset.project_id, name="Brussels", geometry="MULTIPOLYGON EMPTY") db = FakeSession(objects={(Area, dataset.area_id): wrong_area}) with pytest.raises(AppError) as exc_info: DetectionService._validate_model_area_scope(db, dataset, _settings(tmp_path, yolo_validated_area_names="Mol,Kempen")) assert exc_info.value.code == "DETECTION_VALIDATION_SCOPE_UNAVAILABLE" def test_yolo_validation_scope_accepts_bound_mol_area(tmp_path: Path) -> None: dataset = Dataset(id=uuid4(), project_id=uuid4(), name="image.tif", dataset_type="raster", source="test", area_id=uuid4()) area = Area(id=dataset.area_id, project_id=dataset.project_id, name="Gemeente Mol", geometry="MULTIPOLYGON EMPTY") db = FakeSession(objects={(Area, dataset.area_id): area}) DetectionService._validate_model_area_scope(db, dataset, _settings(tmp_path, yolo_validated_area_names="Mol,Kempen")) def test_yolo_run_requires_tile_manifest_path(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() settings = _settings(tmp_path) with pytest.raises(Exception) as exc_info: DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, settings=settings, yolo_adapter_class=AvailableAdapter, ) assert getattr(exc_info.value, "code", None) == "DETECTION_TILE_MANIFEST_REQUIRED" def test_yolo_run_rejects_manifest_over_tile_limit(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path), yolo_max_tiles=1) manifest_path = _manifest(tmp_path, tile_count=2) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=MockYoloAdapter, ) assert result.status == "failed" assert result.error_code == "DETECTION_TILE_LIMIT_EXCEEDED" def test_yolo_run_rejects_missing_tile_manifest_file(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, tile_manifest_path=str(tmp_path / "missing-manifest.json"), settings=settings, yolo_adapter_class=MockYoloAdapter, ) assert result.status == "failed" assert result.error_code == "DETECTION_TILE_MANIFEST_NOT_FOUND" def test_yolo_run_rejects_invalid_tile_manifest_json(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) manifest_path = tmp_path / "manifest.json" manifest_path.write_text("{not-json", encoding="utf-8") result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=MockYoloAdapter, ) assert result.status == "failed" assert result.error_code == "DETECTION_TILE_MANIFEST_INVALID" def test_pixel_bbox_to_epsg4326_polygon_from_gdal_transform() -> None: polygon = pixel_bbox_to_epsg4326_polygon( bbox=[10, 20, 30, 40], tile={ "transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01], "bounds": [4.0, 51.0, 5.0, 52.0], }, crs="EPSG:4326", ) assert polygon.bounds == pytest.approx((4.1, 51.6, 4.3, 51.8)) def test_yolo_run_persists_mocked_georeferenced_detections(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path), yolo_model_version="local-test") manifest_path = _manifest(tmp_path, tile_count=1) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, class_filter=["building"], tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=MockYoloAdapter, ) detections = [item for item in db.added if isinstance(item, Detection)] runs = [item for item in db.added if isinstance(item, AnalysisRun)] jobs = [item for item in db.added if isinstance(item, Job)] assert result.status == "success" assert result.detection_count == 1 assert detections[0].model_name == "yolo-configured" assert detections[0].model_version == "local-test" assert detections[0].class_name == "building" assert detections[0].confidence == 0.91 assert detections[0].source_tile_path.endswith("tile_0000.tif") assert detections[0].bbox_json == {"x_min": 10.0, "y_min": 20.0, "x_max": 30.0, "y_max": 40.0} assert detections[0].properties_json == {"adapter": "mock", "tile_index": 0} assert runs[0].status == "success" assert jobs[0].status == "success" def test_yolo_class_filter_is_case_insensitive_and_persists_canonical_class(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) manifest_path = _manifest(tmp_path, tile_count=1) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, class_filter=["building"], tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=MixedCaseYoloAdapter, ) detections = [item for item in db.added if isinstance(item, Detection)] assert result.status == "success" assert result.detection_count == 1 assert detections[0].class_name == "building" assert detections[0].properties_json["model_class_name"] == "Building" def test_yolo_run_suppresses_cross_tile_duplicate_detections(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path), yolo_duplicate_iou_threshold=0.5) manifest_path = _manifest(tmp_path, tile_count=2) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, class_filter=["building"], tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=OverlappingTileYoloAdapter, ) detections = [item for item in db.added if isinstance(item, Detection)] runs = [item for item in db.added if isinstance(item, AnalysisRun)] assert result.status == "success" assert result.detection_count == 1 assert detections[0].confidence == 0.91 assert detections[0].source_tile_path.endswith("tile_0001.tif") assert runs[0].result_json["raw_detection_count"] == 2 assert runs[0].result_json["suppressed_detection_count"] == 1 assert runs[0].result_json["duplicate_iou_threshold"] == 0.5 def test_yolo_adapter_converts_single_band_tiles_to_rgb_before_prediction(tmp_path: Path) -> None: Image = pytest.importorskip("PIL.Image") tile_path = tmp_path / "single_band_tile.tif" Image.new("L", (16, 16), 128).save(tile_path) model = RecordingPredictModel() settings = _settings(tmp_path, yolo_image_size=64, yolo_device="cpu") detections = YoloDetectionAdapter(settings).predict_tile(model, tile_path, confidence_threshold=0.25) assert detections == [] assert model.seen_sources[0]["mode"] == "RGB" assert model.seen_sources[0]["bands"] == 3 assert model.seen_sources[0]["path"] != str(tile_path) assert model.seen_sources[0]["conf"] == 0.25 assert model.seen_sources[0]["imgsz"] == 64 assert model.seen_sources[0]["device"] == "cpu" assert model.seen_sources[0]["verbose"] is False assert model.seen_sources[0]["max_det"] == 1000 def test_yolo_adapter_uses_configured_max_detections(tmp_path: Path) -> None: Image = pytest.importorskip("PIL.Image") tile_path = tmp_path / "rgb_tile.png" Image.new("RGB", (16, 16), (10, 20, 30)).save(tile_path) model = RecordingPredictModel() settings = _settings(tmp_path, yolo_max_detections=1500) detections = YoloDetectionAdapter(settings).predict_tile(model, tile_path, confidence_threshold=0.25) assert detections == [] assert model.seen_sources[0]["max_det"] == 1500 def test_yolo_adapter_wraps_prediction_runtime_errors(tmp_path: Path) -> None: tile_path = tmp_path / "tile.tif" tile_path.write_bytes(b"not an image but present") settings = _settings(tmp_path) with pytest.raises(AppError) as exc_info: YoloDetectionAdapter(settings).predict_tile(ExplodingPredictModel(), tile_path, confidence_threshold=0.25) assert exc_info.value.code == "DETECTION_INFERENCE_FAILED" assert "Configured YOLO inference failed for a raster tile" in exc_info.value.message assert exc_info.value.details["tile_path"] == str(tile_path)