Add QA evidence map overlay

backend/app/api/routes/quality_checks.py

@@ -7,6 +7,7 @@ from sqlalchemy.orm import Session
 
 from app.db.session import get_db
 from app.schemas.qa import QualityCheckList
+from app.services.quality_evidence_service import QualityEvidenceService
 from app.services.quality_check_service import QualityCheckService
 from app.utils.response import envelope
 
@@ -27,3 +28,12 @@ def list_quality_checks(
         offset=offset,
     )
     return envelope(QualityCheckList(items=items, total=total, limit=limit, offset=offset).model_dump())
+
+
+@router.get("/quality-checks/{quality_check_id}/evidence/geojson", response_model=dict)
+def get_quality_check_evidence_geojson(
+    project_id: UUID,
+    quality_check_id: UUID,
+    db: Session = Depends(get_db),
+) -> dict:
+    return envelope(QualityEvidenceService.evidence_geojson(db, project_id=project_id, quality_check_id=quality_check_id))

backend/app/services/quality_evidence_service.py

@@ -0,0 +1,203 @@
+from __future__ import annotations
+
+from typing import Any
+from uuid import UUID
+
+from geoalchemy2.shape import to_shape
+from shapely.geometry import mapping
+from sqlalchemy.orm import Session
+
+from app.core.errors import AppError
+from app.models import Detection, QualityCheck, Segmentation, VectorFeature
+
+
+class QualityEvidenceService:
+    @staticmethod
+    def evidence_geojson(db: Session, *, project_id: UUID, quality_check_id: UUID) -> dict[str, Any]:
+        quality_check = db.get(QualityCheck, quality_check_id)
+        if not quality_check or quality_check.project_id != project_id:
+            raise AppError(code="QUALITY_CHECK_NOT_FOUND", message="Quality check not found", status_code=404)
+
+        findings = quality_check.findings_json or {}
+        features: list[dict[str, Any]] = []
+        warnings: list[str] = []
+        candidate_index = QualityEvidenceService._candidate_feature_index(db, quality_check)
+        reference_index = QualityEvidenceService._reference_feature_index(db, quality_check)
+
+        for evidence in QualityEvidenceService._evidence_items(findings.get("match_evidence")):
+            candidate_id = QualityEvidenceService._string_value(evidence.get("candidate_feature_id"))
+            reference_id = QualityEvidenceService._string_value(evidence.get("reference_feature_id"))
+            iou = evidence.get("iou")
+            if candidate_id:
+                row = candidate_index.get(candidate_id)
+                if row is not None:
+                    features.append(
+                        QualityEvidenceService._row_to_feature(
+                            row,
+                            role="match_candidate",
+                            quality_check=quality_check,
+                            evidence=evidence,
+                        )
+                    )
+                else:
+                    warnings.append(f"Candidate evidence feature not found: {candidate_id}")
+            if reference_id:
+                row = reference_index.get(reference_id)
+                if row is not None:
+                    features.append(
+                        QualityEvidenceService._row_to_feature(
+                            row,
+                            role="match_reference",
+                            quality_check=quality_check,
+                            evidence={"candidate_feature_id": candidate_id, "reference_feature_id": reference_id, "iou": iou},
+                        )
+                    )
+                else:
+                    warnings.append(f"Reference evidence feature not found: {reference_id}")
+
+        for evidence in QualityEvidenceService._evidence_items(findings.get("false_positive_evidence")):
+            candidate_id = QualityEvidenceService._string_value(evidence.get("candidate_feature_id"))
+            if not candidate_id:
+                continue
+            row = candidate_index.get(candidate_id)
+            if row is not None:
+                features.append(
+                    QualityEvidenceService._row_to_feature(
+                        row,
+                        role="false_positive",
+                        quality_check=quality_check,
+                        evidence=evidence,
+                    )
+                )
+            else:
+                warnings.append(f"False-positive evidence feature not found: {candidate_id}")
+
+        for evidence in QualityEvidenceService._evidence_items(findings.get("false_negative_evidence")):
+            reference_id = QualityEvidenceService._string_value(evidence.get("reference_feature_id"))
+            if not reference_id:
+                continue
+            row = reference_index.get(reference_id)
+            if row is not None:
+                features.append(
+                    QualityEvidenceService._row_to_feature(
+                        row,
+                        role="false_negative",
+                        quality_check=quality_check,
+                        evidence=evidence,
+                    )
+                )
+            else:
+                warnings.append(f"False-negative evidence feature not found: {reference_id}")
+
+        return {
+            "quality_check_id": str(quality_check.id),
+            "project_id": str(quality_check.project_id),
+            "candidate_dataset_id": str(quality_check.candidate_dataset_id) if quality_check.candidate_dataset_id else None,
+            "reference_dataset_id": str(quality_check.reference_dataset_id),
+            "analysis_run_id": str(quality_check.analysis_run_id) if quality_check.analysis_run_id else None,
+            "feature_count": len(features),
+            "warnings": warnings,
+            "geojson": {
+                "type": "FeatureCollection",
+                "features": features,
+            },
+        }
+
+    @staticmethod
+    def _evidence_items(value: Any) -> list[dict[str, Any]]:
+        if not isinstance(value, list):
+            return []
+        return [item for item in value if isinstance(item, dict)]
+
+    @staticmethod
+    def _string_value(value: Any) -> str | None:
+        if value is None:
+            return None
+        text = str(value).strip()
+        return text or None
+
+    @staticmethod
+    def _candidate_feature_index(db: Session, quality_check: QualityCheck) -> dict[str, Any]:
+        index: dict[str, Any] = {}
+        if quality_check.candidate_dataset_id:
+            for row in db.query(VectorFeature).filter(VectorFeature.dataset_id == quality_check.candidate_dataset_id).all():
+                QualityEvidenceService._add_index_keys(index, row)
+            for row in db.query(Detection).filter(Detection.dataset_id == quality_check.candidate_dataset_id).all():
+                QualityEvidenceService._add_index_keys(index, row)
+            for row in db.query(Segmentation).filter(Segmentation.dataset_id == quality_check.candidate_dataset_id).all():
+                QualityEvidenceService._add_index_keys(index, row)
+            if quality_check.analysis_run_id:
+                for row in db.query(Detection).filter(Detection.analysis_run_id == quality_check.analysis_run_id).all():
+                    QualityEvidenceService._add_index_keys(index, row)
+                for row in db.query(Segmentation).filter(Segmentation.analysis_run_id == quality_check.analysis_run_id).all():
+                    QualityEvidenceService._add_index_keys(index, row)
+        elif quality_check.analysis_run_id:
+            for row in db.query(Detection).filter(Detection.analysis_run_id == quality_check.analysis_run_id).all():
+                QualityEvidenceService._add_index_keys(index, row)
+            for row in db.query(Segmentation).filter(Segmentation.analysis_run_id == quality_check.analysis_run_id).all():
+                QualityEvidenceService._add_index_keys(index, row)
+        return index
+
+    @staticmethod
+    def _reference_feature_index(db: Session, quality_check: QualityCheck) -> dict[str, Any]:
+        index: dict[str, Any] = {}
+        for row in db.query(VectorFeature).filter(VectorFeature.dataset_id == quality_check.reference_dataset_id).all():
+            QualityEvidenceService._add_index_keys(index, row)
+        return index
+
+    @staticmethod
+    def _add_index_keys(index: dict[str, Any], row: Any) -> None:
+        for key in QualityEvidenceService._row_identifiers(row):
+            index.setdefault(key, row)
+
+    @staticmethod
+    def _row_identifiers(row: Any) -> set[str]:
+        identifiers = {str(row.id)}
+        source_feature_id = getattr(row, "source_feature_id", None)
+        if source_feature_id:
+            identifiers.add(str(source_feature_id))
+        properties = getattr(row, "properties_json", None) or {}
+        if isinstance(properties, dict):
+            for property_key in ("vector_feature_id", "source_feature_id", "detection_id", "segmentation_id", "id", "name"):
+                value = properties.get(property_key)
+                if value is not None:
+                    identifiers.add(str(value))
+        return identifiers
+
+    @staticmethod
+    def _row_to_feature(row: Any, *, role: str, quality_check: QualityCheck, evidence: dict[str, Any]) -> dict[str, Any]:
+        try:
+            geometry = to_shape(row.geometry)
+        except Exception as exc:
+            raise AppError(
+                code="INVALID_QA_EVIDENCE_GEOMETRY",
+                message="Persisted QA evidence geometry could not be converted to GeoJSON",
+                details={"feature_id": str(getattr(row, "id", ""))},
+                status_code=500,
+            ) from exc
+
+        properties = dict(getattr(row, "properties_json", None) or {})
+        properties.update(
+            {
+                "qa_evidence_role": role,
+                "quality_check_id": str(quality_check.id),
+                "project_id": str(quality_check.project_id),
+                "candidate_dataset_id": str(quality_check.candidate_dataset_id) if quality_check.candidate_dataset_id else None,
+                "reference_dataset_id": str(quality_check.reference_dataset_id),
+                "analysis_run_id": str(quality_check.analysis_run_id) if quality_check.analysis_run_id else None,
+                "feature_id": str(row.id),
+                "dataset_id": str(getattr(row, "dataset_id", "")) if getattr(row, "dataset_id", None) else None,
+                "source_feature_id": getattr(row, "source_feature_id", None),
+                "feature_class": getattr(row, "feature_class", None) or getattr(row, "class_name", None),
+                "candidate_feature_id": QualityEvidenceService._string_value(evidence.get("candidate_feature_id")),
+                "reference_feature_id": QualityEvidenceService._string_value(evidence.get("reference_feature_id")),
+                "iou": evidence.get("iou"),
+            }
+        )
+
+        return {
+            "type": "Feature",
+            "id": f"{role}:{row.id}",
+            "geometry": mapping(geometry),
+            "properties": properties,
+        }

backend/tests/test_sprint112_qa_evidence_overlay.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+import pytest
+from fastapi.testclient import TestClient
+from geoalchemy2.shape import from_shape
+from shapely.geometry import box
+
+from app.core.errors import AppError
+from app.db.session import get_db
+from app.main import app
+from app.models import QualityCheck, VectorFeature
+from app.services.quality_evidence_service import QualityEvidenceService
+
+
+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 and operator.__name__ == "eq":
+                self.rows = [row for row in self.rows if getattr(row, name) == value]
+        return self
+
+    def all(self):
+        return list(self.rows)
+
+
+class FakeSession:
+    def __init__(self, objects=None, query_rows=None) -> None:
+        self.objects = objects or {}
+        self.query_rows = query_rows or {}
+
+    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 _vector_feature(dataset_id, *, feature_id=None, source_feature_id: str, geom=None) -> VectorFeature:
+    return VectorFeature(
+        id=feature_id or uuid4(),
+        dataset_id=dataset_id,
+        source_feature_id=source_feature_id,
+        feature_class="building",
+        properties_json={"name": source_feature_id},
+        geometry=from_shape(geom or box(4.0, 51.0, 4.1, 51.1), srid=4326),
+    )
+
+
+def test_quality_check_evidence_geojson_resolves_persisted_vector_features() -> None:
+    project_id = uuid4()
+    quality_check_id = uuid4()
+    candidate_dataset_id = uuid4()
+    reference_dataset_id = uuid4()
+    candidate_match = _vector_feature(candidate_dataset_id, source_feature_id="candidate-match")
+    candidate_extra = _vector_feature(candidate_dataset_id, source_feature_id="candidate-extra", geom=box(4.4, 51.4, 4.5, 51.5))
+    reference_match = _vector_feature(reference_dataset_id, source_feature_id="reference-match")
+    reference_missing = _vector_feature(reference_dataset_id, source_feature_id="reference-missing", geom=box(4.7, 51.7, 4.8, 51.8))
+    quality_check = QualityCheck(
+        id=quality_check_id,
+        project_id=project_id,
+        candidate_dataset_id=candidate_dataset_id,
+        reference_dataset_id=reference_dataset_id,
+        check_type="candidate_vs_reference",
+        status="ok",
+        findings_json={
+            "match_evidence": [
+                {
+                    "candidate_feature_id": "candidate-match",
+                    "reference_feature_id": "reference-match",
+                    "iou": 1.0,
+                }
+            ],
+            "false_positive_evidence": [{"candidate_feature_id": "candidate-extra"}],
+            "false_negative_evidence": [{"reference_feature_id": "reference-missing"}],
+        },
+    )
+    db = FakeSession(
+        objects={(QualityCheck, quality_check_id): quality_check},
+        query_rows={VectorFeature: [candidate_match, candidate_extra, reference_match, reference_missing]},
+    )
+
+    result = QualityEvidenceService.evidence_geojson(db, project_id=project_id, quality_check_id=quality_check_id)
+
+    assert result["quality_check_id"] == str(quality_check_id)
+    assert result["feature_count"] == 4
+    assert result["geojson"]["type"] == "FeatureCollection"
+    roles = [feature["properties"]["qa_evidence_role"] for feature in result["geojson"]["features"]]
+    assert roles == ["match_candidate", "match_reference", "false_positive", "false_negative"]
+    match_candidate = result["geojson"]["features"][0]
+    assert match_candidate["properties"]["quality_check_id"] == str(quality_check_id)
+    assert match_candidate["properties"]["candidate_feature_id"] == "candidate-match"
+    assert match_candidate["properties"]["reference_feature_id"] == "reference-match"
+    assert match_candidate["properties"]["iou"] == 1.0
+    assert match_candidate["properties"]["source_feature_id"] == "candidate-match"
+
+
+def test_quality_check_evidence_geojson_rejects_cross_project_access() -> None:
+    quality_check_id = uuid4()
+    quality_check = QualityCheck(
+        id=quality_check_id,
+        project_id=uuid4(),
+        reference_dataset_id=uuid4(),
+        check_type="candidate_vs_reference",
+        status="ok",
+        findings_json={},
+    )
+    db = FakeSession(objects={(QualityCheck, quality_check_id): quality_check})
+
+    with pytest.raises(AppError) as exc:
+        QualityEvidenceService.evidence_geojson(db, project_id=uuid4(), quality_check_id=quality_check_id)
+
+    assert exc.value.code == "QUALITY_CHECK_NOT_FOUND"
+
+
+def test_quality_check_evidence_geojson_api_uses_canonical_envelope(monkeypatch) -> None:
+    project_id = uuid4()
+    quality_check_id = uuid4()
+    payload = {
+        "quality_check_id": str(quality_check_id),
+        "project_id": str(project_id),
+        "feature_count": 0,
+        "geojson": {"type": "FeatureCollection", "features": []},
+    }
+
+    monkeypatch.setattr(
+        "app.api.routes.quality_checks.QualityEvidenceService.evidence_geojson",
+        lambda *_args, **_kwargs: payload,
+    )
+    app.dependency_overrides[get_db] = lambda: FakeSession()
+    try:
+        response = TestClient(app).get(f"/api/v1/projects/{project_id}/quality-checks/{quality_check_id}/evidence/geojson")
+    finally:
+        app.dependency_overrides.pop(get_db, None)
+
+    assert response.status_code == 200
+    assert response.json() == {"data": payload}
+
+
+def test_frontend_quality_evidence_overlay_contract_is_wired() -> None:
+    from pathlib import Path
+
+    root = Path(__file__).resolve().parents[2]
+    geo_map = (root / "frontend" / "src" / "components" / "GeoMap.tsx").read_text(encoding="utf-8")
+    map_workspace = (root / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8")
+    qa_api = (root / "frontend" / "src" / "services" / "api" / "qa.ts").read_text(encoding="utf-8")
+
+    assert "qaEvidenceData" in geo_map
+    assert "qa-evidence-fill" in geo_map
+    assert "qa_evidence_role" in geo_map
+    assert "qualityEvidenceGeoJson" in map_workspace
+    assert "getQualityEvidenceGeoJson" in qa_api