334 lines
11 KiB
Python
334 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from geoalchemy2.shape import to_shape
|
|
|
|
from app.core.config import Settings
|
|
from app.models import Dataset, Project, Segmentation
|
|
from app.services.detection_georeferencing import pixel_points_to_epsg4326_polygon
|
|
from app.services.model_registry_service import ModelRegistryService
|
|
from app.services.segmentation_service import SegmentationService
|
|
|
|
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 AvailableSegAdapter:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.settings = settings
|
|
|
|
@staticmethod
|
|
def dependencies_available() -> bool:
|
|
return True
|
|
|
|
def load_model(self, model_path: Path):
|
|
return object()
|
|
|
|
def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]:
|
|
return [
|
|
{
|
|
"class_name": "building",
|
|
"confidence": 0.91,
|
|
"points": [[10.0, 20.0], [30.0, 20.0], [30.0, 40.0], [10.0, 40.0]],
|
|
"bbox": [10.0, 20.0, 30.0, 40.0],
|
|
"properties": {"class_id": 0},
|
|
}
|
|
]
|
|
|
|
|
|
class ClassAgnosticSamAdapter(AvailableSegAdapter):
|
|
def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]:
|
|
return [
|
|
{
|
|
"class_name": "segment",
|
|
"confidence": None,
|
|
"points": [[5.0, 5.0], [25.0, 5.0], [25.0, 25.0], [5.0, 25.0]],
|
|
"bbox": [5.0, 5.0, 25.0, 25.0],
|
|
"properties": {"class_id": -1},
|
|
}
|
|
]
|
|
|
|
|
|
class MissingDependencySegAdapter(AvailableSegAdapter):
|
|
@staticmethod
|
|
def dependencies_available() -> bool:
|
|
return False
|
|
|
|
|
|
def _project_and_dataset(dataset_type: str = "raster"):
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
project = Project(id=project_id, name="Mol")
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="ortho.tif",
|
|
dataset_type=dataset_type,
|
|
source="user_upload",
|
|
storage_path="storage/uploads/ortho.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:
|
|
values = {
|
|
"yolo_seg_enabled": True,
|
|
"yolo_seg_model_path": str(tmp_path / "seg.pt"),
|
|
"sam_enabled": True,
|
|
"sam_model_path": str(tmp_path / "sam.pt"),
|
|
"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_segmentation_models_report_not_configured_when_disabled(tmp_path: Path) -> None:
|
|
settings = _settings(tmp_path, yolo_seg_enabled=False, sam_enabled=False)
|
|
|
|
models = {
|
|
model.model_id: model
|
|
for model in ModelRegistryService.list_segmentation_model_capabilities(settings=settings)
|
|
}
|
|
|
|
assert models["yolo-seg-configured"].configured is False
|
|
assert models["yolo-seg-configured"].status == "not_configured"
|
|
assert models["sam-configured"].configured is False
|
|
assert models["sam-configured"].status == "not_configured"
|
|
|
|
|
|
def test_segmentation_models_report_dependency_unavailable(tmp_path: Path) -> None:
|
|
(tmp_path / "seg.pt").write_bytes(b"weights")
|
|
(tmp_path / "sam.pt").write_bytes(b"weights")
|
|
settings = _settings(tmp_path)
|
|
|
|
models = {
|
|
model.model_id: model
|
|
for model in ModelRegistryService.list_segmentation_model_capabilities(
|
|
settings=settings,
|
|
yolo_seg_adapter_class=MissingDependencySegAdapter,
|
|
sam_adapter_class=MissingDependencySegAdapter,
|
|
)
|
|
}
|
|
|
|
assert models["yolo-seg-configured"].status == "dependency_unavailable"
|
|
assert models["sam-configured"].status == "dependency_unavailable"
|
|
|
|
|
|
def test_segmentation_models_report_configured_with_local_weights(tmp_path: Path) -> None:
|
|
(tmp_path / "seg.pt").write_bytes(b"weights")
|
|
(tmp_path / "sam.pt").write_bytes(b"weights")
|
|
settings = _settings(tmp_path)
|
|
|
|
models = {
|
|
model.model_id: model
|
|
for model in ModelRegistryService.list_segmentation_model_capabilities(
|
|
settings=settings,
|
|
yolo_seg_adapter_class=AvailableSegAdapter,
|
|
sam_adapter_class=ClassAgnosticSamAdapter,
|
|
)
|
|
}
|
|
|
|
assert models["yolo-seg-configured"].configured is True
|
|
assert models["yolo-seg-configured"].status == "configured"
|
|
assert models["sam-configured"].configured is True
|
|
assert models["sam-configured"].status == "configured"
|
|
|
|
|
|
def test_segmentation_dependency_check_uses_real_imports_not_find_spec() -> None:
|
|
source = (ROOT / "backend" / "app" / "services" / "segmentation_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_configured_segmentation_requires_tile_manifest(tmp_path: Path) -> None:
|
|
(tmp_path / "seg.pt").write_bytes(b"weights")
|
|
db, project_id, dataset_id = _project_and_dataset()
|
|
settings = _settings(tmp_path)
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
SegmentationService.run_segmentation(
|
|
db=db,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
model_id="yolo-seg-configured",
|
|
confidence_threshold=0.5,
|
|
settings=settings,
|
|
yolo_seg_adapter_class=AvailableSegAdapter,
|
|
sam_adapter_class=ClassAgnosticSamAdapter,
|
|
)
|
|
|
|
assert getattr(exc_info.value, "code", None) == "SEGMENTATION_TILE_MANIFEST_REQUIRED"
|
|
|
|
|
|
def test_configured_yolo_seg_run_persists_georeferenced_masks(tmp_path: Path) -> None:
|
|
(tmp_path / "seg.pt").write_bytes(b"weights")
|
|
db, project_id, dataset_id = _project_and_dataset()
|
|
settings = _settings(tmp_path)
|
|
manifest_path = _manifest(tmp_path)
|
|
|
|
response = SegmentationService.run_segmentation(
|
|
db=db,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
model_id="yolo-seg-configured",
|
|
confidence_threshold=0.5,
|
|
tile_manifest_path=str(manifest_path),
|
|
settings=settings,
|
|
yolo_seg_adapter_class=AvailableSegAdapter,
|
|
sam_adapter_class=ClassAgnosticSamAdapter,
|
|
)
|
|
|
|
assert response.status == "success"
|
|
assert response.segmentation_count == 1
|
|
persisted = [item for item in db.added if isinstance(item, Segmentation)]
|
|
assert len(persisted) == 1
|
|
segmentation = persisted[0]
|
|
assert segmentation.class_name == "building"
|
|
assert segmentation.confidence == pytest.approx(0.91)
|
|
geometry = to_shape(segmentation.geometry)
|
|
assert geometry.geom_type == "MultiPolygon"
|
|
min_x, min_y, max_x, max_y = geometry.bounds
|
|
assert 4.0 <= min_x <= 5.0
|
|
assert 51.0 <= min_y <= 52.0
|
|
assert max_x <= 5.0
|
|
assert max_y <= 52.0
|
|
assert segmentation.area_m2 is not None and segmentation.area_m2 > 0
|
|
assert segmentation.provenance_json["inference"] == "local"
|
|
assert segmentation.provenance_json["model_id"] == "yolo-seg-configured"
|
|
|
|
|
|
def test_configured_sam_run_is_class_agnostic(tmp_path: Path) -> None:
|
|
(tmp_path / "sam.pt").write_bytes(b"weights")
|
|
db, project_id, dataset_id = _project_and_dataset()
|
|
settings = _settings(tmp_path)
|
|
manifest_path = _manifest(tmp_path)
|
|
|
|
response = SegmentationService.run_segmentation(
|
|
db=db,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
model_id="sam-configured",
|
|
confidence_threshold=0.5,
|
|
tile_manifest_path=str(manifest_path),
|
|
settings=settings,
|
|
yolo_seg_adapter_class=AvailableSegAdapter,
|
|
sam_adapter_class=ClassAgnosticSamAdapter,
|
|
)
|
|
|
|
assert response.status == "success"
|
|
assert response.segmentation_count == 1
|
|
persisted = [item for item in db.added if isinstance(item, Segmentation)]
|
|
assert persisted[0].class_name == "segment"
|
|
assert persisted[0].confidence is None
|
|
|
|
|
|
def test_unconfigured_segmentation_run_fails_closed(tmp_path: Path) -> None:
|
|
db, project_id, dataset_id = _project_and_dataset()
|
|
settings = _settings(tmp_path, yolo_seg_enabled=False)
|
|
manifest_path = _manifest(tmp_path)
|
|
|
|
response = SegmentationService.run_segmentation(
|
|
db=db,
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
model_id="yolo-seg-configured",
|
|
confidence_threshold=0.5,
|
|
tile_manifest_path=str(manifest_path),
|
|
settings=settings,
|
|
yolo_seg_adapter_class=AvailableSegAdapter,
|
|
sam_adapter_class=ClassAgnosticSamAdapter,
|
|
)
|
|
|
|
assert response.status == "failed"
|
|
assert response.error_code == "SEGMENTATION_MODEL_UNAVAILABLE"
|
|
assert not [item for item in db.added if isinstance(item, Segmentation)]
|
|
|
|
|
|
def test_pixel_points_to_epsg4326_polygon_uses_tile_transform() -> None:
|
|
tile = {
|
|
"transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01],
|
|
"bounds": [4.0, 51.0, 5.0, 52.0],
|
|
"pixel_window": [0, 0, 100, 100],
|
|
}
|
|
|
|
polygon = pixel_points_to_epsg4326_polygon(
|
|
points=[[0.0, 0.0], [100.0, 0.0], [100.0, 100.0], [0.0, 100.0]],
|
|
tile=tile,
|
|
crs="EPSG:4326",
|
|
)
|
|
|
|
min_x, min_y, max_x, max_y = polygon.bounds
|
|
assert min_x == pytest.approx(4.0)
|
|
assert max_x == pytest.approx(5.0)
|
|
assert min_y == pytest.approx(51.0)
|
|
assert max_y == pytest.approx(52.0)
|
|
|
|
|
|
def test_pixel_points_to_epsg4326_polygon_rejects_degenerate_input() -> None:
|
|
tile = {"transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01]}
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
pixel_points_to_epsg4326_polygon(points=[[0.0, 0.0], [1.0, 1.0]], tile=tile)
|
|
|
|
assert getattr(exc_info.value, "code", None) == "SEGMENTATION_INVALID_MASK"
|