371 lines
14 KiB
Python
371 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import rasterio
|
|
from fastapi.testclient import TestClient
|
|
from pyproj import Transformer
|
|
from rasterio.io import MemoryFile
|
|
from rasterio.transform import from_origin
|
|
from shapely.geometry import box
|
|
|
|
from app.core.config import Settings
|
|
from app.core.errors import AppError
|
|
from app.db.session import get_db
|
|
from app.main import app
|
|
from app.models import Dataset, Job, Project
|
|
from app.schemas.flood_hazard import FloodHazardAcquireRequest, FloodHazardSelectionRequest
|
|
from app.schemas.assistant import AssistantQueryRequest
|
|
from app.services.geo_assistant_service import GeoAssistantService
|
|
from app.services.flood_hazard_acquisition_service import FloodHazardAcquisitionService
|
|
from app.services.flood_hazard_analysis_service import FloodHazardAnalysisService
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, result=None):
|
|
self.result = result
|
|
|
|
def filter(self, *_args):
|
|
return self
|
|
|
|
def order_by(self, *_args):
|
|
return self
|
|
|
|
def first(self):
|
|
return self.result
|
|
|
|
def all(self):
|
|
return self.result if isinstance(self.result, list) else []
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, rows=None, query_result=None):
|
|
self.rows = rows or {}
|
|
self.query_result = query_result
|
|
self.added = []
|
|
|
|
def get(self, model, row_id):
|
|
row = self.rows.get((model, row_id))
|
|
if row is not None:
|
|
return row
|
|
return next((item for item in self.added if isinstance(item, model) and item.id == row_id), None)
|
|
|
|
def add(self, row):
|
|
self.added.append(row)
|
|
|
|
def commit(self):
|
|
return None
|
|
|
|
def rollback(self):
|
|
return None
|
|
|
|
def refresh(self, row):
|
|
return row
|
|
|
|
def query(self, _model):
|
|
return FakeQuery(self.query_result)
|
|
|
|
|
|
def flood_payload(*, product_key: str = "pluviaal_current_t100", side_m: float = 100.0) -> FloodHazardAcquireRequest:
|
|
transformer = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True)
|
|
min_x, min_y = transformer.transform(200_000, 210_000)
|
|
max_x, max_y = transformer.transform(200_000 + side_m, 210_000 + side_m)
|
|
return FloodHazardAcquireRequest(
|
|
bbox={"min_x": min_x, "min_y": min_y, "max_x": max_x, "max_y": max_y, "crs": "EPSG:4326"},
|
|
product_key=product_key,
|
|
resolution_m=5.0,
|
|
force_refresh=True,
|
|
)
|
|
|
|
|
|
def depth_tiff(*, normalized_metres: bool = False) -> bytes:
|
|
values = np.zeros((20, 20), dtype="float32")
|
|
values[:, :10] = 1.0 if normalized_metres else 100.0
|
|
with MemoryFile() as memory:
|
|
with memory.open(
|
|
driver="GTiff",
|
|
width=20,
|
|
height=20,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:31370",
|
|
transform=from_origin(200_000, 210_100, 5.0, 5.0),
|
|
nodata=-9999.0 if normalized_metres else 0.0,
|
|
) as output:
|
|
if normalized_metres:
|
|
values[:, 10:] = -9999.0
|
|
output.write(values, 1)
|
|
return memory.read()
|
|
|
|
|
|
def test_flood_hazard_registry_is_complete_and_semantically_honest() -> None:
|
|
products = FloodHazardAcquisitionService.list_products()
|
|
|
|
assert len(products) == 12
|
|
assert {item["mechanism"] for item in products} == {"pluviaal", "fluviaal"}
|
|
assert {item["climate_context"] for item in products} == {"huidig klimaat", "klimaatprojectie 2050"}
|
|
assert {item["return_period_years"] for item in products} == {10, 100, 1000}
|
|
assert all(item["coverage_id"].startswith("Overstromingsgevaarkaarten-") for item in products)
|
|
assert all(item["source_value_unit"] == "cm" and item["normalized_value_unit"] == "m" for item in products)
|
|
assert all("geen bathymetrie" in item["limitation_message"] for item in products)
|
|
|
|
|
|
def test_flood_hazard_request_is_bounded_and_rejects_arbitrary_products() -> None:
|
|
settings = Settings(_env_file=None)
|
|
prepared = FloodHazardAcquisitionService._prepared_request(flood_payload(), settings)
|
|
product = prepared["product"]
|
|
url = FloodHazardAcquisitionService._wcs_request_url(
|
|
settings,
|
|
product,
|
|
tuple(prepared["bbox_epsg31370"]),
|
|
prepared["resolution_m"],
|
|
)
|
|
|
|
assert "VERSION=1.1.0" in url
|
|
assert "IDENTIFIER=Overstromingsgevaarkaarten-PLUVIAAL%3Awaterdiepte_PLU_noCC_T100" in url
|
|
assert "GRIDOFFSETS=5%2C-5" in url
|
|
assert prepared["width"] * prepared["height"] <= settings.flood_hazard_max_pixels
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
FloodHazardAcquisitionService._prepared_request(flood_payload(product_key="custom"), settings)
|
|
assert exc_info.value.code == "FLOOD_HAZARD_PRODUCT_NOT_SUPPORTED"
|
|
|
|
|
|
def test_flood_hazard_tiles_stay_below_the_observed_vmm_coverage_limit() -> None:
|
|
prepared = FloodHazardAcquisitionService._prepared_request(flood_payload(side_m=15_000), Settings(_env_file=None))
|
|
tiles = FloodHazardAcquisitionService._tile_bounds(prepared)
|
|
|
|
assert 9 <= len(tiles) <= 16
|
|
assert all((max_x - min_x) <= 5_000 for min_x, _min_y, max_x, _max_y in tiles)
|
|
assert all((max_y - min_y) <= 5_000 for _min_x, min_y, _max_x, max_y in tiles)
|
|
assert all(
|
|
((max_x - min_x) / prepared["resolution_m"]) * ((max_y - min_y) / prepared["resolution_m"])
|
|
<= 1_000_000
|
|
for min_x, min_y, max_x, max_y in tiles
|
|
)
|
|
|
|
|
|
def test_flood_hazard_xml_provider_error_is_exposed_without_losing_the_canonical_error() -> None:
|
|
response = b"""<?xml version="1.0"?>
|
|
<ExceptionReport xmlns="http://www.opengis.net/ows/1.1">
|
|
<Exception exceptionCode="NoApplicableCode">
|
|
<ExceptionText>This request is trying to generate too much data</ExceptionText>
|
|
</Exception>
|
|
</ExceptionReport>"""
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
FloodHazardAcquisitionService._extract_geotiff(response, "application/xml")
|
|
|
|
assert exc_info.value.code == "FLOOD_HAZARD_PROVIDER_INVALID_RESPONSE"
|
|
assert exc_info.value.details["provider_exception"] == "This request is trying to generate too much data"
|
|
|
|
|
|
def test_flood_hazard_normalization_converts_centimetres_and_clips_zero_values() -> None:
|
|
payload = flood_payload()
|
|
prepared = FloodHazardAcquisitionService._prepared_request(payload, Settings(_env_file=None))
|
|
scope = box(
|
|
payload.bbox.min_x,
|
|
payload.bbox.min_y,
|
|
payload.bbox.max_x,
|
|
payload.bbox.max_y,
|
|
)
|
|
|
|
normalized, validation = FloodHazardAcquisitionService._normalize_raster(depth_tiff(), scope, prepared)
|
|
|
|
assert validation["inundated_pixel_count"] == 200
|
|
assert validation["minimum_depth_m"] == pytest.approx(1.0)
|
|
assert validation["maximum_depth_m"] == pytest.approx(1.0)
|
|
with MemoryFile(normalized) as memory, memory.open() as dataset:
|
|
values = dataset.read(1, masked=True)
|
|
assert dataset.crs.to_epsg() == 31370
|
|
assert dataset.nodata == -9999.0
|
|
assert values.count() == 200
|
|
assert float(values.mean()) == pytest.approx(1.0)
|
|
|
|
|
|
def test_flood_hazard_analysis_reports_scenario_metrics_without_claiming_waterbody_volume(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
path = tmp_path / "flood.tif"
|
|
path.write_bytes(depth_tiff(normalized_metres=True))
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="flood.tif",
|
|
dataset_type="raster",
|
|
source="VMM",
|
|
source_name=FloodHazardAcquisitionService.PROVIDER,
|
|
source_metadata={"product_key": "pluviaal_current_t100", "normalized_value_unit": "m"},
|
|
status="ready",
|
|
storage_path=str(path),
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
|
|
result = FloodHazardAnalysisService.analyze(
|
|
db,
|
|
project_id,
|
|
dataset_id,
|
|
FloodHazardSelectionRequest(bbox=flood_payload().bbox),
|
|
settings=Settings(_env_file=None),
|
|
)
|
|
metrics = {item["metric_key"]: item for item in result["summary"]["metrics"]}
|
|
|
|
assert result["inundated_cell_count"] == 200
|
|
assert result["inundated_fraction"] == pytest.approx(0.5)
|
|
assert metrics["modelled_inundated_area_ha"]["metric_value"] == pytest.approx(0.5)
|
|
assert metrics["modelled_depth_mean_m"]["metric_value"] == pytest.approx(1.0)
|
|
assert metrics["modelled_max_depth_area_integral_m3"]["metric_value"] == pytest.approx(5000.0)
|
|
assert "concurrent_flood_volume_m3" in result["unsupported_metrics"]
|
|
assert "geen gelijktijdig" in result["limitation_message"]
|
|
|
|
|
|
def test_flood_hazard_renderer_returns_transparent_png(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
path = tmp_path / "flood.tif"
|
|
path.write_bytes(depth_tiff(normalized_metres=True))
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="flood.tif",
|
|
dataset_type="raster",
|
|
source="VMM",
|
|
source_name=FloodHazardAcquisitionService.PROVIDER,
|
|
source_metadata={"product_key": "pluviaal_current_t100", "normalized_value_unit": "m"},
|
|
status="ready",
|
|
storage_path=str(path),
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
|
|
assert FloodHazardAnalysisService.render_png(db, project_id, dataset_id).startswith(b"\x89PNG\r\n\x1a\n")
|
|
|
|
|
|
def test_flood_hazard_api_uses_canonical_envelopes(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
output_dataset_id = uuid4()
|
|
db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")})
|
|
monkeypatch.setattr(
|
|
FloodHazardAcquisitionService,
|
|
"acquire",
|
|
lambda *_args, **_kwargs: {"output_dataset_id": str(output_dataset_id), "provider": "vmm_flood_hazard", "reused": False},
|
|
)
|
|
monkeypatch.setattr(
|
|
FloodHazardAnalysisService,
|
|
"analyze",
|
|
lambda *_args, **_kwargs: {
|
|
"dataset_id": str(output_dataset_id),
|
|
"inundated_cell_count": 4,
|
|
"summary": {"metric_value": 0.01, "metric_unit": "ha", "metrics": []},
|
|
"unsupported_metrics": ["permanent_water_volume_m3"],
|
|
},
|
|
)
|
|
app.dependency_overrides[get_db] = lambda: db
|
|
try:
|
|
client = TestClient(app)
|
|
products = client.get(f"/api/v1/projects/{project_id}/datasets/flood-hazard/products")
|
|
acquisition = client.post(
|
|
f"/api/v1/projects/{project_id}/datasets/flood-hazard/acquire",
|
|
json=flood_payload().model_dump(mode="json"),
|
|
)
|
|
selection = client.post(
|
|
f"/api/v1/projects/{project_id}/datasets/{output_dataset_id}/raster/flood-hazard/select",
|
|
json={"bbox": flood_payload().bbox.model_dump()},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert products.status_code == 200 and set(products.json()) == {"data"}
|
|
assert products.json()["data"]["total"] == 12
|
|
assert acquisition.status_code == 200 and set(acquisition.json()) == {"data"}
|
|
assert acquisition.json()["data"]["job_type"] == "raster.flood_hazard.acquire"
|
|
assert selection.status_code == 200 and set(selection.json()) == {"data"}
|
|
assert any(isinstance(item, Job) for item in db.added)
|
|
|
|
|
|
def test_geo_assistant_receives_scenario_bound_flood_metrics(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="pluvial.tif",
|
|
dataset_type="raster",
|
|
source="VMM",
|
|
source_name=FloodHazardAcquisitionService.PROVIDER,
|
|
source_metadata={
|
|
"product_key": "pluviaal_current_t100",
|
|
"product_display_name": "Pluviaal - huidig klimaat - middelgrote kans (T100)",
|
|
},
|
|
status="ready",
|
|
)
|
|
db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")}, query_result=[dataset])
|
|
monkeypatch.setattr(
|
|
FloodHazardAnalysisService,
|
|
"analyze",
|
|
lambda *_args, **_kwargs: {
|
|
"product_key": "pluviaal_current_t100",
|
|
"mechanism": "pluviaal",
|
|
"climate_context": "huidig klimaat",
|
|
"probability_class": "middelgrote kans",
|
|
"return_period_years": 100,
|
|
"summary": {
|
|
"metrics": [
|
|
{
|
|
"metric_label": "Gemodelleerd overstroomd oppervlak",
|
|
"metric_value": 12.5,
|
|
"metric_unit": "ha",
|
|
}
|
|
]
|
|
},
|
|
"limitation_message": "Geen werkelijk of gelijktijdig volume.",
|
|
},
|
|
)
|
|
payload = AssistantQueryRequest(question="Wat is het overstromingsgevaar?", bbox=flood_payload().bbox)
|
|
|
|
context, metrics, _series, dataset_ids, warnings, _scope = GeoAssistantService(Settings(_env_file=None))._build_context(
|
|
db,
|
|
project_id=project_id,
|
|
payload=payload,
|
|
)
|
|
|
|
assert warnings == []
|
|
assert dataset_ids == [dataset_id]
|
|
assert metrics[0].theme == "flood_hazard"
|
|
assert "T100" in metrics[0].label
|
|
assert context["rules"]["water_volume_available"] is False
|
|
assert context["rules"]["flood_hazard_scenarios_available"] is True
|
|
assert context["rules"]["flood_depth_area_integral_is_concurrent_volume"] is False
|
|
|
|
|
|
def test_flood_hazard_runtime_contract_is_packaged() -> None:
|
|
for path in (
|
|
ROOT / ".env.example",
|
|
ROOT / "docker-compose.yml",
|
|
ROOT / "docker-compose.unraid.yml",
|
|
ROOT / "deploy" / "unraid" / "geointel.env.example",
|
|
):
|
|
content = path.read_text(encoding="utf-8")
|
|
assert "FLOOD_HAZARD_ENABLED" in content
|
|
assert "FLOOD_HAZARD_WCS_URL" in content
|
|
assert "FLOOD_HAZARD_MAX_PIXELS" in content
|
|
|
|
operator = (ROOT / "scripts" / "provision_mol_flood_hazards.py").read_text(encoding="utf-8")
|
|
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")
|
|
frontend = (ROOT / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
assert "/datasets/flood-hazard/acquire" in operator
|
|
assert "/raster/flood-hazard/select" in operator
|
|
assert "concurrent_flood_volume_m3" in operator
|
|
assert "py_compile scripts/provision_mol_flood_hazards.py" in readiness
|
|
assert "COPY scripts/provision_mol_flood_hazards.py" in dockerfile
|
|
assert "Overstromingsscenario" in frontend
|
|
assert "floodHazardImageUrl" in frontend
|
|
assert "dataset.source_name === 'vmm_flood_hazard'" in frontend
|
|
assert "return theme.id === 'flood_hazard'" in frontend
|