"""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())