Add regional flood hazard provisioning

CHANGELOG.md

@@ -7,6 +7,22 @@
 
 # Changelog
 
+## Sprint 211 Regional VMM flood-hazard provisioning support (2026-07-16)
+
+- Added `provision_regional_flood_hazards.py`, an explicit operator for the
+  approved 28-municipality Kempen scope that provisions official VMM
+  flood-depth scenarios per municipality Area through the existing canonical
+  flood-hazard API.
+- Kept persistence inside the existing Dataset/DatasetVersion/Job raster flow;
+  no direct PostGIS writes, no browser-side provider fetches, no startup fetches
+  and no new schema/API contract were introduced.
+- Added dry-run, member/product subset, resume/reuse and failure-reporting
+  controls so operators can validate Mol or selected municipalities before the
+  full 336 municipality/scenario matrix.
+- Updated Unraid packaging, readiness checks, tests and docs. The semantic
+  limitation remains explicit: VMM water depth is modeled scenario depth, not
+  permanent water volume, current water level or bathymetry.
+
 ## Sprint 210 Regional BWK/Natura 2000 expansion (2026-07-16)
 
 - Added an explicit 28-municipality operator for the official INBO BWK/Natura

backend/README.md

@@ -1246,6 +1246,37 @@ Run all scenarios for Mol after the regional workspace and Mol Area exist:
 docker exec geointel python /app/scripts/provision_mol_flood_hazards.py
 ```
 
+Provision the same official VMM scenario set for every persisted municipality
+Area in the approved Kempen regional workspace:
+
+```bash
+docker exec geointel python /app/scripts/provision_regional_flood_hazards.py \
+  --scope kempen-transport-region
+```
+
+Inspect the planned municipality/scenario matrix without writing data:
+
+```bash
+docker exec geointel python /app/scripts/provision_regional_flood_hazards.py \
+  --scope kempen-transport-region --dry-run
+```
+
+Useful bounded runs:
+
+```bash
+docker exec geointel python /app/scripts/provision_regional_flood_hazards.py \
+  --members Mol,Geel --products pluviaal_current_t100
+docker exec geointel python /app/scripts/provision_regional_flood_hazards.py \
+  --members 13025 --products pluviaal_current_t10,pluviaal_current_t100
+```
+
+The regional operator uses the canonical API only. It requires the geographic
+scope Areas to exist first, persists one ordinary raster Dataset per
+municipality/scenario and reuses existing Datasets unless `--force` is supplied.
+The full Kempen scope with all products means 28 municipalities times 12
+scenario rasters. This is intentionally explicit operator work, not startup
+work and not a browser-side provider fetch.
+
 Use `--products pluviaal_current_t100`, `--resolution-m 5` or `--force` for an
 explicit subset/refresh. `POST .../raster/flood-hazard/select` returns mapped
 inundated hectares, selection share and local modeled maximum-depth statistics.

backend/tests/test_sprint211_regional_flood_hazards.py

@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPTS = ROOT / "scripts"
+
+
+def load_operator():
+    if str(SCRIPTS) not in sys.path:
+        sys.path.insert(0, str(SCRIPTS))
+    spec = importlib.util.spec_from_file_location(
+        "test_provision_regional_flood_hazards",
+        SCRIPTS / "provision_regional_flood_hazards.py",
+    )
+    module = importlib.util.module_from_spec(spec)
+    assert spec and spec.loader
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+class FakeResponse:
+    def __init__(self, payload: dict[str, Any]):
+        self.payload = payload
+        self.url = "http://test.local"
+
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self) -> dict[str, Any]:
+        return self.payload
+
+
+class FakeSession:
+    def __init__(self, module):
+        self.module = module
+        self.posts: list[tuple[str, dict[str, Any]]] = []
+        self.headers: dict[str, str] = {}
+
+    def get(self, url: str, **_kwargs):
+        if url.endswith("/api/v1/projects"):
+            return FakeResponse({"data": {"items": [{"id": "project-1", "name": "Kempen Regional Workbench"}]}})
+        if url.endswith("/areas"):
+            return FakeResponse(
+                {
+                    "data": {
+                        "items": [
+                            {
+                                "id": "area-mol",
+                                "name": "Gemeente Mol - officiele grens",
+                                "geometry": {
+                                    "type": "Polygon",
+                                    "coordinates": [[
+                                        [5.0, 51.0],
+                                        [5.1, 51.0],
+                                        [5.1, 51.1],
+                                        [5.0, 51.1],
+                                        [5.0, 51.0],
+                                    ]],
+                                },
+                            }
+                        ]
+                    }
+                }
+            )
+        if url.endswith("/datasets/flood-hazard/products"):
+            return FakeResponse({"data": {"items": [{"key": key} for key in self.module.PRODUCTS]}})
+        raise AssertionError(url)
+
+    def post(self, url: str, json: dict[str, Any], **_kwargs):
+        self.posts.append((url, json))
+        if url.endswith("/datasets/flood-hazard/acquire"):
+            return FakeResponse(
+                {
+                    "data": {
+                        "id": "job-1",
+                        "status": "success",
+                        "output_dataset_id": "dataset-1",
+                        "result_json": {"reused": True},
+                    }
+                }
+            )
+        if url.endswith("/raster/flood-hazard/select"):
+            return FakeResponse(
+                {
+                    "data": {
+                        "resolution_m": 5.0,
+                        "selected_cell_count": 100,
+                        "inundated_cell_count": 10,
+                        "unsupported_metrics": [
+                            "bathymetry_depth_m",
+                            "permanent_water_volume_m3",
+                            "concurrent_flood_volume_m3",
+                        ],
+                        "summary": {"metrics": [{"metric_key": "modelled_inundated_area_ha", "metric_value": 0.25}]},
+                    }
+                }
+            )
+        raise AssertionError(url)
+
+
+def test_regional_flood_operator_resolves_products_and_members() -> None:
+    module = load_operator()
+
+    products = module.requested_products("pluviaal_current_t100,fluviaal_future_2050_t1000", set(module.PRODUCTS))
+    members = module.requested_members("Mol,13008", module.KEMPEN_TRANSPORT_REGION_SCOPE.members)
+
+    assert products == ["pluviaal_current_t100", "fluviaal_future_2050_t1000"]
+    assert [member.nis_code for member in members] == ["13025", "13008"]
+
+
+def test_regional_flood_operator_rejects_unknown_scope_inputs() -> None:
+    module = load_operator()
+
+    with pytest.raises(RuntimeError, match="Unsupported flood-hazard"):
+        module.requested_products("custom", set(module.PRODUCTS))
+    with pytest.raises(RuntimeError, match="Unknown scope members"):
+        module.requested_members("Atlantis", module.KEMPEN_TRANSPORT_REGION_SCOPE.members)
+
+
+def test_regional_flood_operator_dry_run_uses_canonical_registry(monkeypatch, capsys) -> None:
+    module = load_operator()
+    fake_session = FakeSession(module)
+    monkeypatch.setattr(module.requests, "Session", lambda: fake_session)
+
+    result = module.main(["--members", "Mol", "--products", "pluviaal_current_t100", "--dry-run"])
+    output = capsys.readouterr().out
+
+    assert result == 0
+    assert '"status": "dry_run"' in output
+    assert '"planned_acquisitions": 1' in output
+    assert fake_session.posts == []
+
+
+def test_regional_flood_operator_calls_acquisition_and_selection(monkeypatch, capsys) -> None:
+    module = load_operator()
+    fake_session = FakeSession(module)
+    monkeypatch.setattr(module.requests, "Session", lambda: fake_session)
+
+    result = module.main(["--members", "Mol", "--products", "pluviaal_current_t100"])
+    output = capsys.readouterr().out
+
+    assert result == 0
+    assert '"completed_count": 1' in output
+    assert len(fake_session.posts) == 2
+    acquisition_payload = fake_session.posts[0][1]
+    assert fake_session.posts[0][0].endswith("/datasets/flood-hazard/acquire")
+    assert acquisition_payload["area_id"] == "area-mol"
+    assert acquisition_payload["product_key"] == "pluviaal_current_t100"
+    assert acquisition_payload["bbox"]["crs"] == "EPSG:4326"
+
+
+def test_regional_flood_operator_is_packaged() -> 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")
+    docs = (ROOT / "scripts" / "README.md").read_text(encoding="utf-8")
+
+    assert "py_compile scripts/provision_regional_flood_hazards.py" in readiness
+    assert "COPY scripts/provision_regional_flood_hazards.py" in dockerfile
+    assert "provision_regional_flood_hazards.py" in docs

deploy/unraid/Dockerfile.all-in-one

@@ -76,6 +76,7 @@ COPY scripts/provision_mol_municipality_workspace.py /app/scripts/provision_mol_
 COPY scripts/provision_mol_context_layers.py /app/scripts/provision_mol_context_layers.py
 COPY scripts/provision_mol_dhmv.py /app/scripts/provision_mol_dhmv.py
 COPY scripts/provision_mol_flood_hazards.py /app/scripts/provision_mol_flood_hazards.py
+COPY scripts/provision_regional_flood_hazards.py /app/scripts/provision_regional_flood_hazards.py
 COPY scripts/provision_mol_population_history.py /app/scripts/provision_mol_population_history.py
 COPY scripts/provision_mol_historical_landuse.py /app/scripts/provision_mol_historical_landuse.py
 COPY scripts/provision_regional_historical_landuse.py /app/scripts/provision_regional_historical_landuse.py

docs/CODEX_EXECUTION_LOG.md

@@ -8940,3 +8940,40 @@ Boundaries and next step:
   governed municipality partitions and measured VMM scenario rasters before
   attempting a full-resolution regional DHMV expansion; no water volume or
   inland bathymetry may be inferred from flood depth.
+
+## Sprint 211 - Regional VMM flood-hazard provisioning support (2026-07-16)
+
+Implemented:
+- Added `scripts/provision_regional_flood_hazards.py`, an explicit operator
+  that resolves the approved geographic scope, validates the fixed twelve-item
+  VMM flood-hazard registry and acquires scenarios per persisted municipality
+  Area through the existing canonical API endpoints.
+- Kept all raster persistence inside the existing `FloodHazardAcquisitionService`
+  and Dataset/DatasetVersion/Job flow. The operator performs no direct WCS
+  requests, no direct PostGIS writes and no browser/startup provider fetching.
+- Added `--dry-run`, `--members`, `--products`, `--force` and
+  `--stop-on-error` controls so Mol, selected municipalities or the complete
+  28-member scope can be run safely and resumed.
+- Added packaging/readiness support so the operator is compiled and available
+  in the all-in-one Unraid image.
+- Documented that regional flood coverage remains municipality-partitioned
+  because one monolithic Kempen raster would exceed practical WCS/pixel limits.
+
+Validation:
+- Focused tests cover product/member resolution, dry-run planning, canonical
+  acquisition/selection calls and release packaging.
+
+Known limitations:
+- This pass adds the operational regional provisioner and validation contract.
+  A complete live run of all 336 municipality/scenario acquisitions is long
+  operator work and should be launched deliberately on Tower after reviewing
+  the dry-run matrix.
+- VMM flood depth remains modelled local maximum scenario depth. GeoIntel still
+  must not expose permanent waterbody volume, current water level or
+  bathymetry from this source.
+
+Next:
+- Run the regional VMM operator first for Mol/Geel with `pluviaal_current_t100`
+  on Tower, then expand to the full twelve-scenario municipality matrix if the
+  provider remains stable. After that, prioritize regional DHMV DTM/DSM with
+  the same partitioned operator discipline.

docs/DATA_SOURCES.md

@@ -383,6 +383,34 @@ De officiële audit vond geen publieke, gebiedsdekkende inland-bathymetrie voor
 Mol. Kust- en Noordzeeproducten vallen buiten de ruimtelijke scope. Waterinfo
 stations blijven puntmetingen en GRB-watergeometrie blijft tweedimensionaal.
 
+## Regionale VMM overstromingsscenario's
+
+`scripts/provision_regional_flood_hazards.py` breidt de beheerde
+VMM-overstromingsflow uit naar de goedgekeurde `kempen-transport-region`
+scope. De operator gebruikt de 28 persistente gemeente-Areas die door
+`provision_geographic_scope.py` zijn aangemaakt en roept per gemeente en per
+scenario uitsluitend de bestaande canonical API aan:
+
+- `POST /api/v1/projects/{project_id}/datasets/flood-hazard/acquire`
+- `POST /api/v1/projects/{project_id}/datasets/{dataset_id}/raster/flood-hazard/select`
+
+Deze gemeentepartities zijn bewust. Een volledig regionaal raster in een
+aanvraag zou de publieke WCS- en pixelgrenzen onnodig belasten. In de UI wordt
+een VMM-dataset alleen als overstromingslaag getoond voor het actieve
+werkgebied waaraan die dataset gekoppeld is. Zo blijft Mol bij Mol, Geel bij
+Geel, enzovoort.
+
+De regionale operator ondersteunt `--dry-run`, `--members` en `--products`.
+Een volledige scope met alle twaalf scenario's plant 336 gecontroleerde
+acquisities. Herhaalruns gebruiken bestaande checksummed Datasets via de
+backend-cache zolang de requestidentiteit niet verandert.
+
+Ook regionaal blijft de semantiek onveranderd: VMM-waterdiepte is een
+gemodelleerde maximale lokale diepte per kans- en klimaatscenario. GeoIntel kan
+oppervlakte, gemiddelde/P90/maximale diepte en de diepte-oppervlakte-integraal
+berekenen. Dat is geen gelijktijdig waterbergingsvolume, geen actuele
+waterstand en geen bathymetrie.
+
 ## Gebouwenregister
 
 The governed operator `scripts/provision_buildings_addresses_register.py`

docs/STORAGE_ARCHITECTURE.md

@@ -118,6 +118,15 @@ metadata records the 1 m native product, 5 m default analysis grid, EPSG:31370,
 derived browser views and are never authoritative. No raster binary is stored
 in PostgreSQL and no DHMV file is treated as water depth or volume.
 
+VMM flood-hazard scenario outputs follow the same raster Dataset policy. The
+bounded acquisition service stores one normalized, compressed, Area-clipped
+GeoTIFF per scenario/request identity. `provision_regional_flood_hazards.py`
+does not create a parallel storage layout: every municipality/scenario result
+is an ordinary Dataset and DatasetVersion with WCS request hashes, tile request
+URLs, response/coverage/normalized checksums, EPSG:31370 bounds, scenario
+metadata and explicit unsupported-volume flags. Repeat runs reuse matching
+ready Datasets through the acquisition service cache.
+
 BWK/Natura 2000 evidence lives under
 `storage/operator-evidence/bwk-natura2000-2025/mol/`. The `raw/` directory
 contains immutable WFS pages; the adjacent manifest records their URLs,

docs/TODO.md

@@ -26,6 +26,7 @@
 - [x] Add bounded historical orthophoto acquisition for official 1971-2025 products with a map overlay and no current-GRB QA on old imagery.
 - [x] Add BWK/Natura 2000 through an explicit provider/operator contract.
 - [x] Expand BWK/Natura 2000 state 2025 from Mol to all 28 approved Kempen municipalities with partitioned source evidence.
+- [x] Add a regional VMM flood-hazard operator that provisions official scenario rasters per municipality Area through the canonical API.
 - [x] Add annual agricultural-use parcels through an explicit provider/operator contract.
 - [x] Extend the official 1778/1873/1969 historical buildings, water and roads series from Mol to the approved regional scope with partitioned source audits.
 - [x] Connect a drawn rectangle to bounded official orthophoto acquisition, local configured-YOLO detection and persisted GRB QA.
@@ -72,6 +73,7 @@ pass live Mol validation before regional expansion.
 - [x] Integrate the separate public VMM fluvial/pluvial flood-hazard depth scenarios without presenting them as bathymetry or current water state.
 - [x] Persist scenario identity, source centimetres, normalized metres, WCS checksums, exact Area clipping and positive-depth coverage.
 - [x] Expose mapped inundation area, depth statistics and a clearly named maximum-depth area integral with a prohibition on calling it concurrent volume.
+- [x] Extend the VMM scenario workflow to the approved regional scope using 28 municipality partitions instead of one unsafe monolithic raster.
 - [ ] Define waterbody linkage, surface elevation, bottom elevation and uncertainty propagation before adding any volume metric.
 - [ ] Validate coverage gaps and prohibit extrapolation outside measured/profiled waterbodies.
 - [ ] Add independent GIS review and golden-volume fixtures before exposing the result to users or Ollama.

frontend/README.md

@@ -518,6 +518,14 @@ pattern and a rectangle returns scenario-bound hectare/depth metrics. Raster
 GeoJSON export stays disabled. The UI never labels the maximum-depth area
 integral as current, permanent or concurrent water volume.
 
+Regional VMM provisioning creates one scenario raster per municipality Area.
+The explorer therefore shows only the flood scenarios whose `area_id` matches
+the active work area. This avoids presenting a Mol scenario while the map is
+focused on another municipality. The region-wide Area remains the navigation
+context; municipality Areas are the analysis scope for flood rasters because
+the public WCS and raster cell limits make one monolithic Kempen raster
+operationally unsafe.
+
 ## Useful repository scripts
 
 - `bash scripts/frontend_install.sh`

scripts/README.md

@@ -1579,6 +1579,40 @@ docker exec geointel python /app/scripts/provision_mol_flood_hazards.py --produc
 docker exec geointel python /app/scripts/provision_mol_flood_hazards.py --resolution-m 5 --force
 ```
 
+## Regional VMM flood-hazard scenarios
+
+Provision governed VMM flood-depth scenarios for every persisted municipality
+Area in an approved scope:
+
+```bash
+docker exec geointel python /app/scripts/provision_regional_flood_hazards.py \
+  --scope kempen-transport-region --dry-run
+
+docker exec geointel python /app/scripts/provision_regional_flood_hazards.py \
+  --scope kempen-transport-region
+```
+
+The command requires `provision_geographic_scope.py --scope
+kempen-transport-region` to have created the regional project and member
+Areas. It uses only canonical API calls, validates the backend twelve-product
+registry and runs a full-Area selection smoke after each acquisition. Existing
+scenario Datasets are reused unless `--force` is supplied.
+
+Useful bounded runs while validating source availability:
+
+```bash
+docker exec geointel python /app/scripts/provision_regional_flood_hazards.py \
+  --members Mol --products pluviaal_current_t100
+
+docker exec geointel python /app/scripts/provision_regional_flood_hazards.py \
+  --members Mol,Geel --products pluviaal_current_t10,pluviaal_current_t100
+```
+
+A complete Kempen run plans 28 municipalities times 12 scenario rasters. It can
+take a long time because every VMM WCS tile is bounded, rate-limited and
+validated. This is expected operator work; the app never fetches these rasters
+on page load or map click.
+
 ## Tower deployment
 
 Push the local branch to Gitea, then rebuild the Unraid/Tower Docker runtime:

scripts/provision_regional_flood_hazards.py

@@ -0,0 +1,322 @@
+"""Provision governed VMM flood-depth scenarios for an approved GeoIntel scope.
+
+The command coordinates canonical API calls only. It does not fetch rasters
+directly, does not write database rows directly and does not run implicitly at
+application startup.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from dataclasses import dataclass
+from typing import Any, Iterable, Sequence
+
+import requests
+
+from geographic_scopes import GEOGRAPHIC_SCOPES, KEMPEN_TRANSPORT_REGION_SCOPE, ScopeMember
+
+
+DEFAULT_API_URL = "http://127.0.0.1:8000"
+DEFAULT_SCOPE = KEMPEN_TRANSPORT_REGION_SCOPE.key
+PRODUCTS = tuple(
+    f"{mechanism}_{climate}_t{period}"
+    for mechanism in ("pluviaal", "fluviaal")
+    for climate in ("current", "future_2050")
+    for period in (10, 100, 1000)
+)
+
+
+@dataclass(frozen=True)
+class ResolvedArea:
+    id: str
+    name: str
+    member_name: str
+    nis_code: str
+    geometry: dict[str, Any]
+
+
+def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Provision official VMM flood-depth scenarios for an approved scope.")
+    parser.add_argument("--base-url", default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL))
+    parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE)
+    parser.add_argument("--products", default=",".join(PRODUCTS), help="Comma-separated governed product keys.")
+    parser.add_argument(
+        "--members",
+        default="",
+        help="Optional comma-separated municipality names or NIS codes. Empty means every member in the selected scope.",
+    )
+    parser.add_argument("--resolution-m", type=float, default=5.0)
+    parser.add_argument("--timeout", type=int, default=1800)
+    parser.add_argument("--force", action="store_true")
+    parser.add_argument("--stop-on-error", action="store_true")
+    parser.add_argument("--dry-run", action="store_true", help="Resolve scope/products only; do not call acquisition endpoints.")
+    return parser.parse_args(argv)
+
+
+def unwrap(response: requests.Response) -> Any:
+    response.raise_for_status()
+    payload = response.json()
+    if not isinstance(payload, dict) or "data" not in payload:
+        raise RuntimeError(f"Non-canonical API response from {response.url}")
+    return payload["data"]
+
+
+def coordinates(geometry: dict[str, Any]) -> Iterable[tuple[float, float]]:
+    def walk(value: Any):
+        if isinstance(value, list) and len(value) >= 2 and all(isinstance(item, (int, float)) for item in value[:2]):
+            yield float(value[0]), float(value[1])
+            return
+        if isinstance(value, list):
+            for child in value:
+                yield from walk(child)
+
+    yield from walk(geometry.get("coordinates", []))
+
+
+def geometry_bbox(geometry: dict[str, Any]) -> dict[str, float | str]:
+    points = list(coordinates(geometry))
+    if not points:
+        raise RuntimeError("Persisted Area geometry contains no coordinates")
+    xs = [point[0] for point in points]
+    ys = [point[1] for point in points]
+    return {"min_x": min(xs), "min_y": min(ys), "max_x": max(xs), "max_y": max(ys), "crs": "EPSG:4326"}
+
+
+def requested_products(raw_products: str, registry_keys: set[str]) -> list[str]:
+    products = [item.strip().lower() for item in raw_products.split(",") if item.strip()]
+    invalid = sorted(set(products) - registry_keys)
+    if invalid:
+        raise RuntimeError(f"Unsupported flood-hazard product keys: {', '.join(invalid)}")
+    if not products:
+        raise RuntimeError("At least one flood-hazard product key is required")
+    return products
+
+
+def requested_members(raw_members: str, members: tuple[ScopeMember, ...]) -> list[ScopeMember]:
+    if not raw_members.strip():
+        return list(members)
+    index: dict[str, ScopeMember] = {}
+    for member in members:
+        index[member.name.casefold()] = member
+        index[member.nis_code] = member
+    selected: list[ScopeMember] = []
+    unknown: list[str] = []
+    for token in [item.strip() for item in raw_members.split(",") if item.strip()]:
+        member = index.get(token.casefold()) or index.get(token)
+        if member is None:
+            unknown.append(token)
+        elif member not in selected:
+            selected.append(member)
+    if unknown:
+        raise RuntimeError(f"Unknown scope members: {', '.join(unknown)}")
+    if not selected:
+        raise RuntimeError("At least one scope member is required")
+    return selected
+
+
+def resolve_project(session: requests.Session, base_url: str, project_name: str) -> dict[str, Any]:
+    projects = unwrap(session.get(f"{base_url}/api/v1/projects", params={"limit": 200, "offset": 0}, timeout=60))["items"]
+    project = next((item for item in projects if item["name"] == project_name), None)
+    if project is None:
+        raise RuntimeError(f"Project {project_name!r} was not found. Run provision_geographic_scope.py first.")
+    return project
+
+
+def resolve_areas(
+    session: requests.Session,
+    base_url: str,
+    project_id: str,
+    members: Sequence[ScopeMember],
+) -> list[ResolvedArea]:
+    areas = unwrap(
+        session.get(
+            f"{base_url}/api/v1/projects/{project_id}/areas",
+            params={"limit": 500, "offset": 0},
+            timeout=60,
+        )
+    )["items"]
+    resolved: list[ResolvedArea] = []
+    missing: list[str] = []
+    for member in members:
+        name_token = member.name.casefold()
+        nis_token = member.nis_code
+        area = next(
+            (
+                item
+                for item in areas
+                if name_token in str(item.get("name", "")).casefold()
+                or nis_token in str(item.get("name", ""))
+                or nis_token in str(item.get("source_metadata", ""))
+            ),
+            None,
+        )
+        if area is None:
+            missing.append(f"{member.name} ({member.nis_code})")
+            continue
+        if not area.get("geometry"):
+            raise RuntimeError(f"Area {area.get('name')!r} has no geometry in the canonical API response")
+        resolved.append(
+            ResolvedArea(
+                id=area["id"],
+                name=area["name"],
+                member_name=member.name,
+                nis_code=member.nis_code,
+                geometry=area["geometry"],
+            )
+        )
+    if missing:
+        raise RuntimeError(
+            "Persisted municipality Areas are missing: "
+            + ", ".join(missing)
+            + ". Run provision_geographic_scope.py for the selected scope first."
+        )
+    return resolved
+
+
+def validate_registry(session: requests.Session, base_url: str, project_id: str) -> set[str]:
+    registry = unwrap(session.get(f"{base_url}/api/v1/projects/{project_id}/datasets/flood-hazard/products", timeout=60))["items"]
+    registry_keys = {item["key"] for item in registry}
+    if registry_keys != set(PRODUCTS):
+        raise RuntimeError("Backend flood-hazard registry does not expose the governed twelve-product set")
+    return registry_keys
+
+
+def acquire_one(
+    session: requests.Session,
+    base_url: str,
+    project_id: str,
+    area: ResolvedArea,
+    product_key: str,
+    resolution_m: float,
+    timeout: int,
+    force: bool,
+) -> dict[str, Any]:
+    bbox = geometry_bbox(area.geometry)
+    job = unwrap(
+        session.post(
+            f"{base_url}/api/v1/projects/{project_id}/datasets/flood-hazard/acquire",
+            json={
+                "bbox": bbox,
+                "area_id": area.id,
+                "product_key": product_key,
+                "resolution_m": resolution_m,
+                "force_refresh": force,
+            },
+            timeout=timeout,
+        )
+    )
+    if job.get("status") != "success" or not job.get("output_dataset_id"):
+        raise RuntimeError(f"Flood-hazard acquisition failed for {area.member_name} {product_key}: {job.get('error_message') or job}")
+    analysis = unwrap(
+        session.post(
+            f"{base_url}/api/v1/projects/{project_id}/datasets/{job['output_dataset_id']}/raster/flood-hazard/select",
+            json={"bbox": bbox, "area_id": area.id},
+            timeout=timeout,
+        )
+    )
+    unsupported = set(analysis.get("unsupported_metrics", []))
+    required_unsupported = {"bathymetry_depth_m", "permanent_water_volume_m3", "concurrent_flood_volume_m3"}
+    if required_unsupported - unsupported:
+        raise RuntimeError("Flood-hazard contract must keep bathymetry and definitive water volumes unavailable")
+    return {
+        "member_name": area.member_name,
+        "nis_code": area.nis_code,
+        "area_id": area.id,
+        "area_name": area.name,
+        "product_key": product_key,
+        "dataset_id": job["output_dataset_id"],
+        "reused": bool((job.get("result_json") or {}).get("reused")),
+        "resolution_m": analysis["resolution_m"],
+        "selected_cell_count": analysis["selected_cell_count"],
+        "inundated_cell_count": analysis["inundated_cell_count"],
+        "metrics": analysis["summary"]["metrics"],
+    }
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = parse_args(argv)
+    scope = GEOGRAPHIC_SCOPES[args.scope]
+    base_url = args.base_url.rstrip("/")
+    selected_members = requested_members(args.members, scope.members)
+
+    session = requests.Session()
+    session.headers.update({"User-Agent": "GeoIntel-Regional-VMM-Flood-Hazard-Operator/1.0"})
+    project = resolve_project(session, base_url, scope.project_name)
+    registry_keys = validate_registry(session, base_url, project["id"])
+    products = requested_products(args.products, registry_keys)
+    areas = resolve_areas(session, base_url, project["id"], selected_members)
+
+    if args.dry_run:
+        print(
+            json.dumps(
+                {
+                    "status": "dry_run",
+                    "scope": scope.key,
+                    "project_id": project["id"],
+                    "member_count": len(areas),
+                    "product_count": len(products),
+                    "planned_acquisitions": len(areas) * len(products),
+                    "members": [{"name": area.member_name, "nis_code": area.nis_code, "area_id": area.id} for area in areas],
+                    "products": products,
+                },
+                ensure_ascii=False,
+                indent=2,
+            )
+        )
+        return 0
+
+    results: list[dict[str, Any]] = []
+    failures: list[dict[str, Any]] = []
+    for area in areas:
+        for product_key in products:
+            try:
+                results.append(
+                    acquire_one(
+                        session,
+                        base_url,
+                        project["id"],
+                        area,
+                        product_key,
+                        args.resolution_m,
+                        args.timeout,
+                        args.force,
+                    )
+                )
+            except Exception as exc:  # noqa: BLE001 - operator summary should retain every failed member/product.
+                failure = {
+                    "member_name": area.member_name,
+                    "nis_code": area.nis_code,
+                    "area_id": area.id,
+                    "product_key": product_key,
+                    "error": str(exc),
+                }
+                failures.append(failure)
+                print(json.dumps({"status": "failed_item", **failure}, ensure_ascii=False), file=sys.stderr)
+                if args.stop_on_error:
+                    raise
+
+    print(
+        json.dumps(
+            {
+                "status": "ok" if not failures else "partial",
+                "scope": scope.key,
+                "project_id": project["id"],
+                "member_count": len(areas),
+                "product_count": len(products),
+                "completed_count": len(results),
+                "failure_count": len(failures),
+                "products": results,
+                "failures": failures,
+            },
+            ensure_ascii=False,
+            indent=2,
+        )
+    )
+    return 0 if not failures else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_readiness_check.sh

@@ -55,6 +55,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_agricultural_parcel_history.py
 ${PYTHON_BIN} -m py_compile scripts/provision_buildings_addresses_register.py
 ${PYTHON_BIN} -m py_compile scripts/provision_mol_dhmv.py
 ${PYTHON_BIN} -m py_compile scripts/provision_mol_flood_hazards.py
+${PYTHON_BIN} -m py_compile scripts/provision_regional_flood_hazards.py
 ${PYTHON_BIN} -m py_compile scripts/provision_regional_timeseries.py
 ${PYTHON_BIN} -m py_compile scripts/geographic_scopes.py
 ${PYTHON_BIN} -m py_compile scripts/provision_geographic_scope.py