"""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/mol") DEFAULT_PAGE_LIMIT = 1000 DEFAULT_MAX_BUILDINGS = 100_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 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 params: dict[str, str] | None = { "f": "application/geo+json", "bbox": ",".join(f"{value:.8f}" for value in bbox), "limit": str(page_limit), } seen_urls: set[str] = set() seen_ids: set[str] = set() features: list[dict[str, Any]] = [] pages: list[dict[str, Any]] = [] duplicate_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 []) 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 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, "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) -> 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("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]]: if not force: reusable = reusable_snapshot(snapshot_dir) 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": sha256_bytes(json.dumps(mapping(boundary_wgs84), sort_keys=True).encode("utf-8")), "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 / 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())