206 lines
7.1 KiB
Python
206 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from app.models import Dataset, SourceRegistry, SourceSnapshot
|
|
from app.services.qa_service import QaService
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, datasets=None, areas=None):
|
|
self.datasets = {item.id: item for item in (datasets or [])}
|
|
self.areas = {item.id: item for item in (areas or [])}
|
|
|
|
def get(self, model, item_id):
|
|
if model.__name__ == "Dataset":
|
|
return self.datasets.get(item_id)
|
|
if model.__name__ == "Area":
|
|
return self.areas.get(item_id)
|
|
return None
|
|
|
|
|
|
def _feature(feature_id: str, coordinates: list[list[list[float]]]) -> dict:
|
|
return {
|
|
"type": "Feature",
|
|
"id": feature_id,
|
|
"properties": {"source_feature_id": feature_id},
|
|
"geometry": {
|
|
"type": "Polygon",
|
|
"coordinates": coordinates,
|
|
},
|
|
}
|
|
|
|
|
|
def _write_dataset(path: Path, coordinates: list[list[list[float]]]) -> None:
|
|
payload = {
|
|
"type": "FeatureCollection",
|
|
"features": [_feature("feature-1", coordinates)],
|
|
}
|
|
path.write_text(json.dumps(payload), encoding="utf-8")
|
|
|
|
|
|
def _write_features(path: Path, features: list[dict]) -> None:
|
|
path.write_text(json.dumps({"type": "FeatureCollection", "features": features}), encoding="utf-8")
|
|
|
|
|
|
def _authoritative_reference(dataset: Dataset) -> Dataset:
|
|
"""Give QA reference fixtures the same durable authority proof as GRB."""
|
|
|
|
source_id = uuid4()
|
|
snapshot_id = uuid4()
|
|
checksum = sha256(Path(str(dataset.storage_path)).read_bytes()).hexdigest()
|
|
source = SourceRegistry(
|
|
id=source_id,
|
|
source_key="grb",
|
|
display_name="GRB test reference",
|
|
classification="authoritative",
|
|
authority_name="Digitaal Vlaanderen",
|
|
authority_scope_json={"zone": "Flanders"},
|
|
usage_policy_json={"ground_truth_allowed": True, "validation_authority": {"building_validation": "primary"}},
|
|
)
|
|
snapshot = SourceSnapshot(
|
|
id=snapshot_id,
|
|
source_registry_id=source_id,
|
|
snapshot_key=f"qa-grb-{dataset.id}",
|
|
checksum_sha256=checksum,
|
|
ingest_status="ingested",
|
|
freshness_status="current",
|
|
)
|
|
dataset.source = "grb"
|
|
dataset.source_name = "grb"
|
|
dataset.dataset_role = "reference"
|
|
dataset.status = "ready"
|
|
dataset.checksum_sha256 = checksum
|
|
dataset.source_registry_id = source_id
|
|
dataset.source_snapshot_id = snapshot_id
|
|
dataset.data_contract_key = "geointel.vector.geojson"
|
|
dataset.data_contract_version = "1.0.0"
|
|
dataset.validation_status = "passed"
|
|
dataset.provenance_status = "complete"
|
|
dataset.lineage_status = "complete"
|
|
dataset.quarantine_status = "not_quarantined"
|
|
dataset.source_registry = source
|
|
dataset.source_snapshot = snapshot
|
|
return dataset
|
|
|
|
|
|
def test_qa_compare_candidate_with_reference_returns_metrics(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
candidate_id = uuid4()
|
|
reference_id = uuid4()
|
|
candidate_path = tmp_path / "candidate.geojson"
|
|
reference_path = tmp_path / "reference.geojson"
|
|
polygon = [[[4.0, 51.0], [4.1, 51.0], [4.1, 51.1], [4.0, 51.1], [4.0, 51.0]]]
|
|
_write_dataset(candidate_path, polygon)
|
|
_write_dataset(reference_path, polygon)
|
|
|
|
candidate = Dataset(
|
|
id=candidate_id,
|
|
project_id=project_id,
|
|
name="candidate.geojson",
|
|
dataset_type="vector",
|
|
source="test",
|
|
storage_path=str(candidate_path),
|
|
crs="EPSG:4326",
|
|
metadata_json={"crs_assumed": False},
|
|
)
|
|
reference = _authoritative_reference(Dataset(
|
|
id=reference_id,
|
|
project_id=project_id,
|
|
name="reference.geojson",
|
|
dataset_type="vector",
|
|
source="test",
|
|
storage_path=str(reference_path),
|
|
crs="EPSG:4326",
|
|
metadata_json={"crs_assumed": False},
|
|
))
|
|
|
|
result = QaService.compare_candidate_with_reference(
|
|
db=FakeSession([candidate, reference]),
|
|
project_id=project_id,
|
|
candidate_dataset_id=candidate_id,
|
|
reference_dataset_id=reference_id,
|
|
iou_threshold=0.5,
|
|
)
|
|
|
|
assert result.status == "ok"
|
|
assert result.matches == 1
|
|
assert result.false_positives == 0
|
|
assert result.false_negatives == 0
|
|
assert result.precision == 1.0
|
|
assert result.recall == 1.0
|
|
assert result.f1_score == 1.0
|
|
|
|
|
|
def test_qa_compare_candidate_with_reference_returns_feature_level_evidence(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
candidate_id = uuid4()
|
|
reference_id = uuid4()
|
|
candidate_path = tmp_path / "candidate.geojson"
|
|
reference_path = tmp_path / "reference.geojson"
|
|
matched_candidate = [[[4.0, 51.0], [4.1, 51.0], [4.1, 51.1], [4.0, 51.1], [4.0, 51.0]]]
|
|
matched_reference = [[[4.0, 51.0], [4.1, 51.0], [4.1, 51.1], [4.0, 51.1], [4.0, 51.0]]]
|
|
false_positive = [[[4.5, 51.5], [4.6, 51.5], [4.6, 51.6], [4.5, 51.6], [4.5, 51.5]]]
|
|
false_negative = [[[4.8, 51.8], [4.9, 51.8], [4.9, 51.9], [4.8, 51.9], [4.8, 51.8]]]
|
|
_write_features(candidate_path, [_feature("candidate-match", matched_candidate), _feature("candidate-extra", false_positive)])
|
|
_write_features(reference_path, [_feature("reference-match", matched_reference), _feature("reference-missing", false_negative)])
|
|
|
|
candidate = Dataset(
|
|
id=candidate_id,
|
|
project_id=project_id,
|
|
name="candidate.geojson",
|
|
dataset_type="vector",
|
|
source="test",
|
|
storage_path=str(candidate_path),
|
|
crs="EPSG:4326",
|
|
metadata_json={"crs_assumed": False},
|
|
)
|
|
reference = _authoritative_reference(Dataset(
|
|
id=reference_id,
|
|
project_id=project_id,
|
|
name="reference.geojson",
|
|
dataset_type="vector",
|
|
source="test",
|
|
storage_path=str(reference_path),
|
|
crs="EPSG:4326",
|
|
metadata_json={"crs_assumed": False},
|
|
))
|
|
|
|
result = QaService.compare_candidate_with_reference(
|
|
db=FakeSession([candidate, reference]),
|
|
project_id=project_id,
|
|
candidate_dataset_id=candidate_id,
|
|
reference_dataset_id=reference_id,
|
|
iou_threshold=0.5,
|
|
)
|
|
|
|
assert result.matches == 1
|
|
assert result.false_positives == 1
|
|
assert result.false_negatives == 1
|
|
assert result.match_evidence == [
|
|
{
|
|
"candidate_feature_id": "candidate-match",
|
|
"reference_feature_id": "reference-match",
|
|
"iou": 1.0,
|
|
}
|
|
]
|
|
assert result.false_positive_evidence == [{"candidate_feature_id": "candidate-extra"}]
|
|
assert result.false_negative_evidence == [{"reference_feature_id": "reference-missing"}]
|
|
|
|
|
|
def test_dataset_reference_metadata_migration_declares_required_columns() -> None:
|
|
migration_path = Path(__file__).parents[1] / "alembic" / "versions" / "202606120001_add_dataset_reference_metadata.py"
|
|
migration_text = migration_path.read_text(encoding="utf-8")
|
|
for column_name in (
|
|
"dataset_role",
|
|
"source_name",
|
|
"reference_layer_name",
|
|
"source_metadata",
|
|
"provenance_metadata",
|
|
"imported_at",
|
|
):
|
|
assert column_name in migration_text
|