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
122 lines
4.4 KiB
Python
122 lines
4.4 KiB
Python
"""A selection finer than the source raster must answer, not return zero.
|
|
|
|
End-to-end counterpart to ``test_raster_cell_selection``: the analysis reads a
|
|
real GeoTIFF, so it proves the fallback survives the clip/mask path the service
|
|
actually uses rather than only the helper in isolation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
np = pytest.importorskip("numpy")
|
|
rasterio = pytest.importorskip("rasterio")
|
|
|
|
from pyproj import Transformer # noqa: E402 - optional rasterio gate precedes geospatial imports
|
|
from rasterio.transform import from_origin # noqa: E402 - optional rasterio gate precedes geospatial imports
|
|
|
|
from app.core.config import Settings # noqa: E402 - optional rasterio gate precedes app imports
|
|
from app.models import Dataset # noqa: E402 - optional rasterio gate precedes app imports
|
|
from app.schemas.flood_hazard import FloodHazardSelectionRequest # noqa: E402 - optional rasterio gate precedes app imports
|
|
from app.services.flood_hazard_acquisition_service import ( # noqa: E402 - optional rasterio gate precedes app imports
|
|
FloodHazardAcquisitionService,
|
|
)
|
|
from app.services.flood_hazard_analysis_service import ( # noqa: E402 - optional rasterio gate precedes app imports
|
|
FloodHazardAnalysisService,
|
|
)
|
|
|
|
|
|
TO_4326 = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True)
|
|
PRODUCT_KEY = "fluviaal_current_t100"
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, objects):
|
|
self.objects = objects
|
|
|
|
def get(self, model, item_id):
|
|
return self.objects.get((model, item_id))
|
|
|
|
|
|
def _write_raster(path: Path, *, resolution: float, depth: float) -> None:
|
|
values = np.full((4, 4), depth, dtype="float32")
|
|
with rasterio.open(
|
|
path,
|
|
"w",
|
|
driver="GTiff",
|
|
width=4,
|
|
height=4,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:31370",
|
|
transform=from_origin(200_000, 210_000, resolution, resolution),
|
|
nodata=-9999.0,
|
|
) as output:
|
|
output.write(values, 1)
|
|
|
|
|
|
def _dataset(project_id, dataset_id, path: Path) -> Dataset:
|
|
return Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="vmm-flood.tif",
|
|
dataset_type="raster",
|
|
source="vmm",
|
|
source_name=FloodHazardAcquisitionService.PROVIDER,
|
|
status="ready",
|
|
storage_path=str(path),
|
|
source_metadata={"product_key": PRODUCT_KEY, "normalized_value_unit": "m"},
|
|
)
|
|
|
|
|
|
def _bbox_for(min_x: float, min_y: float, max_x: float, max_y: float) -> dict:
|
|
left, bottom = TO_4326.transform(min_x, min_y)
|
|
right, top = TO_4326.transform(max_x, max_y)
|
|
return {"min_x": left, "min_y": bottom, "max_x": right, "max_y": top, "crs": "EPSG:4326"}
|
|
|
|
|
|
def _analyze(tmp_path: Path, bbox: dict, *, resolution: float = 100.0) -> dict:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
path = tmp_path / "flood.tif"
|
|
_write_raster(path, resolution=resolution, depth=2.0)
|
|
dataset = _dataset(project_id, dataset_id, path)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
|
|
return FloodHazardAnalysisService.analyze(
|
|
db,
|
|
project_id,
|
|
dataset_id,
|
|
FloodHazardSelectionRequest(bbox=bbox),
|
|
settings=Settings(_env_file=None),
|
|
)
|
|
|
|
|
|
def test_a_selection_smaller_than_one_cell_reports_the_cell_it_touches(tmp_path: Path) -> None:
|
|
# A 40 x 30 m rectangle wholly inside one 100 m cell: no cell centre falls
|
|
# inside it, so the centre rule alone would report an empty selection.
|
|
result = _analyze(tmp_path, _bbox_for(200_010, 209_960, 200_050, 209_990))
|
|
|
|
assert result["inundated_cell_count"] == 1
|
|
assert result["inundated_fraction"] == pytest.approx(1.0)
|
|
assert "kleiner dan één rastercel" in result["coverage_warning"]
|
|
|
|
|
|
def test_a_normal_selection_is_unaffected(tmp_path: Path) -> None:
|
|
result = _analyze(tmp_path, _bbox_for(200_000, 209_700, 200_300, 210_000))
|
|
|
|
assert result["inundated_cell_count"] >= 9
|
|
assert result["coverage_warning"] is None
|
|
|
|
|
|
def test_the_reported_area_matches_the_cells_that_were_analysed(tmp_path: Path) -> None:
|
|
result = _analyze(tmp_path, _bbox_for(200_010, 209_960, 200_050, 209_990))
|
|
metrics = {item["metric_key"]: item["metric_value"] for item in result["summary"]["metrics"]}
|
|
|
|
# One 100 x 100 m cell, not the 0.12 ha that was drawn.
|
|
assert metrics["modelled_inundated_area_ha"] == pytest.approx(1.0)
|
|
assert metrics["selection_area_ha"] == pytest.approx(1.0)
|