"""Provision VHA cross-section profiles for every persisted Flemish municipality. The operator is resumable and calls only canonical GeoIntel API endpoints. Regional completeness is finalized server-side only when every municipality has either one ready profile Dataset or an explicit no-profile source result. """ from __future__ import annotations import argparse import hashlib import json import os import sys from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable import requests DEFAULT_API_URL = "http://127.0.0.1:8000" DEFAULT_PROJECT_NAME = "Flanders Regional Workbench" DEFAULT_MANIFEST_PATH = Path("/app/storage/operator-data/bathymetry/vha-flanders/manifest.json") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Provision partitioned VHA profiles for Flanders.") parser.add_argument("--base-url", default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL)) parser.add_argument("--project-name", default=DEFAULT_PROJECT_NAME) parser.add_argument( "--manifest-path", type=Path, default=Path(os.environ.get("GEOINTEL_VHA_FLANDERS_MANIFEST", DEFAULT_MANIFEST_PATH)), ) parser.add_argument("--timeout", type=int, default=900) parser.add_argument("--force", action="store_true") parser.add_argument("--continue-on-error", action="store_true") parser.add_argument("--members", nargs="*", default=[]) parser.add_argument("--max-partitions", type=int, default=0) return parser.parse_args() def unwrap(response: requests.Response) -> Any: try: payload = response.json() except ValueError as exc: raise RuntimeError(f"GeoIntel returned non-JSON ({response.status_code}): {response.text[:300]}") from exc if not response.ok: error_code = payload.get("error") if isinstance(payload, dict) else None message = payload.get("message") if isinstance(payload, dict) else None raise RuntimeError(f"{error_code or response.status_code}: {message or response.text[:300]}") if not isinstance(payload, dict) or "data" not in payload: raise RuntimeError(f"Non-canonical API response from {response.url}") return payload["data"] def list_paginated(session: requests.Session, url: str, *, timeout: int) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] offset = 0 total: int | None = None while True: page = unwrap(session.get(url, params={"limit": 200, "offset": offset}, timeout=timeout)) page_items = list(page.get("items") or []) items.extend(page_items) total = int(page.get("total") or 0) if total is None else total if not page_items or len(items) >= total: break offset += len(page_items) if total is not None and len(items) != total: raise RuntimeError(f"Paginated API returned {len(items)} of {total} records for {url}") return items 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 municipality Area 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 area_identity(area: dict[str, Any]) -> str: canonical = { "id": area.get("id"), "name": area.get("name"), "geometry": area.get("geometry"), } return hashlib.sha256( json.dumps(canonical, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") ).hexdigest() def write_manifest(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".partial") temporary.write_text( json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8", ) temporary.replace(path) def load_manifest(path: Path) -> dict[str, Any]: if not path.is_file(): return {} try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} return payload if isinstance(payload, dict) else {} def source_error(response: requests.Response) -> tuple[str | None, str]: try: payload = response.json() except ValueError: return None, response.text[:300] if not isinstance(payload, dict): return None, response.text[:300] return ( str(payload.get("error")) if payload.get("error") else None, str(payload.get("message") or response.text[:300]), ) def acquire_partition( session: requests.Session, *, base_url: str, project_id: str, area: dict[str, Any], timeout: int, force: bool, ) -> dict[str, Any]: bbox = geometry_bbox(area["geometry"]) response = session.post( f"{base_url}/api/v1/projects/{project_id}/datasets/bathymetry/profiles/acquire", json={"bbox": bbox, "area_id": area["id"], "force_refresh": force}, timeout=timeout, ) if response.status_code == 404: code, message = source_error(response) if code == "BATHYMETRY_NO_PROFILES": return { "status": "no_profiles", "area_id": area["id"], "area_name": area["name"], "area_identity_sha256": area_identity(area), "bbox": bbox, "message": message, "completed_at": datetime.now(timezone.utc).isoformat(), } data = unwrap(response) if data.get("status") != "success" or not data.get("output_dataset_id"): raise RuntimeError(f"VHA acquisition failed for {area['name']}: {data.get('error_message') or data}") result = data.get("result_json") or {} return { "status": "complete", "area_id": area["id"], "area_name": area["name"], "area_identity_sha256": area_identity(area), "bbox": bbox, "dataset_id": data["output_dataset_id"], "reused": bool(result.get("reused")), "profile_count": int(result.get("profile_count") or 0), "document_count": int(result.get("document_count") or 0), "structured_depth_count": int(result.get("structured_depth_count") or 0), "measurement_date_min": result.get("measurement_date_min"), "measurement_date_max": result.get("measurement_date_max"), "completed_at": datetime.now(timezone.utc).isoformat(), } def manifest_identity( *, project_id: str, partitions: list[dict[str, Any]], ) -> tuple[dict[str, Any], str]: identity = { "schema_version": 1, "source": "vmm_vha_bathymetry_profiles", "partition_scope_key": "flanders", "project_id": project_id, "partitions": [ { key: partition.get(key) for key in ( "area_id", "area_name", "area_identity_sha256", "status", "dataset_id", "profile_count", "document_count", "structured_depth_count", "measurement_date_min", "measurement_date_max", ) } for partition in sorted(partitions, key=lambda item: str(item["area_id"])) ], } digest = hashlib.sha256( json.dumps(identity, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") ).hexdigest() return identity, digest def main() -> int: args = parse_args() if args.max_partitions < 0: print(json.dumps({"status": "error", "message": "--max-partitions cannot be negative"}), file=sys.stderr) return 2 base_url = args.base_url.rstrip("/") session = requests.Session() session.headers.update({"User-Agent": "GeoIntel-VHA-Flanders-Operator/1.0"}) manifest = load_manifest(args.manifest_path) try: projects = list_paginated(session, f"{base_url}/api/v1/projects", timeout=60) project = next((item for item in projects if item.get("name") == args.project_name), None) if project is None: raise RuntimeError( f"Project {args.project_name!r} was not found; run provision_flanders_geographic_scope.py first" ) all_areas = list_paginated( session, f"{base_url}/api/v1/projects/{project['id']}/areas", timeout=120, ) municipalities = sorted( (area for area in all_areas if str(area.get("name") or "").casefold().startswith("gemeente ")), key=lambda item: str(item["name"]).casefold(), ) if not 270 <= len(municipalities) <= 300: raise RuntimeError( f"Flanders workspace exposes {len(municipalities)} municipality Areas; expected 270..300" ) requested_names = {name.casefold() for name in args.members} selected = [ area for area in municipalities if not requested_names or str(area["name"]).removeprefix("Gemeente ").split(" - ", 1)[0].casefold() in requested_names ] if requested_names: found = { str(area["name"]).removeprefix("Gemeente ").split(" - ", 1)[0].casefold() for area in selected } missing = sorted(requested_names - found) if missing: raise RuntimeError(f"Unknown Flanders municipality selections: {', '.join(missing)}") if args.max_partitions: selected = selected[: args.max_partitions] prior_by_area = { str(item.get("area_id")): item for item in (manifest.get("partitions") or []) if isinstance(item, dict) and item.get("area_id") } current: list[dict[str, Any]] = [] failures: list[dict[str, Any]] = [] observed_at = datetime.now(timezone.utc).isoformat() working_manifest = { "schema_version": 1, "status": "running", "source": "vmm_vha_bathymetry_profiles", "partition_scope_key": "flanders", "project_id": project["id"], "project_name": project["name"], "municipality_inventory_count": len(municipalities), "selected_partition_count": len(selected), "observed_at": observed_at, "partitions": current, } for index, area in enumerate(selected, start=1): identity = area_identity(area) prior = prior_by_area.get(str(area["id"])) if ( not args.force and prior and prior.get("area_identity_sha256") == identity and prior.get("status") in {"complete", "no_profiles"} ): result = prior else: try: result = acquire_partition( session, base_url=base_url, project_id=str(project["id"]), area=area, timeout=args.timeout, force=args.force, ) except (RuntimeError, requests.RequestException) as exc: result = { "status": "failed", "area_id": area["id"], "area_name": area["name"], "area_identity_sha256": identity, "message": str(exc), "failed_at": datetime.now(timezone.utc).isoformat(), } failures.append(result) current.append(result) working_manifest["completed_partition_count"] = index working_manifest["partitions"] = current write_manifest(args.manifest_path, working_manifest) if result["status"] == "failed" and not args.continue_on_error: raise RuntimeError(f"Partition failed for {area['name']}: {result['message']}") full_inventory_selected = len(selected) == len(municipalities) and not requested_names and not args.max_partitions complete = not failures and all(item["status"] in {"complete", "no_profiles"} for item in current) identity_payload, identity_sha = manifest_identity( project_id=str(project["id"]), partitions=current, ) finalization = None if complete and full_inventory_selected: dataset_ids = [item["dataset_id"] for item in current if item["status"] == "complete"] no_profile_area_ids = [item["area_id"] for item in current if item["status"] == "no_profiles"] finalization = unwrap( session.post( f"{base_url}/api/v1/projects/{project['id']}/datasets/bathymetry/profiles/partitions/finalize", json={ "partition_scope_key": "flanders", "expected_area_ids": [area["id"] for area in municipalities], "dataset_ids": dataset_ids, "no_profile_area_ids": no_profile_area_ids, "manifest_sha256": identity_sha, "observed_at": observed_at, }, timeout=args.timeout, ) ) working_manifest.update( { "status": "complete" if complete else "failed", "completed_at": datetime.now(timezone.utc).isoformat(), "coverage_manifest": identity_payload, "coverage_manifest_sha256": identity_sha, "regional_finalized": finalization is not None, "finalization": finalization, "failed_partition_count": len(failures), } ) write_manifest(args.manifest_path, working_manifest) except (OSError, RuntimeError, ValueError, KeyError, requests.RequestException) as exc: print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr) return 1 print( json.dumps( { "status": working_manifest["status"], "project_id": project["id"], "municipality_inventory_count": len(municipalities), "processed_partition_count": len(current), "data_partition_count": sum(item["status"] == "complete" for item in current), "no_profile_partition_count": sum(item["status"] == "no_profiles" for item in current), "failed_partition_count": len(failures), "profile_count": sum(int(item.get("profile_count") or 0) for item in current), "document_count": sum(int(item.get("document_count") or 0) for item in current), "regional_finalized": finalization is not None, "manifest_path": str(args.manifest_path), "manifest_sha256": identity_sha, "finalization": finalization, }, ensure_ascii=False, indent=2, ) ) return 0 if working_manifest["status"] == "complete" else 1 if __name__ == "__main__": raise SystemExit(main())