Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
198 lines
7.1 KiB
Python
198 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import sys
|
|
|
|
import numpy as np
|
|
from pyproj import Transformer
|
|
import rasterio
|
|
from rasterio.transform import from_origin
|
|
from shapely.geometry import Polygon, shape
|
|
from shapely.ops import transform
|
|
from tests.frontend_contract import read_map_workspace
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def load_provisioner():
|
|
script_path = ROOT / "scripts" / "provision_official_landuse_timeseries.py"
|
|
spec = importlib.util.spec_from_file_location("official_landuse_provisioner", 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 test_official_landuse_wcs_contract_is_categorical_and_deterministic() -> None:
|
|
module = load_provisioner()
|
|
|
|
params = module.build_wcs_params(2025, (196594.1, 205064.2, 210910.7, 223974.9))
|
|
|
|
assert module.SUPPORTED_YEARS == (2013, 2016, 2019, 2022, 2025)
|
|
assert module.LAND_USE_CLASSES[12] == "Bos"
|
|
assert params == {
|
|
"SERVICE": "WCS",
|
|
"VERSION": "1.0.0",
|
|
"REQUEST": "GetCoverage",
|
|
"COVERAGE": "lu:lu_landgebruik_vlaa_2025_v3",
|
|
"CRS": "EPSG:31370",
|
|
"BBOX": "196590.000,205060.000,210920.000,223980.000",
|
|
"RESX": "10",
|
|
"RESY": "10",
|
|
"FORMAT": "image/tiff",
|
|
"RESPONSE_CRS": "EPSG:31370",
|
|
}
|
|
assert module.series_key(module.THEMES[0], "Mol") == "department-omgeving:land-use:forest:mol"
|
|
|
|
|
|
def test_official_landuse_polygonization_clips_and_preserves_provenance(tmp_path: Path) -> None:
|
|
module = load_provisioner()
|
|
raster_path = tmp_path / "landuse.tif"
|
|
values = np.array(
|
|
[
|
|
[1, 1, 1, 1, 1, 1],
|
|
[1, 12, 12, 1, 1, 1],
|
|
[1, 12, 12, 1, 12, 1],
|
|
[1, 1, 1, 1, 12, 1],
|
|
[1, 1, 1, 1, 1, 1],
|
|
[1, 1, 1, 1, 1, 1],
|
|
],
|
|
dtype="int32",
|
|
)
|
|
with rasterio.open(
|
|
raster_path,
|
|
"w",
|
|
driver="GTiff",
|
|
width=6,
|
|
height=6,
|
|
count=1,
|
|
dtype="int32",
|
|
crs="EPSG:31370",
|
|
transform=from_origin(200000, 210000, 10, 10),
|
|
nodata=-9999,
|
|
) as destination:
|
|
destination.write(values, 1)
|
|
|
|
to_wgs84 = Transformer.from_crs(31370, 4326, always_xy=True)
|
|
boundary_metric = Polygon(
|
|
[(200005, 209945), (200055, 209945), (200055, 209995), (200005, 209995), (200005, 209945)]
|
|
)
|
|
boundary = transform(to_wgs84.transform, boundary_metric)
|
|
|
|
payload, stats = module.polygonize_snapshot(
|
|
raster_path=raster_path,
|
|
boundary=boundary,
|
|
year=2025,
|
|
theme=module.THEMES[0],
|
|
municipality_name="Mol",
|
|
nis_code="13025",
|
|
scope_key="mol",
|
|
max_features=100,
|
|
)
|
|
|
|
assert payload["type"] == "FeatureCollection"
|
|
assert payload["crs"]["properties"]["name"] == "EPSG:4326"
|
|
assert payload["source_coverage_id"] == "lu:lu_landgebruik_vlaa_2025_v3"
|
|
assert stats["source_pixel_count"] == 6
|
|
assert stats["source_pixel_area_m2"] == 600
|
|
assert stats["feature_count"] == 2
|
|
assert 0 < stats["polygon_area_m2"] <= 600
|
|
assert stats["class_histogram"] == {"1": 30, "12": 6}
|
|
for feature in payload["features"]:
|
|
geometry = shape(feature["geometry"])
|
|
properties = feature["properties"]
|
|
assert geometry.is_valid
|
|
assert geometry.within(boundary.buffer(1e-9))
|
|
assert properties["source_name"] == "department_omgeving_land_use"
|
|
assert properties["land_use_class_ids"] == [12]
|
|
assert properties["source_resolution_m"] == 10.0
|
|
assert properties["source_raster_sha256"] == stats["raster_sha256"]
|
|
|
|
|
|
def test_official_landuse_metadata_keeps_modern_series_separate(tmp_path: Path) -> None:
|
|
module = load_provisioner()
|
|
theme = module.THEMES[0]
|
|
snapshot = module.PreparedSnapshot(
|
|
year=2022,
|
|
theme=theme,
|
|
raster_path=tmp_path / "source.tif",
|
|
vector_path=tmp_path / "forest.geojson",
|
|
manifest_path=tmp_path / "forest.manifest.json",
|
|
feature_count=42,
|
|
raster_sha256="a" * 64,
|
|
vector_sha256="b" * 64,
|
|
)
|
|
args = SimpleNamespace(scope_key="mol", municipality_name="Mol", nis_code="13025")
|
|
|
|
source = module.build_source_metadata(args, snapshot)
|
|
provenance = module.build_provenance_metadata(args, snapshot)
|
|
|
|
assert source["temporal_series_label"] == "Moderne landgebruikskaart (10 m)"
|
|
assert source["selection_aggregation"]["method"] == "intersection_area"
|
|
assert source["identity_stable"] is False
|
|
assert source["land_use_class_names"] == ["Bos"]
|
|
assert "10 m" in source["selection_aggregation"]["warning"]
|
|
assert provenance["operator_explicit_fetch"] is True
|
|
assert provenance["coverage_id"] == "lu:lu_landgebruik_vlaa_2022_v3"
|
|
assert "historical-landuse" not in module.series_key(theme, "mol")
|
|
|
|
|
|
def test_official_landuse_operator_paginates_within_api_limit() -> None:
|
|
module = load_provisioner()
|
|
|
|
class Response:
|
|
ok = True
|
|
status_code = 200
|
|
text = ""
|
|
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
|
|
def json(self):
|
|
return {"data": self.payload}
|
|
|
|
class Session:
|
|
def __init__(self) -> None:
|
|
self.calls = []
|
|
|
|
def get(self, url, *, params, timeout):
|
|
self.calls.append((url, params, timeout))
|
|
offset = params["offset"]
|
|
page_items = [{"id": index} for index in range(offset, min(offset + 200, 405))]
|
|
return Response({"items": page_items, "total": 405, "limit": 200, "offset": offset})
|
|
|
|
session = Session()
|
|
items = module.list_paginated_items(session, "http://backend/api/v1/projects", timeout=30)
|
|
|
|
assert len(items) == 405
|
|
assert [call[1] for call in session.calls] == [
|
|
{"limit": 200, "offset": 0},
|
|
{"limit": 200, "offset": 200},
|
|
{"limit": 200, "offset": 400},
|
|
]
|
|
|
|
|
|
def test_official_landuse_operator_is_packaged_and_readiness_checked() -> None:
|
|
readiness = (ROOT / "scripts/run_readiness_check.sh").read_text(encoding="utf-8")
|
|
dockerfile = (ROOT / "deploy/unraid/Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
workspace = read_map_workspace()
|
|
geo_map = (ROOT / "frontend/src/components/GeoMap.tsx").read_text(encoding="utf-8")
|
|
premium_css = (ROOT / "frontend/src/styles/premium.css").read_text(encoding="utf-8")
|
|
|
|
assert "py_compile scripts/provision_official_landuse_timeseries.py" in readiness
|
|
assert "COPY scripts/provision_official_landuse_timeseries.py" in dockerfile
|
|
assert "department_omgeving_land_use' ? 90_000" in workspace
|
|
assert "activeTemporalSeriesGroups.length > 1" in workspace
|
|
assert "dataFillColor={activeThemeMapStyle.fill}" in workspace
|
|
assert "forest: { fill: '#347950', line: '#225f3b' }" in workspace
|
|
assert "datasetFillColor(dataFillColor)" in geo_map
|
|
assert ".workbench-main > .geo-explorer" in premium_css
|
|
assert "Moderne landgebruikskaart (10 m)" in (ROOT / "scripts/provision_official_landuse_timeseries.py").read_text(
|
|
encoding="utf-8"
|
|
)
|