audit: establish accuracy phase 1 baseline
This commit is contained in:
Jens
@@ -0,0 +1,271 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from uuid import UUID
|
||||
|
||||
from geoalchemy2.shape import to_shape
|
||||
from shapely.geometry import Point, box
|
||||
|
||||
import sys
|
||||
|
||||
|
||||
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
||||
BACKEND_ROOT = REPOSITORY_ROOT / "backend"
|
||||
if str(BACKEND_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(BACKEND_ROOT))
|
||||
|
||||
from app.core.config import Settings # noqa: E402
|
||||
from app.schemas.area import AreaUpdate # noqa: E402
|
||||
from app.services.coverage_registry_service import ( # noqa: E402
|
||||
CoverageRegistryService,
|
||||
SOURCE_DEFINITIONS,
|
||||
)
|
||||
from app.services.detection_service import DetectionService # noqa: E402
|
||||
from app.services.vector_feature_service import VectorFeatureService # noqa: E402
|
||||
|
||||
|
||||
FIXED_UUID = UUID("00000000-0000-4000-8000-000000000001")
|
||||
|
||||
|
||||
def _dataset(*, layer: str, bbox_values: list[float], source_name: str = "spw_picc") -> SimpleNamespace:
|
||||
return SimpleNamespace(
|
||||
id=FIXED_UUID,
|
||||
status="ready",
|
||||
source_name=source_name,
|
||||
reference_layer_name=layer,
|
||||
source_metadata={"coverage_zones": ["wallonia"], "bbox_epsg4326": bbox_values},
|
||||
provenance_metadata={},
|
||||
observed_at=None,
|
||||
source_version="forensic-reproduction",
|
||||
resolution_json=None,
|
||||
checksum_sha256="forensic-only",
|
||||
)
|
||||
|
||||
|
||||
def _coverage_cross_theme_contamination() -> dict:
|
||||
definition = next(
|
||||
item for item in SOURCE_DEFINITIONS if item.contract.source_name == "spw_geoportail"
|
||||
)
|
||||
selection = box(4.50, 50.50, 4.70, 50.60)
|
||||
building = _dataset(layer="buildings", bbox_values=[4.55, 50.52, 4.56, 50.53])
|
||||
road = _dataset(layer="roads", bbox_values=[4.50, 50.50, 4.70, 50.60])
|
||||
matches, fully_covered = CoverageRegistryService._matching_datasets(
|
||||
[building, road],
|
||||
definition,
|
||||
"buildings",
|
||||
"wallonia",
|
||||
selection,
|
||||
)
|
||||
observed_ids = [str(item.id) for item in matches]
|
||||
reproduced = observed_ids == [str(building.id)] and fully_covered is True
|
||||
return {
|
||||
"id": "P1-COV-001",
|
||||
"severity": "critical",
|
||||
"source": "backend/app/services/coverage_registry_service.py:481-505",
|
||||
"expected": "A small buildings partition remains partial; a roads bbox cannot complete buildings coverage.",
|
||||
"observed": {"matched_dataset_ids": observed_ids, "fully_covered": fully_covered},
|
||||
"reproduced": reproduced,
|
||||
}
|
||||
|
||||
|
||||
def _meter_buffer_as_degrees() -> dict:
|
||||
geometry = Point(5.0, 51.0)
|
||||
buffered = geometry.buffer(100.0)
|
||||
bounds = [float(value) for value in buffered.bounds]
|
||||
reproduced = round(bounds[2] - bounds[0], 6) == 200.0
|
||||
return {
|
||||
"id": "P1-CRS-001",
|
||||
"severity": "critical",
|
||||
"source": "backend/app/services/vector_operations_service.py:179-203",
|
||||
"expected": "A 100 metre buffer is projected to a metric CRS and spans roughly hundreds of metres.",
|
||||
"observed": {"bounds_epsg4326": bounds, "longitude_span_degrees": bounds[2] - bounds[0]},
|
||||
"reproduced": reproduced,
|
||||
}
|
||||
|
||||
|
||||
def _lambert_feature_mislabeled() -> dict:
|
||||
row = VectorFeatureService._feature_row(
|
||||
FIXED_UUID,
|
||||
{
|
||||
"type": "Feature",
|
||||
"geometry": {"type": "Point", "coordinates": [150000.0, 210000.0]},
|
||||
"properties": {},
|
||||
},
|
||||
0,
|
||||
None,
|
||||
)
|
||||
assert row is not None
|
||||
geometry = to_shape(row.geometry)
|
||||
reproduced = (
|
||||
int(row.geometry.srid) == 4326
|
||||
and float(geometry.x) == 150000.0
|
||||
and float(geometry.y) == 210000.0
|
||||
)
|
||||
return {
|
||||
"id": "P1-CRS-002",
|
||||
"severity": "critical",
|
||||
"source": "backend/app/services/vector_feature_service.py:271-299",
|
||||
"expected": "Non-WGS84 input is transformed to EPSG:4326 or rejected before persistence.",
|
||||
"observed": {
|
||||
"stored_srid": int(row.geometry.srid),
|
||||
"stored_coordinates": [float(geometry.x), float(geometry.y)],
|
||||
},
|
||||
"reproduced": reproduced,
|
||||
}
|
||||
|
||||
|
||||
def _authority_spoof() -> dict:
|
||||
selection = box(4.9, 50.9, 5.0, 51.0)
|
||||
dataset = SimpleNamespace(
|
||||
id=FIXED_UUID,
|
||||
status="ready",
|
||||
source_name="grb",
|
||||
reference_layer_name="buildings",
|
||||
source_metadata={
|
||||
"coverage_zones": ["flanders"],
|
||||
"bbox_epsg4326": [4.8, 50.8, 5.1, 51.1],
|
||||
},
|
||||
provenance_metadata={"provided_by": "caller"},
|
||||
observed_at=None,
|
||||
source_version="caller-provided",
|
||||
resolution_json=None,
|
||||
checksum_sha256="caller-provided",
|
||||
crs="EPSG:4326",
|
||||
)
|
||||
item = CoverageRegistryService._resolve_item(
|
||||
zone="flanders",
|
||||
theme="buildings",
|
||||
datasets=[dataset],
|
||||
selection=selection,
|
||||
)
|
||||
authority = item.evidence[0].authority_level if item.evidence else None
|
||||
reproduced = item.status == "operational" and authority == "authoritative"
|
||||
return {
|
||||
"id": "P1-AUTH-001",
|
||||
"severity": "critical",
|
||||
"source": (
|
||||
"backend/app/api/routes/datasets.py:142-163; "
|
||||
"backend/app/services/coverage_registry_service.py:463-559"
|
||||
),
|
||||
"expected": "Only server-attested source identities can produce authoritative operational coverage.",
|
||||
"observed": {
|
||||
"caller_controlled_source_name": dataset.source_name,
|
||||
"coverage_status": item.status,
|
||||
"reported_authority": authority,
|
||||
},
|
||||
"reproduced": reproduced,
|
||||
}
|
||||
|
||||
|
||||
def _mutable_name_model_scope() -> dict:
|
||||
area = SimpleNamespace(name="Mol validation bypass", geometry=box(-75.0, 35.0, -74.9, 35.1))
|
||||
dataset = SimpleNamespace(id=FIXED_UUID, area_id=FIXED_UUID)
|
||||
|
||||
class FakeSession:
|
||||
@staticmethod
|
||||
def get(_model, _identifier):
|
||||
return area
|
||||
|
||||
settings = Settings().model_copy(
|
||||
update={"yolo_validated_area_names": "Mol,Kempen"}
|
||||
)
|
||||
accepted = True
|
||||
try:
|
||||
DetectionService._validate_model_area_scope(FakeSession(), dataset, settings)
|
||||
except Exception:
|
||||
accepted = False
|
||||
return {
|
||||
"id": "P1-AI-001",
|
||||
"severity": "critical",
|
||||
"source": "backend/app/services/detection_service.py:223-232",
|
||||
"expected": "Validation scope is bound to immutable geometry/source/checksum evidence.",
|
||||
"observed": {
|
||||
"area_name": area.name,
|
||||
"geometry_bounds": [float(value) for value in area.geometry.bounds],
|
||||
"accepted": accepted,
|
||||
},
|
||||
"reproduced": accepted,
|
||||
}
|
||||
|
||||
|
||||
def _mutable_name_legal_scope() -> dict:
|
||||
selection = box(4.9, 50.9, 5.0, 51.0)
|
||||
flemish_geometry = box(2.5, 50.7, 5.9, 51.5)
|
||||
canonical = [SimpleNamespace(name="Flanders", geometry=flemish_geometry)]
|
||||
renamed = [SimpleNamespace(name="Vlaanderen", geometry=flemish_geometry)]
|
||||
canonical_zones, canonical_outside = CoverageRegistryService._intersected_zones(
|
||||
canonical, selection
|
||||
)
|
||||
renamed_zones, renamed_outside = CoverageRegistryService._intersected_zones(
|
||||
renamed, selection
|
||||
)
|
||||
reproduced = (
|
||||
canonical_zones == ["flanders"]
|
||||
and canonical_outside is False
|
||||
and renamed_zones == []
|
||||
and renamed_outside is True
|
||||
)
|
||||
return {
|
||||
"id": "P1-COV-002",
|
||||
"severity": "high",
|
||||
"source": "backend/app/services/coverage_registry_service.py:56-65,425-447",
|
||||
"expected": "Renaming an Area cannot change its legal coverage-zone identity.",
|
||||
"observed": {
|
||||
"canonical": {"zones": canonical_zones, "outside": canonical_outside},
|
||||
"renamed_same_geometry": {"zones": renamed_zones, "outside": renamed_outside},
|
||||
},
|
||||
"reproduced": reproduced,
|
||||
}
|
||||
|
||||
|
||||
def _area_patch_ignores_geometry() -> dict:
|
||||
payload = AreaUpdate.model_validate(
|
||||
{
|
||||
"name": "Renamed",
|
||||
"geometry": {
|
||||
"type": "Polygon",
|
||||
"coordinates": [[[4.0, 50.0], [5.0, 50.0], [5.0, 51.0], [4.0, 50.0]]],
|
||||
},
|
||||
}
|
||||
)
|
||||
parsed = payload.model_dump()
|
||||
reproduced = "geometry" not in parsed
|
||||
return {
|
||||
"id": "P1-API-001",
|
||||
"severity": "high",
|
||||
"source": "backend/app/schemas/area.py:15-17; backend/app/services/area_service.py:154-171",
|
||||
"expected": "PATCH /areas/{area_id} either validates and applies geometry or rejects the field.",
|
||||
"observed": {"parsed_payload": parsed, "geometry_silently_ignored": reproduced},
|
||||
"reproduced": reproduced,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
findings = [
|
||||
_coverage_cross_theme_contamination(),
|
||||
_meter_buffer_as_degrees(),
|
||||
_lambert_feature_mislabeled(),
|
||||
_authority_spoof(),
|
||||
_mutable_name_model_scope(),
|
||||
_mutable_name_legal_scope(),
|
||||
_area_patch_ignores_geometry(),
|
||||
]
|
||||
reproduced_count = sum(bool(item["reproduced"]) for item in findings)
|
||||
payload = {
|
||||
"schema_version": 1,
|
||||
"purpose": "Read-only deterministic reproductions of Phase-1 contract violations.",
|
||||
"findings": findings,
|
||||
"summary": {
|
||||
"total": len(findings),
|
||||
"reproduced": reproduced_count,
|
||||
"all_reproduced": reproduced_count == len(findings),
|
||||
},
|
||||
}
|
||||
print(json.dumps(payload, indent=2, sort_keys=True))
|
||||
return 0 if reproduced_count == len(findings) else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user