feat: provision regional Kempen buildings

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`,

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)

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,

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

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