feat: scope detection QA to inference coverage
GeoIntel CI / docs-smoke (push) Canceled after 0s
GeoIntel CI / contract-smoke (push) Canceled after 0s

This commit is contained in:
Codex
2026-07-14 11:25:28 +02:00
parent aae6315981
commit 948e50b5e1
17 changed files with 762 additions and 7 deletions
@@ -28,6 +28,9 @@ def test_real_data_detection_qa_smoke_requires_operator_inputs_and_checks_full_c
assert "/api/v1/detection/yolo/preflight" in script
assert "/api/v1/detection/run" in script
assert "/qa/reference" in script
assert "persisted_tile_manifest_union" in script
assert "box_to_footprint_diagnostics" in script
assert "candidate_polygon_vs_reference_footprint_iou" in script
assert "/api/v1/exports/geojson" in script
assert "Response is not a canonical GeoIntel data envelope" in script
assert "No local model assets are available" in script
@@ -0,0 +1,80 @@
from __future__ import annotations
from uuid import uuid4
import pytest
from pyproj import Transformer
from shapely.geometry import box
from app.core.errors import AppError
from app.services.detection_qa_service import DetectionQaService
def test_tile_coverage_transforms_projected_manifest_bounds_to_epsg4326() -> None:
dataset_id = uuid4()
to_lambert = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True)
left, bottom = to_lambert.transform(5.11, 51.18)
right, top = to_lambert.transform(5.13, 51.20)
manifest = {
"source_dataset_id": str(dataset_id),
"crs": "EPSG:31370",
"tiles": [{"bounds": [left, bottom, right, top], "crs": "EPSG:31370"}],
}
coverage = DetectionQaService.build_tile_coverage(
manifest,
manifest_path="/app/storage/tiles/manifest.json",
expected_dataset_id=dataset_id,
)
min_x, min_y, max_x, max_y = coverage.geometry.bounds
assert min_x == pytest.approx(5.11, abs=0.001)
assert min_y == pytest.approx(51.18, abs=0.001)
assert max_x == pytest.approx(5.13, abs=0.001)
assert max_y == pytest.approx(51.20, abs=0.001)
assert coverage.tile_count == 1
def test_tile_coverage_rejects_manifest_for_different_dataset() -> None:
manifest = {
"source_dataset_id": str(uuid4()),
"crs": "EPSG:4326",
"tiles": [{"bounds": [5.0, 51.0, 5.1, 51.1]}],
}
with pytest.raises(AppError) as exc_info:
DetectionQaService.build_tile_coverage(
manifest,
manifest_path="/app/storage/tiles/manifest.json",
expected_dataset_id=uuid4(),
)
assert exc_info.value.code == "DETECTION_QA_COVERAGE_MISMATCH"
def test_coverage_filter_reports_outside_and_boundary_clipped_population() -> None:
dataset_id = uuid4()
coverage = DetectionQaService.build_tile_coverage(
{
"source_dataset_id": str(dataset_id),
"crs": "EPSG:4326",
"tiles": [{"bounds": [0.0, 0.0, 1.0, 1.0]}],
},
manifest_path="/app/storage/tiles/manifest.json",
expected_dataset_id=dataset_id,
)
population = DetectionQaService.filter_population(
[
({"id": "inside"}, box(0.1, 0.1, 0.2, 0.2)),
({"id": "crossing"}, box(0.8, 0.8, 1.2, 1.2)),
({"id": "outside"}, box(2.0, 2.0, 3.0, 3.0)),
],
coverage,
)
assert population.raw_count == 3
assert population.evaluated_count == 2
assert population.excluded_outside_count == 1
assert population.clipped_boundary_count == 1
assert population.geometries[1][1].bounds == pytest.approx((0.8, 0.8, 1.0, 1.0))
@@ -6,7 +6,7 @@ from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
from geoalchemy2.shape import from_shape
from shapely.geometry import box
from shapely.geometry import Polygon, box
from app.main import app
from app.db.session import get_db
@@ -274,3 +274,150 @@ def test_detection_qa_no_match_case_persists_zero_scores() -> None:
assert result["precision"] == 0.0
assert result["recall"] == 0.0
assert result["f1_score"] == 0.0
def _coverage_manifest(tmp_path, dataset_id, bounds=(-1.0, -1.0, 3.0, 3.0)):
manifest_path = tmp_path / "manifest.json"
manifest_path.write_text(
json.dumps(
{
"source_dataset_id": str(dataset_id),
"crs": "EPSG:4326",
"tiles": [
{
"index": 0,
"path": "tile_0000.tif",
"bounds": list(bounds),
"crs": "EPSG:4326",
}
],
}
),
encoding="utf-8",
)
return manifest_path
def test_detection_qa_excludes_references_outside_persisted_tile_coverage(tmp_path) -> None:
project_id = uuid4()
dataset_id = uuid4()
reference_dataset_id = uuid4()
analysis_run_id = uuid4()
manifest_path = _coverage_manifest(tmp_path, dataset_id, bounds=(0.0, 0.0, 1.0, 1.0))
detection = _detection(project_id, dataset_id, analysis_run_id, geom=box(0.1, 0.1, 0.9, 0.9))
reference_dataset = Dataset(
id=reference_dataset_id,
project_id=project_id,
name="reference.geojson",
dataset_type="vector",
source="manual",
dataset_role="reference",
)
inside_reference = VectorFeature(
id=uuid4(),
dataset_id=reference_dataset_id,
feature_class="building",
geometry=from_shape(box(0.1, 0.1, 0.9, 0.9), srid=4326),
)
outside_reference = VectorFeature(
id=uuid4(),
dataset_id=reference_dataset_id,
feature_class="building",
geometry=from_shape(box(10.0, 10.0, 11.0, 11.0), 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",
model_name="yolo-configured",
parameters_json={
"model_id": "yolo-configured",
"tile_manifest_path": str(manifest_path),
},
),
(Dataset, reference_dataset_id): reference_dataset,
},
query_rows={Detection: [detection], VectorFeature: [inside_reference, outside_reference]},
)
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_check = next(item for item in db.added if isinstance(item, QualityCheck))
assert result["matches"] == 1
assert result["false_negatives"] == 0
assert result["reference_feature_count_raw"] == 2
assert result["reference_feature_count"] == 1
assert result["coverage"]["applied"] is True
assert result["coverage"]["reference_excluded_outside_count"] == 1
assert quality_check.parameters_json["coverage_policy"] == "persisted_tile_manifest_union"
assert quality_check.findings_json["coverage"] == result["coverage"]
def test_detection_qa_reports_box_to_footprint_diagnostic_without_changing_strict_metrics(tmp_path) -> None:
project_id = uuid4()
dataset_id = uuid4()
reference_dataset_id = uuid4()
analysis_run_id = uuid4()
manifest_path = _coverage_manifest(tmp_path, dataset_id)
detection = _detection(project_id, dataset_id, analysis_run_id, geom=box(0.0, 0.0, 2.0, 2.0))
l_shaped_footprint = Polygon(
[(0.0, 0.0), (2.0, 0.0), (2.0, 0.4), (0.4, 0.4), (0.4, 2.0), (0.0, 2.0), (0.0, 0.0)]
)
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(l_shaped_footprint, 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",
model_name="yolo-configured",
parameters_json={
"model_id": "yolo-configured",
"tile_manifest_path": str(manifest_path),
},
),
(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,
)
diagnostics = result["box_to_footprint_diagnostics"]
quality_check = next(item for item in db.added if isinstance(item, QualityCheck))
assert result["matches"] == 0
assert result["false_positives"] == 1
assert result["false_negatives"] == 1
assert diagnostics["diagnostic_only"] is True
assert diagnostics["envelope_matches"] == 1
assert diagnostics["possible_box_to_footprint_mismatch_count"] == 1
assert quality_check.findings_json["box_to_footprint_diagnostics"] == diagnostics