277 lines
9.8 KiB
Python
277 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from geoalchemy2.shape import from_shape
|
|
from shapely.geometry import box
|
|
|
|
from app.main import app
|
|
from app.db.session import get_db
|
|
from app.models import AnalysisRun, Dataset, Detection, Job, Metric, Project, QualityCheck, VectorFeature
|
|
from app.services.detection_service import DetectionService
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, rows):
|
|
self.rows = list(rows)
|
|
|
|
def filter(self, *criteria):
|
|
for criterion in criteria:
|
|
left = getattr(criterion, "left", None)
|
|
right = getattr(criterion, "right", None)
|
|
operator = getattr(criterion, "operator", None)
|
|
name = getattr(left, "name", None)
|
|
value = getattr(right, "value", right)
|
|
if name and operator:
|
|
if operator.__name__ == "eq":
|
|
self.rows = [row for row in self.rows if getattr(row, name) == value]
|
|
elif operator.__name__ == "ge":
|
|
self.rows = [row for row in self.rows if getattr(row, name) >= value]
|
|
return self
|
|
|
|
def order_by(self, *_args):
|
|
return self
|
|
|
|
def all(self):
|
|
return list(self.rows)
|
|
|
|
def first(self):
|
|
return self.rows[0] if self.rows else None
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, objects=None, query_rows=None) -> None:
|
|
self.objects = objects or {}
|
|
self.query_rows = query_rows or {}
|
|
self.added = []
|
|
self.commits = 0
|
|
self.refreshes = []
|
|
|
|
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) -> 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)
|
|
|
|
|
|
def _detection(project_id, dataset_id, analysis_run_id, class_name="building", confidence=0.91, geom=None):
|
|
return Detection(
|
|
id=uuid4(),
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
analysis_run_id=analysis_run_id,
|
|
job_id=uuid4(),
|
|
model_name="yolo-configured",
|
|
model_version="local-test",
|
|
class_name=class_name,
|
|
confidence=confidence,
|
|
geometry=from_shape(geom or box(4.0, 51.0, 4.1, 51.1), srid=4326),
|
|
bbox_json={"x_min": 1, "y_min": 2, "x_max": 10, "y_max": 12},
|
|
source_tile_path="storage/tiles/tile_0000.tif",
|
|
)
|
|
|
|
|
|
def test_detection_geojson_feature_collection_shape() -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
analysis_run_id = uuid4()
|
|
detection = _detection(project_id, dataset_id, analysis_run_id)
|
|
db = FakeSession(
|
|
objects={(AnalysisRun, analysis_run_id): AnalysisRun(id=analysis_run_id, project_id=project_id, analysis_type="detection", status="success", parameters_json={})},
|
|
query_rows={Detection: [detection]},
|
|
)
|
|
|
|
feature_collection = DetectionService.detections_to_geojson(db, analysis_run_id=analysis_run_id)
|
|
|
|
assert feature_collection["type"] == "FeatureCollection"
|
|
assert len(feature_collection["features"]) == 1
|
|
feature = feature_collection["features"][0]
|
|
assert feature["geometry"]["type"] == "Polygon"
|
|
assert feature["properties"]["detection_id"] == str(detection.id)
|
|
assert feature["properties"]["class_name"] == "building"
|
|
assert feature["properties"]["confidence"] == 0.91
|
|
assert feature["properties"]["model_name"] == "yolo-configured"
|
|
assert feature["properties"]["analysis_run_id"] == str(analysis_run_id)
|
|
assert feature["properties"]["dataset_id"] == str(dataset_id)
|
|
assert feature["properties"]["job_id"] == str(detection.job_id)
|
|
assert feature["properties"]["source_tile_path"] == "storage/tiles/tile_0000.tif"
|
|
assert feature["properties"]["bbox_json"] == {"x_min": 1, "y_min": 2, "x_max": 10, "y_max": 12}
|
|
|
|
|
|
def test_detection_list_filters_by_dataset_class_and_confidence() -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
other_dataset_id = uuid4()
|
|
analysis_run_id = uuid4()
|
|
rows = [
|
|
_detection(project_id, dataset_id, analysis_run_id, "building", 0.91),
|
|
_detection(project_id, dataset_id, analysis_run_id, "road", 0.95),
|
|
_detection(project_id, dataset_id, analysis_run_id, "building", 0.25),
|
|
_detection(project_id, other_dataset_id, analysis_run_id, "building", 0.99),
|
|
]
|
|
db = FakeSession(
|
|
objects={(AnalysisRun, analysis_run_id): AnalysisRun(id=analysis_run_id, project_id=project_id, analysis_type="detection", status="success", parameters_json={})},
|
|
query_rows={Detection: rows},
|
|
)
|
|
|
|
result = DetectionService.list_detections(
|
|
db,
|
|
analysis_run_id=analysis_run_id,
|
|
dataset_id=dataset_id,
|
|
class_name="building",
|
|
min_confidence=0.5,
|
|
)
|
|
|
|
assert result.total == 1
|
|
assert result.items[0].class_name == "building"
|
|
assert result.items[0].confidence == 0.91
|
|
|
|
|
|
def test_detection_detail_returns_one_detection() -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
analysis_run_id = uuid4()
|
|
detection = _detection(project_id, dataset_id, analysis_run_id)
|
|
db = FakeSession(objects={(Detection, detection.id): detection})
|
|
|
|
result = DetectionService.get_detection(db, detection.id)
|
|
|
|
assert result.id == detection.id
|
|
assert result.class_name == "building"
|
|
|
|
|
|
def test_detection_models_api_envelope_still_canonical() -> None:
|
|
response = TestClient(app).get("/api/v1/detection/models")
|
|
|
|
assert response.status_code == 200
|
|
assert "data" in response.json()
|
|
assert "models" in response.json()["data"]
|
|
|
|
|
|
def test_detection_geojson_api_uses_canonical_envelope(monkeypatch) -> None:
|
|
analysis_run_id = uuid4()
|
|
|
|
monkeypatch.setattr(
|
|
"app.api.routes.detection.DetectionService.detections_to_geojson",
|
|
lambda *_args, **_kwargs: {"type": "FeatureCollection", "features": []},
|
|
)
|
|
app.dependency_overrides[get_db] = lambda: FakeSession()
|
|
try:
|
|
response = TestClient(app).get(f"/api/v1/detection/runs/{analysis_run_id}/geojson")
|
|
finally:
|
|
app.dependency_overrides.pop(get_db, None)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"data": {"type": "FeatureCollection", "features": []}}
|
|
|
|
|
|
def test_detection_qa_persists_quality_check_and_metrics() -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
reference_dataset_id = uuid4()
|
|
analysis_run_id = uuid4()
|
|
detection = _detection(project_id, dataset_id, analysis_run_id, geom=box(0, 0, 1, 1))
|
|
reference_dataset = Dataset(
|
|
id=reference_dataset_id,
|
|
project_id=project_id,
|
|
name="reference.geojson",
|
|
dataset_type="vector",
|
|
source="manual",
|
|
dataset_role="reference",
|
|
)
|
|
reference_feature = VectorFeature(
|
|
id=uuid4(),
|
|
dataset_id=reference_dataset_id,
|
|
feature_class="building",
|
|
geometry=from_shape(box(0, 0, 1, 1), srid=4326),
|
|
)
|
|
db = FakeSession(
|
|
objects={
|
|
(AnalysisRun, analysis_run_id): AnalysisRun(id=analysis_run_id, project_id=project_id, dataset_id=dataset_id, analysis_type="detection", status="success", parameters_json={}),
|
|
(Dataset, reference_dataset_id): reference_dataset,
|
|
},
|
|
query_rows={Detection: [detection], VectorFeature: [reference_feature]},
|
|
)
|
|
|
|
result = DetectionService.compare_detections_with_reference(
|
|
db=db,
|
|
analysis_run_id=analysis_run_id,
|
|
reference_dataset_id=reference_dataset_id,
|
|
iou_threshold=0.5,
|
|
)
|
|
|
|
quality_checks = [item for item in db.added if isinstance(item, QualityCheck)]
|
|
metrics = [item for item in db.added if isinstance(item, Metric)]
|
|
assert result["matches"] == 1
|
|
assert result["precision"] == 1.0
|
|
assert result["recall"] == 1.0
|
|
assert result["f1_score"] == 1.0
|
|
assert result["quality_check_id"] == str(quality_checks[0].id)
|
|
assert quality_checks[0].analysis_run_id == analysis_run_id
|
|
assert quality_checks[0].reference_dataset_id == reference_dataset_id
|
|
assert [metric.metric_key for metric in metrics] == [
|
|
"precision",
|
|
"recall",
|
|
"f1",
|
|
"mean_iou",
|
|
"false_positive_count",
|
|
"false_negative_count",
|
|
]
|
|
|
|
|
|
def test_detection_qa_no_match_case_persists_zero_scores() -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
reference_dataset_id = uuid4()
|
|
analysis_run_id = uuid4()
|
|
detection = _detection(project_id, dataset_id, analysis_run_id, geom=box(0, 0, 1, 1))
|
|
reference_dataset = Dataset(
|
|
id=reference_dataset_id,
|
|
project_id=project_id,
|
|
name="reference.geojson",
|
|
dataset_type="vector",
|
|
source="manual",
|
|
dataset_role="reference",
|
|
)
|
|
reference_feature = VectorFeature(
|
|
id=uuid4(),
|
|
dataset_id=reference_dataset_id,
|
|
feature_class="building",
|
|
geometry=from_shape(box(10, 10, 11, 11), srid=4326),
|
|
)
|
|
db = FakeSession(
|
|
objects={
|
|
(AnalysisRun, analysis_run_id): AnalysisRun(id=analysis_run_id, project_id=project_id, dataset_id=dataset_id, analysis_type="detection", status="success", parameters_json={}),
|
|
(Dataset, reference_dataset_id): reference_dataset,
|
|
},
|
|
query_rows={Detection: [detection], VectorFeature: [reference_feature]},
|
|
)
|
|
|
|
result = DetectionService.compare_detections_with_reference(
|
|
db=db,
|
|
analysis_run_id=analysis_run_id,
|
|
reference_dataset_id=reference_dataset_id,
|
|
iou_threshold=0.5,
|
|
)
|
|
|
|
assert result["matches"] == 0
|
|
assert result["false_positives"] == 1
|
|
assert result["false_negatives"] == 1
|
|
assert result["precision"] == 0.0
|
|
assert result["recall"] == 0.0
|
|
assert result["f1_score"] == 0.0
|