183 lines
7.7 KiB
Python
183 lines
7.7 KiB
Python
"""Provision governed Flemish thematic rasters through the GeoIntel API.
|
|
|
|
The safe default loads all five products for the official Mol municipality
|
|
Area. Use --all-municipalities with an explicitly named regional project to
|
|
load every persisted municipality Area. The operator never writes to PostGIS
|
|
or storage directly and never accepts an arbitrary external service URL.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
|
|
DEFAULT_API_URL = "http://127.0.0.1:8000"
|
|
DEFAULT_PROJECT_NAME = "Mol Municipality Workbench"
|
|
DEFAULT_AREA_FRAGMENT = "Gemeente Mol"
|
|
DEFAULT_PRODUCTS = (
|
|
"space_occupation_2025",
|
|
"open_space_2022",
|
|
"population_density_2019",
|
|
"node_value_2022",
|
|
"service_level_2022",
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Provision governed Flemish thematic raster products.")
|
|
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("--area", default=DEFAULT_AREA_FRAGMENT, help="Case-insensitive Area name fragment.")
|
|
parser.add_argument("--products", default=",".join(DEFAULT_PRODUCTS))
|
|
parser.add_argument("--all-municipalities", action="store_true", help="Process every Area whose name starts with 'Gemeente '.")
|
|
parser.add_argument("--force-refresh", action="store_true")
|
|
parser.add_argument("--timeout", type=int, default=900)
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
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 HTTP {response.status_code}: {response.text[:300]}") from exc
|
|
if not response.ok:
|
|
error = payload.get("error") if isinstance(payload, dict) else None
|
|
message = error.get("message") if isinstance(error, dict) else response.text[:300]
|
|
raise RuntimeError(f"GeoIntel HTTP {response.status_code}: {message}")
|
|
return payload.get("data") if isinstance(payload, dict) and "data" in payload else payload
|
|
|
|
|
|
def paged_items(session: requests.Session, url: str, timeout: int) -> list[dict[str, Any]]:
|
|
items: list[dict[str, Any]] = []
|
|
offset = 0
|
|
while True:
|
|
separator = "&" if "?" in url else "?"
|
|
page = unwrap(session.get(f"{url}{separator}limit=200&offset={offset}", timeout=timeout))
|
|
rows = list(page.get("items") or [])
|
|
items.extend(rows)
|
|
total = int(page.get("total") or 0)
|
|
if not rows or len(items) >= total:
|
|
return items
|
|
offset += len(rows)
|
|
|
|
|
|
def geometry_bbox(geometry: dict[str, Any]) -> dict[str, Any]:
|
|
points: list[tuple[float, float]] = []
|
|
|
|
def visit(value: Any) -> None:
|
|
if isinstance(value, list) and len(value) >= 2 and all(isinstance(item, (int, float)) for item in value[:2]):
|
|
points.append((float(value[0]), float(value[1])))
|
|
return
|
|
if isinstance(value, list):
|
|
for item in value:
|
|
visit(item)
|
|
|
|
visit(geometry.get("coordinates"))
|
|
if not points:
|
|
raise RuntimeError("Persisted Area geometry contains no coordinates")
|
|
return {
|
|
"min_x": min(point[0] for point in points),
|
|
"min_y": min(point[1] for point in points),
|
|
"max_x": max(point[0] for point in points),
|
|
"max_y": max(point[1] for point in points),
|
|
"crs": "EPSG:4326",
|
|
}
|
|
|
|
|
|
def find_project(projects: list[dict[str, Any]], name: str) -> dict[str, Any]:
|
|
matches = [project for project in projects if str(project.get("name", "")).casefold() == name.casefold()]
|
|
if len(matches) != 1:
|
|
raise RuntimeError(f"Expected exactly one project named {name!r}, found {len(matches)}")
|
|
return matches[0]
|
|
|
|
|
|
def select_areas(areas: list[dict[str, Any]], fragment: str, all_municipalities: bool) -> list[dict[str, Any]]:
|
|
if all_municipalities:
|
|
selected = [area for area in areas if str(area.get("name", "")).casefold().startswith("gemeente ")]
|
|
else:
|
|
selected = [area for area in areas if fragment.casefold() in str(area.get("name", "")).casefold()]
|
|
if not selected:
|
|
raise RuntimeError("No persisted Area matches the requested scope")
|
|
selected.sort(key=lambda item: str(item.get("name", "")).casefold())
|
|
return selected
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
base_url = args.base_url.rstrip("/")
|
|
requested_products = [value.strip() for value in args.products.split(",") if value.strip()]
|
|
if not requested_products:
|
|
raise RuntimeError("Select at least one thematic raster product")
|
|
session = requests.Session()
|
|
session.headers.update({"User-Agent": "GeoIntel-Thematic-Raster-Operator/1.0"})
|
|
|
|
projects = paged_items(session, f"{base_url}/api/v1/projects", args.timeout)
|
|
project = find_project(projects, args.project_name)
|
|
project_id = str(project["id"])
|
|
areas = paged_items(session, f"{base_url}/api/v1/projects/{project_id}/areas", args.timeout)
|
|
selected_areas = select_areas(areas, args.area, args.all_municipalities)
|
|
registry = unwrap(session.get(f"{base_url}/api/v1/projects/{project_id}/datasets/thematic-raster/products", timeout=args.timeout))
|
|
products = {str(item["key"]): item for item in registry.get("items") or []}
|
|
unknown = sorted(set(requested_products) - set(products))
|
|
if unknown:
|
|
raise RuntimeError(f"Products are not present in the canonical registry: {', '.join(unknown)}")
|
|
|
|
print(json.dumps({
|
|
"status": "planned" if args.dry_run else "running",
|
|
"project_id": project_id,
|
|
"project_name": project["name"],
|
|
"area_count": len(selected_areas),
|
|
"products": requested_products,
|
|
}, ensure_ascii=False))
|
|
if args.dry_run:
|
|
for area in selected_areas:
|
|
print(json.dumps({"area_id": area["id"], "area_name": area["name"], "bbox": geometry_bbox(area["geometry"])}, ensure_ascii=False))
|
|
return 0
|
|
|
|
results: list[dict[str, Any]] = []
|
|
for area in selected_areas:
|
|
bbox = geometry_bbox(area["geometry"])
|
|
for product_key in requested_products:
|
|
acquisition = unwrap(session.post(
|
|
f"{base_url}/api/v1/projects/{project_id}/datasets/thematic-raster/acquire",
|
|
json={
|
|
"bbox": bbox,
|
|
"area_id": area["id"],
|
|
"product_key": product_key,
|
|
"force_refresh": args.force_refresh,
|
|
},
|
|
timeout=args.timeout,
|
|
))
|
|
if acquisition.get("status") != "success" or not acquisition.get("output_dataset_id"):
|
|
raise RuntimeError(f"Acquisition failed for {area['name']} / {product_key}: {acquisition}")
|
|
dataset_id = str(acquisition["output_dataset_id"])
|
|
analysis = unwrap(session.post(
|
|
f"{base_url}/api/v1/projects/{project_id}/datasets/{dataset_id}/raster/thematic/select",
|
|
json={"bbox": bbox, "area_id": area["id"]},
|
|
timeout=args.timeout,
|
|
))
|
|
result = {
|
|
"area_id": area["id"],
|
|
"area_name": area["name"],
|
|
"product_key": product_key,
|
|
"dataset_id": dataset_id,
|
|
"reused": bool((acquisition.get("result_json") or {}).get("reused")),
|
|
"metric": analysis.get("summary"),
|
|
"coverage_ratio": analysis.get("coverage_ratio"),
|
|
}
|
|
results.append(result)
|
|
print(json.dumps(result, ensure_ascii=False))
|
|
|
|
print(json.dumps({"status": "complete", "dataset_count": len(results), "project_id": project_id}, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|