Add regional DHMV provisioning

scripts/README.md

@@ -1556,6 +1556,32 @@ docker exec geointel python /app/scripts/provision_mol_dhmv.py --resolution-m 5
 Do not use DHMV output as water depth or water volume. The command fails when
 the API no longer reports those metrics as explicitly unsupported.
 
+## Regional DHMV terrain rasters
+
+Plan and provision the governed DTM/DSM pair for all 28 persisted Kempen
+municipality Areas:
+
+```bash
+docker exec geointel python /app/scripts/provision_regional_dhmv.py \
+  --scope kempen-transport-region --dry-run
+
+docker exec geointel python /app/scripts/provision_regional_dhmv.py \
+  --scope kempen-transport-region
+```
+
+The complete run plans 56 canonical acquisitions. `--members Mol,Geel` and
+`--products dtm_1m` provide bounded validation subsets; `--stop-on-error`
+turns the default complete failure report into fail-fast behavior. Existing
+checksummed request identities are reused unless `--force` is explicit. Each
+successful item is reported immediately so a long operator run remains
+observable. Every output is still an ordinary Area-linked Dataset and
+DatasetVersion produced by the existing DHMV service.
+
+The regional operator deliberately retains municipality partitions. It does
+not assemble a monolithic Kempen height raster, does not claim annual terrain
+change and rejects any terrain-analysis response that stops listing water
+depth and water volume as unsupported.
+
 ## Mol VMM flood-hazard scenarios
 
 Acquire and validate all twelve official VMM fluvial/pluvial flood-depth

scripts/provision_regional_dhmv.py

@@ -0,0 +1,337 @@
+"""Provision governed DHMV II rasters for an approved GeoIntel scope.
+
+The command coordinates canonical API calls only. It does not fetch rasters
+directly, write database rows directly or run implicitly at 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 = ("dtm_1m", "dsm_1m")
+
+
+@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 DHMV II rasters 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 scope member.",
+    )
+    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 acquire data.")
+    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 DHMV product keys: {', '.join(invalid)}")
+    if not products:
+        raise RuntimeError("At least one DHMV 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: list[dict[str, Any]] = []
+    limit = 200
+    offset = 0
+    while True:
+        page = unwrap(
+            session.get(
+                f"{base_url}/api/v1/projects/{project_id}/areas",
+                params={"limit": limit, "offset": offset},
+                timeout=60,
+            )
+        )
+        page_items = list(page["items"])
+        areas.extend(page_items)
+        total = int(page.get("total", len(areas)))
+        if len(areas) >= total or len(page_items) < limit:
+            break
+        offset += limit
+
+    resolved: list[ResolvedArea] = []
+    missing: list[str] = []
+    for member in members:
+        area = next(
+            (
+                item
+                for item in areas
+                if member.name.casefold() in str(item.get("name", "")).casefold()
+                or member.nis_code in str(item.get("name", ""))
+                or member.nis_code 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/dhmv/products", timeout=60))["items"]
+    registry_keys = {item["key"] for item in registry}
+    if registry_keys != set(PRODUCTS):
+        raise RuntimeError("Backend DHMV registry does not expose the governed DTM/DSM 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/dhmv/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"DHMV 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/terrain/select",
+            json={"bbox": bbox, "area_id": area.id},
+            timeout=timeout,
+        )
+    )
+    if sorted(analysis.get("unsupported_metrics", [])) != ["water_depth_m", "water_volume_m3"]:
+        raise RuntimeError("DHMV terrain contract must explicitly keep water depth and volume 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"],
+        "sample_count": analysis["sample_count"],
+        "coverage_ratio": analysis["coverage_ratio"],
+        "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-DHMV-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:
+                result = acquire_one(
+                    session,
+                    base_url,
+                    project["id"],
+                    area,
+                    product_key,
+                    args.resolution_m,
+                    args.timeout,
+                    args.force,
+                )
+                results.append(result)
+                print(
+                    json.dumps(
+                        {
+                            "status": "completed_item",
+                            "completed": len(results),
+                            "planned": len(areas) * len(products),
+                            "member_name": area.member_name,
+                            "product_key": product_key,
+                            "dataset_id": result["dataset_id"],
+                            "reused": result["reused"],
+                        },
+                        ensure_ascii=False,
+                    ),
+                    flush=True,
+                )
+            except Exception as exc:  # noqa: BLE001 - retain every failed member/product in the operator summary.
+                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, flush=True)
+                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/provision_regional_flood_hazards.py

@@ -283,17 +283,31 @@ def main(argv: Sequence[str] | None = None) -> int:
     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,
-                    )
+                result = acquire_one(
+                    session,
+                    base_url,
+                    project["id"],
+                    area,
+                    product_key,
+                    args.resolution_m,
+                    args.timeout,
+                    args.force,
+                )
+                results.append(result)
+                print(
+                    json.dumps(
+                        {
+                            "status": "completed_item",
+                            "completed": len(results),
+                            "planned": len(areas) * len(products),
+                            "member_name": area.member_name,
+                            "product_key": product_key,
+                            "dataset_id": result["dataset_id"],
+                            "reused": result["reused"],
+                        },
+                        ensure_ascii=False,
+                    ),
+                    flush=True,
                 )
             except Exception as exc:  # noqa: BLE001 - operator summary should retain every failed member/product.
                 failure = {
@@ -304,7 +318,7 @@ def main(argv: Sequence[str] | None = None) -> int:
                     "error": str(exc),
                 }
                 failures.append(failure)
-                print(json.dumps({"status": "failed_item", **failure}, ensure_ascii=False), file=sys.stderr)
+                print(json.dumps({"status": "failed_item", **failure}, ensure_ascii=False), file=sys.stderr, flush=True)
                 if args.stop_on_error:
                     raise
 

scripts/run_readiness_check.sh

@@ -54,6 +54,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_regional_bwk_natura2000.py
 ${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_regional_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_thematic_rasters.py