364 lines
13 KiB
Python
364 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from shapely.geometry import Point, box, mapping
|
|
from shapely.ops import transform as transform_geometry
|
|
|
|
from app.models import Dataset
|
|
from app.schemas.operations import VectorSelectionSummary
|
|
from app.services.vector_feature_service import VectorFeatureService
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
BBOX = {"min_x": 5.0, "min_y": 51.1, "max_x": 5.3, "max_y": 51.4, "crs": "EPSG:4326"}
|
|
|
|
|
|
def load_operator():
|
|
script_path = ROOT / "scripts" / "provision_buildings_addresses_register.py"
|
|
spec = importlib.util.spec_from_file_location("buildings_addresses_register_operator", script_path)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def building_feature(object_id: str, geometry, status: str = "Gerealiseerd") -> dict: # noqa: ANN001
|
|
return {
|
|
"type": "Feature",
|
|
"id": f"Gebouw.{object_id}",
|
|
"geometry": mapping(geometry),
|
|
"properties": {
|
|
"ObjectId": int(object_id),
|
|
"VersieId": "2026-07-15T08:00:00+02:00",
|
|
"GeometrieMethode": "IngemetenGRB",
|
|
"GebouwStatus": status,
|
|
},
|
|
}
|
|
|
|
|
|
def unit_feature(object_id: str, building_id: str, point: Point) -> dict:
|
|
return {
|
|
"type": "Feature",
|
|
"id": f"Gebouweenheid.{object_id}",
|
|
"geometry": mapping(point),
|
|
"properties": {
|
|
"ObjectId": int(object_id),
|
|
"GebouwObjectId": int(building_id),
|
|
"GebouweenheidStatus": "Gerealiseerd",
|
|
"Functie": "NietGekend",
|
|
},
|
|
}
|
|
|
|
|
|
def address_feature(object_id: str, point: Point) -> dict:
|
|
return {
|
|
"type": "Feature",
|
|
"id": f"Adres.{object_id}",
|
|
"geometry": mapping(point),
|
|
"properties": {
|
|
"ObjectId": int(object_id),
|
|
"AdresStatus": "InGebruik",
|
|
"PositieSpecificatie": "Gebouweenheid",
|
|
"VolledigAdres": "Teststraat 1 bus 2, 2400 Mol",
|
|
"Straatnaam": "Teststraat",
|
|
"Huisnummer": "1",
|
|
"Busnummer": "2",
|
|
},
|
|
}
|
|
|
|
|
|
class ScalarQuery:
|
|
def __init__(self, value: float):
|
|
self.value = value
|
|
|
|
def filter(self, *args): # noqa: ANN002, ARG002
|
|
return self
|
|
|
|
def scalar(self):
|
|
return self.value
|
|
|
|
|
|
class SequenceScalarSession:
|
|
def __init__(self, values: list[float]):
|
|
self.values = iter(values)
|
|
|
|
def query(self, *args): # noqa: ANN002, ARG002
|
|
return ScalarQuery(next(self.values))
|
|
|
|
|
|
class OfficialResponse:
|
|
status_code = 200
|
|
|
|
def __init__(self, payload: dict, url: str):
|
|
self.payload = payload
|
|
self.url = url
|
|
self.content = json.dumps(payload).encode("utf-8")
|
|
|
|
def json(self):
|
|
return self.payload
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
|
|
class TwoPageOfficialSession:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
self.params = []
|
|
|
|
def get(self, url, *, params, timeout): # noqa: ANN001, ARG002
|
|
self.calls += 1
|
|
self.params.append(params)
|
|
if self.calls == 1:
|
|
payload = {
|
|
"type": "FeatureCollection",
|
|
"features": [building_feature("1", box(5.10, 51.20, 5.101, 51.201))],
|
|
"links": [{"rel": "next", "href": f"{url}?startIndex=1"}],
|
|
}
|
|
elif self.calls == 2:
|
|
payload = {
|
|
"type": "FeatureCollection",
|
|
"features": [building_feature("2", box(5.102, 51.20, 5.103, 51.201))],
|
|
"links": [],
|
|
}
|
|
else:
|
|
payload = {"type": "FeatureCollection", "features": [], "links": []}
|
|
return OfficialResponse(payload, f"{url}?page={self.calls}")
|
|
|
|
|
|
def normalized_fixture(module): # noqa: ANN001
|
|
boundary_wgs84 = box(5.09, 51.19, 5.12, 51.22)
|
|
boundary_lambert72 = transform_geometry(module.TO_LAMBERT72.transform, boundary_wgs84)
|
|
polygon = box(5.10, 51.20, 5.105, 51.205)
|
|
buildings, summary = module.normalize_buildings(
|
|
[building_feature("100", polygon)],
|
|
boundary_lambert72,
|
|
)
|
|
return boundary_wgs84, boundary_lambert72, polygon, buildings, summary
|
|
|
|
|
|
def test_official_collection_pagination_retains_checksummed_pages(tmp_path: Path) -> None:
|
|
module = load_operator()
|
|
session = TwoPageOfficialSession()
|
|
raw_dir = tmp_path / "raw"
|
|
|
|
features, summary = module.fetch_collection(
|
|
session,
|
|
url=module.BUILDING_ITEMS_URL,
|
|
name="buildings",
|
|
bbox=(5.0, 51.0, 5.2, 51.2),
|
|
raw_dir=raw_dir,
|
|
page_limit=1,
|
|
max_features=10,
|
|
timeout=30,
|
|
)
|
|
|
|
assert [feature["properties"]["ObjectId"] for feature in features] == [1, 2]
|
|
assert summary["page_count"] == 3
|
|
assert summary["pagination_fallback_count"] == 1
|
|
assert session.params[2]["startIndex"] == "2"
|
|
assert all((tmp_path / page["path"]).is_file() for page in summary["pages"])
|
|
assert all(len(page["sha256"]) == 64 for page in summary["pages"])
|
|
|
|
|
|
def test_buildings_are_clipped_in_lambert72_and_keep_lifecycle_status() -> None:
|
|
module = load_operator()
|
|
boundary = box(5.10, 51.20, 5.11, 51.21)
|
|
boundary_lambert72 = transform_geometry(module.TO_LAMBERT72.transform, boundary)
|
|
source = box(5.095, 51.195, 5.105, 51.205)
|
|
|
|
buildings, summary = module.normalize_buildings(
|
|
[building_feature("100", source, "InAanbouw")],
|
|
boundary_lambert72,
|
|
)
|
|
|
|
assert summary == {"rejected_or_outside_count": 0, "clipped_count": 1}
|
|
record = buildings["100"]
|
|
assert record["status_key"] == "under_construction"
|
|
assert record["was_clipped"] is True
|
|
assert record["geometry_wgs84"].difference(boundary.buffer(1e-7)).area < 1e-12
|
|
assert record["area_ha"] > 0
|
|
|
|
|
|
def test_official_unit_relation_and_exact_address_position_are_aggregated_without_labels() -> None:
|
|
module = load_operator()
|
|
_, boundary_lambert72, polygon, buildings, _ = normalized_fixture(module)
|
|
point = polygon.centroid
|
|
units, unit_summary = module.normalize_units(
|
|
[unit_feature("200", "100", point)],
|
|
boundary_lambert72,
|
|
buildings,
|
|
)
|
|
address_counts, address_summary = module.link_addresses(
|
|
[address_feature("300", point)],
|
|
boundary_lambert72,
|
|
buildings,
|
|
units,
|
|
)
|
|
module.reconcile_with_grb(
|
|
buildings,
|
|
[{"source_feature_id": "GRB.1", "geometry_wgs84": polygon}],
|
|
)
|
|
output, totals = module.build_output_features(
|
|
buildings,
|
|
units,
|
|
address_counts,
|
|
observed_date=module.date(2026, 7, 15),
|
|
area_name="Gemeente Mol - officiële grens",
|
|
)
|
|
|
|
assert unit_summary["orphan_building_count"] == 0
|
|
assert address_summary["match_method_counts"] == {"unit_position_exact": 1}
|
|
assert totals["linked_unit_count"] == 1
|
|
assert totals["linked_address_count"] == 1
|
|
properties = output[0]["properties"]
|
|
assert properties["unit_count"] == 1
|
|
assert properties["active_address_count"] == 1
|
|
assert properties["grb_match_status"] == "matched"
|
|
for prohibited in ("VolledigAdres", "Straatnaam", "Huisnummer", "Busnummer", "HuisnummerLabel"):
|
|
assert prohibited not in properties
|
|
|
|
|
|
def test_ambiguous_unit_position_is_reported_and_never_forced() -> None:
|
|
module = load_operator()
|
|
boundary = box(5.09, 51.19, 5.12, 51.22)
|
|
boundary_lambert72 = transform_geometry(module.TO_LAMBERT72.transform, boundary)
|
|
point = Point(5.105, 51.205)
|
|
buildings, _ = module.normalize_buildings(
|
|
[
|
|
building_feature("100", box(5.10, 51.20, 5.106, 51.21)),
|
|
building_feature("101", box(5.104, 51.20, 5.11, 51.21)),
|
|
],
|
|
boundary_lambert72,
|
|
)
|
|
units, _ = module.normalize_units(
|
|
[unit_feature("200", "100", point), unit_feature("201", "101", point)],
|
|
boundary_lambert72,
|
|
buildings,
|
|
)
|
|
|
|
counts, summary = module.link_addresses(
|
|
[address_feature("300", point)],
|
|
boundary_lambert72,
|
|
buildings,
|
|
units,
|
|
)
|
|
|
|
assert summary["ambiguous_address_count"] == 1
|
|
assert summary["matched_address_count"] == 0
|
|
assert not counts
|
|
|
|
|
|
def test_grb_reconciliation_distinguishes_exact_and_unmatched_geometry() -> None:
|
|
module = load_operator()
|
|
_, _, polygon, buildings, _ = normalized_fixture(module)
|
|
buildings["101"] = {
|
|
**buildings["100"],
|
|
"object_id": "101",
|
|
"geometry_wgs84": box(5.11, 51.21, 5.115, 51.215),
|
|
"geometry_lambert72": transform_geometry(
|
|
module.TO_LAMBERT72.transform,
|
|
box(5.11, 51.21, 5.115, 51.215),
|
|
),
|
|
}
|
|
|
|
summary = module.reconcile_with_grb(
|
|
buildings,
|
|
[{"source_feature_id": "GRB.1", "geometry_wgs84": polygon}],
|
|
)
|
|
|
|
assert buildings["100"]["grb_match_method"] == "exact_geometry"
|
|
assert buildings["100"]["grb_match_confidence"] == 1.0
|
|
assert buildings["101"]["grb_match_status"] == "unmatched"
|
|
assert summary["match_status_counts"] == {"matched": 1, "unmatched": 1}
|
|
assert summary["match_rate"] == 0.5
|
|
|
|
|
|
def test_status_and_relation_metrics_use_filtered_server_owned_aggregations() -> None:
|
|
module = load_operator()
|
|
metrics = module.selection_metrics()
|
|
assert {item["metric_key"] for item in metrics} >= {
|
|
"registered_building_count",
|
|
"realized_building_count",
|
|
"building_unit_count",
|
|
"linked_address_count",
|
|
"active_address_count",
|
|
"grb_matched_building_count",
|
|
}
|
|
status_metrics = [item for item in metrics if item["metric_key"].endswith("building_count")]
|
|
assert any(item.get("filter_property") == "building_status_key" for item in status_metrics)
|
|
assert "huishoudens" in next(item for item in metrics if item["metric_key"] == "linked_address_count")["warning"]
|
|
|
|
|
|
def test_filtered_feature_count_and_numeric_relations_validate_as_selection_summary() -> None:
|
|
module = load_operator()
|
|
dataset = Dataset(
|
|
id=uuid4(),
|
|
project_id=uuid4(),
|
|
name="buildings_addresses_register.geojson",
|
|
dataset_type="vector",
|
|
dataset_role="reference",
|
|
source_name=module.SOURCE_NAME,
|
|
reference_layer_name="building_registry",
|
|
source_metadata={
|
|
"theme": "buildings",
|
|
"semantic_metrics": False,
|
|
"selection_aggregation": {
|
|
"metric_key": "building_footprint_area",
|
|
"method": "intersection_area",
|
|
"label": "Gebouwgrondoppervlakte",
|
|
"unit": "ha",
|
|
"geometry_dimension": 2,
|
|
},
|
|
"selection_metrics": module.selection_metrics(),
|
|
},
|
|
)
|
|
session = SequenceScalarSession([100_000, 2, 0, 1, 0, 4, 3, 4, 3, 2])
|
|
|
|
result = VectorFeatureService.summarize_features_by_bbox(
|
|
session,
|
|
dataset=dataset,
|
|
bbox=BBOX,
|
|
total_feature_count=3,
|
|
full_dataset_area=True,
|
|
)
|
|
|
|
metrics = {item["metric_key"]: item for item in result["metrics"]}
|
|
assert result["metric_value"] == 10.0
|
|
assert metrics["registered_building_count"]["metric_value"] == 3
|
|
assert metrics["realized_building_count"]["metric_value"] == 2
|
|
assert metrics["building_unit_count"]["metric_value"] == 4
|
|
assert metrics["active_address_count"]["metric_value"] == 3
|
|
assert metrics["grb_matched_building_count"]["metric_value"] == 2
|
|
VectorSelectionSummary(**result)
|
|
|
|
|
|
def test_operator_is_canonical_packaged_and_mol_scoped_in_explorer() -> None:
|
|
operator = (ROOT / "scripts/provision_buildings_addresses_register.py").read_text(encoding="utf-8")
|
|
service = (ROOT / "backend/app/services/vector_feature_service.py").read_text(encoding="utf-8")
|
|
dockerfile = (ROOT / "deploy/unraid/Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
readiness = (ROOT / "scripts/run_readiness_check.sh").read_text(encoding="utf-8")
|
|
workspace = (ROOT / "frontend/src/components/map/MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
catalog = (ROOT / "frontend/src/components/datasets/SourceCatalogPanel.tsx").read_text(encoding="utf-8")
|
|
display = (ROOT / "frontend/src/lib/datasetDisplay.ts").read_text(encoding="utf-8")
|
|
|
|
assert "/datasets/upload" in operator
|
|
assert "VectorFeature" not in operator
|
|
assert "INSERT INTO vector_features" not in operator
|
|
assert "VolledigAdres" in operator and '"VolledigAdres", "Straatnaam"' in operator
|
|
assert '"provision_buildings_addresses_register.py"' in service
|
|
assert "COPY scripts/provision_buildings_addresses_register.py" in dockerfile
|
|
assert "py_compile scripts/provision_buildings_addresses_register.py" in readiness
|
|
assert "datasetCoversSelectedArea" in workspace
|
|
assert "digitaal_vlaanderen_buildings_addresses_register' ? 1_200_000" in workspace
|
|
assert "Gebouwen- en Adressenregister" in catalog
|
|
assert "building_registry: 'Gebouwenregister'" in display
|
|
assert "digitaal_vlaanderen_buildings_addresses_register: 'Digitaal Vlaanderen'" in display
|