JensandClaude Opus 5
dd87a62e8f
Four ways a selection produced a confident number about a different area than the operator drew: Flood hazard divided the inundated cells by every cell in the drawn rectangle, including cells the VMM raster does not model at all. A selection reaching past the modelled extent therefore reported a diluted risk share, turning missing data into an implied absence of risk. Terrain, bathymetry and thematic raster already divided by valid cells; flood hazard was the outlier. It now reports the three populations separately, states model coverage next to the drawn area, and returns a null fraction rather than a zero when nothing was modelled. geometry_mask selects a cell when its centre falls inside the geometry, so a rectangle smaller than one cell — or one landing between four centres — selected nothing and the analysis returned zeros indistinguishable on screen from "we looked and there is nothing here". On a 100 m population raster a 40 m rectangle over a city block reported no inhabitants. Selection now falls back to the touched cells and says that it did, since the answer then covers more ground than was requested. rasterio.mask applies the same centre rule when cropping, so that call is widened too; the cells that count are still decided by the centre rule wherever it selects anything. The object count treated any feature touching the selection as whole, while intersection_area clipped it — two headline numbers on one panel describing different populations. The count stays whole-feature, which is what "objecten" means to an operator, but now reports how many the edge cuts and is marked an estimate when it does. The area_weighted_sum branch reuses that same count instead of issuing its own near-identical query. Partitioned selection de-duplicated the count on source_feature_id but returned the raw rows, so a building on a municipal boundary was counted once and drawn twice. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
562 lines
22 KiB
Python
562 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import numpy as np
|
|
import pytest
|
|
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,
|
|
FloodHazardPartitionSelectionRequest,
|
|
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 edge_depth_tiff(*, left: float, top: float, x_resolution: float, y_resolution: float = 5.0) -> bytes:
|
|
values = np.full((20, 20), 100.0, dtype="float32")
|
|
with MemoryFile() as memory:
|
|
with memory.open(
|
|
driver="GTiff",
|
|
width=20,
|
|
height=20,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:31370",
|
|
transform=from_origin(left, top, x_resolution, y_resolution),
|
|
nodata=0.0,
|
|
) as output:
|
|
output.write(values, 1)
|
|
return memory.read()
|
|
|
|
|
|
def normalized_depth_tiff(*, left: float, top: float, value: float) -> bytes:
|
|
values = np.full((20, 20), value, dtype="float32")
|
|
with MemoryFile() as memory:
|
|
with memory.open(
|
|
driver="GTiff",
|
|
width=20,
|
|
height=20,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:31370",
|
|
transform=from_origin(left, top, 5.0, 5.0),
|
|
nodata=-9999.0,
|
|
) as output:
|
|
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_mosaic_harmonizes_only_bounded_wcs_edge_grid_rounding() -> None:
|
|
regular = edge_depth_tiff(left=200_000, top=210_100, x_resolution=5.0)
|
|
rounded_edge = edge_depth_tiff(
|
|
left=200_100,
|
|
top=210_100,
|
|
x_resolution=4.76555,
|
|
y_resolution=5.0008,
|
|
)
|
|
diagnostics: dict[str, object] = {}
|
|
|
|
mosaic = FloodHazardAcquisitionService._mosaic_geotiffs(
|
|
[regular, rounded_edge],
|
|
expected_resolution_m=5.0,
|
|
diagnostics=diagnostics,
|
|
)
|
|
|
|
with MemoryFile(mosaic) as memory, memory.open() as dataset:
|
|
assert dataset.res == pytest.approx((5.0, 5.0))
|
|
assert diagnostics["harmonized_tile_indexes"] == [1]
|
|
assert diagnostics["harmonization_method"] == "rasterio_merge_target_resolution"
|
|
|
|
unsafe_edge = edge_depth_tiff(left=200_100, top=210_100, x_resolution=4.5)
|
|
with pytest.raises(AppError) as exc_info:
|
|
FloodHazardAcquisitionService._mosaic_geotiffs(
|
|
[regular, unsafe_edge],
|
|
expected_resolution_m=5.0,
|
|
)
|
|
assert exc_info.value.code == "FLOOD_HAZARD_TILE_MISMATCH"
|
|
assert exc_info.value.details["invalid_resolution_tiles"] == [
|
|
{"tile_index": 1, "resolution": [4.5, 5.0]}
|
|
]
|
|
|
|
|
|
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
|
|
# The fixture models the left half and marks the right half nodata. All of
|
|
# the modelled half is wet, and the model covers half the selection. The
|
|
# earlier 0.5 conflated "not modelled" with "modelled dry" and reported
|
|
# half the risk that the model actually describes.
|
|
assert result["valid_cell_count"] == 200
|
|
assert result["no_data_cell_count"] == 200
|
|
assert result["inundated_fraction"] == pytest.approx(1.0)
|
|
assert result["data_coverage_ratio"] == pytest.approx(0.5)
|
|
assert "50.0%" in result["coverage_warning"]
|
|
assert metrics["modelled_inundated_share_pct"]["metric_value"] == pytest.approx(100.0)
|
|
assert metrics["model_coverage_pct"]["metric_value"] == pytest.approx(50.0)
|
|
assert metrics["modelled_inundated_area_ha"]["metric_value"] == pytest.approx(0.5)
|
|
assert metrics["modelled_area_ha"]["metric_value"] == pytest.approx(0.5)
|
|
assert metrics["selection_area_ha"]["metric_value"] == pytest.approx(1.0)
|
|
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_partitioned_flood_analysis_is_exact_across_municipality_boundaries(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
transformer = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True)
|
|
min_x, min_y = transformer.transform(200_000, 210_000)
|
|
middle_x, _ = transformer.transform(200_100, 210_000)
|
|
max_x, max_y = transformer.transform(200_200, 210_100)
|
|
paths = [tmp_path / "left-flood.tif", tmp_path / "right-flood.tif"]
|
|
paths[0].write_bytes(normalized_depth_tiff(left=200_000, top=210_100, value=1.0))
|
|
paths[1].write_bytes(normalized_depth_tiff(left=200_100, top=210_100, value=2.0))
|
|
datasets = [
|
|
Dataset(
|
|
id=uuid4(),
|
|
project_id=project_id,
|
|
area_id=uuid4(),
|
|
name=path.name,
|
|
dataset_type="raster",
|
|
source="VMM",
|
|
source_name=FloodHazardAcquisitionService.PROVIDER,
|
|
source_metadata={
|
|
"product_key": "pluviaal_current_t100",
|
|
"normalized_value_unit": "m",
|
|
"bbox_epsg4326": [left, min_y, right, max_y],
|
|
},
|
|
status="ready",
|
|
storage_path=str(path),
|
|
)
|
|
for path, left, right in (
|
|
(paths[0], min_x, middle_x),
|
|
(paths[1], middle_x, max_x),
|
|
)
|
|
]
|
|
db = FakeSession(query_result=datasets)
|
|
payload = FloodHazardPartitionSelectionRequest(
|
|
bbox={"min_x": min_x, "min_y": min_y, "max_x": max_x, "max_y": max_y, "crs": "EPSG:4326"},
|
|
product_key="pluviaal_current_t100",
|
|
)
|
|
|
|
result = FloodHazardAnalysisService.analyze_partitions(
|
|
db,
|
|
project_id,
|
|
payload,
|
|
settings=Settings(_env_file=None),
|
|
)
|
|
metrics = {item["metric_key"]: item["metric_value"] for item in result["summary"]["metrics"]}
|
|
|
|
assert result["partition_count"] == 2
|
|
assert set(result["dataset_ids"]) == {str(dataset.id) for dataset in datasets}
|
|
assert result["inundated_cell_count"] >= 790
|
|
assert result["inundated_fraction"] == pytest.approx(1.0)
|
|
assert metrics["modelled_depth_mean_m"] == pytest.approx(1.5, abs=0.01)
|
|
assert metrics["modelled_depth_p90_m"] == 2.0
|
|
assert metrics["modelled_inundated_area_ha"] == pytest.approx(2.0, abs=0.03)
|
|
assert "2 persistente gemeentelijke rasterpartities" 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),
|
|
"product_key": "pluviaal_current_t100",
|
|
"mechanism": "pluviaal",
|
|
"climate_context": "huidig klimaat",
|
|
"probability_class": "middelgrote kans",
|
|
"return_period_years": 100,
|
|
"selection_bbox": flood_payload().bbox.model_dump(),
|
|
"selected_cell_count": 10,
|
|
"inundated_cell_count": 4,
|
|
"inundated_fraction": 0.4,
|
|
"resolution_m": 5.0,
|
|
"summary": {
|
|
"metric_label": "Overstroomde oppervlakte",
|
|
"metric_value": 0.01,
|
|
"metric_unit": "ha",
|
|
"aggregation_method": "positive_depth_area",
|
|
"primary_metric_key": "inundated_area_ha",
|
|
"metrics": [],
|
|
},
|
|
"unsupported_metrics": ["permanent_water_volume_m3"],
|
|
"limitation_message": "Scenario depth is not bathymetry.",
|
|
"generated_at": "2026-07-18T00:00:00Z",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
FloodHazardAnalysisService,
|
|
"analyze_partitions",
|
|
lambda *_args, **_kwargs: {
|
|
"dataset_id": str(output_dataset_id),
|
|
"dataset_ids": [str(output_dataset_id)],
|
|
"partition_count": 1,
|
|
"product_key": "pluviaal_current_t100",
|
|
"mechanism": "pluviaal",
|
|
"climate_context": "huidig klimaat",
|
|
"probability_class": "middelgrote kans",
|
|
"return_period_years": 100,
|
|
"selection_bbox": flood_payload().bbox.model_dump(),
|
|
"selected_cell_count": 10,
|
|
"inundated_cell_count": 4,
|
|
"inundated_fraction": 0.4,
|
|
"resolution_m": 5.0,
|
|
"summary": {
|
|
"metric_label": "Overstroomde oppervlakte",
|
|
"metric_value": 0.01,
|
|
"metric_unit": "ha",
|
|
"aggregation_method": "positive_depth_area",
|
|
"primary_metric_key": "inundated_area_ha",
|
|
"metrics": [],
|
|
},
|
|
"unsupported_metrics": ["permanent_water_volume_m3"],
|
|
"limitation_message": "Scenario depth is not bathymetry.",
|
|
"generated_at": "2026-07-18T00:00:00Z",
|
|
},
|
|
)
|
|
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()},
|
|
)
|
|
regional_selection = client.post(
|
|
f"/api/v1/projects/{project_id}/datasets/raster/flood-hazard/select",
|
|
json={"bbox": flood_payload().bbox.model_dump(), "product_key": "pluviaal_current_t100"},
|
|
)
|
|
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 regional_selection.status_code == 200 and set(regional_selection.json()) == {"data"}
|
|
assert regional_selection.json()["data"]["partition_count"] == 1
|
|
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
|