diff --git a/CHANGELOG.md b/CHANGELOG.md index 304fc957..1fe7703b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ # Changelog +## Sprint 190 Regional Kempen GRB buildings (2026-07-14) + +- Added an explicit regional GRB building operator that fetches the approved Kempen scope in 28 resumable municipality partitions and follows every OGC API pagination link. +- Assigned boundary-crossing GRB features to exactly one partition using maximum municipality intersection, with deterministic NIS-code tie breaking and one retained source identity. +- Added streaming artifact copy and batch-wise partition indexing through `DatasetService` and `VectorFeatureService`, avoiding one giant multipart parse while preserving one normal regional dataset for existing map and PostGIS selection flows. +- Added truncation guards, source/checksum manifests, immutable observation dates, duplicate-source rejection and focused service/operator tests. +- Kept the provider endpoint contract unchanged and retained explicit operator-only fetching; no source request runs during application startup or interactive map use. + ## Sprint 189 Official Kempen operational scope (2026-07-14) - Defined `Kempen` operationally as the official 28-municipality Vlaamse vervoerregio, with an explicit warning that this policy boundary is not the wider cultural or landscape Kempen. diff --git a/backend/README.md b/backend/README.md index 46a6fca6..6db4be20 100644 --- a/backend/README.md +++ b/backend/README.md @@ -956,6 +956,22 @@ Use `--fetch-only` for a source/geometry/checksum audit. The scope pass does not fetch thematic GRB, population or land-use data; those remain separate, bounded operator jobs. +Provision the regional GRB building theme after the scope pass: + +```bash +docker exec geointel python /app/scripts/provision_regional_grb_buildings.py \ + --scope kempen-transport-region +``` + +The operator retains 28 checksummed municipality partitions but exposes one +normal regional reference dataset. `StorageService` copies the combined +artifact without materializing it as upload bytes; `DatasetService` creates +the Dataset and immutable DatasetVersion; `VectorFeatureService` validates and +flushes partition features in bounded batches. The transaction must index the +exact manifest feature count or it rolls back and removes the managed copy. +No public API contract or provider readiness claim is changed by this +operator-only path. + ## Temporal Mol data and evolution Dataset uploads accept `temporal_series_key`, `observed_at`, `valid_from`, diff --git a/backend/app/services/dataset_service.py b/backend/app/services/dataset_service.py index 6c171229..e2bf38e8 100644 --- a/backend/app/services/dataset_service.py +++ b/backend/app/services/dataset_service.py @@ -12,7 +12,7 @@ from fastapi import UploadFile from sqlalchemy.orm import Session from app.core.errors import AppError -from app.models import Dataset, DatasetVersion, Project +from app.models import Area, Dataset, DatasetVersion, Project from app.schemas.dataset import ( DatasetCreateResponse, DatasetStorageResponse, @@ -410,6 +410,136 @@ class DatasetService: return DatasetService._to_response(dataset) + @staticmethod + def import_partitioned_vector_artifact( + db: Session, + *, + project_id: UUID, + area_id: UUID, + artifact_path: str | Path, + partition_paths: list[str | Path], + original_filename: str, + source: str, + dataset_role: str, + source_name: str, + reference_layer_name: str | None, + metadata_json: dict[str, Any], + source_metadata: dict[str, Any], + provenance_metadata: dict[str, Any], + temporal_series_key: str, + observed_at: datetime, + temporal_granularity: str = "snapshot", + source_version: str | None = None, + batch_size: int = 1000, + ) -> DatasetCreateResponse: + if not db.get(Project, project_id): + raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) + area = db.get(Area, area_id) + if not area: + raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404) + if area.project_id != project_id: + raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400) + if not partition_paths: + raise AppError( + code="INVALID_GEOJSON_PARTITIONS", + message="At least one GeoJSON partition is required", + status_code=400, + ) + + filename = DatasetService._validate_upload_filename(original_filename) + if DatasetService._extension_for_path(filename) not in DatasetService.VECTOR_EXTENSIONS: + raise AppError(code="INVALID_UPLOAD", message="Vector artifacts require .geojson or .json files", status_code=415) + normalized_role = DatasetService._normalize_dataset_role(dataset_role) + temporal = DatasetService._validate_temporal_metadata( + temporal_series_key=temporal_series_key, + observed_at=observed_at, + valid_from=observed_at, + valid_to=None, + temporal_granularity=temporal_granularity, + source_version=source_version, + ) + metadata = dict(metadata_json) + expected_feature_count = int(metadata.get("feature_count") or 0) + if expected_feature_count <= 0: + raise AppError( + code="INVALID_GEOJSON_PARTITIONS", + message="Partition metadata must declare a positive feature_count", + status_code=400, + ) + + dataset_id = uuid.uuid4() + storage_info = StorageService.persist_dataset_file_from_path( + project_id=str(project_id), + dataset_id=str(dataset_id), + dataset_type="vector", + original_filename=filename, + source_path=artifact_path, + content_type="application/geo+json", + ) + dataset = Dataset( + id=dataset_id, + project_id=project_id, + area_id=area_id, + name=filename, + dataset_type="vector", + source=source, + dataset_role=normalized_role, + source_name=source_name, + reference_layer_name=reference_layer_name if normalized_role == "reference" else None, + source_metadata=source_metadata, + provenance_metadata=provenance_metadata, + imported_at=datetime.now(timezone.utc), + **temporal, + storage_path=storage_info["storage_path"], + original_filename=storage_info["original_filename"], + stored_filename=storage_info["stored_filename"], + content_type=storage_info["content_type"], + size_bytes=storage_info["size_bytes"], + checksum_sha256=storage_info["checksum_sha256"], + crs=str(metadata.get("crs") or "EPSG:4326"), + bounds_json=metadata.get("bounds_json"), + metadata_json=metadata, + status="ready", + ) + try: + db.add(dataset) + db.add( + DatasetVersion( + dataset_id=dataset.id, + version=1, + storage_path=dataset.storage_path, + source_version=dataset.source_version, + observed_at=dataset.observed_at, + valid_from=dataset.valid_from, + checksum_sha256=dataset.checksum_sha256, + source_metadata=dataset.source_metadata, + provenance_metadata=dataset.provenance_metadata, + ) + ) + persisted_count = VectorFeatureService.persist_geojson_partitions( + db, + dataset.id, + partition_paths, + feature_class=reference_layer_name if normalized_role == "reference" else None, + batch_size=batch_size, + ) + if persisted_count != expected_feature_count: + raise AppError( + code="PARTITION_FEATURE_COUNT_MISMATCH", + message=( + f"Regional artifact declares {expected_feature_count} features but " + f"{persisted_count} queryable features were indexed" + ), + status_code=400, + ) + db.commit() + db.refresh(dataset) + except Exception: + db.rollback() + StorageService.remove_dataset_file(storage_info["storage_path"]) + raise + return DatasetService._to_response(dataset) + @staticmethod def refresh_metadata(db: Session, dataset_id: UUID) -> DatasetCreateResponse: dataset = DatasetService._get_dataset(db, dataset_id) diff --git a/backend/app/services/storage_service.py b/backend/app/services/storage_service.py index fc3694f3..62a091f2 100644 --- a/backend/app/services/storage_service.py +++ b/backend/app/services/storage_service.py @@ -95,6 +95,39 @@ class StorageService: } return metadata + @staticmethod + def persist_dataset_file_from_path( + project_id: str, + dataset_id: str, + dataset_type: str, + original_filename: str, + source_path: str | Path, + content_type: str | None, + ) -> dict[str, Any]: + source = Path(source_path).resolve() + if not source.is_file(): + raise FileNotFoundError(f"Dataset source artifact does not exist: {source}") + + normalized_type = StorageService.normalize_dataset_type(dataset_type) + file_path = Path(StorageService.dataset_file_path(project_id, dataset_id, normalized_type, original_filename)) + file_path.parent.mkdir(parents=True, exist_ok=True) + digest = hashlib.sha256() + size_bytes = 0 + with source.open("rb") as input_stream, file_path.open("wb") as output_stream: + for chunk in iter(lambda: input_stream.read(8 * 1024 * 1024), b""): + output_stream.write(chunk) + digest.update(chunk) + size_bytes += len(chunk) + + return { + "original_filename": StorageService._safe_filename(original_filename), + "stored_filename": file_path.name, + "content_type": content_type or "application/octet-stream", + "size_bytes": size_bytes, + "checksum_sha256": digest.hexdigest(), + "storage_path": str(file_path), + } + @staticmethod def persist_file( storage_path: str, diff --git a/backend/app/services/vector_feature_service.py b/backend/app/services/vector_feature_service.py index 4795e4f4..b423c9a1 100644 --- a/backend/app/services/vector_feature_service.py +++ b/backend/app/services/vector_feature_service.py @@ -1,6 +1,8 @@ from __future__ import annotations -from typing import Any +import json +from pathlib import Path +from typing import Any, Iterable from uuid import UUID from geoalchemy2.functions import ST_Intersects, ST_MakeEnvelope @@ -17,6 +19,37 @@ from app.models import Dataset, VectorFeature class VectorFeatureService: + @staticmethod + def _feature_row(dataset_id: UUID, feature: dict[str, Any], index: int, feature_class: str | None) -> VectorFeature | None: + geometry_payload = feature.get("geometry") + if geometry_payload is None: + return None + try: + geometry = shape(geometry_payload) + except Exception as exc: + raise AppError(code="INVALID_GEOJSON", message=f"Invalid feature geometry at index {index}", status_code=400) from exc + if geometry.is_empty: + return None + if not geometry.is_valid: + geometry = make_valid(geometry) + if geometry.is_empty or not geometry.is_valid: + raise AppError(code="INVALID_GEOMETRY", message=f"Invalid feature geometry at index {index}", status_code=400) + if geometry.has_z: + geometry = transform_geometry(lambda x, y, z=None: (x, y), geometry) + + properties = feature.get("properties") if isinstance(feature.get("properties"), dict) else {} + source_feature_id = feature.get("id") + if source_feature_id is None: + source_feature_id = properties.get("id") or properties.get("source_feature_id") + + return VectorFeature( + dataset_id=dataset_id, + feature_class=feature_class, + source_feature_id=str(source_feature_id) if source_feature_id is not None else None, + properties_json=properties, + geometry=from_shape(geometry, srid=4326), + ) + @staticmethod def _normalize_selection_bbox(bbox: dict[str, Any]) -> dict[str, float | str]: try: @@ -263,34 +296,9 @@ class VectorFeatureService: for index, feature in enumerate(features): if not isinstance(feature, dict): raise AppError(code="INVALID_GEOJSON", message=f"Feature {index} must be an object", status_code=400) - geometry_payload = feature.get("geometry") - if geometry_payload is None: + row = VectorFeatureService._feature_row(dataset_id, feature, index, feature_class) + if row is None: continue - try: - geometry = shape(geometry_payload) - except Exception as exc: - raise AppError(code="INVALID_GEOJSON", message=f"Invalid feature geometry at index {index}", status_code=400) from exc - if geometry.is_empty: - continue - if not geometry.is_valid: - geometry = make_valid(geometry) - if geometry.is_empty or not geometry.is_valid: - raise AppError(code="INVALID_GEOMETRY", message=f"Invalid feature geometry at index {index}", status_code=400) - if geometry.has_z: - geometry = transform_geometry(lambda x, y, z=None: (x, y), geometry) - - properties = feature.get("properties") if isinstance(feature.get("properties"), dict) else {} - source_feature_id = feature.get("id") - if source_feature_id is None: - source_feature_id = properties.get("id") or properties.get("source_feature_id") - - row = VectorFeature( - dataset_id=dataset_id, - feature_class=feature_class, - source_feature_id=str(source_feature_id) if source_feature_id is not None else None, - properties_json=properties, - geometry=from_shape(geometry, srid=4326), - ) db.add(row) persisted.append(row) @@ -298,3 +306,69 @@ class VectorFeatureService: db.flush() db.commit() return persisted + + @staticmethod + def persist_geojson_partitions( + db, + dataset_id: UUID, + partition_paths: Iterable[str | Path], + feature_class: str | None = None, + *, + batch_size: int = 1000, + ) -> int: + if batch_size <= 0: + raise ValueError("batch_size must be positive") + + persisted_count = 0 + source_feature_ids: set[str] = set() + for partition_path in partition_paths: + path = Path(partition_path) + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise AppError( + code="INVALID_GEOJSON_PARTITION", + message=f"Could not read GeoJSON partition {path.name}", + status_code=400, + ) from exc + features = payload.get("features") + if payload.get("type") != "FeatureCollection" or not isinstance(features, list): + raise AppError( + code="INVALID_GEOJSON_PARTITION", + message=f"GeoJSON partition {path.name} must be a FeatureCollection", + status_code=400, + ) + + batch: list[VectorFeature] = [] + for index, feature in enumerate(features): + if not isinstance(feature, dict): + raise AppError( + code="INVALID_GEOJSON_PARTITION", + message=f"Feature {index} in {path.name} must be an object", + status_code=400, + ) + row = VectorFeatureService._feature_row(dataset_id, feature, index, feature_class) + if row is None: + continue + if row.source_feature_id: + if row.source_feature_id in source_feature_ids: + raise AppError( + code="DUPLICATE_SOURCE_FEATURE", + message=f"Duplicate source feature {row.source_feature_id} across regional partitions", + status_code=400, + ) + source_feature_ids.add(row.source_feature_id) + db.add(row) + batch.append(row) + persisted_count += 1 + if len(batch) >= batch_size: + db.flush() + for persisted in batch: + db.expunge(persisted) + batch.clear() + if batch: + db.flush() + for persisted in batch: + db.expunge(persisted) + + return persisted_count diff --git a/backend/tests/test_sprint190_regional_grb_buildings.py b/backend/tests/test_sprint190_regional_grb_buildings.py new file mode 100644 index 00000000..a281699e --- /dev/null +++ b/backend/tests/test_sprint190_regional_grb_buildings.py @@ -0,0 +1,195 @@ +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path +import sys +from uuid import uuid4 + +import pytest +from shapely.geometry import Polygon +from shapely.ops import unary_union + +from app.core.errors import AppError +from app.services.storage_service import StorageService +from app.services.vector_feature_service import VectorFeatureService + + +ROOT = Path(__file__).resolve().parents[2] +SCRIPTS = ROOT / "scripts" + + +def load_operator(): + scripts_path = str(SCRIPTS) + if scripts_path not in sys.path: + sys.path.insert(0, scripts_path) + spec = importlib.util.spec_from_file_location( + "provision_regional_grb_buildings_test", + SCRIPTS / "provision_regional_grb_buildings.py", + ) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def feature(feature_id: str, polygon: Polygon) -> dict: + return { + "type": "Feature", + "id": feature_id, + "geometry": polygon.__geo_interface__, + "properties": {"UIDN": feature_id}, + } + + +def test_partition_assignment_is_deterministic_and_has_no_cross_member_duplicates() -> None: + operator = load_operator() + scopes = importlib.import_module("geographic_scopes") + alpha = scopes.ScopeMember("Alpha", "10001") + beta = scopes.ScopeMember("Beta", "10002") + scope = scopes.GeographicScope( + key="test-region", + display_name="Test region", + project_name="Test project", + project_region="Test", + area_name="Test operation boundary", + authority_name="Test authority", + authority_url="https://example.test/scope", + scope_type="policy_region", + limitation_message="Test limitation.", + members=(alpha, beta), + ) + alpha_boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)]) + beta_boundary = Polygon([(1, 0), (2, 0), (2, 1), (1, 1)]) + members = [(alpha, alpha_boundary), (beta, beta_boundary)] + region = unary_union([alpha_boundary, beta_boundary]) + alpha_building = feature("GBG.alpha", Polygon([(0.1, 0.1), (0.2, 0.1), (0.2, 0.2), (0.1, 0.2)])) + beta_building = feature("GBG.beta", Polygon([(1.2, 0.1), (1.3, 0.1), (1.3, 0.2), (1.2, 0.2)])) + crossing = feature("GBG.crossing", Polygon([(0.8, 0.3), (1.1, 0.3), (1.1, 0.5), (0.8, 0.5)])) + page = ({"type": "FeatureCollection", "features": [alpha_building, beta_building, crossing]}, "https://example.test/grb") + + alpha_features, alpha_summary = operator.build_partition_features( + [page], member=alpha, members=members, regional_boundary=region, scope=scope, max_features=10 + ) + beta_features, beta_summary = operator.build_partition_features( + [page], member=beta, members=members, regional_boundary=region, scope=scope, max_features=10 + ) + + assert {item["id"] for item in alpha_features} == {"GBG.alpha", "GBG.crossing"} + assert {item["id"] for item in beta_features} == {"GBG.beta"} + assert alpha_summary["reference_truncated"] is False + assert beta_summary["reference_truncated"] is False + assert alpha_features[1]["properties"]["partition_assignment"] == "maximum_boundary_intersection" + + +def test_combined_artifact_streams_partitions_and_rejects_duplicate_source_ids(tmp_path: Path) -> None: + operator = load_operator() + scope = importlib.import_module("geographic_scopes").KEMPEN_TRANSPORT_REGION_SCOPE + first = tmp_path / "first.geojson" + second = tmp_path / "second.geojson" + first.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.1", Polygon([(4, 51), (4.01, 51), (4.01, 51.01), (4, 51.01)]))]}), encoding="utf-8") + second.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.2", Polygon([(4.1, 51), (4.11, 51), (4.11, 51.01), (4.1, 51.01)]))]}), encoding="utf-8") + combined = tmp_path / "combined.geojson" + + summary = operator.write_combined_artifact( + combined, + scope=scope, + observed_date=operator.date(2026, 7, 14), + partition_paths=[first, second], + expected_feature_count=2, + ) + + payload = json.loads(combined.read_text(encoding="utf-8")) + assert summary["feature_count"] == 2 + assert summary["sha256"] == operator.sha256_file(combined) + assert [item["id"] for item in payload["features"]] == ["GBG.1", "GBG.2"] + + second.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.1", Polygon([(4.1, 51), (4.11, 51), (4.11, 51.01), (4.1, 51.01)]))]}), encoding="utf-8") + with pytest.raises(RuntimeError, match="Duplicate regional source feature"): + operator.write_combined_artifact( + combined, + scope=scope, + observed_date=operator.date(2026, 7, 14), + partition_paths=[first, second], + expected_feature_count=2, + ) + + +class FakeDb: + def __init__(self) -> None: + self.rows = [] + self.flush_count = 0 + self.expunge_count = 0 + + def add(self, row) -> None: + self.rows.append(row) + + def flush(self) -> None: + self.flush_count += 1 + + def expunge(self, row) -> None: + assert row in self.rows + self.expunge_count += 1 + + +def test_partition_persistence_batches_rows_and_guards_source_identity(tmp_path: Path) -> None: + first = tmp_path / "one.geojson" + second = tmp_path / "two.geojson" + first.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.1", Polygon([(4, 51), (4.01, 51), (4.01, 51.01), (4, 51.01)]))]}), encoding="utf-8") + second.write_text(json.dumps({"type": "FeatureCollection", "features": [feature("GBG.2", Polygon([(4.1, 51), (4.11, 51), (4.11, 51.01), (4.1, 51.01)]))]}), encoding="utf-8") + db = FakeDb() + + persisted = VectorFeatureService.persist_geojson_partitions( + db, + uuid4(), + [first, second], + feature_class="buildings", + batch_size=1, + ) + + assert persisted == 2 + assert db.flush_count == 2 + assert db.expunge_count == 2 + assert {row.source_feature_id for row in db.rows} == {"GBG.1", "GBG.2"} + + second.write_text(first.read_text(encoding="utf-8"), encoding="utf-8") + with pytest.raises(AppError) as error: + VectorFeatureService.persist_geojson_partitions(FakeDb(), uuid4(), [first, second]) + assert error.value.code == "DUPLICATE_SOURCE_FEATURE" + + +def test_storage_service_copies_large_artifacts_without_loading_them_as_upload_bytes(tmp_path: Path, monkeypatch) -> None: + source = tmp_path / "source.geojson" + source.write_bytes((b"0123456789abcdef" * 1024 * 1024) + b"tail") + storage_root = tmp_path / "storage" + monkeypatch.setattr(StorageService, "_base_dir", staticmethod(lambda: storage_root)) + + metadata = StorageService.persist_dataset_file_from_path( + "project", + "dataset", + "vector", + "regional.geojson", + source, + "application/geo+json", + ) + + stored = Path(metadata["storage_path"]) + assert stored.read_bytes() == source.read_bytes() + assert metadata["size_bytes"] == source.stat().st_size + assert metadata["checksum_sha256"] == load_operator().sha256_file(source) + + +def test_regional_operator_is_packaged_documented_and_uses_service_boundaries() -> None: + dockerfile = (ROOT / "deploy/unraid/Dockerfile.all-in-one").read_text(encoding="utf-8") + readiness = (ROOT / "scripts/run_readiness_check.sh").read_text(encoding="utf-8") + operator = (SCRIPTS / "provision_regional_grb_buildings.py").read_text(encoding="utf-8") + dataset_service = (ROOT / "backend/app/services/dataset_service.py").read_text(encoding="utf-8") + + assert "provision_regional_grb_buildings.py" in dockerfile + assert "py_compile scripts/provision_regional_grb_buildings.py" in readiness + assert "DatasetService.import_partitioned_vector_artifact" in operator + assert "VectorFeatureService.persist_geojson_partitions" in dataset_service + assert "insert into vector_features" not in operator.lower() + assert "db.add(VectorFeature" not in operator diff --git a/deploy/unraid/Dockerfile.all-in-one b/deploy/unraid/Dockerfile.all-in-one index c35db4ee..55058bea 100644 --- a/deploy/unraid/Dockerfile.all-in-one +++ b/deploy/unraid/Dockerfile.all-in-one @@ -79,6 +79,7 @@ COPY scripts/provision_mol_historical_landuse.py /app/scripts/provision_mol_hist COPY scripts/provision_official_landuse_timeseries.py /app/scripts/provision_official_landuse_timeseries.py COPY scripts/geographic_scopes.py /app/scripts/geographic_scopes.py COPY scripts/provision_geographic_scope.py /app/scripts/provision_geographic_scope.py +COPY scripts/provision_regional_grb_buildings.py /app/scripts/provision_regional_grb_buildings.py COPY scripts/export_operator_yolo_tile_dataset.py /app/scripts/export_operator_yolo_tile_dataset.py COPY scripts/audit_operator_yolo_dataset_quality.py /app/scripts/audit_operator_yolo_dataset_quality.py COPY scripts/render_operator_yolo_label_qa_contact_sheets.py /app/scripts/render_operator_yolo_label_qa_contact_sheets.py diff --git a/docs/CODEX_EXECUTION_LOG.md b/docs/CODEX_EXECUTION_LOG.md index c333bc71..5369a485 100644 --- a/docs/CODEX_EXECUTION_LOG.md +++ b/docs/CODEX_EXECUTION_LOG.md @@ -7884,3 +7884,20 @@ Live operational proof: Next: - Build bounded, idempotent regional theme ingestion in municipality-sized partitions, starting with current buildings and retaining per-member provenance before exposing any Kempen-wide metric in the explorer. + +## Sprint 190 Regional Kempen GRB buildings (2026-07-14) + +Implemented: +- Added `provision_regional_grb_buildings.py` for the approved 28-member scope, with complete OGC pagination, per-member safety caps, resumable checksummed artifacts and an explicit total cap. +- Added deterministic cross-boundary ownership by maximum member intersection and NIS-code tie breaking, preventing duplicate source identities without cutting the final geometry at internal municipality borders. +- Added streaming managed-artifact copy plus batch-wise partition indexing through DatasetService and VectorFeatureService. One regional Dataset remains compatible with existing viewport, selection, QA and export ownership boundaries. +- Added exact manifest/index count enforcement, duplicate source-feature rejection and rollback/storage cleanup on import failure. +- Kept fetching operator-triggered and local service import container-only. No startup fetch, browser provider fetch, direct vector-feature SQL or public API change was introduced. + +Validation before deployment: +- Backend/application and operator compilation passed. +- New focused operator/storage/batch-index suite passed 5 tests. +- Dataset, vector persistence, Mol municipality and Kempen scope regression selection passed 37 tests. + +Next: +- Run the complete readiness gate, deploy to Tower, execute the full 28-partition source job, verify exact PostGIS counts and perform full-region plus rectangle browser selections. diff --git a/docs/DATABASE_IMPLEMENTATION_PLAN.md b/docs/DATABASE_IMPLEMENTATION_PLAN.md index 801f5d95..0ef3a246 100644 --- a/docs/DATABASE_IMPLEMENTATION_PLAN.md +++ b/docs/DATABASE_IMPLEMENTATION_PLAN.md @@ -224,6 +224,16 @@ Provider integration is a contract layer only in Sprint 7B. Providers do not wri GRB and OSM live imports are intentionally `not_configured` in Sprint 7B. Manual and fixture reference datasets use existing upload and fixture flows. +Large explicit operator imports may invoke DatasetService directly inside the +backend container when a single multipart upload would require loading the +entire regional artifact into memory. The regional GRB building operator still +creates a normal Dataset and DatasetVersion and delegates every queryable row +to VectorFeatureService. It copies the immutable combined artifact in a +stream, loads one retained municipality partition at a time, flushes bounded +feature batches and commits only when the indexed count matches the manifest. +It does not expose a direct SQL/provider write path and does not alter the +public provider endpoint's `not_configured` status. + ## Geometry normalization - User-drawn polygons arrive as EPSG:4326. diff --git a/docs/DATASET_STRATEGY.md b/docs/DATASET_STRATEGY.md index 30351904..dc77bdd8 100644 --- a/docs/DATASET_STRATEGY.md +++ b/docs/DATASET_STRATEGY.md @@ -106,6 +106,24 @@ boundaries. no interactive request or application startup may download the entire region or silently substitute missing data. +### Regional building partition policy + +The current GRB building snapshot is fetched in one resumable partition per +registered municipality. Partitions are source/provenance artifacts, not 28 +independent user-facing layers. A building intersecting more than one member +boundary is owned by the member with the largest intersection area; the lower +NIS code resolves exact ties. The retained geometry is clipped to the regional +union, preserving complete in-scope coverage while guaranteeing one source ID +per regional dataset. + +After all partitions pass validation, one combined immutable GeoJSON artifact +is copied into managed storage and indexed in bounded batches through the +normal DatasetService/VectorFeatureService ownership boundary. The resulting +reference dataset uses `source_name=grb`, `reference_layer_name=buildings`, +`coverage_scope=kempen-transport-region` and an exact `feature_count` +selection aggregation. Existing viewport and bbox selection APIs remain the +only interactive delivery path. + ## User-uploaded raster strategy V1 must support controlled local datasets because public raster access and model compatibility can be difficult. diff --git a/docs/DATA_SOURCES.md b/docs/DATA_SOURCES.md index 0e17bb33..5adc04c1 100644 --- a/docs/DATA_SOURCES.md +++ b/docs/DATA_SOURCES.md @@ -52,6 +52,21 @@ De scope-operator haalt geen thematische gegevens op. GRB, bevolking en landgebruik volgen als afzonderlijke begrensde imports; ontbrekende regionale thema's blijven zichtbaar onbeschikbaar. +De actuele regionale gebouwlaag wordt afzonderlijk opgehaald door +`scripts/provision_regional_grb_buildings.py`. De operator bevraagt +`GRB/GBG` per officiële gemeentegrens, volgt alle pagina's en weigert elk +veiligheidslimiet als truncatie te verbergen. Grensoverschrijdende objecten +worden aan één gemeentepartitie toegewezen op basis van de grootste overlap; +de uiteindelijke geometrie wordt uitsluitend tegen de volledige regiogrens +gesneden. Daardoor bevat het regionale dataset elke GRB-bronidentiteit exact +één keer. + +De 28 bronpartities, paginabronnen, aantallen en checksums blijven als +operatorbewijs bewaard. GeoIntel indexeert ze batchgewijs als één regionaal +referentiedataset via DatasetService en VectorFeatureService. De publieke GRB +provider blijft `not_configured`: dit is een bewuste operatorrun en geen live +download vanuit een browseractie of applicatiestart. + ### Mol population history `scripts/provision_mol_population_history.py` imports official Statbel diff --git a/scripts/README.md b/scripts/README.md index 46375c10..6c1e1230 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1286,6 +1286,39 @@ This command provisions boundaries only. Regional buildings, population, land use, roads, water and parcels must be added by bounded source operators; missing themes remain unavailable and are never filled with synthetic values. +### Regional Kempen GRB buildings + +After the scope foundation exists, fetch and persist the complete current GRB +`GBG` building layer for the approved region: + +```bash +docker exec -it geointel python3 /app/scripts/provision_regional_grb_buildings.py \ + --scope kempen-transport-region +``` + +Use `--fetch-only` to build and inspect artifacts without touching the +database. The operator uses one bounded request/pagination sequence per +official municipality and writes resumable partitions below +`/app/storage/operator-data/regional-themes/kempen-transport-region/buildings/`. +Every source feature is assigned to exactly one partition using its largest +intersection with the member boundaries; exact ties use the lowest NIS code. +This avoids duplicate building counts at shared borders while retaining the +feature clipped only to the complete regional scope. + +Once all 28 partitions are complete, the operator streams one combined +GeoJSON artifact and invokes `DatasetService`/`VectorFeatureService` inside +the container. Features are indexed in batches as one normal regional +reference dataset, so the existing viewport and rectangle-selection paths do +not need a parallel API. The service import is intentionally local-only and +refuses a remote backend URL. Repeat runs reuse checksummed artifacts and the +persisted dataset; use a new `--observed-date` for a newer immutable snapshot. + +Safety limits can be adjusted explicitly with +`--max-features-per-member`, `--max-total-features`, `--page-limit` and +`--batch-size`. Exceeding a limit fails the run instead of producing a +truncated dataset. The command never runs during startup or an interactive +map query. + ## Tower deployment Push the local branch to Gitea, then rebuild the Unraid/Tower Docker runtime: diff --git a/scripts/provision_regional_grb_buildings.py b/scripts/provision_regional_grb_buildings.py new file mode 100644 index 00000000..813345fa --- /dev/null +++ b/scripts/provision_regional_grb_buildings.py @@ -0,0 +1,743 @@ +"""Provision one queryable GRB building dataset for an approved regional scope. + +Source requests are partitioned by official municipality boundaries and can be +resumed per partition. The retained partition artifacts are then indexed as one +Dataset through DatasetService and VectorFeatureService. This operator never +runs during application startup and never writes directly to vector_features. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sys +import unicodedata +from datetime import date, datetime, time, timezone +from pathlib import Path +from typing import Any, Iterable +from urllib.parse import urlparse +from uuid import UUID + +import requests +from requests.adapters import HTTPAdapter +from shapely.geometry import GeometryCollection, MultiPolygon, Polygon, mapping, shape +from shapely.ops import unary_union +from shapely.validation import make_valid +from urllib3.util.retry import Retry + +from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope, ScopeMember +from provision_geographic_scope import fetch_scope_members + + +GRB_GBG_ITEMS_URL = "https://geo.api.vlaanderen.be/GRB/ogc/features/v1/collections/GBG/items" +GRB_ATTRIBUTION = "Bron: Grootschalig Referentie Bestand Vlaanderen, Digitaal Vlaanderen" +DEFAULT_SCOPE_KEY = "kempen-transport-region" +DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-data/regional-themes") +DEFAULT_API_URL = "http://127.0.0.1:8000" +DEFAULT_PAGE_LIMIT = 1000 +DEFAULT_MAX_FEATURES_PER_MEMBER = 100000 +DEFAULT_MAX_TOTAL_FEATURES = 1500000 +GEOJSON_CRS = {"type": "name", "properties": {"name": "EPSG:4326"}} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Provision municipality-partitioned regional GRB buildings.") + parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY) + parser.add_argument("--observed-date", type=date.fromisoformat, default=date.today()) + parser.add_argument("--base-url", default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL)) + parser.add_argument( + "--output-root", + type=Path, + default=Path(os.environ.get("GEOINTEL_REGIONAL_THEME_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)), + ) + parser.add_argument("--page-limit", type=int, default=DEFAULT_PAGE_LIMIT) + parser.add_argument("--max-features-per-member", type=int, default=DEFAULT_MAX_FEATURES_PER_MEMBER) + parser.add_argument("--max-total-features", type=int, default=DEFAULT_MAX_TOTAL_FEATURES) + parser.add_argument("--request-timeout", type=int, default=180) + parser.add_argument("--api-timeout", type=int, default=180) + parser.add_argument("--batch-size", type=int, default=1000) + parser.add_argument("--force", action="store_true", help="Refetch every municipality partition for this date.") + parser.add_argument("--fetch-only", action="store_true", help="Build and validate artifacts without persistence.") + return parser.parse_args() + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def observed_at(value: date) -> datetime: + return datetime.combine(value, time.min, tzinfo=timezone.utc) + + +def safe_slug(value: str) -> str: + normalized = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii") + return "-".join(part for part in "".join(char.lower() if char.isalnum() else " " for char in normalized).split()) + + +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_json_atomic(path: Path, payload: dict[str, 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( + payload, + ensure_ascii=False, + indent=2 if pretty else None, + separators=None if pretty else (",", ":"), + sort_keys=pretty, + ), + encoding="utf-8", + ) + temporary.replace(path) + + +def normalize_polygonal(geometry): + if geometry is None or 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 build_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-Regional-GRB-Buildings-Operator/1.0"}) + session.mount("https://", adapter) + session.mount("http://", adapter) + return session + + +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 iter_grb_pages( + session: requests.Session, + 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 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() + if payload.get("type") != "FeatureCollection": + raise RuntimeError("GRB returned a non-FeatureCollection response") + yield payload, response.url + url = next_page_url(payload) + + +def build_member_geometries( + scope: GeographicScope, + source_features: list[dict[str, Any]], +) -> tuple[list[tuple[ScopeMember, Any]], Any]: + if len(source_features) != len(scope.members): + raise RuntimeError(f"Expected {len(scope.members)} scope boundaries, received {len(source_features)}") + members: list[tuple[ScopeMember, Any]] = [] + for member, source_feature in zip(scope.members, source_features, strict=True): + properties = source_feature.get("properties") or {} + if str(properties.get("NISCODE") or "") != member.nis_code: + raise RuntimeError(f"VRBG scope member mismatch for {member.name}") + geometry = normalize_polygonal(shape(source_feature.get("geometry"))) + if geometry is None: + raise RuntimeError(f"Invalid official boundary for {member.name}") + members.append((member, geometry)) + regional_boundary = normalize_polygonal(unary_union([geometry for _, geometry in members])) + if regional_boundary is None: + raise RuntimeError("The regional scope union is invalid") + return members, regional_boundary + + +def bounds_overlap(left: tuple[float, float, float, float], right: tuple[float, float, float, float]) -> bool: + return left[0] <= right[2] and left[2] >= right[0] and left[1] <= right[3] and left[3] >= right[1] + + +def assign_owner_nis(source_geometry, members: list[tuple[ScopeMember, Any]]) -> str | None: + candidates: list[tuple[float, str]] = [] + source_bounds = source_geometry.bounds + for member, boundary in members: + if not bounds_overlap(source_bounds, boundary.bounds) or not source_geometry.intersects(boundary): + continue + intersection = source_geometry.intersection(boundary) + if not intersection.is_empty and intersection.area > 0: + candidates.append((float(intersection.area), member.nis_code)) + if not candidates: + return None + candidates.sort(key=lambda item: (-item[0], item[1])) + return candidates[0][1] + + +def build_partition_features( + pages: Iterable[tuple[dict[str, Any], str]], + *, + member: ScopeMember, + members: list[tuple[ScopeMember, Any]], + regional_boundary, + scope: GeographicScope, + 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 + assigned_elsewhere_count = 0 + outside_scope_count = 0 + clipped_to_scope_count = 0 + + for payload, source_url in pages: + source_urls.append(source_url) + for source_feature in payload.get("features") or []: + bbox_feature_count += 1 + 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) + source_geometry = normalize_polygonal(shape(source_feature.get("geometry"))) + if source_geometry is None or not source_geometry.intersects(regional_boundary): + outside_scope_count += 1 + continue + owner_nis = assign_owner_nis(source_geometry, members) + if owner_nis != member.nis_code: + assigned_elsewhere_count += 1 + continue + clipped_geometry = source_geometry + if not source_geometry.within(regional_boundary): + clipped_geometry = normalize_polygonal(source_geometry.intersection(regional_boundary)) + clipped_to_scope_count += 1 + if clipped_geometry is None: + outside_scope_count += 1 + continue + if len(features) >= max_features: + raise RuntimeError( + f"{member.name} exceeds --max-features-per-member={max_features}; refusing truncated output" + ) + properties = dict(source_feature.get("properties") or {}) + properties.update( + { + "source_name": "grb", + "source_feature_id": feature_id, + "reference_layer_name": "buildings", + "layer_type": "building", + "theme": "buildings", + "authority_level": "authoritative", + "coverage_scope": scope.key, + "scope_type": scope.scope_type, + "partition_scope": "municipality", + "partition_municipality": member.name, + "partition_nis_code": member.nis_code, + "partition_assignment": "maximum_boundary_intersection", + "clipped_to_regional_scope": clipped_geometry is not source_geometry, + "attribution": GRB_ATTRIBUTION, + } + ) + features.append( + { + "type": "Feature", + "id": feature_id, + "geometry": mapping(clipped_geometry), + "properties": properties, + } + ) + + if not features: + raise RuntimeError(f"No GRB buildings were assigned to {member.name}") + return features, { + "municipality": member.name, + "nis_code": member.nis_code, + "pages_fetched": len(source_urls), + "source_urls": source_urls, + "bbox_feature_count": bbox_feature_count, + "feature_count": len(features), + "assigned_elsewhere_count": assigned_elsewhere_count, + "outside_scope_count": outside_scope_count, + "clipped_to_scope_count": clipped_to_scope_count, + "reference_truncated": False, + } + + +def partition_filename(member: ScopeMember, observed_date: date) -> str: + return f"{member.nis_code}_{safe_slug(member.name)}_grb_buildings_{observed_date.isoformat()}.geojson" + + +def combined_filename(scope: GeographicScope, observed_date: date) -> str: + return f"grb_buildings_{scope.key.replace('-', '_')}_{observed_date.isoformat()}.geojson" + + +def write_partition( + path: Path, + *, + scope: GeographicScope, + member: ScopeMember, + features: list[dict[str, Any]], + generated_at: str, + source_url: str, +) -> None: + write_json_atomic( + path, + { + "type": "FeatureCollection", + "name": f"GRB buildings - {member.name} partition of {scope.display_name}", + "crs": GEOJSON_CRS, + "features": features, + "source": "Digitaal Vlaanderen GRB OGC API Features collection GBG", + "source_url": source_url, + "attribution": GRB_ATTRIBUTION, + "coverage_scope": scope.key, + "partition_municipality": member.name, + "partition_nis_code": member.nis_code, + "generated_at": generated_at, + }, + ) + + +def write_combined_artifact( + path: Path, + *, + scope: GeographicScope, + observed_date: date, + partition_paths: list[Path], + expected_feature_count: int, +) -> dict[str, Any]: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(f"{path.suffix}.partial") + header = { + "type": "FeatureCollection", + "name": f"GRB buildings - {scope.display_name}", + "crs": GEOJSON_CRS, + "source": "Digitaal Vlaanderen GRB OGC API Features collection GBG", + "source_url": GRB_GBG_ITEMS_URL, + "attribution": GRB_ATTRIBUTION, + "coverage_scope": scope.key, + "scope_type": scope.scope_type, + "partition_count": len(partition_paths), + "partition_strategy": "municipality_bbox_maximum_boundary_intersection", + "observed_at": observed_date.isoformat(), + } + encoded_header = json.dumps(header, ensure_ascii=False, separators=(",", ":")) + seen_ids: set[str] = set() + written = 0 + first = True + with temporary.open("w", encoding="utf-8", newline="") as output: + output.write(encoded_header[:-1]) + output.write(',"features":[') + for partition_path in partition_paths: + payload = json.loads(partition_path.read_text(encoding="utf-8")) + if payload.get("type") != "FeatureCollection" or not isinstance(payload.get("features"), list): + raise RuntimeError(f"Invalid partition artifact: {partition_path}") + for feature in payload["features"]: + feature_id = str(feature.get("id") or (feature.get("properties") or {}).get("source_feature_id") or "") + if not feature_id: + raise RuntimeError(f"Partition feature without source identity in {partition_path.name}") + if feature_id in seen_ids: + raise RuntimeError(f"Duplicate regional source feature {feature_id} in {partition_path.name}") + seen_ids.add(feature_id) + if not first: + output.write(",") + output.write(json.dumps(feature, ensure_ascii=False, separators=(",", ":"))) + first = False + written += 1 + output.write("]}") + if written != expected_feature_count: + temporary.unlink(missing_ok=True) + raise RuntimeError(f"Expected {expected_feature_count} combined features, wrote {written}") + temporary.replace(path) + return { + "feature_count": written, + "size_bytes": path.stat().st_size, + "sha256": sha256_file(path), + } + + +def reusable_manifest( + manifest_path: Path, + artifact_path: Path, + partition_dir: Path, + expected_member_count: int, +) -> dict[str, Any] | None: + if not manifest_path.is_file() or not artifact_path.is_file(): + return None + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + if manifest.get("status") != "complete" or len(manifest.get("partitions") or []) != expected_member_count: + return None + if sha256_file(artifact_path) != manifest.get("artifact_sha256"): + return None + for summary in manifest["partitions"]: + path = partition_dir / str(summary.get("filename") or "") + if not path.is_file() or sha256_file(path) != summary.get("sha256"): + return None + return manifest + + +def prepare_artifacts( + args: argparse.Namespace, + scope: GeographicScope, +) -> tuple[Path, list[Path], Path, dict[str, Any]]: + if args.page_limit <= 0 or args.max_features_per_member <= 0 or args.max_total_features <= 0: + raise RuntimeError("Page and feature limits must be positive") + observation_dir = args.output_root / scope.key / "buildings" / args.observed_date.isoformat() + partition_dir = observation_dir / "partitions" + artifact_path = observation_dir / combined_filename(scope, args.observed_date) + manifest_path = observation_dir / "regional_buildings_manifest.json" + partition_dir.mkdir(parents=True, exist_ok=True) + + if not args.force: + existing = reusable_manifest(manifest_path, artifact_path, partition_dir, len(scope.members)) + if existing: + paths = [partition_dir / summary["filename"] for summary in existing["partitions"]] + return artifact_path, paths, manifest_path, existing + + existing_manifest: dict[str, Any] = {} + if manifest_path.is_file() and not args.force: + existing_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + existing_by_nis = { + str(item.get("nis_code")): item + for item in existing_manifest.get("partitions") or [] + if isinstance(item, dict) + } + generated_at = utc_now() + with build_source_session() as session: + source_features, vrbg_source_url = fetch_scope_members(session, scope, args.request_timeout) + members, regional_boundary = build_member_geometries(scope, source_features) + summaries: list[dict[str, Any]] = [] + for member, boundary in members: + path = partition_dir / partition_filename(member, args.observed_date) + reusable = existing_by_nis.get(member.nis_code) + if ( + reusable + and path.is_file() + and reusable.get("filename") == path.name + and reusable.get("sha256") == sha256_file(path) + ): + summaries.append(reusable) + continue + features, summary = build_partition_features( + iter_grb_pages( + session, + boundary.bounds, + page_limit=args.page_limit, + timeout=args.request_timeout, + ), + member=member, + members=members, + regional_boundary=regional_boundary, + scope=scope, + max_features=args.max_features_per_member, + ) + write_partition( + path, + scope=scope, + member=member, + features=features, + generated_at=generated_at, + source_url=summary["source_urls"][0], + ) + summary.update({"filename": path.name, "size_bytes": path.stat().st_size, "sha256": sha256_file(path)}) + summaries.append(summary) + progress = { + "schema_version": 1, + "status": "in_progress", + "scope": scope.key, + "theme": "buildings", + "observed_at": args.observed_date.isoformat(), + "generated_at": generated_at, + "vrbg_source_url": vrbg_source_url, + "grb_source_url": GRB_GBG_ITEMS_URL, + "partitions": summaries, + } + write_json_atomic(manifest_path, progress, pretty=True) + + total_features = sum(int(summary["feature_count"]) for summary in summaries) + if total_features > args.max_total_features: + raise RuntimeError( + f"Regional building count {total_features} exceeds --max-total-features={args.max_total_features}" + ) + partition_paths = [partition_dir / summary["filename"] for summary in summaries] + artifact = write_combined_artifact( + artifact_path, + scope=scope, + observed_date=args.observed_date, + partition_paths=partition_paths, + expected_feature_count=total_features, + ) + manifest = { + "schema_version": 1, + "status": "complete", + "scope": scope.key, + "scope_type": scope.scope_type, + "scope_authority_url": scope.authority_url, + "scope_limitation": scope.limitation_message, + "theme": "buildings", + "observed_at": args.observed_date.isoformat(), + "generated_at": generated_at, + "member_count": len(scope.members), + "feature_count": total_features, + "reference_truncated": False, + "partition_strategy": "municipality_bbox_maximum_boundary_intersection", + "partition_assignment_rule": "largest geometry intersection; NIS code resolves exact ties", + "vrbg_source_url": vrbg_source_url, + "grb_source_url": GRB_GBG_ITEMS_URL, + "artifact_filename": artifact_path.name, + "artifact_size_bytes": artifact["size_bytes"], + "artifact_sha256": artifact["sha256"], + "bounds_json": { + "min_x": float(regional_boundary.bounds[0]), + "min_y": float(regional_boundary.bounds[1]), + "max_x": float(regional_boundary.bounds[2]), + "max_y": float(regional_boundary.bounds[3]), + }, + "partitions": summaries, + "attribution": GRB_ATTRIBUTION, + } + write_json_atomic(manifest_path, manifest, pretty=True) + return artifact_path, partition_paths, manifest_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.status_code}): {response.text[:500]}") from exc + if not response.ok: + raise RuntimeError(f"GeoIntel API request failed ({response.status_code}): {json.dumps(payload)[: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 list_paginated_items(session: requests.Session, url: str, timeout: int) -> list[dict[str, Any]]: + items: list[dict[str, Any]] = [] + offset = 0 + while True: + page = response_data(session.get(url, params={"limit": 200, "offset": offset}, timeout=timeout)) + page_items = list(page.get("items") or []) + items.extend(page_items) + total = int(page.get("total") or len(items)) + if not page_items or len(items) >= total: + return items + offset += len(page_items) + + +def ensure_backend_path() -> None: + repository_root = Path(__file__).resolve().parents[1] + backend_root = repository_root / "backend" if (repository_root / "backend" / "app").is_dir() else repository_root + if str(backend_root) not in sys.path: + sys.path.insert(0, str(backend_root)) + + +def provision_dataset( + args: argparse.Namespace, + scope: GeographicScope, + artifact_path: Path, + partition_paths: list[Path], + manifest_path: Path, + manifest: dict[str, Any], +) -> dict[str, Any]: + parsed_base = urlparse(args.base_url) + if parsed_base.hostname not in {"127.0.0.1", "localhost", "::1"}: + raise RuntimeError("Partitioned service import must run inside the GeoIntel container against its local backend") + with requests.Session() as session: + projects = list_paginated_items(session, f"{args.base_url.rstrip('/')}/api/v1/projects", args.api_timeout) + project = next((item for item in projects if item.get("name") == scope.project_name), None) + if not project: + raise RuntimeError("Regional scope project is missing; run provision_geographic_scope.py first") + project_id = str(project["id"]) + areas = list_paginated_items( + session, + f"{args.base_url.rstrip('/')}/api/v1/projects/{project_id}/areas", + args.api_timeout, + ) + area = next((item for item in areas if item.get("name") == scope.area_name), None) + if not area: + raise RuntimeError("Regional scope Area is missing; run provision_geographic_scope.py first") + datasets = list_paginated_items( + session, + f"{args.base_url.rstrip('/')}/api/v1/projects/{project_id}/datasets", + args.api_timeout, + ) + existing = next((item for item in datasets if item.get("original_filename") == artifact_path.name), None) + if existing: + persisted_checksum = (existing.get("provenance_metadata") or {}).get("artifact_sha256") + if persisted_checksum and persisted_checksum != manifest["artifact_sha256"]: + raise RuntimeError( + f"Immutable dataset {artifact_path.name} checksum changed; use a new --observed-date for refreshed GRB data" + ) + return { + "dataset_id": str(existing["id"]), + "feature_count": existing.get("feature_count"), + "reused": True, + } + + ensure_backend_path() + from app.db.session import SessionLocal + from app.services.dataset_service import DatasetService + + partition_checksums = { + summary["nis_code"]: summary["sha256"] + for summary in manifest["partitions"] + } + metadata_json = { + "feature_count": manifest["feature_count"], + "feature_geometry_count": manifest["feature_count"], + "geometry_types": ["MultiPolygon", "Polygon"], + "bounds_json": manifest["bounds_json"], + "approximate_area_m2": None, + "invalid_features": 0, + "z_dimension_feature_count": 0, + "canonical_storage_dimension": "2D", + "crs": "EPSG:4326", + "crs_assumed": False, + "extracted_at": manifest["generated_at"], + } + source_metadata = { + "provider": "Digitaal Vlaanderen", + "collection": "GRB/GBG", + "authority_level": "authoritative", + "theme": "buildings", + "layer_type": "regional_buildings", + "coverage_scope": scope.key, + "scope_type": scope.scope_type, + "scope_authority": scope.authority_name, + "scope_authority_url": scope.authority_url, + "scope_limitation": scope.limitation_message, + "member_count": len(scope.members), + "member_nis_codes": list(scope.nis_codes), + "feature_count": manifest["feature_count"], + "partition_count": len(partition_paths), + "partition_strategy": manifest["partition_strategy"], + "selection_aggregation": { + "method": "feature_count", + "label": "Gebouwen", + "unit": "objecten", + "is_estimate": False, + }, + "attribution": GRB_ATTRIBUTION, + } + provenance_metadata = { + "operator_tool": "provision_regional_grb_buildings.py", + "operator_explicit_fetch": True, + "manifest_path": str(manifest_path), + "source_url": GRB_GBG_ITEMS_URL, + "artifact_sha256": manifest["artifact_sha256"], + "artifact_size_bytes": manifest["artifact_size_bytes"], + "partition_checksums": partition_checksums, + "partition_assignment_rule": manifest["partition_assignment_rule"], + "reference_truncated": False, + } + with SessionLocal() as db: + dataset = DatasetService.import_partitioned_vector_artifact( + db, + project_id=UUID(project_id), + area_id=UUID(str(area["id"])), + artifact_path=artifact_path, + partition_paths=partition_paths, + original_filename=artifact_path.name, + source="operator_official_import", + dataset_role="reference", + source_name="grb", + reference_layer_name="buildings", + metadata_json=metadata_json, + source_metadata=source_metadata, + provenance_metadata=provenance_metadata, + temporal_series_key=f"grb:buildings:{scope.key}", + observed_at=observed_at(args.observed_date), + temporal_granularity="snapshot", + source_version=args.observed_date.isoformat(), + batch_size=args.batch_size, + ) + if dataset.checksum_sha256 != manifest["artifact_sha256"]: + raise RuntimeError("Managed dataset checksum differs from the retained regional artifact") + return {"dataset_id": str(dataset.id), "feature_count": dataset.feature_count, "reused": False} + + +def main() -> int: + args = parse_args() + scope = GEOGRAPHIC_SCOPES[args.scope] + try: + artifact_path, partition_paths, manifest_path, manifest = prepare_artifacts(args, scope) + persistence = None if args.fetch_only else provision_dataset( + args, + scope, + artifact_path, + partition_paths, + manifest_path, + manifest, + ) + except (OSError, RuntimeError, ValueError, KeyError, requests.RequestException) as exc: + print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr) + return 1 + print( + json.dumps( + { + "status": "ok", + "mode": "fetch_only" if args.fetch_only else "provisioned", + "scope": scope.key, + "theme": "buildings", + "observed_at": args.observed_date.isoformat(), + "member_count": manifest["member_count"], + "feature_count": manifest["feature_count"], + "artifact_size_bytes": manifest["artifact_size_bytes"], + "artifact_path": str(artifact_path), + "manifest_path": str(manifest_path), + "reference_truncated": manifest["reference_truncated"], + "persistence": persistence, + }, + ensure_ascii=False, + indent=2, + ) + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/run_readiness_check.sh b/scripts/run_readiness_check.sh index 01942646..9017be9f 100755 --- a/scripts/run_readiness_check.sh +++ b/scripts/run_readiness_check.sh @@ -49,6 +49,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_mol_historical_landuse.py ${PYTHON_BIN} -m py_compile scripts/provision_official_landuse_timeseries.py ${PYTHON_BIN} -m py_compile scripts/geographic_scopes.py ${PYTHON_BIN} -m py_compile scripts/provision_geographic_scope.py +${PYTHON_BIN} -m py_compile scripts/provision_regional_grb_buildings.py ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_dataset.py ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_tile_dataset.py ${PYTHON_BIN} -m py_compile scripts/audit_operator_yolo_dataset_quality.py