612 lines
24 KiB
Python
612 lines
24 KiB
Python
"""Provision a complete, authoritative Mol municipality map workspace.
|
|
|
|
The script is an explicit operator tool. It fetches the official Mol boundary
|
|
and GRB buildings, writes auditable artifacts, and imports them through the
|
|
existing GeoIntel API. It is never called automatically by application startup.
|
|
"""
|
|
|
|
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
|
|
from pyproj import Transformer
|
|
from shapely.geometry import GeometryCollection, MultiPolygon, Polygon, mapping, shape
|
|
from shapely.ops import transform, unary_union
|
|
from shapely.validation import make_valid
|
|
|
|
|
|
MUNICIPALITY_NAME = "Mol"
|
|
MUNICIPALITY_NIS_CODE = "13025"
|
|
PROJECT_NAME = "Mol Municipality Workbench"
|
|
PROJECT_REGION = "Mol, Kempen"
|
|
AREA_NAME = "Gemeente Mol - officiële grens"
|
|
BOUNDARY_FILENAME = "mol_municipality_boundary.geojson"
|
|
BUILDINGS_FILENAME = "mol_grb_gbg_buildings.geojson"
|
|
MANIFEST_FILENAME = "mol_municipality_manifest.json"
|
|
VRBG_ITEMS_URL = "https://geo.api.vlaanderen.be/VRBG/ogc/features/v1/collections/Refgem/items"
|
|
GRB_GBG_ITEMS_URL = "https://geo.api.vlaanderen.be/GRB/ogc/features/v1/collections/GBG/items"
|
|
VRBG_ATTRIBUTION = "Bron: Voorlopig referentiebestand gemeentegrenzen, Digitaal Vlaanderen"
|
|
GRB_ATTRIBUTION = "Bron: Grootschalig Referentie Bestand Vlaanderen, Digitaal Vlaanderen"
|
|
DEFAULT_OUTPUT_DIR = Path("/app/storage/operator-data/mol-municipality")
|
|
DEFAULT_API_URL = "http://127.0.0.1:8000"
|
|
DEFAULT_PAGE_LIMIT = 1000
|
|
DEFAULT_MAX_FEATURES = 100000
|
|
GEOJSON_CRS = {"type": "name", "properties": {"name": "EPSG:4326"}}
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Fetch and provision the complete official Mol municipality workspace.",
|
|
)
|
|
parser.add_argument(
|
|
"--output-dir",
|
|
type=Path,
|
|
default=Path(os.environ.get("MOL_MUNICIPALITY_OUTPUT_DIR", DEFAULT_OUTPUT_DIR)),
|
|
help="Persistent directory for the boundary, GRB buildings and provenance manifest.",
|
|
)
|
|
parser.add_argument(
|
|
"--base-url",
|
|
default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL),
|
|
help="GeoIntel backend URL. The in-container direct backend avoids reverse-proxy timeouts during large imports.",
|
|
)
|
|
parser.add_argument(
|
|
"--project-name",
|
|
default=PROJECT_NAME,
|
|
help="Exact idempotent project name used for the municipality workspace.",
|
|
)
|
|
parser.add_argument(
|
|
"--page-limit",
|
|
type=int,
|
|
default=int(os.environ.get("MOL_GRB_PAGE_LIMIT", str(DEFAULT_PAGE_LIMIT))),
|
|
help="GRB OGC API page size.",
|
|
)
|
|
parser.add_argument(
|
|
"--max-features",
|
|
type=int,
|
|
default=int(os.environ.get("MOL_GRB_MAX_FEATURES", str(DEFAULT_MAX_FEATURES))),
|
|
help="Safety cap. The script fails instead of writing a truncated municipality dataset.",
|
|
)
|
|
parser.add_argument(
|
|
"--request-timeout",
|
|
type=int,
|
|
default=int(os.environ.get("MOL_MUNICIPALITY_REQUEST_TIMEOUT", "180")),
|
|
help="Timeout per official source request in seconds.",
|
|
)
|
|
parser.add_argument(
|
|
"--import-timeout",
|
|
type=int,
|
|
default=int(os.environ.get("MOL_MUNICIPALITY_IMPORT_TIMEOUT", "1800")),
|
|
help="Timeout for each GeoIntel API import request in seconds.",
|
|
)
|
|
parser.add_argument("--force", action="store_true", help="Refetch official source artifacts even when complete local artifacts exist.")
|
|
parser.add_argument("--fetch-only", action="store_true", help="Prepare artifacts without creating or updating the GeoIntel workspace.")
|
|
return parser.parse_args()
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def write_json(path: Path, payload: dict[str, Any], *, pretty: bool = False) -> None:
|
|
path.write_text(
|
|
json.dumps(
|
|
payload,
|
|
ensure_ascii=False,
|
|
indent=2 if pretty else None,
|
|
separators=None if pretty else (",", ":"),
|
|
sort_keys=pretty,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def next_page_url(payload: dict[str, Any]) -> str | None:
|
|
links = payload.get("links") or []
|
|
for link in links:
|
|
if link.get("rel") == "next" and "geo+json" in str(link.get("type", "")).lower():
|
|
return str(link["href"])
|
|
for link in links:
|
|
if link.get("rel") == "next" and link.get("href"):
|
|
return str(link["href"])
|
|
return None
|
|
|
|
|
|
def normalize_polygonal(geometry):
|
|
if geometry.is_empty:
|
|
return None
|
|
if not geometry.is_valid:
|
|
geometry = make_valid(geometry)
|
|
if isinstance(geometry, (Polygon, MultiPolygon)):
|
|
return geometry
|
|
if isinstance(geometry, GeometryCollection):
|
|
polygons = [part for part in geometry.geoms if isinstance(part, (Polygon, MultiPolygon)) and not part.is_empty]
|
|
if polygons:
|
|
merged = unary_union(polygons)
|
|
return merged if isinstance(merged, (Polygon, MultiPolygon)) and not merged.is_empty else None
|
|
return None
|
|
|
|
|
|
def fetch_mol_boundary(session: requests.Session, timeout: int) -> tuple[dict[str, Any], Any, str]:
|
|
params = {
|
|
"f": "application/geo+json",
|
|
"limit": "10",
|
|
"filter": "NAAM='Mol'",
|
|
"filter-lang": "cql2-text",
|
|
}
|
|
response = session.get(VRBG_ITEMS_URL, params=params, timeout=timeout)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
matches = [
|
|
feature
|
|
for feature in payload.get("features") or []
|
|
if str((feature.get("properties") or {}).get("NAAM", "")).casefold() == MUNICIPALITY_NAME.casefold()
|
|
]
|
|
if len(matches) != 1:
|
|
raise RuntimeError(f"Expected exactly one official Mol boundary, received {len(matches)}")
|
|
|
|
source_feature = matches[0]
|
|
properties = dict(source_feature.get("properties") or {})
|
|
if str(properties.get("NISCODE")) != MUNICIPALITY_NIS_CODE:
|
|
raise RuntimeError(
|
|
f"Official Mol boundary NIS code drifted: expected {MUNICIPALITY_NIS_CODE}, received {properties.get('NISCODE')}"
|
|
)
|
|
boundary = normalize_polygonal(shape(source_feature.get("geometry")))
|
|
if boundary is None or not boundary.is_valid:
|
|
raise RuntimeError("Official Mol boundary is empty, non-polygonal or invalid")
|
|
|
|
source_url = response.url
|
|
properties.update(
|
|
{
|
|
"source_name": "vrbg",
|
|
"source_feature_id": str(source_feature.get("id") or MUNICIPALITY_NIS_CODE),
|
|
"layer_type": "municipality_boundary",
|
|
"authority_level": "authoritative",
|
|
"coverage_scope": "municipality",
|
|
"municipality": MUNICIPALITY_NAME,
|
|
"nis_code": MUNICIPALITY_NIS_CODE,
|
|
"attribution": VRBG_ATTRIBUTION,
|
|
"source_url": source_url,
|
|
}
|
|
)
|
|
feature = {
|
|
"type": "Feature",
|
|
"id": str(source_feature.get("id") or f"Refgem.{MUNICIPALITY_NIS_CODE}"),
|
|
"geometry": mapping(boundary),
|
|
"properties": properties,
|
|
}
|
|
return feature, boundary, source_url
|
|
|
|
|
|
def iter_grb_pages(
|
|
session: requests.Session,
|
|
boundary_bounds: tuple[float, float, float, float],
|
|
*,
|
|
page_limit: int,
|
|
timeout: int,
|
|
) -> Iterable[tuple[dict[str, Any], str]]:
|
|
params = {
|
|
"f": "application/geo+json",
|
|
"limit": str(page_limit),
|
|
"bbox": ",".join(f"{value:.8f}" for value in boundary_bounds),
|
|
}
|
|
url: str | None = GRB_GBG_ITEMS_URL
|
|
seen_urls: set[str] = set()
|
|
first_request = True
|
|
while url:
|
|
if url in seen_urls:
|
|
raise RuntimeError(f"GRB pagination loop detected: {url}")
|
|
seen_urls.add(url)
|
|
response = session.get(url, params=params if first_request else None, timeout=timeout)
|
|
first_request = False
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
yield payload, response.url
|
|
url = next_page_url(payload)
|
|
|
|
|
|
def build_municipality_buildings(
|
|
pages: Iterable[tuple[dict[str, Any], str]],
|
|
boundary,
|
|
*,
|
|
max_features: int,
|
|
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
|
features: list[dict[str, Any]] = []
|
|
source_urls: list[str] = []
|
|
seen_ids: set[str] = set()
|
|
bbox_feature_count = 0
|
|
outside_boundary_count = 0
|
|
clipped_at_boundary_count = 0
|
|
|
|
for page, source_url in pages:
|
|
source_urls.append(source_url)
|
|
for source_feature in page.get("features") or []:
|
|
feature_id = str(source_feature.get("id") or "")
|
|
if not feature_id:
|
|
feature_id = hashlib.sha256(
|
|
json.dumps(source_feature.get("geometry"), sort_keys=True).encode("utf-8")
|
|
).hexdigest()
|
|
if feature_id in seen_ids:
|
|
continue
|
|
seen_ids.add(feature_id)
|
|
bbox_feature_count += 1
|
|
|
|
source_geometry = normalize_polygonal(shape(source_feature.get("geometry")))
|
|
if source_geometry is None or not source_geometry.intersects(boundary):
|
|
outside_boundary_count += 1
|
|
continue
|
|
within_boundary = source_geometry.within(boundary)
|
|
clipped_geometry = source_geometry if within_boundary else normalize_polygonal(source_geometry.intersection(boundary))
|
|
if clipped_geometry is None:
|
|
outside_boundary_count += 1
|
|
continue
|
|
if not within_boundary:
|
|
clipped_at_boundary_count += 1
|
|
if len(features) >= max_features:
|
|
raise RuntimeError(
|
|
f"Mol contains more than the configured {max_features} GRB features; refusing a truncated municipality dataset"
|
|
)
|
|
|
|
properties = dict(source_feature.get("properties") or {})
|
|
properties.update(
|
|
{
|
|
"source_name": "grb",
|
|
"source_feature_id": feature_id,
|
|
"reference_layer_name": "buildings",
|
|
"layer_type": "building",
|
|
"authority_level": "authoritative",
|
|
"coverage_scope": "municipality",
|
|
"municipality": MUNICIPALITY_NAME,
|
|
"nis_code": MUNICIPALITY_NIS_CODE,
|
|
"clipped_to_municipality": not within_boundary,
|
|
"attribution": GRB_ATTRIBUTION,
|
|
}
|
|
)
|
|
features.append(
|
|
{
|
|
"type": "Feature",
|
|
"id": feature_id,
|
|
"geometry": mapping(clipped_geometry),
|
|
"properties": properties,
|
|
}
|
|
)
|
|
|
|
if not features:
|
|
raise RuntimeError("No GRB buildings intersect the official Mol municipality boundary")
|
|
return features, {
|
|
"pages_fetched": len(source_urls),
|
|
"source_urls": source_urls,
|
|
"bbox_feature_count": bbox_feature_count,
|
|
"municipality_feature_count": len(features),
|
|
"outside_boundary_count": outside_boundary_count,
|
|
"clipped_at_boundary_count": clipped_at_boundary_count,
|
|
"reference_truncated": False,
|
|
}
|
|
|
|
|
|
def municipality_area_km2(boundary) -> float:
|
|
transformer = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True)
|
|
return float(transform(transformer.transform, boundary).area / 1_000_000)
|
|
|
|
|
|
def prepare_artifacts(args: argparse.Namespace) -> tuple[Path, Path, dict[str, Any]]:
|
|
output_dir: Path = args.output_dir
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
boundary_path = output_dir / BOUNDARY_FILENAME
|
|
buildings_path = output_dir / BUILDINGS_FILENAME
|
|
manifest_path = output_dir / MANIFEST_FILENAME
|
|
|
|
if not args.force and boundary_path.exists() and buildings_path.exists() and manifest_path.exists():
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
if manifest.get("status") == "complete" and int(manifest.get("municipality_feature_count") or 0) > 0:
|
|
return boundary_path, buildings_path, manifest
|
|
|
|
if args.page_limit <= 0 or args.max_features <= 0:
|
|
raise RuntimeError("--page-limit and --max-features must be positive integers")
|
|
|
|
with requests.Session() as session:
|
|
session.headers.update({"User-Agent": "GeoIntel-Mol-Municipality-Operator/1.0"})
|
|
boundary_feature, boundary, boundary_source_url = fetch_mol_boundary(session, args.request_timeout)
|
|
buildings, building_summary = build_municipality_buildings(
|
|
iter_grb_pages(
|
|
session,
|
|
boundary.bounds,
|
|
page_limit=args.page_limit,
|
|
timeout=args.request_timeout,
|
|
),
|
|
boundary,
|
|
max_features=args.max_features,
|
|
)
|
|
|
|
generated_at = utc_now()
|
|
boundary_payload = {
|
|
"type": "FeatureCollection",
|
|
"name": "Official municipality boundary - Mol",
|
|
"crs": GEOJSON_CRS,
|
|
"features": [boundary_feature],
|
|
"source": "Digitaal Vlaanderen VRBG OGC API Features collection Refgem",
|
|
"source_url": boundary_source_url,
|
|
"attribution": VRBG_ATTRIBUTION,
|
|
"municipality": MUNICIPALITY_NAME,
|
|
"nis_code": MUNICIPALITY_NIS_CODE,
|
|
"coverage_scope": "municipality",
|
|
"generated_at": generated_at,
|
|
}
|
|
buildings_payload = {
|
|
"type": "FeatureCollection",
|
|
"name": "GRB buildings - complete municipality Mol",
|
|
"crs": GEOJSON_CRS,
|
|
"features": buildings,
|
|
"source": "Digitaal Vlaanderen GRB OGC API Features collection GBG",
|
|
"source_url": building_summary["source_urls"][0],
|
|
"attribution": GRB_ATTRIBUTION,
|
|
"municipality": MUNICIPALITY_NAME,
|
|
"nis_code": MUNICIPALITY_NIS_CODE,
|
|
"coverage_scope": "municipality",
|
|
"reference_truncated": False,
|
|
"generated_at": generated_at,
|
|
}
|
|
write_json(boundary_path, boundary_payload)
|
|
write_json(buildings_path, buildings_payload)
|
|
|
|
manifest = {
|
|
"schema_version": 1,
|
|
"status": "complete",
|
|
"generated_at": generated_at,
|
|
"municipality": MUNICIPALITY_NAME,
|
|
"nis_code": MUNICIPALITY_NIS_CODE,
|
|
"area_km2": municipality_area_km2(boundary),
|
|
"wgs84_bbox": list(boundary.bounds),
|
|
"boundary_path": str(boundary_path),
|
|
"boundary_sha256": sha256_file(boundary_path),
|
|
"buildings_path": str(buildings_path),
|
|
"buildings_sha256": sha256_file(buildings_path),
|
|
"page_limit": args.page_limit,
|
|
"max_features": args.max_features,
|
|
"boundary_source_url": boundary_source_url,
|
|
"attribution": {"boundary": VRBG_ATTRIBUTION, "buildings": GRB_ATTRIBUTION},
|
|
**building_summary,
|
|
}
|
|
write_json(manifest_path, manifest, pretty=True)
|
|
return boundary_path, buildings_path, manifest
|
|
|
|
|
|
def response_data(response: requests.Response) -> Any:
|
|
try:
|
|
payload = response.json()
|
|
except ValueError as exc:
|
|
raise RuntimeError(f"GeoIntel API returned non-JSON response ({response.status_code}): {response.text[:500]}") from exc
|
|
if not response.ok:
|
|
raise RuntimeError(f"GeoIntel API request failed ({response.status_code}): {json.dumps(payload, ensure_ascii=False)[:1000]}")
|
|
if not isinstance(payload, dict) or "data" not in payload:
|
|
raise RuntimeError("GeoIntel API response does not use the canonical data envelope")
|
|
return payload["data"]
|
|
|
|
|
|
def find_or_create_project(session: requests.Session, base_url: str, project_name: str, timeout: int) -> dict[str, Any]:
|
|
projects = response_data(session.get(f"{base_url}/api/v1/projects", params={"limit": 200}, timeout=timeout))
|
|
existing = next((item for item in projects.get("items") or [] if item.get("name") == project_name), None)
|
|
if existing:
|
|
return existing
|
|
return response_data(
|
|
session.post(
|
|
f"{base_url}/api/v1/projects",
|
|
json={
|
|
"name": project_name,
|
|
"description": (
|
|
"Complete municipality workspace for Mol using the official Digitaal Vlaanderen municipality boundary "
|
|
"and the full GRB GBG building reference layer clipped to NIS 13025."
|
|
),
|
|
"region": PROJECT_REGION,
|
|
},
|
|
timeout=timeout,
|
|
)
|
|
)
|
|
|
|
|
|
def find_or_create_area(
|
|
session: requests.Session,
|
|
base_url: str,
|
|
project_id: str,
|
|
boundary_path: Path,
|
|
timeout: int,
|
|
) -> dict[str, Any]:
|
|
areas = response_data(
|
|
session.get(f"{base_url}/api/v1/projects/{project_id}/areas", params={"limit": 200}, timeout=timeout)
|
|
)
|
|
existing = next((item for item in areas.get("items") or [] if item.get("name") == AREA_NAME), None)
|
|
if existing:
|
|
return existing
|
|
boundary_payload = json.loads(boundary_path.read_text(encoding="utf-8"))
|
|
geometry = boundary_payload["features"][0]["geometry"]
|
|
return response_data(
|
|
session.post(
|
|
f"{base_url}/api/v1/projects/{project_id}/areas",
|
|
json={"name": AREA_NAME, "crs": "EPSG:4326", "geometry": geometry},
|
|
timeout=timeout,
|
|
)
|
|
)
|
|
|
|
|
|
def upload_dataset(
|
|
session: requests.Session,
|
|
base_url: str,
|
|
project_id: str,
|
|
area_id: str,
|
|
path: Path,
|
|
*,
|
|
dataset_role: str,
|
|
source_name: str,
|
|
reference_layer_name: str | None,
|
|
source_metadata: dict[str, Any],
|
|
provenance_metadata: dict[str, Any],
|
|
timeout: int,
|
|
) -> dict[str, Any]:
|
|
form = {
|
|
"dataset_type": "vector",
|
|
"source": "operator_official_import",
|
|
"dataset_role": dataset_role,
|
|
"source_name": source_name,
|
|
"source_metadata_json": json.dumps(source_metadata, ensure_ascii=False),
|
|
"provenance_metadata_json": json.dumps(provenance_metadata, ensure_ascii=False),
|
|
"area_id": area_id,
|
|
}
|
|
if reference_layer_name:
|
|
form["reference_layer_name"] = reference_layer_name
|
|
with path.open("rb") as handle:
|
|
response = session.post(
|
|
f"{base_url}/api/v1/projects/{project_id}/datasets/upload",
|
|
data=form,
|
|
files={"file": (path.name, handle, "application/geo+json")},
|
|
timeout=timeout,
|
|
)
|
|
return response_data(response)
|
|
|
|
|
|
def provision_workspace(
|
|
args: argparse.Namespace,
|
|
boundary_path: Path,
|
|
buildings_path: Path,
|
|
manifest: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
base_url = args.base_url.rstrip("/")
|
|
with requests.Session() as session:
|
|
project = find_or_create_project(session, base_url, args.project_name, args.import_timeout)
|
|
project_id = str(project["id"])
|
|
area = find_or_create_area(session, base_url, project_id, boundary_path, args.import_timeout)
|
|
area_id = str(area["id"])
|
|
dataset_list = response_data(
|
|
session.get(
|
|
f"{base_url}/api/v1/projects/{project_id}/datasets",
|
|
params={"limit": 200},
|
|
timeout=args.import_timeout,
|
|
)
|
|
)
|
|
datasets = list(dataset_list.get("items") or [])
|
|
|
|
common_provenance = {
|
|
"operator_tool": "provision_mol_municipality_workspace.py",
|
|
"operator_explicit_fetch": True,
|
|
"municipality": MUNICIPALITY_NAME,
|
|
"nis_code": MUNICIPALITY_NIS_CODE,
|
|
"coverage_scope": "municipality",
|
|
"manifest_path": str(args.output_dir / MANIFEST_FILENAME),
|
|
"manifest_generated_at": manifest["generated_at"],
|
|
"reference_truncated": False,
|
|
}
|
|
|
|
building_dataset = next((item for item in datasets if item.get("original_filename") == buildings_path.name), None)
|
|
if not building_dataset:
|
|
building_dataset = upload_dataset(
|
|
session,
|
|
base_url,
|
|
project_id,
|
|
area_id,
|
|
buildings_path,
|
|
dataset_role="reference",
|
|
source_name="grb",
|
|
reference_layer_name="buildings",
|
|
source_metadata={
|
|
"provider": "Digitaal Vlaanderen",
|
|
"collection": "GRB/GBG",
|
|
"authority_level": "authoritative",
|
|
"coverage_scope": "municipality",
|
|
"municipality": MUNICIPALITY_NAME,
|
|
"nis_code": MUNICIPALITY_NIS_CODE,
|
|
"feature_count": manifest["municipality_feature_count"],
|
|
"pages_fetched": manifest["pages_fetched"],
|
|
"attribution": GRB_ATTRIBUTION,
|
|
},
|
|
provenance_metadata={
|
|
**common_provenance,
|
|
"source_url": GRB_GBG_ITEMS_URL,
|
|
"artifact_sha256": manifest["buildings_sha256"],
|
|
},
|
|
timeout=args.import_timeout,
|
|
)
|
|
|
|
boundary_dataset = next((item for item in datasets if item.get("original_filename") == boundary_path.name), None)
|
|
if not boundary_dataset:
|
|
boundary_dataset = upload_dataset(
|
|
session,
|
|
base_url,
|
|
project_id,
|
|
area_id,
|
|
boundary_path,
|
|
dataset_role="source",
|
|
source_name="vrbg",
|
|
reference_layer_name=None,
|
|
source_metadata={
|
|
"provider": "Digitaal Vlaanderen",
|
|
"collection": "VRBG/Refgem",
|
|
"authority_level": "authoritative",
|
|
"coverage_scope": "municipality",
|
|
"layer_type": "municipality_boundary",
|
|
"municipality": MUNICIPALITY_NAME,
|
|
"nis_code": MUNICIPALITY_NIS_CODE,
|
|
"attribution": VRBG_ATTRIBUTION,
|
|
},
|
|
provenance_metadata={
|
|
**common_provenance,
|
|
"source_url": manifest["boundary_source_url"],
|
|
"artifact_sha256": manifest["boundary_sha256"],
|
|
},
|
|
timeout=args.import_timeout,
|
|
)
|
|
|
|
return {
|
|
"project_id": project_id,
|
|
"project_name": project.get("name"),
|
|
"area_id": area_id,
|
|
"area_name": area.get("name"),
|
|
"boundary_dataset_id": str(boundary_dataset["id"]),
|
|
"boundary_feature_count": boundary_dataset.get("feature_count"),
|
|
"building_dataset_id": str(building_dataset["id"]),
|
|
"building_feature_count": building_dataset.get("feature_count"),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
try:
|
|
boundary_path, buildings_path, manifest = prepare_artifacts(args)
|
|
workspace = None if args.fetch_only else provision_workspace(args, boundary_path, buildings_path, manifest)
|
|
except (OSError, RuntimeError, requests.RequestException, ValueError, KeyError) as exc:
|
|
print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr)
|
|
return 1
|
|
|
|
result = {
|
|
"status": "ok",
|
|
"mode": "fetch_only" if args.fetch_only else "provisioned",
|
|
"manifest_path": str(args.output_dir / MANIFEST_FILENAME),
|
|
"municipality": MUNICIPALITY_NAME,
|
|
"nis_code": MUNICIPALITY_NIS_CODE,
|
|
"area_km2": manifest["area_km2"],
|
|
"wgs84_bbox": manifest["wgs84_bbox"],
|
|
"municipality_feature_count": manifest["municipality_feature_count"],
|
|
"pages_fetched": manifest["pages_fetched"],
|
|
"reference_truncated": manifest["reference_truncated"],
|
|
"workspace": workspace,
|
|
}
|
|
print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|