1242 lines
51 KiB
Python
1242 lines
51 KiB
Python
"""Provision a privacy-minimized Buildings and Addresses Register snapshot.
|
|
|
|
The explicit operator reads the official Digitaal Vlaanderen OGC API Features
|
|
collections for buildings, building units and addresses. It retains raw,
|
|
checksummed source pages as operator evidence, but the queryable Dataset only
|
|
contains building polygons and aggregate relation counts. Street names, house
|
|
numbers, box numbers and complete address labels are never copied into
|
|
``vector_features``.
|
|
|
|
Building-unit relations use the official ``GebouwObjectId``. Address-to-
|
|
building relations use the official address position and are classified as an
|
|
exact unit-position match, an unambiguous containing-building match, ambiguous
|
|
or unmatched. GRB reconciliation uses the latest persisted regional GRB
|
|
manifest and records exact/high-IoU/review/unmatched outcomes per building.
|
|
All persistence flows through the canonical Dataset upload route.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
import unicodedata
|
|
from collections import Counter, defaultdict
|
|
from datetime import date, datetime, time, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
import requests
|
|
from pyproj import Transformer
|
|
from requests.adapters import HTTPAdapter
|
|
from shapely.geometry import MultiPolygon, Point, Polygon, mapping, shape
|
|
from shapely.ops import transform as transform_geometry
|
|
from shapely.ops import unary_union
|
|
from shapely.strtree import STRtree
|
|
from shapely.validation import make_valid
|
|
from urllib3.util.retry import Retry
|
|
|
|
|
|
BUILDING_ITEMS_URL = (
|
|
"https://geo.api.vlaanderen.be/Gebouwenregister/ogc/features/v1/collections/Gebouw/items"
|
|
)
|
|
UNIT_ITEMS_URL = (
|
|
"https://geo.api.vlaanderen.be/Gebouwenregister/ogc/features/v1/collections/Gebouweenheid/items"
|
|
)
|
|
ADDRESS_ITEMS_URL = (
|
|
"https://geo.api.vlaanderen.be/Adressenregister/ogc/features/v1/collections/Adres/items"
|
|
)
|
|
CATALOG_URL = "https://www.vlaanderen.be/datavindplaats/catalogus/gebouwen-en-adressenregister"
|
|
BUILDING_CATALOG_URL = "https://www.vlaanderen.be/datavindplaats/catalogus/gebouwenregister"
|
|
CHANGE_NOTICE_URL = (
|
|
"https://www.vlaanderen.be/digitaal-vlaanderen/"
|
|
"belangrijke-wijzigingen-bij-het-gebouwen-en-adressenregister-in-de-zomer-van-2026"
|
|
)
|
|
ATTRIBUTION = "Bron: Digitaal Vlaanderen"
|
|
SOURCE_NAME = "digitaal_vlaanderen_buildings_addresses_register"
|
|
DEFAULT_PROJECT_NAME = "Kempen Regional Workbench"
|
|
DEFAULT_AREA_NAME = "Gemeente Mol - officiele grens"
|
|
DEFAULT_API_URL = "http://127.0.0.1:8000"
|
|
DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-evidence/buildings-addresses-register")
|
|
DEFAULT_PAGE_LIMIT = 1000
|
|
DEFAULT_MAX_BUILDINGS = 150_000
|
|
DEFAULT_MAX_UNITS = 200_000
|
|
DEFAULT_MAX_ADDRESSES = 200_000
|
|
SCHEMA_VERSION = 1
|
|
TO_LAMBERT72 = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True)
|
|
TO_WGS84 = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True)
|
|
|
|
BUILDING_STATUS_KEYS = {
|
|
"gerealiseerd": "realized",
|
|
"gepland": "planned",
|
|
"nietgerealiseerd": "not_realized",
|
|
"inaanbouw": "under_construction",
|
|
"gehistoreerd": "historical",
|
|
}
|
|
BUILDING_STATUS_LABELS = {
|
|
"realized": "Gerealiseerd",
|
|
"planned": "Gepland",
|
|
"not_realized": "Niet gerealiseerd",
|
|
"under_construction": "In aanbouw",
|
|
"historical": "Gehistoreerd",
|
|
"unknown": "Onbekend",
|
|
}
|
|
UNIT_STATUS_KEYS = {
|
|
"gerealiseerd": "realized",
|
|
"gepland": "planned",
|
|
"nietgerealiseerd": "not_realized",
|
|
"gehistoreerd": "historical",
|
|
}
|
|
ADDRESS_STATUS_KEYS = {
|
|
"ingebruik": "in_use",
|
|
"voorgesteld": "proposed",
|
|
"gehistoreerd": "historical",
|
|
"afgekeurd": "rejected",
|
|
"inonderzoek": "under_review",
|
|
}
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Provision an official Buildings and Addresses Register snapshot for a persisted Area."
|
|
)
|
|
parser.add_argument("--project-name", default=DEFAULT_PROJECT_NAME)
|
|
parser.add_argument("--area-name", default=DEFAULT_AREA_NAME)
|
|
parser.add_argument("--observed-date", type=date.fromisoformat, default=date.today())
|
|
parser.add_argument("--base-url", default=DEFAULT_API_URL)
|
|
parser.add_argument("--output-root", type=Path, default=DEFAULT_OUTPUT_ROOT)
|
|
parser.add_argument("--page-limit", type=int, default=DEFAULT_PAGE_LIMIT)
|
|
parser.add_argument("--max-buildings", type=int, default=DEFAULT_MAX_BUILDINGS)
|
|
parser.add_argument("--max-units", type=int, default=DEFAULT_MAX_UNITS)
|
|
parser.add_argument("--max-addresses", type=int, default=DEFAULT_MAX_ADDRESSES)
|
|
parser.add_argument("--request-timeout", type=int, default=180)
|
|
parser.add_argument("--api-timeout", type=int, default=300)
|
|
parser.add_argument("--force", action="store_true")
|
|
parser.add_argument("--fetch-only", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def observed_at(value: date) -> str:
|
|
return datetime.combine(value, time.min, tzinfo=timezone.utc).isoformat()
|
|
|
|
|
|
def sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def write_bytes_atomic(path: Path, value: bytes) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(f"{path.suffix}.partial")
|
|
temporary.write_bytes(value)
|
|
temporary.replace(path)
|
|
|
|
|
|
def write_json_atomic(path: Path, value: Any, *, pretty: bool = False) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(f"{path.suffix}.partial")
|
|
temporary.write_text(
|
|
json.dumps(
|
|
value,
|
|
ensure_ascii=False,
|
|
indent=2 if pretty else None,
|
|
separators=None if pretty else (",", ":"),
|
|
sort_keys=pretty,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
temporary.replace(path)
|
|
|
|
|
|
def source_session() -> requests.Session:
|
|
retry = Retry(
|
|
total=5,
|
|
connect=5,
|
|
read=5,
|
|
status=5,
|
|
backoff_factor=1.0,
|
|
status_forcelist=(429, 500, 502, 503, 504),
|
|
allowed_methods=frozenset({"GET"}),
|
|
raise_on_status=True,
|
|
)
|
|
adapter = HTTPAdapter(max_retries=retry)
|
|
session = requests.Session()
|
|
session.headers.update({"User-Agent": "GeoIntel-Buildings-Addresses-Register-Operator/1.0"})
|
|
session.mount("https://", adapter)
|
|
session.mount("http://", adapter)
|
|
return session
|
|
|
|
|
|
def response_data(response: requests.Response) -> Any:
|
|
try:
|
|
payload = response.json()
|
|
except ValueError as exc:
|
|
raise RuntimeError(
|
|
f"GeoIntel API returned non-JSON ({response.status_code}): {response.text[:300]}"
|
|
) from exc
|
|
if not response.ok:
|
|
raise RuntimeError(
|
|
f"GeoIntel API failed ({response.status_code}): {json.dumps(payload, ensure_ascii=False)[:800]}"
|
|
)
|
|
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 paginated_items(session: requests.Session, url: str, *, timeout: int) -> list[dict[str, Any]]:
|
|
items: list[dict[str, Any]] = []
|
|
offset = 0
|
|
total: int | None = None
|
|
while total is None or offset < total:
|
|
page = response_data(session.get(url, params={"limit": 200, "offset": offset}, timeout=timeout))
|
|
page_items = list(page.get("items") or [])
|
|
page_total = int(page.get("total") or 0)
|
|
if total is None:
|
|
total = page_total
|
|
elif page_total != total:
|
|
raise RuntimeError("GeoIntel pagination total changed while reading the workspace")
|
|
items.extend(page_items)
|
|
if not page_items:
|
|
break
|
|
offset += len(page_items)
|
|
if total is not None and len(items) != total:
|
|
raise RuntimeError(f"GeoIntel pagination returned {len(items)} of {total} items")
|
|
return items
|
|
|
|
|
|
def polygonal_geometry(geometry):
|
|
if geometry is None or geometry.is_empty:
|
|
return None
|
|
if not geometry.is_valid:
|
|
geometry = make_valid(geometry)
|
|
polygons: list[Polygon] = []
|
|
|
|
def collect(candidate) -> None:
|
|
if candidate is None or candidate.is_empty:
|
|
return
|
|
if isinstance(candidate, Polygon):
|
|
polygons.append(candidate)
|
|
elif isinstance(candidate, MultiPolygon):
|
|
polygons.extend(part for part in candidate.geoms if not part.is_empty)
|
|
elif hasattr(candidate, "geoms"):
|
|
for part in candidate.geoms:
|
|
collect(part)
|
|
|
|
collect(geometry)
|
|
if not polygons:
|
|
return None
|
|
result = unary_union(polygons)
|
|
if not result.is_valid:
|
|
result = make_valid(result)
|
|
return result if not result.is_empty and result.is_valid else None
|
|
|
|
|
|
def normalize_area_name(value: str) -> str:
|
|
return unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii").casefold()
|
|
|
|
|
|
def area_storage_key(value: str) -> str:
|
|
normalized = normalize_area_name(value)
|
|
if normalized.startswith("gemeente "):
|
|
municipality = normalized.removeprefix("gemeente ").split(" - ", maxsplit=1)[0].strip()
|
|
if municipality:
|
|
return "-".join(municipality.split())
|
|
key = "-".join(part for part in compact_key(value).split() if part)
|
|
return key or "area"
|
|
|
|
|
|
def boundary_checksum(boundary_wgs84) -> str:
|
|
return sha256_bytes(
|
|
json.dumps(mapping(boundary_wgs84), sort_keys=True).encode("utf-8")
|
|
)
|
|
|
|
|
|
def locate_workspace(
|
|
session: requests.Session,
|
|
base_url: str,
|
|
project_name: str,
|
|
area_name: str,
|
|
timeout: int,
|
|
) -> tuple[str, str, Any, list[dict[str, Any]]]:
|
|
projects = paginated_items(session, f"{base_url}/api/v1/projects", timeout=timeout)
|
|
project = next((item for item in projects if item.get("name") == project_name), None)
|
|
if not project:
|
|
raise RuntimeError(f"Project {project_name!r} is missing")
|
|
project_id = str(project["id"])
|
|
areas = paginated_items(session, f"{base_url}/api/v1/projects/{project_id}/areas", timeout=timeout)
|
|
target = normalize_area_name(area_name)
|
|
area = next((item for item in areas if normalize_area_name(str(item.get("name") or "")) == target), None)
|
|
if not area or not area.get("geometry"):
|
|
raise RuntimeError(f"Persisted Area {area_name!r} with geometry is missing")
|
|
boundary = polygonal_geometry(shape(area["geometry"]))
|
|
if boundary is None:
|
|
raise RuntimeError("Persisted Area geometry is invalid or not polygonal")
|
|
datasets = paginated_items(session, f"{base_url}/api/v1/projects/{project_id}/datasets", timeout=timeout)
|
|
return project_id, str(area["id"]), boundary, datasets
|
|
|
|
|
|
def next_page_url(payload: dict[str, Any]) -> str | None:
|
|
for link in payload.get("links") or []:
|
|
if str(link.get("rel") or "").lower() == "next" and link.get("href"):
|
|
return str(link["href"])
|
|
return None
|
|
|
|
|
|
def fetch_collection(
|
|
session: requests.Session,
|
|
*,
|
|
url: str,
|
|
name: str,
|
|
bbox: tuple[float, float, float, float],
|
|
raw_dir: Path,
|
|
page_limit: int,
|
|
max_features: int,
|
|
timeout: int,
|
|
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
|
request_url: str | None = url
|
|
base_params: dict[str, str] = {
|
|
"f": "application/geo+json",
|
|
"bbox": ",".join(f"{value:.8f}" for value in bbox),
|
|
"limit": str(page_limit),
|
|
}
|
|
params: dict[str, str] | None = dict(base_params)
|
|
seen_urls: set[str] = set()
|
|
seen_ids: set[str] = set()
|
|
features: list[dict[str, Any]] = []
|
|
pages: list[dict[str, Any]] = []
|
|
duplicate_count = 0
|
|
retrieved_count = 0
|
|
pagination_fallback_count = 0
|
|
while request_url:
|
|
request_key = requests.Request("GET", request_url, params=params).prepare().url or request_url
|
|
if request_key in seen_urls:
|
|
raise RuntimeError(f"{name} pagination loop detected")
|
|
seen_urls.add(request_key)
|
|
response = session.get(request_url, params=params, timeout=timeout)
|
|
params = None
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
if not isinstance(payload, dict) or payload.get("type") != "FeatureCollection":
|
|
raise RuntimeError(f"{name} returned an invalid FeatureCollection")
|
|
raw_bytes = response.content
|
|
page_path = raw_dir / f"{name}_page_{len(pages) + 1:05d}.json"
|
|
write_bytes_atomic(page_path, raw_bytes)
|
|
page_features = list(payload.get("features") or [])
|
|
retrieved_count += len(page_features)
|
|
pages.append(
|
|
{
|
|
"path": str(page_path.relative_to(raw_dir.parent)),
|
|
"sha256": sha256_bytes(raw_bytes),
|
|
"size_bytes": len(raw_bytes),
|
|
"feature_count": len(page_features),
|
|
"source_url": response.url,
|
|
}
|
|
)
|
|
for feature in page_features:
|
|
identity = str(feature.get("id") or (feature.get("properties") or {}).get("ObjectId") or "")
|
|
if not identity:
|
|
raise RuntimeError(f"{name} returned a feature without stable identity")
|
|
if identity in seen_ids:
|
|
duplicate_count += 1
|
|
continue
|
|
seen_ids.add(identity)
|
|
features.append(feature)
|
|
if len(features) > max_features:
|
|
raise RuntimeError(
|
|
f"{name} exceeds the {max_features} feature safety limit; refusing truncated output"
|
|
)
|
|
request_url = next_page_url(payload)
|
|
if request_url is None and len(page_features) == page_limit:
|
|
# The production Address Register currently stops emitting `next`
|
|
# after a service-side window while higher startIndex values remain
|
|
# available. A full page is therefore not proof of completion.
|
|
request_url = url
|
|
params = {**base_params, "startIndex": str(retrieved_count)}
|
|
pagination_fallback_count += 1
|
|
if not features:
|
|
raise RuntimeError(f"{name} returned no source features for the Area bbox")
|
|
return features, {
|
|
"collection": name,
|
|
"page_count": len(pages),
|
|
"bbox_feature_count": len(features),
|
|
"duplicate_count": duplicate_count,
|
|
"pagination_fallback_count": pagination_fallback_count,
|
|
"pages": pages,
|
|
}
|
|
|
|
|
|
def compact_key(value: Any) -> str:
|
|
normalized = unicodedata.normalize("NFKD", str(value or "")).encode("ascii", "ignore").decode("ascii")
|
|
return "".join(character.lower() for character in normalized if character.isalnum())
|
|
|
|
|
|
def normalize_buildings(
|
|
source_features: Iterable[dict[str, Any]],
|
|
boundary_lambert72,
|
|
) -> tuple[dict[str, dict[str, Any]], dict[str, int]]:
|
|
records: dict[str, dict[str, Any]] = {}
|
|
rejected = 0
|
|
clipped = 0
|
|
for feature in source_features:
|
|
properties = dict(feature.get("properties") or {})
|
|
object_id = str(properties.get("ObjectId") or "").strip()
|
|
geometry_payload = feature.get("geometry")
|
|
source_wgs84 = polygonal_geometry(shape(geometry_payload)) if geometry_payload else None
|
|
source_lambert72 = (
|
|
polygonal_geometry(transform_geometry(TO_LAMBERT72.transform, source_wgs84))
|
|
if source_wgs84 is not None
|
|
else None
|
|
)
|
|
if not object_id or source_lambert72 is None or not source_lambert72.intersects(boundary_lambert72):
|
|
rejected += 1
|
|
continue
|
|
was_clipped = not source_lambert72.within(boundary_lambert72)
|
|
clipped_lambert72 = polygonal_geometry(source_lambert72.intersection(boundary_lambert72))
|
|
if clipped_lambert72 is None or clipped_lambert72.area <= 0:
|
|
rejected += 1
|
|
continue
|
|
if was_clipped:
|
|
clipped += 1
|
|
clipped_wgs84 = polygonal_geometry(transform_geometry(TO_WGS84.transform, clipped_lambert72))
|
|
if clipped_wgs84 is None:
|
|
rejected += 1
|
|
continue
|
|
raw_status = str(properties.get("GebouwStatus") or "").strip()
|
|
status_key = BUILDING_STATUS_KEYS.get(compact_key(raw_status), "unknown")
|
|
records[object_id] = {
|
|
"object_id": object_id,
|
|
"version_id": str(properties.get("VersieId") or ""),
|
|
"geometry_method": str(properties.get("GeometrieMethode") or ""),
|
|
"status_key": status_key,
|
|
"status_label": BUILDING_STATUS_LABELS[status_key],
|
|
"geometry_wgs84": clipped_wgs84,
|
|
"geometry_lambert72": clipped_lambert72,
|
|
"area_ha": float(clipped_lambert72.area) / 10_000.0,
|
|
"was_clipped": was_clipped,
|
|
}
|
|
if not records:
|
|
raise RuntimeError("No valid Buildings Register polygons intersect the persisted Area")
|
|
return records, {"rejected_or_outside_count": rejected, "clipped_count": clipped}
|
|
|
|
|
|
def normalize_units(
|
|
source_features: Iterable[dict[str, Any]],
|
|
boundary_lambert72,
|
|
buildings: dict[str, dict[str, Any]],
|
|
) -> tuple[list[dict[str, Any]], dict[str, int]]:
|
|
units: list[dict[str, Any]] = []
|
|
outside = 0
|
|
orphan = 0
|
|
for feature in source_features:
|
|
properties = dict(feature.get("properties") or {})
|
|
geometry_payload = feature.get("geometry")
|
|
point_wgs84 = shape(geometry_payload) if geometry_payload else None
|
|
if not isinstance(point_wgs84, Point) or point_wgs84.is_empty:
|
|
outside += 1
|
|
continue
|
|
point_lambert72 = transform_geometry(TO_LAMBERT72.transform, point_wgs84)
|
|
if not boundary_lambert72.covers(point_lambert72):
|
|
outside += 1
|
|
continue
|
|
building_id = str(properties.get("GebouwObjectId") or "").strip()
|
|
if building_id not in buildings:
|
|
orphan += 1
|
|
status_key = UNIT_STATUS_KEYS.get(compact_key(properties.get("GebouweenheidStatus")), "unknown")
|
|
units.append(
|
|
{
|
|
"object_id": str(properties.get("ObjectId") or ""),
|
|
"building_id": building_id,
|
|
"status_key": status_key,
|
|
"function_key": compact_key(properties.get("Functie")) or "unknown",
|
|
"point_wgs84": point_wgs84,
|
|
}
|
|
)
|
|
return units, {"outside_count": outside, "orphan_building_count": orphan}
|
|
|
|
|
|
def coordinate_key(point: Point) -> tuple[float, float]:
|
|
return round(float(point.x), 7), round(float(point.y), 7)
|
|
|
|
|
|
def link_addresses(
|
|
source_features: Iterable[dict[str, Any]],
|
|
boundary_lambert72,
|
|
buildings: dict[str, dict[str, Any]],
|
|
units: list[dict[str, Any]],
|
|
) -> tuple[dict[str, Counter[str]], dict[str, Any]]:
|
|
unit_positions: dict[tuple[float, float], list[dict[str, Any]]] = defaultdict(list)
|
|
for unit in units:
|
|
if unit["building_id"] in buildings:
|
|
unit_positions[coordinate_key(unit["point_wgs84"])].append(unit)
|
|
|
|
building_ids = list(buildings)
|
|
building_geometries = [buildings[building_id]["geometry_wgs84"] for building_id in building_ids]
|
|
building_tree = STRtree(building_geometries)
|
|
counts: dict[str, Counter[str]] = defaultdict(Counter)
|
|
method_counts: Counter[str] = Counter()
|
|
status_counts: Counter[str] = Counter()
|
|
retained_address_count = 0
|
|
for feature in source_features:
|
|
properties = dict(feature.get("properties") or {})
|
|
geometry_payload = feature.get("geometry")
|
|
point_wgs84 = shape(geometry_payload) if geometry_payload else None
|
|
if not isinstance(point_wgs84, Point) or point_wgs84.is_empty:
|
|
continue
|
|
point_lambert72 = transform_geometry(TO_LAMBERT72.transform, point_wgs84)
|
|
if not boundary_lambert72.covers(point_lambert72):
|
|
continue
|
|
retained_address_count += 1
|
|
status_key = ADDRESS_STATUS_KEYS.get(compact_key(properties.get("AdresStatus")), "unknown")
|
|
status_counts[status_key] += 1
|
|
specificity = compact_key(properties.get("PositieSpecificatie"))
|
|
matched_building_id: str | None = None
|
|
method = "unmatched"
|
|
if specificity == "gebouweenheid":
|
|
candidates = unit_positions.get(coordinate_key(point_wgs84), [])
|
|
candidate_buildings = {candidate["building_id"] for candidate in candidates}
|
|
if len(candidate_buildings) == 1:
|
|
matched_building_id = next(iter(candidate_buildings))
|
|
method = "unit_position_exact" if len(candidates) == 1 else "unit_position_building_unambiguous"
|
|
elif len(candidate_buildings) > 1:
|
|
method = "ambiguous_unit_position"
|
|
|
|
if matched_building_id is None and method == "unmatched":
|
|
containing = [
|
|
int(index)
|
|
for index in building_tree.query(point_wgs84)
|
|
if building_geometries[int(index)].covers(point_wgs84)
|
|
]
|
|
if len(containing) == 1:
|
|
matched_building_id = building_ids[containing[0]]
|
|
method = "building_contains"
|
|
elif len(containing) > 1:
|
|
method = "ambiguous_building_contains"
|
|
|
|
method_counts[method] += 1
|
|
if matched_building_id is not None:
|
|
counts[matched_building_id]["address_count"] += 1
|
|
counts[matched_building_id][f"address_status_{status_key}"] += 1
|
|
counts[matched_building_id][f"address_match_{method}"] += 1
|
|
|
|
return counts, {
|
|
"retained_address_count": retained_address_count,
|
|
"match_method_counts": dict(sorted(method_counts.items())),
|
|
"status_counts": dict(sorted(status_counts.items())),
|
|
"matched_address_count": sum(
|
|
value for key, value in method_counts.items() if key in {
|
|
"unit_position_exact", "unit_position_building_unambiguous", "building_contains"
|
|
}
|
|
),
|
|
"ambiguous_address_count": sum(value for key, value in method_counts.items() if key.startswith("ambiguous")),
|
|
"unmatched_address_count": int(method_counts.get("unmatched", 0)),
|
|
"privacy_field_violations": 0,
|
|
}
|
|
|
|
|
|
def find_grb_dataset(datasets: list[dict[str, Any]]) -> dict[str, Any]:
|
|
candidates = [
|
|
dataset
|
|
for dataset in datasets
|
|
if dataset.get("status") == "ready"
|
|
and dataset.get("source_name") == "grb"
|
|
and dataset.get("reference_layer_name") == "buildings"
|
|
and (dataset.get("provenance_metadata") or {}).get("manifest_path")
|
|
]
|
|
if not candidates:
|
|
raise RuntimeError("A persisted regional GRB buildings Dataset with manifest evidence is required")
|
|
candidates.sort(key=lambda item: str(item.get("observed_at") or item.get("imported_at") or ""), reverse=True)
|
|
return candidates[0]
|
|
|
|
|
|
def load_grb_reference(
|
|
grb_dataset: dict[str, Any],
|
|
boundary_wgs84,
|
|
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
|
provenance = grb_dataset.get("provenance_metadata") or {}
|
|
manifest_path = Path(str(provenance.get("manifest_path") or ""))
|
|
if not manifest_path.is_file():
|
|
raise RuntimeError(f"Persisted GRB manifest is missing: {manifest_path}")
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
if manifest.get("status") != "complete" or manifest.get("reference_truncated") is not False:
|
|
raise RuntimeError("Persisted GRB manifest is incomplete or truncated")
|
|
artifact_path = manifest_path.parent / str(manifest.get("artifact_filename") or "")
|
|
if not artifact_path.is_file() or sha256_file(artifact_path) != manifest.get("artifact_sha256"):
|
|
raise RuntimeError("Persisted GRB combined artifact checksum is invalid")
|
|
|
|
records: list[dict[str, Any]] = []
|
|
partition_checksums: dict[str, str] = {}
|
|
partition_dir = manifest_path.parent / "partitions"
|
|
for summary in manifest.get("partitions") or []:
|
|
partition_path = partition_dir / str(summary.get("filename") or "")
|
|
expected_checksum = str(summary.get("sha256") or "")
|
|
if not partition_path.is_file() or sha256_file(partition_path) != expected_checksum:
|
|
raise RuntimeError(f"Persisted GRB partition checksum is invalid: {partition_path.name}")
|
|
partition_checksums[partition_path.name] = expected_checksum
|
|
payload = json.loads(partition_path.read_text(encoding="utf-8"))
|
|
for feature in payload.get("features") or []:
|
|
geometry_payload = feature.get("geometry")
|
|
geometry = polygonal_geometry(shape(geometry_payload)) if geometry_payload else None
|
|
if geometry is None or not geometry.intersects(boundary_wgs84):
|
|
continue
|
|
properties = feature.get("properties") or {}
|
|
identity = str(feature.get("id") or properties.get("source_feature_id") or "")
|
|
if identity:
|
|
records.append({"source_feature_id": identity, "geometry_wgs84": geometry})
|
|
if not records:
|
|
raise RuntimeError("Persisted GRB evidence contains no buildings intersecting the Area")
|
|
return records, {
|
|
"dataset_id": str(grb_dataset["id"]),
|
|
"observed_at": grb_dataset.get("observed_at"),
|
|
"manifest_path": str(manifest_path),
|
|
"artifact_sha256": manifest["artifact_sha256"],
|
|
"partition_checksums": partition_checksums,
|
|
"feature_count": len(records),
|
|
}
|
|
|
|
|
|
def reconcile_with_grb(
|
|
buildings: dict[str, dict[str, Any]],
|
|
grb_records: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
grb_metric = [
|
|
polygonal_geometry(transform_geometry(TO_LAMBERT72.transform, record["geometry_wgs84"]))
|
|
for record in grb_records
|
|
]
|
|
valid_indexes = [index for index, geometry in enumerate(grb_metric) if geometry is not None]
|
|
geometries = [grb_metric[index] for index in valid_indexes]
|
|
tree = STRtree(geometries)
|
|
exact_lookup: dict[str, list[int]] = defaultdict(list)
|
|
for tree_index, geometry in enumerate(geometries):
|
|
exact_lookup[geometry.wkb_hex].append(tree_index)
|
|
|
|
target_to_buildings: dict[str, list[str]] = defaultdict(list)
|
|
match_counts: Counter[str] = Counter()
|
|
for building_id, building in buildings.items():
|
|
geometry = building["geometry_lambert72"]
|
|
best_tree_index: int | None = None
|
|
best_iou = 0.0
|
|
method = "unmatched"
|
|
exact = exact_lookup.get(geometry.wkb_hex, [])
|
|
if len(exact) == 1 and geometries[exact[0]].equals(geometry):
|
|
best_tree_index = exact[0]
|
|
best_iou = 1.0
|
|
method = "exact_geometry"
|
|
else:
|
|
for candidate in tree.query(geometry):
|
|
tree_index = int(candidate)
|
|
candidate_geometry = geometries[tree_index]
|
|
intersection_area = geometry.intersection(candidate_geometry).area
|
|
if intersection_area <= 0:
|
|
continue
|
|
union_area = geometry.union(candidate_geometry).area
|
|
iou = float(intersection_area / union_area) if union_area > 0 else 0.0
|
|
if iou > best_iou:
|
|
best_iou = iou
|
|
best_tree_index = tree_index
|
|
if (
|
|
best_tree_index is not None
|
|
and best_iou >= 0.99999999
|
|
and geometry.hausdorff_distance(geometries[best_tree_index]) <= 0.001
|
|
):
|
|
method = "exact_geometry"
|
|
best_iou = 1.0
|
|
elif best_iou >= 0.98:
|
|
method = "high_iou"
|
|
elif best_iou >= 0.80:
|
|
method = "review_iou"
|
|
else:
|
|
best_tree_index = None
|
|
best_iou = 0.0
|
|
|
|
if best_tree_index is None:
|
|
building["grb_match_status"] = "unmatched"
|
|
building["grb_match_method"] = "unmatched"
|
|
building["grb_match_confidence"] = 0.0
|
|
building["grb_source_feature_id"] = None
|
|
continue
|
|
source_index = valid_indexes[best_tree_index]
|
|
target_id = grb_records[source_index]["source_feature_id"]
|
|
building["grb_match_status"] = "matched" if method != "review_iou" else "review"
|
|
building["grb_match_method"] = method
|
|
building["grb_match_confidence"] = round(best_iou, 8)
|
|
building["grb_source_feature_id"] = target_id
|
|
target_to_buildings[target_id].append(building_id)
|
|
|
|
duplicate_targets = {target for target, ids in target_to_buildings.items() if len(ids) > 1}
|
|
for target_id in duplicate_targets:
|
|
for building_id in target_to_buildings[target_id]:
|
|
building = buildings[building_id]
|
|
building["grb_match_status"] = "ambiguous"
|
|
building["grb_match_method"] = "duplicate_grb_target"
|
|
|
|
for building in buildings.values():
|
|
match_counts[building["grb_match_status"]] += 1
|
|
matched_targets = {
|
|
building["grb_source_feature_id"]
|
|
for building in buildings.values()
|
|
if building["grb_match_status"] == "matched" and building["grb_source_feature_id"]
|
|
}
|
|
realized = [building for building in buildings.values() if building["status_key"] == "realized"]
|
|
realized_matched = sum(1 for building in realized if building["grb_match_status"] == "matched")
|
|
return {
|
|
"match_status_counts": dict(sorted(match_counts.items())),
|
|
"matched_grb_feature_count": len(matched_targets),
|
|
"unmatched_grb_feature_count": max(0, len(grb_records) - len(matched_targets)),
|
|
"duplicate_grb_target_count": len(duplicate_targets),
|
|
"match_rate": round(match_counts.get("matched", 0) / len(buildings), 8),
|
|
"realized_match_rate": round(realized_matched / len(realized), 8) if realized else 0.0,
|
|
}
|
|
|
|
|
|
def build_output_features(
|
|
buildings: dict[str, dict[str, Any]],
|
|
units: list[dict[str, Any]],
|
|
address_counts: dict[str, Counter[str]],
|
|
*,
|
|
observed_date: date,
|
|
area_name: str,
|
|
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
|
unit_counts: dict[str, Counter[str]] = defaultdict(Counter)
|
|
for unit in units:
|
|
building_id = unit["building_id"]
|
|
if building_id not in buildings:
|
|
continue
|
|
unit_counts[building_id]["unit_count"] += 1
|
|
unit_counts[building_id][f"unit_status_{unit['status_key']}"] += 1
|
|
|
|
features: list[dict[str, Any]] = []
|
|
status_counts: Counter[str] = Counter()
|
|
total_area_ha = 0.0
|
|
total_units = 0
|
|
total_addresses = 0
|
|
prohibited_output_fields = {"VolledigAdres", "Straatnaam", "Huisnummer", "Busnummer", "HuisnummerLabel"}
|
|
for building_id, building in buildings.items():
|
|
units_for_building = unit_counts[building_id]
|
|
addresses_for_building = address_counts[building_id]
|
|
properties = {
|
|
"source_name": SOURCE_NAME,
|
|
"source_feature_id": f"Gebouwenregister:Gebouw:{building_id}",
|
|
"reference_layer_name": "building_registry",
|
|
"theme": "buildings",
|
|
"authority_level": "authoritative",
|
|
"coverage_scope": "municipality",
|
|
"coverage_area": area_name,
|
|
"observed_at": observed_date.isoformat(),
|
|
"attribution": ATTRIBUTION,
|
|
"building_object_id": building_id,
|
|
"building_version_id": building["version_id"],
|
|
"building_status_key": building["status_key"],
|
|
"building_status_label": building["status_label"],
|
|
"building_geometry_method": building["geometry_method"],
|
|
"clipped_to_area": building["was_clipped"],
|
|
"building_area_ha": round(building["area_ha"], 8),
|
|
"unit_count": int(units_for_building.get("unit_count", 0)),
|
|
"realized_unit_count": int(units_for_building.get("unit_status_realized", 0)),
|
|
"planned_unit_count": int(units_for_building.get("unit_status_planned", 0)),
|
|
"historical_unit_count": int(units_for_building.get("unit_status_historical", 0)),
|
|
"not_realized_unit_count": int(units_for_building.get("unit_status_not_realized", 0)),
|
|
"unknown_unit_count": int(units_for_building.get("unit_status_unknown", 0)),
|
|
"address_count": int(addresses_for_building.get("address_count", 0)),
|
|
"active_address_count": int(addresses_for_building.get("address_status_in_use", 0)),
|
|
"grb_match_status": building["grb_match_status"],
|
|
"grb_match_method": building["grb_match_method"],
|
|
"grb_match_confidence": building["grb_match_confidence"],
|
|
"grb_source_feature_id": building["grb_source_feature_id"],
|
|
"privacy_profile": "aggregate_counts_only",
|
|
}
|
|
if prohibited_output_fields.intersection(properties):
|
|
raise RuntimeError("Privacy-minimized output unexpectedly contains address label fields")
|
|
feature_id = properties["source_feature_id"]
|
|
features.append(
|
|
{
|
|
"type": "Feature",
|
|
"id": feature_id,
|
|
"geometry": mapping(building["geometry_wgs84"]),
|
|
"properties": properties,
|
|
}
|
|
)
|
|
status_counts[building["status_key"]] += 1
|
|
total_area_ha += building["area_ha"]
|
|
total_units += properties["unit_count"]
|
|
total_addresses += properties["address_count"]
|
|
return features, {
|
|
"building_status_counts": dict(sorted(status_counts.items())),
|
|
"building_area_ha": round(total_area_ha, 6),
|
|
"linked_unit_count": total_units,
|
|
"linked_address_count": total_addresses,
|
|
}
|
|
|
|
|
|
def reusable_snapshot(
|
|
snapshot_dir: Path,
|
|
*,
|
|
area_name: str,
|
|
observed_date: date,
|
|
expected_boundary_checksum: str,
|
|
) -> tuple[Path, Path, dict[str, Any]] | None:
|
|
artifact_path = snapshot_dir / "buildings_addresses_register.geojson"
|
|
manifest_path = snapshot_dir / "buildings_addresses_register.manifest.json"
|
|
if not artifact_path.is_file() or not manifest_path.is_file():
|
|
return None
|
|
try:
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return None
|
|
if manifest.get("schema_version") != SCHEMA_VERSION:
|
|
return None
|
|
if manifest.get("coverage_area") != area_name or manifest.get("observed_at") != observed_date.isoformat():
|
|
return None
|
|
if manifest.get("boundary_sha256") != expected_boundary_checksum:
|
|
return None
|
|
if manifest.get("artifact_sha256") != sha256_file(artifact_path):
|
|
return None
|
|
for collection in manifest.get("collections") or []:
|
|
for page in collection.get("pages") or []:
|
|
page_path = snapshot_dir / str(page.get("path") or "")
|
|
if not page_path.is_file() or page.get("sha256") != sha256_file(page_path):
|
|
return None
|
|
for filename, checksum in (manifest.get("grb_reference") or {}).get("partition_checksums", {}).items():
|
|
grb_manifest = Path(str((manifest.get("grb_reference") or {}).get("manifest_path") or ""))
|
|
partition_path = grb_manifest.parent / "partitions" / filename
|
|
if not partition_path.is_file() or sha256_file(partition_path) != checksum:
|
|
return None
|
|
return artifact_path, manifest_path, manifest
|
|
|
|
|
|
def prepare_snapshot(
|
|
session: requests.Session,
|
|
*,
|
|
boundary_wgs84,
|
|
datasets: list[dict[str, Any]],
|
|
snapshot_dir: Path,
|
|
area_name: str,
|
|
observed_date: date,
|
|
page_limit: int,
|
|
max_buildings: int,
|
|
max_units: int,
|
|
max_addresses: int,
|
|
timeout: int,
|
|
force: bool,
|
|
) -> tuple[Path, Path, dict[str, Any]]:
|
|
expected_boundary_checksum = boundary_checksum(boundary_wgs84)
|
|
if not force:
|
|
reusable = reusable_snapshot(
|
|
snapshot_dir,
|
|
area_name=area_name,
|
|
observed_date=observed_date,
|
|
expected_boundary_checksum=expected_boundary_checksum,
|
|
)
|
|
if reusable:
|
|
return reusable
|
|
snapshot_dir.mkdir(parents=True, exist_ok=True)
|
|
raw_dir = snapshot_dir / "raw"
|
|
raw_dir.mkdir(parents=True, exist_ok=True)
|
|
boundary_lambert72 = polygonal_geometry(transform_geometry(TO_LAMBERT72.transform, boundary_wgs84))
|
|
if boundary_lambert72 is None:
|
|
raise RuntimeError("Area boundary could not be transformed to EPSG:31370")
|
|
|
|
buildings_raw, buildings_source = fetch_collection(
|
|
session,
|
|
url=BUILDING_ITEMS_URL,
|
|
name="buildings",
|
|
bbox=boundary_wgs84.bounds,
|
|
raw_dir=raw_dir,
|
|
page_limit=page_limit,
|
|
max_features=max_buildings,
|
|
timeout=timeout,
|
|
)
|
|
units_raw, units_source = fetch_collection(
|
|
session,
|
|
url=UNIT_ITEMS_URL,
|
|
name="building_units",
|
|
bbox=boundary_wgs84.bounds,
|
|
raw_dir=raw_dir,
|
|
page_limit=page_limit,
|
|
max_features=max_units,
|
|
timeout=timeout,
|
|
)
|
|
addresses_raw, addresses_source = fetch_collection(
|
|
session,
|
|
url=ADDRESS_ITEMS_URL,
|
|
name="addresses",
|
|
bbox=boundary_wgs84.bounds,
|
|
raw_dir=raw_dir,
|
|
page_limit=page_limit,
|
|
max_features=max_addresses,
|
|
timeout=timeout,
|
|
)
|
|
buildings, building_filter = normalize_buildings(buildings_raw, boundary_lambert72)
|
|
units, unit_filter = normalize_units(units_raw, boundary_lambert72, buildings)
|
|
address_counts, address_relations = link_addresses(
|
|
addresses_raw,
|
|
boundary_lambert72,
|
|
buildings,
|
|
units,
|
|
)
|
|
if address_relations["privacy_field_violations"]:
|
|
raise RuntimeError("Address privacy output contract failed")
|
|
|
|
grb_dataset = find_grb_dataset(datasets)
|
|
grb_records, grb_reference = load_grb_reference(grb_dataset, boundary_wgs84)
|
|
grb_reconciliation = reconcile_with_grb(buildings, grb_records)
|
|
output_features, output_summary = build_output_features(
|
|
buildings,
|
|
units,
|
|
address_counts,
|
|
observed_date=observed_date,
|
|
area_name=area_name,
|
|
)
|
|
generated_at = utc_now()
|
|
artifact = {
|
|
"type": "FeatureCollection",
|
|
"name": f"Gebouwen- en Adressenregister - {area_name} - {observed_date.isoformat()}",
|
|
"features": output_features,
|
|
"source": "Digitaal Vlaanderen Buildings and Addresses Register OGC API Features",
|
|
"source_urls": [BUILDING_ITEMS_URL, UNIT_ITEMS_URL, ADDRESS_ITEMS_URL],
|
|
"attribution": ATTRIBUTION,
|
|
"catalog_url": CATALOG_URL,
|
|
"observed_at": observed_date.isoformat(),
|
|
"coverage_area": area_name,
|
|
"privacy_profile": "aggregate_counts_only",
|
|
"reference_truncated": False,
|
|
"generated_at": generated_at,
|
|
}
|
|
artifact_path = snapshot_dir / "buildings_addresses_register.geojson"
|
|
write_json_atomic(artifact_path, artifact)
|
|
manifest = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"status": "complete",
|
|
"observed_at": observed_date.isoformat(),
|
|
"generated_at": generated_at,
|
|
"coverage_area": area_name,
|
|
"crs_source": "EPSG:4326",
|
|
"crs_clip": "EPSG:31370",
|
|
"crs_persisted": "EPSG:4326",
|
|
"boundary_sha256": expected_boundary_checksum,
|
|
"boundary_bbox_wgs84": list(boundary_wgs84.bounds),
|
|
"collections": [buildings_source, units_source, addresses_source],
|
|
"building_filter": building_filter,
|
|
"unit_filter": unit_filter,
|
|
"address_relations": address_relations,
|
|
"grb_reference": grb_reference,
|
|
"grb_reconciliation": grb_reconciliation,
|
|
"feature_count": len(output_features),
|
|
**output_summary,
|
|
"privacy_profile": {
|
|
"queryable_output": "building polygons with aggregate relation counts",
|
|
"excluded_fields": ["VolledigAdres", "Straatnaam", "Huisnummer", "Busnummer", "HuisnummerLabel"],
|
|
"personal_data_exposed": False,
|
|
},
|
|
"reference_truncated": False,
|
|
"catalog_url": CATALOG_URL,
|
|
"building_catalog_url": BUILDING_CATALOG_URL,
|
|
"change_notice_url": CHANGE_NOTICE_URL,
|
|
"attribution": ATTRIBUTION,
|
|
"artifact_path": str(artifact_path),
|
|
"artifact_sha256": sha256_file(artifact_path),
|
|
"artifact_size_bytes": artifact_path.stat().st_size,
|
|
"limitations": [
|
|
"This is a continuously updated register captured as a dated snapshot, not an historical annual series.",
|
|
"Address-to-building links are classified by exact unit position or polygon containment; ambiguous and unmatched addresses are never forced.",
|
|
"Address counts are not households, residents, dwellings or population.",
|
|
"Building-unit counts describe registered functional units and do not imply residential use.",
|
|
"GRB matching is evidence for geometric reconciliation; register lifecycle status remains separate from GRB footprint geometry.",
|
|
],
|
|
}
|
|
manifest_path = snapshot_dir / "buildings_addresses_register.manifest.json"
|
|
write_json_atomic(manifest_path, manifest, pretty=True)
|
|
return artifact_path, manifest_path, manifest
|
|
|
|
|
|
def selection_metrics() -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"metric_key": "registered_building_count",
|
|
"method": "feature_count",
|
|
"label": "Gebouwen in het register",
|
|
"unit": "gebouwen",
|
|
},
|
|
{
|
|
"metric_key": "realized_building_count",
|
|
"method": "feature_count",
|
|
"label": "Gerealiseerde gebouwen",
|
|
"unit": "gebouwen",
|
|
"filter_property": "building_status_key",
|
|
"filter_values": ["realized"],
|
|
},
|
|
{
|
|
"metric_key": "under_construction_building_count",
|
|
"method": "feature_count",
|
|
"label": "Gebouwen in aanbouw",
|
|
"unit": "gebouwen",
|
|
"filter_property": "building_status_key",
|
|
"filter_values": ["under_construction"],
|
|
},
|
|
{
|
|
"metric_key": "planned_building_count",
|
|
"method": "feature_count",
|
|
"label": "Geplande gebouwen",
|
|
"unit": "gebouwen",
|
|
"filter_property": "building_status_key",
|
|
"filter_values": ["planned"],
|
|
},
|
|
{
|
|
"metric_key": "historical_building_count",
|
|
"method": "feature_count",
|
|
"label": "Gehistoreerde gebouwen",
|
|
"unit": "gebouwen",
|
|
"filter_property": "building_status_key",
|
|
"filter_values": ["historical"],
|
|
},
|
|
{
|
|
"metric_key": "building_unit_count",
|
|
"method": "sum",
|
|
"property": "unit_count",
|
|
"label": "Geregistreerde gebouweenheden",
|
|
"unit": "eenheden",
|
|
"warning": "Gebouweenheden zijn functionele registereenheden en niet automatisch woningen.",
|
|
},
|
|
{
|
|
"metric_key": "realized_building_unit_count",
|
|
"method": "sum",
|
|
"property": "realized_unit_count",
|
|
"label": "Gerealiseerde gebouweenheden",
|
|
"unit": "eenheden",
|
|
},
|
|
{
|
|
"metric_key": "linked_address_count",
|
|
"method": "sum",
|
|
"property": "address_count",
|
|
"label": "Gekoppelde adressen",
|
|
"unit": "adressen",
|
|
"warning": "Adressen zijn geen huishoudens, woningen, inwoners of bevolkingsmeting.",
|
|
},
|
|
{
|
|
"metric_key": "active_address_count",
|
|
"method": "sum",
|
|
"property": "active_address_count",
|
|
"label": "Adressen in gebruik",
|
|
"unit": "adressen",
|
|
"warning": "Adresstatus beschrijft het registerobject en zegt niets over bewoning.",
|
|
},
|
|
{
|
|
"metric_key": "grb_matched_building_count",
|
|
"method": "feature_count",
|
|
"label": "Gebouwen met bevestigde GRB-match",
|
|
"unit": "gebouwen",
|
|
"filter_property": "grb_match_status",
|
|
"filter_values": ["matched"],
|
|
},
|
|
]
|
|
|
|
|
|
def upload_snapshot(
|
|
session: requests.Session,
|
|
*,
|
|
base_url: str,
|
|
project_id: str,
|
|
area_id: str,
|
|
artifact_path: Path,
|
|
manifest_path: Path,
|
|
manifest: dict[str, Any],
|
|
observed_date: date,
|
|
timeout: int,
|
|
) -> dict[str, Any]:
|
|
warning = (
|
|
"Gebouwoppervlakte is grondvlak, geen vloeroppervlakte of volume. Adressen zijn geen huishoudens, "
|
|
"woningen, inwoners of bevolkingsmeting."
|
|
)
|
|
source_metadata = {
|
|
"provider": "Digitaal Vlaanderen",
|
|
"theme": "buildings",
|
|
"layer_name": "Gebouwen- en Adressenregister",
|
|
"layer_type": "building_registry",
|
|
"authority_level": "authoritative",
|
|
"coverage_scope": "municipality",
|
|
"coverage_area": manifest["coverage_area"],
|
|
"feature_count": manifest["feature_count"],
|
|
"building_unit_count": manifest["linked_unit_count"],
|
|
"linked_address_count": manifest["linked_address_count"],
|
|
"address_match_method_counts": manifest["address_relations"]["match_method_counts"],
|
|
"unmatched_address_count": manifest["address_relations"]["unmatched_address_count"],
|
|
"ambiguous_address_count": manifest["address_relations"]["ambiguous_address_count"],
|
|
"grb_match_rate": manifest["grb_reconciliation"]["match_rate"],
|
|
"realized_grb_match_rate": manifest["grb_reconciliation"]["realized_match_rate"],
|
|
"geometry_clipped_to_area": True,
|
|
"identity_stable": True,
|
|
"semantic_metrics": False,
|
|
"privacy_profile": "aggregate_counts_only",
|
|
"attribution": ATTRIBUTION,
|
|
"catalog_url": CATALOG_URL,
|
|
"selection_aggregation": {
|
|
"metric_key": "building_footprint_area",
|
|
"method": "intersection_area",
|
|
"label": "Gebouwgrondoppervlakte",
|
|
"unit": "ha",
|
|
"geometry_dimension": 2,
|
|
"warning": warning,
|
|
},
|
|
"selection_metrics": selection_metrics(),
|
|
}
|
|
provenance_metadata = {
|
|
"operator_tool": "provision_buildings_addresses_register.py",
|
|
"operator_explicit_fetch": True,
|
|
"geometry_clipped_to_area": True,
|
|
"source_urls": [BUILDING_ITEMS_URL, UNIT_ITEMS_URL, ADDRESS_ITEMS_URL],
|
|
"catalog_url": CATALOG_URL,
|
|
"building_catalog_url": BUILDING_CATALOG_URL,
|
|
"change_notice_url": CHANGE_NOTICE_URL,
|
|
"manifest_path": str(manifest_path),
|
|
"artifact_sha256": manifest["artifact_sha256"],
|
|
"raw_page_checksums": {
|
|
page["path"]: page["sha256"]
|
|
for collection in manifest["collections"]
|
|
for page in collection["pages"]
|
|
},
|
|
"grb_reference": manifest["grb_reference"],
|
|
"grb_reconciliation": manifest["grb_reconciliation"],
|
|
"privacy_profile": manifest["privacy_profile"],
|
|
"reference_truncated": False,
|
|
"generated_at": manifest["generated_at"],
|
|
"limitations": manifest["limitations"],
|
|
}
|
|
with artifact_path.open("rb") as handle:
|
|
response = session.post(
|
|
f"{base_url}/api/v1/projects/{project_id}/datasets/upload",
|
|
data={
|
|
"dataset_type": "vector",
|
|
"source": "operator_official_import",
|
|
"dataset_role": "reference",
|
|
"source_name": SOURCE_NAME,
|
|
"reference_layer_name": "building_registry",
|
|
"source_metadata_json": json.dumps(source_metadata, ensure_ascii=False),
|
|
"provenance_metadata_json": json.dumps(provenance_metadata, ensure_ascii=False),
|
|
"area_id": area_id,
|
|
"temporal_series_key": f"buildings-addresses-register:{area_id}",
|
|
"observed_at": observed_at(observed_date),
|
|
"temporal_granularity": "snapshot",
|
|
"source_version": observed_date.isoformat(),
|
|
},
|
|
files={"file": (artifact_path.name, handle, "application/geo+json")},
|
|
timeout=timeout,
|
|
)
|
|
return response_data(response)
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
if (
|
|
args.page_limit < 1
|
|
or args.page_limit > 5000
|
|
or min(args.max_buildings, args.max_units, args.max_addresses) < args.page_limit
|
|
):
|
|
print(json.dumps({"status": "error", "message": "Invalid page or feature safety limits"}), file=sys.stderr)
|
|
return 2
|
|
base_url = args.base_url.rstrip("/")
|
|
snapshot_dir = args.output_root / area_storage_key(args.area_name) / args.observed_date.isoformat()
|
|
try:
|
|
with requests.Session() as api_session:
|
|
project_id, area_id, boundary, datasets = locate_workspace(
|
|
api_session,
|
|
base_url,
|
|
args.project_name,
|
|
args.area_name,
|
|
args.api_timeout,
|
|
)
|
|
with source_session() as official_session:
|
|
artifact_path, manifest_path, manifest = prepare_snapshot(
|
|
official_session,
|
|
boundary_wgs84=boundary,
|
|
datasets=datasets,
|
|
snapshot_dir=snapshot_dir,
|
|
area_name=args.area_name,
|
|
observed_date=args.observed_date,
|
|
page_limit=args.page_limit,
|
|
max_buildings=args.max_buildings,
|
|
max_units=args.max_units,
|
|
max_addresses=args.max_addresses,
|
|
timeout=args.request_timeout,
|
|
force=args.force,
|
|
)
|
|
existing = next(
|
|
(
|
|
item
|
|
for item in datasets
|
|
if item.get("source_name") == SOURCE_NAME
|
|
and item.get("source_version") == args.observed_date.isoformat()
|
|
and str(item.get("area_id") or "") == area_id
|
|
),
|
|
None,
|
|
)
|
|
if existing:
|
|
persisted_checksum = str(existing.get("checksum_sha256") or "")
|
|
if persisted_checksum and persisted_checksum != manifest["artifact_sha256"]:
|
|
raise RuntimeError(
|
|
"A different register snapshot is already persisted for this date and Area"
|
|
)
|
|
persistence = {
|
|
"status": "existing",
|
|
"dataset_id": str(existing["id"]),
|
|
"feature_count": existing.get("feature_count") or manifest["feature_count"],
|
|
}
|
|
elif args.fetch_only:
|
|
persistence = {"status": "prepared", "dataset_id": None, "feature_count": manifest["feature_count"]}
|
|
else:
|
|
dataset = upload_snapshot(
|
|
api_session,
|
|
base_url=base_url,
|
|
project_id=project_id,
|
|
area_id=area_id,
|
|
artifact_path=artifact_path,
|
|
manifest_path=manifest_path,
|
|
manifest=manifest,
|
|
observed_date=args.observed_date,
|
|
timeout=args.api_timeout,
|
|
)
|
|
persistence = {
|
|
"status": "created",
|
|
"dataset_id": str(dataset["id"]),
|
|
"feature_count": dataset.get("feature_count") or manifest["feature_count"],
|
|
}
|
|
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": "ok",
|
|
"mode": "fetch_only" if args.fetch_only else "provisioned",
|
|
"observed_at": args.observed_date.isoformat(),
|
|
"area": args.area_name,
|
|
"feature_count": manifest["feature_count"],
|
|
"building_area_ha": manifest["building_area_ha"],
|
|
"linked_unit_count": manifest["linked_unit_count"],
|
|
"linked_address_count": manifest["linked_address_count"],
|
|
"address_relations": manifest["address_relations"],
|
|
"grb_reconciliation": manifest["grb_reconciliation"],
|
|
"artifact_path": str(artifact_path),
|
|
"manifest_path": str(manifest_path),
|
|
"persistence": persistence,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|