Add governed bathymetry profile workflow

backend/app/api/routes/datasets.py

@@ -27,6 +27,7 @@ from app.schemas import (
     FloodHazardAcquireRequest,
     FloodHazardPartitionSelectionRequest,
     FloodHazardSelectionRequest,
+    BathymetryProfileAcquireRequest,
     ThematicRasterAcquireRequest,
     ThematicRasterSelectionRequest,
     VectorBBoxResponse,
@@ -52,6 +53,7 @@ from app.services.dhmv_acquisition_service import DhmvAcquisitionService
 from app.services.terrain_analysis_service import TerrainAnalysisService
 from app.services.flood_hazard_acquisition_service import FloodHazardAcquisitionService
 from app.services.flood_hazard_analysis_service import FloodHazardAnalysisService
+from app.services.bathymetry_profile_acquisition_service import BathymetryProfileAcquisitionService
 from app.services.thematic_raster_acquisition_service import ThematicRasterAcquisitionService
 from app.services.thematic_raster_analysis_service import ThematicRasterAnalysisService
 from app.utils.response import envelope
@@ -213,6 +215,30 @@ def list_flood_hazard_products(project_id: UUID, db: Session = Depends(get_db)):
     return envelope({"items": items, "total": len(items)})
 
 
+@router.get("/datasets/bathymetry/sources", response_model=dict)
+def list_bathymetry_sources(project_id: UUID, db: Session = Depends(get_db)):
+    if not db.get(Project, project_id):
+        raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
+    items = BathymetryProfileAcquisitionService.list_sources()
+    return envelope({"items": items, "total": len(items)})
+
+
+@router.post("/datasets/bathymetry/profiles/acquire", response_model=dict)
+def acquire_bounded_bathymetry_profiles(
+    project_id: UUID,
+    payload: BathymetryProfileAcquireRequest,
+    db: Session = Depends(get_db),
+):
+    job = JobService.run_sync_job(
+        db=db,
+        project_id=project_id,
+        job_type="vector.bathymetry_profiles.acquire",
+        parameters=payload.model_dump(mode="json"),
+        operation=lambda: BathymetryProfileAcquisitionService.acquire(db, project_id, payload),
+    )
+    return envelope(job)
+
+
 @router.post("/datasets/thematic-raster/acquire", response_model=dict)
 def acquire_bounded_thematic_raster(
     project_id: UUID,

backend/app/core/config.py

@@ -90,6 +90,39 @@ class Settings(BaseSettings):
     flood_hazard_max_pixels: int = Field(default=12_000_000, ge=1, validation_alias="FLOOD_HAZARD_MAX_PIXELS")
     flood_hazard_timeout_seconds: int = Field(default=300, ge=1, validation_alias="FLOOD_HAZARD_TIMEOUT_SECONDS")
     flood_hazard_max_response_mb: int = Field(default=160, ge=1, validation_alias="FLOOD_HAZARD_MAX_RESPONSE_MB")
+    bathymetry_profiles_enabled: bool = Field(default=True, validation_alias="BATHYMETRY_PROFILES_ENABLED")
+    bathymetry_profiles_layer_url: str = Field(
+        default="https://vha.waterinfo.be/arcgis/rest/services/digitale_atlas/MapServer/0",
+        validation_alias="BATHYMETRY_PROFILES_LAYER_URL",
+    )
+    bathymetry_watercourse_layer_url: str = Field(
+        default="https://vha.waterinfo.be/arcgis/rest/services/digitale_atlas/MapServer/1",
+        validation_alias="BATHYMETRY_WATERCOURSE_LAYER_URL",
+    )
+    bathymetry_profiles_page_size: int = Field(
+        default=1000,
+        ge=1,
+        le=2000,
+        validation_alias="BATHYMETRY_PROFILES_PAGE_SIZE",
+    )
+    bathymetry_profiles_max_features: int = Field(
+        default=50_000,
+        ge=1,
+        le=250_000,
+        validation_alias="BATHYMETRY_PROFILES_MAX_FEATURES",
+    )
+    bathymetry_profiles_timeout_seconds: int = Field(
+        default=120,
+        ge=1,
+        le=600,
+        validation_alias="BATHYMETRY_PROFILES_TIMEOUT_SECONDS",
+    )
+    bathymetry_profiles_max_response_mb: int = Field(
+        default=32,
+        ge=1,
+        le=256,
+        validation_alias="BATHYMETRY_PROFILES_MAX_RESPONSE_MB",
+    )
     thematic_raster_enabled: bool = Field(default=True, validation_alias="THEMATIC_RASTER_ENABLED")
     thematic_raster_wcs_url: str = Field(
         default="https://www.mercator.vlaanderen.be/raadpleegdienstenmercatorpubliek/wcs",

backend/app/schemas/__init__.py

@@ -65,6 +65,11 @@ from .flood_hazard import (
     FloodHazardSelectionResponse,
     FloodHazardSelectionSummary,
 )
+from .bathymetry import (
+    BathymetryProfileAcquireRequest,
+    BathymetryProfileAcquisitionResult,
+    BathymetrySourceRead,
+)
 from .thematic_raster import (
     ThematicRasterAcquireRequest,
     ThematicRasterAcquisitionResult,
@@ -202,6 +207,9 @@ __all__ = [
     "FloodHazardSelectionRequest",
     "FloodHazardSelectionResponse",
     "FloodHazardSelectionSummary",
+    "BathymetryProfileAcquireRequest",
+    "BathymetryProfileAcquisitionResult",
+    "BathymetrySourceRead",
     "ThematicRasterAcquireRequest",
     "ThematicRasterAcquisitionResult",
     "ThematicRasterMetric",

backend/app/schemas/bathymetry.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from typing import Literal
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+from .operations import VectorSelectionBBox
+
+
+class BathymetryProfileAcquireRequest(BaseModel):
+    bbox: VectorSelectionBBox
+    area_id: UUID | None = None
+    force_refresh: bool = False
+
+
+class BathymetrySourceRead(BaseModel):
+    key: str
+    display_name: str
+    owner: str
+    authority_level: Literal["authoritative", "contextual"]
+    geographic_coverage: str
+    data_kind: str
+    query_modes: list[str]
+    vertical_reference: str
+    horizontal_crs: str
+    native_resolution: str | None = None
+    integration_status: Literal["operational", "available_not_integrated", "catalog_only"]
+    acquisition_supported: bool
+    configured: bool
+    service_url: str | None = None
+    catalog_url: str
+    attribution: str
+    license_note: str
+    limitation_message: str
+
+
+class BathymetryProfileAcquisitionResult(BaseModel):
+    output_dataset_id: UUID
+    reused: bool
+    provider: str
+    profile_count: int = Field(ge=0)
+    document_count: int = Field(ge=0)
+    structured_depth_count: int = Field(ge=0)
+    structured_width_count: int = Field(ge=0)
+    watercourse_count: int = Field(ge=0)
+    bbox_epsg4326: list[float]
+    clipped_to_area_id: UUID | None = None
+    measurement_date_min: str | None = None
+    measurement_date_max: str | None = None
+    attribution: str
+    limitation_message: str

backend/app/services/bathymetry_profile_acquisition_service.py

@@ -0,0 +1,671 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+import hashlib
+import json
+import math
+from typing import Any, Callable
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+from uuid import UUID
+
+from geoalchemy2.shape import to_shape
+from shapely.geometry import Point, box, mapping
+
+from app.core.config import Settings, get_settings
+from app.core.errors import AppError
+from app.models import Area, Dataset, Project
+from app.schemas.bathymetry import (
+    BathymetryProfileAcquireRequest,
+    BathymetryProfileAcquisitionResult,
+    BathymetrySourceRead,
+)
+from app.services.dataset_service import DatasetService
+
+
+class BathymetryProfileAcquisitionService:
+    PROVIDER = "vmm_vha_bathymetry_profiles"
+    SOURCE_VERSION = "VHA digitale atlas ArcGIS MapServer"
+    PROFILE_OUT_FIELDS = (
+        "OBJECTID,vhag,atlaspunt,opg_kruinb,opg_vloerb,d_opmeti,"
+        "hyperlink,bron,kunstwerkid,opg_diepte"
+    )
+    ATTRIBUTION = "Vlaamse Milieumaatschappij (VMM), Vlaamse Hydrografische Atlas"
+    LICENSE_NOTE = "Hergebruik volgens de voorwaarden van de Vlaamse overheid en de bronmetadata."
+    LIMITATION = (
+        "Dwarsprofielen zijn historische puntmetingen met bronafhankelijke meetdatum en verticale referentie. "
+        "Ze vormen geen continue actuele bodemkaart en ondersteunen zonder gelijktijdig waterpeil geen "
+        "gebiedsdekkend of actueel watervolume."
+    )
+    _SOURCES = (
+        {
+            "key": "vha_inland_profiles",
+            "display_name": "VHA dwarsprofielen binnenwater",
+            "owner": "Vlaamse Milieumaatschappij",
+            "authority_level": "authoritative",
+            "geographic_coverage": "Vlaanderen, puntlocaties op gekarteerde waterlopen",
+            "data_kind": "dwarsprofielpunten met meetvelden en brondocumenten",
+            "query_modes": ["bbox", "persisted_area"],
+            "vertical_reference": "document-specific; niet uniform als één peilreferentie te behandelen",
+            "horizontal_crs": "EPSG:4326",
+            "native_resolution": None,
+            "integration_status": "operational",
+            "acquisition_supported": True,
+            "configured": True,
+            "service_url": "https://vha.waterinfo.be/arcgis/rest/services/digitale_atlas/MapServer/0",
+            "catalog_url": "https://www.vlaanderen.be/datavindplaats/catalogus/vlaamse-hydrografische-atlas-waterlopen",
+            "attribution": ATTRIBUTION,
+            "license_note": LICENSE_NOTE,
+            "limitation_message": LIMITATION,
+        },
+        {
+            "key": "mdk_bcp_bathymetry",
+            "display_name": "Dieptemodel Belgisch Continentaal Plat",
+            "owner": "Agentschap Maritieme Dienstverlening en Kust",
+            "authority_level": "authoritative",
+            "geographic_coverage": "Belgisch Continentaal Plat en Noordzee",
+            "data_kind": "continu bathymetrisch raster",
+            "query_modes": ["wcs", "wmts", "bounded_raster"],
+            "vertical_reference": "LAT",
+            "horizontal_crs": "bronafhankelijk; expliciet per WCS-respons",
+            "native_resolution": "20 x 20 m",
+            "integration_status": "available_not_integrated",
+            "acquisition_supported": False,
+            "configured": False,
+            "service_url": "https://bathy.agentschapmdk.be/spatialfusionserver/services/ows/wcs/WCS_Public",
+            "catalog_url": "https://www.vlaanderen.be/datavindplaats/catalogus/dieptemodel-van-de-zeebodem-belgisch-continentaal-plat-noordzee",
+            "attribution": "Agentschap Maritieme Dienstverlening en Kust",
+            "license_note": "Zie de officiële datasetmetadata en gebruiksvoorwaarden.",
+            "limitation_message": (
+                "Niet operationeel in GeoIntel totdat WCS, maritieme begrenzing, tegels, LAT-semantiek en "
+                "servercertificaten in een live integratietest zijn gevalideerd."
+            ),
+        },
+        {
+            "key": "spw_walloon_waterway_bathymetry",
+            "display_name": "Bathymétrie des voies navigables et lacs-réservoirs",
+            "owner": "Service public de Wallonie",
+            "authority_level": "authoritative",
+            "geographic_coverage": "Waalse bevaarbare waterwegen en stuwmeren met uitgevoerde opmetingen",
+            "data_kind": "bodemhoogteraster en XYZ-puntenwolk",
+            "query_modes": ["download", "arcgis_map_service"],
+            "vertical_reference": "mDNG",
+            "horizontal_crs": "EPSG:3812; visualisatieservice kan EPSG:31370 aanbieden",
+            "native_resolution": "0,5 m",
+            "integration_status": "available_not_integrated",
+            "acquisition_supported": False,
+            "configured": False,
+            "service_url": "https://geoservices.wallonie.be/arcgis/rest/services/EAU/BATHY/MapServer",
+            "catalog_url": "https://geoportail.wallonie.be/catalogue/c450c28f-d357-48af-8423-62d524632cf9.html",
+            "attribution": "Service public de Wallonie",
+            "license_note": "CC BY 4.0 volgens de officiële Geoportail-metadata.",
+            "limitation_message": (
+                "Dekking en meetjaar verschillen per vaarweg of reservoir. Integratie vereist een beheerde "
+                "download- en mosaïekstroom plus expliciete omzetting van mDNG."
+            ),
+        },
+        {
+            "key": "port_antwerp_bathymetry",
+            "display_name": "Havenbathymetrie Antwerpen-Brugge",
+            "owner": "Port of Antwerp-Bruges",
+            "authority_level": "contextual",
+            "geographic_coverage": "Gepubliceerde havenzones en meetcampagnes",
+            "data_kind": "periodieke peilingen",
+            "query_modes": ["catalog"],
+            "vertical_reference": "product-specific",
+            "horizontal_crs": "product-specific",
+            "native_resolution": None,
+            "integration_status": "catalog_only",
+            "acquisition_supported": False,
+            "configured": False,
+            "service_url": None,
+            "catalog_url": "https://data.gov.be/nl/datasets",
+            "attribution": "Port of Antwerp-Bruges",
+            "license_note": "Per publicatie te verifiëren.",
+            "limitation_message": (
+                "Alleen als cataloguskandidaat geregistreerd; er is nog geen stabiel, publiek en machineleesbaar "
+                "acquisitiecontract in GeoIntel gevalideerd."
+            ),
+        },
+    )
+
+    @staticmethod
+    def list_sources() -> list[dict[str, Any]]:
+        return [BathymetrySourceRead(**item).model_dump() for item in BathymetryProfileAcquisitionService._SOURCES]
+
+    @staticmethod
+    def _validate_bbox(payload: BathymetryProfileAcquireRequest) -> tuple[float, float, float, float]:
+        bbox = payload.bbox
+        if bbox.crs.upper() != "EPSG:4326":
+            raise AppError(
+                code="BATHYMETRY_INVALID_CRS",
+                message="Bathymetry profile acquisition requires EPSG:4326",
+                status_code=400,
+            )
+        values = (bbox.min_x, bbox.min_y, bbox.max_x, bbox.max_y)
+        if not all(math.isfinite(value) for value in values):
+            raise AppError(code="BATHYMETRY_INVALID_BBOX", message="Bounding box values must be finite", status_code=400)
+        if bbox.min_x >= bbox.max_x or bbox.min_y >= bbox.max_y:
+            raise AppError(code="BATHYMETRY_INVALID_BBOX", message="Bounding box has no area", status_code=400)
+        if bbox.min_x < -180 or bbox.max_x > 180 or bbox.min_y < -90 or bbox.max_y > 90:
+            raise AppError(code="BATHYMETRY_INVALID_BBOX", message="Bounding box is outside EPSG:4326", status_code=400)
+        return values
+
+    @staticmethod
+    def _scope_geometry(db, project_id: UUID, area_id: UUID | None, bbox_values: tuple[float, float, float, float]):
+        if not db.get(Project, project_id):
+            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
+        selection = box(*bbox_values)
+        if area_id is None:
+            return selection
+        area = db.get(Area, area_id)
+        if area is None:
+            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)
+        area_geometry = area.geometry if hasattr(area.geometry, "__geo_interface__") else to_shape(area.geometry)
+        intersection = area_geometry.intersection(selection)
+        if intersection.is_empty:
+            raise AppError(
+                code="BATHYMETRY_SCOPE_EMPTY",
+                message="The requested bounding box does not intersect the selected area",
+                status_code=400,
+            )
+        return intersection
+
+    @staticmethod
+    def _query_url(base_url: str, parameters: dict[str, Any]) -> str:
+        return f"{base_url}?{urlencode(parameters)}"
+
+    @staticmethod
+    def _fetch_json(
+        url: str,
+        settings: Settings,
+        opener: Callable[..., Any] | None,
+    ) -> tuple[dict[str, Any], str]:
+        request = Request(
+            url,
+            headers={
+                "Accept": "application/json",
+                "User-Agent": "GeoIntel/1.0 bathymetry-profile-acquisition",
+            },
+        )
+        try:
+            with (opener or urlopen)(request, timeout=settings.bathymetry_profiles_timeout_seconds) as response:
+                limit = settings.bathymetry_profiles_max_response_mb * 1024 * 1024
+                content = response.read(limit + 1)
+        except HTTPError as exc:
+            raise AppError(
+                code="BATHYMETRY_PROVIDER_HTTP_ERROR",
+                message="VHA profile service returned an HTTP error",
+                details={"status_code": exc.code},
+                status_code=502,
+            ) from exc
+        except (TimeoutError, URLError, OSError) as exc:
+            raise AppError(
+                code="BATHYMETRY_PROVIDER_UNAVAILABLE",
+                message="VHA profile service is unavailable",
+                status_code=502,
+            ) from exc
+        if len(content) > limit:
+            raise AppError(
+                code="BATHYMETRY_PROVIDER_RESPONSE_TOO_LARGE",
+                message="VHA profile response exceeded the configured size limit",
+                status_code=502,
+            )
+        try:
+            payload = json.loads(content.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise AppError(
+                code="BATHYMETRY_PROVIDER_INVALID_RESPONSE",
+                message="VHA profile service returned invalid JSON",
+                status_code=502,
+            ) from exc
+        if not isinstance(payload, dict) or payload.get("error"):
+            raise AppError(
+                code="BATHYMETRY_PROVIDER_INVALID_RESPONSE",
+                message="VHA profile service returned an ArcGIS error",
+                details={"provider_error": payload.get("error") if isinstance(payload, dict) else None},
+                status_code=502,
+            )
+        return payload, hashlib.sha256(content).hexdigest()
+
+    @staticmethod
+    def _base_spatial_parameters(bbox_values: tuple[float, float, float, float]) -> dict[str, str]:
+        return {
+            "f": "json",
+            "where": "1=1",
+            "geometry": ",".join(f"{value:.12g}" for value in bbox_values),
+            "geometryType": "esriGeometryEnvelope",
+            "inSR": "4326",
+            "outSR": "4326",
+            "spatialRel": "esriSpatialRelIntersects",
+        }
+
+    @staticmethod
+    def _fetch_profiles(
+        bbox_values: tuple[float, float, float, float],
+        settings: Settings,
+        opener: Callable[..., Any] | None,
+    ) -> tuple[list[dict[str, Any]], dict[str, Any]]:
+        base = settings.bathymetry_profiles_layer_url.rstrip("/") + "/query"
+        count_url = BathymetryProfileAcquisitionService._query_url(
+            base,
+            {
+                **BathymetryProfileAcquisitionService._base_spatial_parameters(bbox_values),
+                "returnCountOnly": "true",
+                "returnGeometry": "false",
+            },
+        )
+        count_payload, count_sha = BathymetryProfileAcquisitionService._fetch_json(count_url, settings, opener)
+        candidate_count = int(count_payload.get("count") or 0)
+        if candidate_count > settings.bathymetry_profiles_max_features:
+            raise AppError(
+                code="BATHYMETRY_SCOPE_TOO_LARGE",
+                message="The requested profile scope exceeds the configured feature limit; acquire smaller area partitions",
+                details={
+                    "candidate_count": candidate_count,
+                    "max_features": settings.bathymetry_profiles_max_features,
+                },
+                status_code=413,
+            )
+
+        features: list[dict[str, Any]] = []
+        response_hashes: list[str] = []
+        request_urls: list[str] = [count_url]
+        offset = 0
+        while offset < candidate_count:
+            page_url = BathymetryProfileAcquisitionService._query_url(
+                base,
+                {
+                    **BathymetryProfileAcquisitionService._base_spatial_parameters(bbox_values),
+                    "outFields": BathymetryProfileAcquisitionService.PROFILE_OUT_FIELDS,
+                    "returnGeometry": "true",
+                    "orderByFields": "OBJECTID",
+                    "resultOffset": str(offset),
+                    "resultRecordCount": str(settings.bathymetry_profiles_page_size),
+                },
+            )
+            page, page_sha = BathymetryProfileAcquisitionService._fetch_json(page_url, settings, opener)
+            page_features = page.get("features")
+            if not isinstance(page_features, list):
+                raise AppError(
+                    code="BATHYMETRY_PROVIDER_INVALID_RESPONSE",
+                    message="VHA profile response does not contain a feature list",
+                    status_code=502,
+                )
+            features.extend(item for item in page_features if isinstance(item, dict))
+            response_hashes.append(page_sha)
+            request_urls.append(page_url)
+            if not page_features:
+                break
+            offset += len(page_features)
+        if len(features) != candidate_count:
+            raise AppError(
+                code="BATHYMETRY_PROVIDER_INCOMPLETE_RESPONSE",
+                message="VHA profile pagination did not return the announced number of records",
+                details={"expected": candidate_count, "received": len(features)},
+                status_code=502,
+            )
+        return features, {
+            "candidate_count": candidate_count,
+            "request_urls": request_urls,
+            "response_sha256": [count_sha, *response_hashes],
+        }
+
+    @staticmethod
+    def _fetch_watercourse_names(
+        vhag_codes: set[int],
+        settings: Settings,
+        opener: Callable[..., Any] | None,
+    ) -> tuple[dict[int, dict[str, str | None]], list[str], list[str]]:
+        if not vhag_codes:
+            return {}, [], []
+        base = settings.bathymetry_watercourse_layer_url.rstrip("/") + "/query"
+        names: dict[int, dict[str, str | None]] = {}
+        urls: list[str] = []
+        hashes: list[str] = []
+        ordered_codes = sorted(vhag_codes)
+        for start in range(0, len(ordered_codes), 100):
+            chunk = ordered_codes[start : start + 100]
+            offset = 0
+            while True:
+                url = BathymetryProfileAcquisitionService._query_url(
+                    base,
+                    {
+                        "f": "json",
+                        "where": f"\"wlasvl.vhag\" IN ({','.join(str(code) for code in chunk)})",
+                        "outFields": "wlasvl.vhag,VHAG_TABEL.naam,VHAG_TABEL.namen",
+                        "returnGeometry": "false",
+                        "orderByFields": "wlasvl.vhag",
+                        "resultOffset": str(offset),
+                        "resultRecordCount": str(settings.bathymetry_profiles_page_size),
+                    },
+                )
+                payload, response_sha = BathymetryProfileAcquisitionService._fetch_json(url, settings, opener)
+                urls.append(url)
+                hashes.append(response_sha)
+                page_features = payload.get("features")
+                if not isinstance(page_features, list):
+                    raise AppError(
+                        code="BATHYMETRY_PROVIDER_INVALID_RESPONSE",
+                        message="VHA watercourse response does not contain a feature list",
+                        status_code=502,
+                    )
+                for feature in page_features:
+                    attributes = feature.get("attributes") if isinstance(feature, dict) else None
+                    if not isinstance(attributes, dict):
+                        continue
+                    raw_code = attributes.get("wlasvl.vhag")
+                    if raw_code is None:
+                        continue
+                    code = int(raw_code)
+                    if code not in names:
+                        names[code] = {
+                            "name": attributes.get("VHAG_TABEL.naam"),
+                            "alternative_names": attributes.get("VHAG_TABEL.namen"),
+                        }
+                if not payload.get("exceededTransferLimit") or not page_features:
+                    break
+                offset += len(page_features)
+        return names, urls, hashes
+
+    @staticmethod
+    def _date_from_arcgis(value: Any) -> str | None:
+        if not isinstance(value, (int, float)) or not math.isfinite(float(value)):
+            return None
+        try:
+            return datetime.fromtimestamp(float(value) / 1000.0, tz=UTC).date().isoformat()
+        except (OverflowError, OSError, ValueError):
+            return None
+
+    @staticmethod
+    def _numeric_or_none(value: Any) -> float | None:
+        if value is None:
+            return None
+        try:
+            normalized = float(value)
+        except (TypeError, ValueError):
+            return None
+        return normalized if math.isfinite(normalized) else None
+
+    @staticmethod
+    def _document_url(value: Any) -> str | None:
+        if not isinstance(value, str) or not value.strip():
+            return None
+        normalized = value.strip()
+        if normalized.startswith("http://vha.waterinfo.be/"):
+            normalized = "https://" + normalized[len("http://") :]
+        return normalized if normalized.startswith("https://vha.waterinfo.be/") else None
+
+    @staticmethod
+    def _normalize_features(
+        raw_features: list[dict[str, Any]],
+        scope_geometry,
+        watercourse_names: dict[int, dict[str, str | None]],
+    ) -> tuple[dict[str, Any], dict[str, Any]]:
+        normalized: list[dict[str, Any]] = []
+        dates: list[str] = []
+        watercourse_codes: set[int] = set()
+        document_count = 0
+        depth_count = 0
+        width_count = 0
+        for raw_feature in raw_features:
+            attributes = raw_feature.get("attributes")
+            geometry = raw_feature.get("geometry")
+            if not isinstance(attributes, dict) or not isinstance(geometry, dict):
+                continue
+            x = BathymetryProfileAcquisitionService._numeric_or_none(geometry.get("x"))
+            y = BathymetryProfileAcquisitionService._numeric_or_none(geometry.get("y"))
+            if x is None or y is None:
+                continue
+            point = Point(x, y)
+            if not scope_geometry.covers(point):
+                continue
+            raw_vhag = attributes.get("vhag")
+            vhag = int(raw_vhag) if isinstance(raw_vhag, (int, float)) else None
+            if vhag is not None:
+                watercourse_codes.add(vhag)
+            names = watercourse_names.get(vhag or -1, {})
+            measurement_date = BathymetryProfileAcquisitionService._date_from_arcgis(attributes.get("d_opmeti"))
+            if measurement_date:
+                dates.append(measurement_date)
+            document_url = BathymetryProfileAcquisitionService._document_url(attributes.get("hyperlink"))
+            depth = BathymetryProfileAcquisitionService._numeric_or_none(attributes.get("opg_diepte"))
+            crown_width = BathymetryProfileAcquisitionService._numeric_or_none(attributes.get("opg_kruinb"))
+            floor_width = BathymetryProfileAcquisitionService._numeric_or_none(attributes.get("opg_vloerb"))
+            if document_url:
+                document_count += 1
+            if depth is not None:
+                depth_count += 1
+            if crown_width is not None or floor_width is not None:
+                width_count += 1
+            object_id = str(attributes.get("OBJECTID"))
+            normalized.append(
+                {
+                    "type": "Feature",
+                    "id": object_id,
+                    "properties": {
+                        "source_feature_id": object_id,
+                        "provider_record_id": object_id,
+                        "watercourse_vhag": vhag,
+                        "watercourse_name": names.get("name") or (f"VHA-waterloop {vhag}" if vhag else "Onbekende waterloop"),
+                        "watercourse_alternative_names": names.get("alternative_names"),
+                        "profile_number": attributes.get("atlaspunt"),
+                        "measurement_date": measurement_date,
+                        "recorded_depth_m": depth,
+                        "recorded_crown_width_m": crown_width,
+                        "recorded_floor_width_m": floor_width,
+                        "source_document_url": document_url,
+                        "document_available": document_url is not None,
+                        "structured_depth_available": depth is not None,
+                        "source_code": attributes.get("bron"),
+                        "structure_id": attributes.get("kunstwerkid"),
+                        "provider": BathymetryProfileAcquisitionService.PROVIDER,
+                        "measurement_semantics": "historical_cross_section_profile_point",
+                        "vertical_reference": "document-specific",
+                    },
+                    "geometry": mapping(point),
+                }
+            )
+        return (
+            {
+                "type": "FeatureCollection",
+                "name": "vha_bathymetry_profiles",
+                "crs": {"type": "name", "properties": {"name": "EPSG:4326"}},
+                "features": normalized,
+            },
+            {
+                "profile_count": len(normalized),
+                "document_count": document_count,
+                "structured_depth_count": depth_count,
+                "structured_width_count": width_count,
+                "watercourse_count": len(watercourse_codes),
+                "measurement_date_min": min(dates) if dates else None,
+                "measurement_date_max": max(dates) if dates else None,
+            },
+        )
+
+    @staticmethod
+    def _cached_dataset(db, project_id: UUID, filename: str) -> Dataset | None:
+        return (
+            db.query(Dataset)
+            .filter(
+                Dataset.project_id == project_id,
+                Dataset.name == filename,
+                Dataset.source_name == BathymetryProfileAcquisitionService.PROVIDER,
+                Dataset.status == "ready",
+            )
+            .order_by(Dataset.created_at.desc())
+            .first()
+        )
+
+    @staticmethod
+    def _result(dataset: Dataset, *, reused: bool) -> dict[str, Any]:
+        metadata = dataset.source_metadata if isinstance(dataset.source_metadata, dict) else {}
+        return BathymetryProfileAcquisitionResult(
+            output_dataset_id=dataset.id,
+            reused=reused,
+            provider=BathymetryProfileAcquisitionService.PROVIDER,
+            profile_count=int(metadata.get("profile_count") or 0),
+            document_count=int(metadata.get("document_count") or 0),
+            structured_depth_count=int(metadata.get("structured_depth_count") or 0),
+            structured_width_count=int(metadata.get("structured_width_count") or 0),
+            watercourse_count=int(metadata.get("watercourse_count") or 0),
+            bbox_epsg4326=list(metadata.get("bbox_epsg4326") or []),
+            clipped_to_area_id=dataset.area_id,
+            measurement_date_min=metadata.get("measurement_date_min"),
+            measurement_date_max=metadata.get("measurement_date_max"),
+            attribution=BathymetryProfileAcquisitionService.ATTRIBUTION,
+            limitation_message=BathymetryProfileAcquisitionService.LIMITATION,
+        ).model_dump(mode="json")
+
+    @staticmethod
+    def acquire(
+        db,
+        project_id: UUID,
+        payload: BathymetryProfileAcquireRequest,
+        *,
+        settings: Settings | None = None,
+        opener: Callable[..., Any] | None = None,
+    ) -> dict[str, Any]:
+        resolved_settings = settings or get_settings()
+        if not resolved_settings.bathymetry_profiles_enabled:
+            raise AppError(
+                code="BATHYMETRY_NOT_CONFIGURED",
+                message="VHA bathymetry profile acquisition is disabled",
+                status_code=503,
+            )
+        bbox_values = BathymetryProfileAcquisitionService._validate_bbox(payload)
+        scope_geometry = BathymetryProfileAcquisitionService._scope_geometry(
+            db, project_id, payload.area_id, bbox_values
+        )
+        exact_bbox = tuple(float(value) for value in scope_geometry.bounds)
+        request_identity = {
+            "provider": BathymetryProfileAcquisitionService.PROVIDER,
+            "bbox_epsg4326": list(exact_bbox),
+            "area_id": str(payload.area_id) if payload.area_id else None,
+            "source_version": BathymetryProfileAcquisitionService.SOURCE_VERSION,
+        }
+        request_hash = hashlib.sha256(json.dumps(request_identity, sort_keys=True).encode()).hexdigest()
+        filename = f"vha_bathymetry_profiles_{request_hash[:12]}.geojson"
+        if not payload.force_refresh:
+            cached = BathymetryProfileAcquisitionService._cached_dataset(db, project_id, filename)
+            if cached is not None:
+                return BathymetryProfileAcquisitionService._result(cached, reused=True)
+
+        raw_features, fetch_provenance = BathymetryProfileAcquisitionService._fetch_profiles(
+            exact_bbox, resolved_settings, opener
+        )
+        vhag_codes = {
+            int(feature["attributes"]["vhag"])
+            for feature in raw_features
+            if isinstance(feature.get("attributes"), dict)
+            and isinstance(feature["attributes"].get("vhag"), (int, float))
+        }
+        names, name_urls, name_hashes = BathymetryProfileAcquisitionService._fetch_watercourse_names(
+            vhag_codes, resolved_settings, opener
+        )
+        feature_collection, summary = BathymetryProfileAcquisitionService._normalize_features(
+            raw_features, scope_geometry, names
+        )
+        if summary["profile_count"] == 0:
+            raise AppError(
+                code="BATHYMETRY_NO_PROFILES",
+                message="No VHA bathymetry profiles intersect the requested area",
+                status_code=404,
+            )
+        artifact = json.dumps(feature_collection, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+        acquired_at = datetime.now(UTC)
+        source_metadata = {
+            "provider": BathymetryProfileAcquisitionService.PROVIDER,
+            "service": "ArcGIS MapServer",
+            "source_version": BathymetryProfileAcquisitionService.SOURCE_VERSION,
+            "theme": "bathymetry",
+            "layer_name": "bathymetry_profiles",
+            "coverage_scope": "municipality" if payload.area_id else "bounded_selection",
+            "bbox_epsg4326": list(exact_bbox),
+            **summary,
+            "selection_aggregation": {
+                "metric_key": "profile_count",
+                "method": "feature_count",
+                "label": "Dwarsprofielen",
+                "unit": "profielen",
+            },
+            "selection_metrics": [
+                {
+                    "metric_key": "recorded_depth_mean_m",
+                    "method": "mean",
+                    "property": "recorded_depth_m",
+                    "label": "Gemiddelde geregistreerde diepte",
+                    "unit": "m",
+                    "warning": "Alleen profielen met een gestructureerde dieptewaarde; meetdata kunnen verschillen.",
+                },
+                {
+                    "metric_key": "recorded_depth_min_m",
+                    "method": "min",
+                    "property": "recorded_depth_m",
+                    "label": "Kleinste geregistreerde diepte",
+                    "unit": "m",
+                },
+                {
+                    "metric_key": "recorded_depth_max_m",
+                    "method": "max",
+                    "property": "recorded_depth_m",
+                    "label": "Grootste geregistreerde diepte",
+                    "unit": "m",
+                },
+                {
+                    "metric_key": "recorded_crown_width_mean_m",
+                    "method": "mean",
+                    "property": "recorded_crown_width_m",
+                    "label": "Gemiddelde geregistreerde kruinbreedte",
+                    "unit": "m",
+                },
+                {
+                    "metric_key": "recorded_floor_width_mean_m",
+                    "method": "mean",
+                    "property": "recorded_floor_width_m",
+                    "label": "Gemiddelde geregistreerde vloerbreedte",
+                    "unit": "m",
+                },
+            ],
+            "attribution": BathymetryProfileAcquisitionService.ATTRIBUTION,
+            "license_note": BathymetryProfileAcquisitionService.LICENSE_NOTE,
+            "limitation_message": BathymetryProfileAcquisitionService.LIMITATION,
+            "volume_supported": False,
+        }
+        dataset = DatasetService.import_vector_bytes(
+            db,
+            project_id=project_id,
+            area_id=payload.area_id,
+            filename=filename,
+            content=artifact,
+            source="VHA digitale atlas dwarsprofielen",
+            source_name=BathymetryProfileAcquisitionService.PROVIDER,
+            dataset_role="reference",
+            reference_layer_name="bathymetry_profiles",
+            source_version=BathymetryProfileAcquisitionService.SOURCE_VERSION,
+            source_metadata=source_metadata,
+            provenance_metadata={
+                "acquisition": "explicit_bounded_arcgis_feature_query",
+                "acquired_at": acquired_at.isoformat(),
+                "request_hash": request_hash,
+                "profile_query_urls": fetch_provenance["request_urls"],
+                "watercourse_name_query_urls": name_urls,
+                "response_sha256": [*fetch_provenance["response_sha256"], *name_hashes],
+                "artifact_sha256": hashlib.sha256(artifact).hexdigest(),
+                "candidate_count": fetch_provenance["candidate_count"],
+                "exact_profile_count": summary["profile_count"],
+                "clipped_to_area_id": str(payload.area_id) if payload.area_id else None,
+                "scope_geometry_type": scope_geometry.geom_type,
+                "vertical_reference": "document-specific",
+                "bathymetric_surface_available": False,
+                "water_surface_level_available": False,
+                "water_volume_available": False,
+                "limitation_message": BathymetryProfileAcquisitionService.LIMITATION,
+            },
+        )
+        persisted = db.get(Dataset, dataset.id)
+        return BathymetryProfileAcquisitionService._result(persisted, reused=False)

backend/app/services/dataset_service.py

@@ -410,6 +410,126 @@ class DatasetService:
 
         return DatasetService._to_response(dataset)
 
+    @staticmethod
+    def import_vector_bytes(
+        db: Session,
+        *,
+        project_id: UUID,
+        filename: str,
+        content: bytes,
+        source: str,
+        source_name: str,
+        dataset_role: str,
+        reference_layer_name: str | None,
+        source_metadata: dict[str, Any],
+        provenance_metadata: dict[str, Any],
+        area_id: UUID | None = None,
+        temporal_series_key: str | None = None,
+        observed_at: datetime | None = None,
+        valid_from: datetime | None = None,
+        valid_to: datetime | None = None,
+        temporal_granularity: str | None = None,
+        source_version: str | None = None,
+        content_type: str = "application/geo+json",
+    ) -> DatasetCreateResponse:
+        if not db.get(Project, project_id):
+            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
+        if area_id is not None:
+            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 content:
+            raise AppError(code="INVALID_UPLOAD", message="Vector artifact is empty", status_code=400)
+        safe_filename = DatasetService._validate_upload_filename(filename)
+        if DatasetService._extension_for_path(safe_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)
+        normalized_source_name = (source_name or "").strip() or ("manual" if normalized_role == "reference" else None)
+        temporal = DatasetService._validate_temporal_metadata(
+            temporal_series_key=temporal_series_key,
+            observed_at=observed_at,
+            valid_from=valid_from,
+            valid_to=valid_to,
+            temporal_granularity=temporal_granularity,
+            source_version=source_version,
+        )
+        try:
+            text = content.decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise AppError(code="INVALID_UPLOAD", message="Vector artifact must be UTF-8 encoded", status_code=400) from exc
+        try:
+            metadata = parse_geojson_payload(text)
+            vector_payload = json.loads(text)
+        except (ValueError, json.JSONDecodeError) as exc:
+            raise AppError(code="INVALID_GEOJSON", message=str(exc), status_code=400) from exc
+
+        dataset_id = uuid.uuid4()
+        storage_info = StorageService.persist_dataset_file(
+            project_id=str(project_id),
+            dataset_id=str(dataset_id),
+            dataset_type="vector",
+            original_filename=safe_filename,
+            content=content,
+            content_type=content_type,
+        )
+        dataset = Dataset(
+            id=dataset_id,
+            project_id=project_id,
+            area_id=area_id,
+            name=safe_filename,
+            dataset_type="vector",
+            source=source,
+            dataset_role=normalized_role,
+            source_name=normalized_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=metadata.get("crs"),
+            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,
+                    valid_to=dataset.valid_to,
+                    checksum_sha256=dataset.checksum_sha256,
+                    source_metadata=dataset.source_metadata,
+                    provenance_metadata=dataset.provenance_metadata,
+                )
+            )
+            VectorFeatureService.persist_geojson_features(
+                db=db,
+                dataset_id=dataset.id,
+                payload=vector_payload,
+                feature_class=reference_layer_name if normalized_role == "reference" else None,
+                commit=False,
+            )
+            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 import_raster_bytes(
         db: Session,

backend/app/services/vector_feature_service.py

@@ -36,6 +36,8 @@ SEMANTIC_METRICS_DISABLED_OPERATOR_TOOLS = {
     # line metrics (road/watercourse length) would therefore be meaningless.
     "provision_regional_historical_landuse.py",
 }
+PROPERTY_AGGREGATION_METHODS = {"sum", "mean", "area_weighted_sum"}
+PROPERTY_EXTREMA_METHODS = {"min", "max"}
 
 PRECLIPPED_MUNICIPALITY_PARTITION_OPERATOR_TOOLS = {
     "provision_regional_bwk_natura2000.py",
@@ -581,7 +583,7 @@ class VectorFeatureService:
             length_m = db.query(func.coalesce(func.sum(length_expression), 0.0)).filter(*metric_filter).scalar()
             divisor = 1_000.0 if unit == "km" else 1.0
             metric_value = float(length_m or 0.0) / divisor
-        elif method in {"sum", "mean", "area_weighted_sum"}:
+        elif method in PROPERTY_AGGREGATION_METHODS | PROPERTY_EXTREMA_METHODS:
             property_name = str(config.get("property") or "").strip()
             if not property_name:
                 raise AppError(
@@ -599,7 +601,11 @@ class VectorFeatureService:
                 )
                 coverage_ratio = intersection_area / func.nullif(source_area, 0.0)
                 value_expression = numeric_value * coverage_ratio
-            aggregate_function = func.avg if method == "mean" else func.sum
+            aggregate_function = {
+                "mean": func.avg,
+                "min": func.min,
+                "max": func.max,
+            }.get(method, func.sum)
             aggregate_value = (
                 db.query(func.coalesce(aggregate_function(value_expression), 0.0))
                 .filter(*metric_filter)