Add Flemish bathymetry partition workflow

backend/app/services/bathymetry_profile_acquisition_service.py

@@ -15,8 +15,10 @@ 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.models import Area, Dataset, DatasetVersion, Project
 from app.schemas.bathymetry import (
+    BathymetryPartitionFinalizeRequest,
+    BathymetryPartitionFinalizationResult,
     BathymetryProfileAcquireRequest,
     BathymetryProfileAcquisitionResult,
     BathymetrySourceRead,
@@ -70,16 +72,16 @@ class BathymetryProfileAcquisitionService:
             "vertical_reference": "LAT",
             "horizontal_crs": "bronafhankelijk; expliciet per WCS-respons",
             "native_resolution": "20 x 20 m",
-            "integration_status": "available_not_integrated",
+            "integration_status": "probe_only",
             "acquisition_supported": False,
             "configured": False,
-            "service_url": "https://bathy.agentschapmdk.be/spatialfusionserver/services/ows/wcs/WCS_Public",
+            "service_url": "https://bathy.agentschapmdk.be/spatialfusionserver/services/ows/wcs/EL_wcs",
             "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."
+                "Alleen een read-only GetCapabilities-probe is beschikbaar. Rasteracquisitie blijft uit totdat "
+                "WCS, maritieme begrenzing, tegels, LAT-semantiek en servercertificaten live zijn gevalideerd."
             ),
         },
         {
@@ -404,6 +406,8 @@ class BathymetryProfileAcquisitionService:
         raw_features: list[dict[str, Any]],
         scope_geometry,
         watercourse_names: dict[int, dict[str, str | None]],
+        *,
+        partition_properties: dict[str, Any] | None = None,
     ) -> tuple[dict[str, Any], dict[str, Any]]:
         normalized: list[dict[str, Any]] = []
         dates: list[str] = []
@@ -465,6 +469,7 @@ class BathymetryProfileAcquisitionService:
                         "provider": BathymetryProfileAcquisitionService.PROVIDER,
                         "measurement_semantics": "historical_cross_section_profile_point",
                         "vertical_reference": "document-specific",
+                        **(partition_properties or {}),
                     },
                     "geometry": mapping(point),
                 }
@@ -487,6 +492,17 @@ class BathymetryProfileAcquisitionService:
             },
         )
 
+    @staticmethod
+    def _municipality_name(area: Area | None) -> str | None:
+        if area is None:
+            return None
+        normalized = str(area.name or "").strip()
+        prefix = "Gemeente "
+        if not normalized.casefold().startswith(prefix.casefold()):
+            return None
+        municipality = normalized[len(prefix) :].split(" - ", 1)[0].strip()
+        return municipality or None
+
     @staticmethod
     def _cached_dataset(db, project_id: UUID, filename: str) -> Dataset | None:
         return (
@@ -541,6 +557,8 @@ class BathymetryProfileAcquisitionService:
         scope_geometry = BathymetryProfileAcquisitionService._scope_geometry(
             db, project_id, payload.area_id, bbox_values
         )
+        area = db.get(Area, payload.area_id) if payload.area_id else None
+        municipality = BathymetryProfileAcquisitionService._municipality_name(area)
         exact_bbox = tuple(float(value) for value in scope_geometry.bounds)
         request_identity = {
             "provider": BathymetryProfileAcquisitionService.PROVIDER,
@@ -568,7 +586,16 @@ class BathymetryProfileAcquisitionService:
             vhag_codes, resolved_settings, opener
         )
         feature_collection, summary = BathymetryProfileAcquisitionService._normalize_features(
-            raw_features, scope_geometry, names
+            raw_features,
+            scope_geometry,
+            names,
+            partition_properties={
+                "partition_area_id": str(area.id),
+                "partition_area_name": area.name,
+                **({"municipality": municipality} if municipality else {}),
+            }
+            if area
+            else None,
         )
         if summary["profile_count"] == 0:
             raise AppError(
@@ -585,6 +612,11 @@ class BathymetryProfileAcquisitionService:
             "theme": "bathymetry",
             "layer_name": "bathymetry_profiles",
             "coverage_scope": "municipality" if payload.area_id else "bounded_selection",
+            "partition_area_id": str(area.id) if area else None,
+            "partition_area_name": area.name if area else None,
+            "municipality": municipality,
+            "partitioned_source_audit": False,
+            "regional_partitions_complete": False,
             "bbox_epsg4326": list(exact_bbox),
             **summary,
             "selection_aggregation": {
@@ -669,3 +701,160 @@ class BathymetryProfileAcquisitionService:
         )
         persisted = db.get(Dataset, dataset.id)
         return BathymetryProfileAcquisitionService._result(persisted, reused=False)
+
+    @staticmethod
+    def finalize_partitions(
+        db,
+        project_id: UUID,
+        payload: BathymetryPartitionFinalizeRequest,
+    ) -> dict[str, Any]:
+        if not db.get(Project, project_id):
+            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
+
+        expected_area_ids = set(payload.expected_area_ids)
+        dataset_ids = set(payload.dataset_ids)
+        no_profile_area_ids = set(payload.no_profile_area_ids)
+
+        areas: dict[UUID, Area] = {}
+        for area_id in expected_area_ids:
+            area = db.get(Area, area_id)
+            if area is None or area.project_id != project_id:
+                raise AppError(
+                    code="BATHYMETRY_PARTITION_AREA_INVALID",
+                    message="Every expected partition Area must belong to the project",
+                    details={"area_id": str(area_id)},
+                    status_code=400,
+                )
+            if BathymetryProfileAcquisitionService._municipality_name(area) is None:
+                raise AppError(
+                    code="BATHYMETRY_PARTITION_AREA_INVALID",
+                    message="Bathymetry partitions must use persisted municipality Areas",
+                    details={"area_id": str(area_id), "area_name": area.name},
+                    status_code=400,
+                )
+            areas[area_id] = area
+
+        if not no_profile_area_ids.issubset(expected_area_ids):
+            raise AppError(
+                code="BATHYMETRY_PARTITION_MANIFEST_INVALID",
+                message="No-profile partitions must be part of the expected Area set",
+                status_code=400,
+            )
+
+        datasets: list[Dataset] = []
+        data_area_ids: set[UUID] = set()
+        for dataset_id in payload.dataset_ids:
+            dataset = db.get(Dataset, dataset_id)
+            if (
+                dataset is None
+                or dataset.project_id != project_id
+                or dataset.source_name != BathymetryProfileAcquisitionService.PROVIDER
+                or dataset.status != "ready"
+                or dataset.area_id not in expected_area_ids
+            ):
+                raise AppError(
+                    code="BATHYMETRY_PARTITION_DATASET_INVALID",
+                    message="Every partition Dataset must be a ready VHA profile Dataset scoped to an expected Area",
+                    details={"dataset_id": str(dataset_id)},
+                    status_code=400,
+                )
+            if dataset.area_id in data_area_ids:
+                raise AppError(
+                    code="BATHYMETRY_PARTITION_DATASET_DUPLICATE",
+                    message="A complete manifest may reference only one profile Dataset per Area",
+                    details={"area_id": str(dataset.area_id)},
+                    status_code=400,
+                )
+            data_area_ids.add(dataset.area_id)
+            datasets.append(dataset)
+
+        accounted_area_ids = data_area_ids.union(no_profile_area_ids)
+        if accounted_area_ids != expected_area_ids:
+            raise AppError(
+                code="BATHYMETRY_PARTITION_MANIFEST_INCOMPLETE",
+                message="Every expected Area must have one ready Dataset or an explicit no-profile result",
+                details={
+                    "missing_area_ids": sorted(str(value) for value in expected_area_ids - accounted_area_ids),
+                    "unexpected_area_ids": sorted(str(value) for value in accounted_area_ids - expected_area_ids),
+                },
+                status_code=400,
+            )
+
+        profile_count = 0
+        document_count = 0
+        structured_depth_count = 0
+        dates_min: list[str] = []
+        dates_max: list[str] = []
+        shared_metadata = {
+            "partition_scope_key": payload.partition_scope_key,
+            "partition_count": len(expected_area_ids),
+            "data_partition_count": len(datasets),
+            "no_profile_partition_count": len(no_profile_area_ids),
+            "partition_manifest_sha256": payload.manifest_sha256,
+            "partition_manifest_observed_at": payload.observed_at.isoformat(),
+            "partitioned_source_audit": True,
+            "regional_partitions_complete": True,
+        }
+        shared_provenance = {
+            "partition_manifest_sha256": payload.manifest_sha256,
+            "partition_manifest_observed_at": payload.observed_at.isoformat(),
+            "partition_scope_key": payload.partition_scope_key,
+            "regional_partitions_complete": True,
+            "no_profile_area_ids": sorted(str(value) for value in no_profile_area_ids),
+        }
+
+        for dataset in datasets:
+            source_metadata = dict(dataset.source_metadata or {})
+            provenance_metadata = dict(dataset.provenance_metadata or {})
+            profile_count += int(source_metadata.get("profile_count") or 0)
+            document_count += int(source_metadata.get("document_count") or 0)
+            structured_depth_count += int(source_metadata.get("structured_depth_count") or 0)
+            if source_metadata.get("measurement_date_min"):
+                dates_min.append(str(source_metadata["measurement_date_min"]))
+            if source_metadata.get("measurement_date_max"):
+                dates_max.append(str(source_metadata["measurement_date_max"]))
+            area = areas[dataset.area_id]
+            source_metadata.update(
+                {
+                    **shared_metadata,
+                    "coverage_scope": payload.partition_scope_key,
+                    "partition_area_id": str(area.id),
+                    "partition_area_name": area.name,
+                    "municipality": BathymetryProfileAcquisitionService._municipality_name(area),
+                }
+            )
+            provenance_metadata.update(shared_provenance)
+            dataset.source_metadata = source_metadata
+            dataset.provenance_metadata = provenance_metadata
+
+        if dataset_ids:
+            versions = (
+                db.query(DatasetVersion)
+                .filter(DatasetVersion.dataset_id.in_(dataset_ids))
+                .all()
+            )
+            for version in versions:
+                version.source_metadata = dict(
+                    next(dataset.source_metadata for dataset in datasets if dataset.id == version.dataset_id)
+                )
+                version.provenance_metadata = dict(
+                    next(dataset.provenance_metadata for dataset in datasets if dataset.id == version.dataset_id)
+                )
+
+        db.commit()
+        return BathymetryPartitionFinalizationResult(
+            partition_scope_key=payload.partition_scope_key,
+            regional_partitions_complete=True,
+            partition_count=len(expected_area_ids),
+            data_partition_count=len(datasets),
+            no_profile_partition_count=len(no_profile_area_ids),
+            profile_count=profile_count,
+            document_count=document_count,
+            structured_depth_count=structured_depth_count,
+            measurement_date_min=min(dates_min) if dates_min else None,
+            measurement_date_max=max(dates_max) if dates_max else None,
+            dataset_ids=payload.dataset_ids,
+            manifest_sha256=payload.manifest_sha256,
+            observed_at=payload.observed_at,
+            limitation_message=BathymetryProfileAcquisitionService.LIMITATION,
+        ).model_dump(mode="json")

backend/app/services/mdk_bathymetry_probe_service.py

@@ -0,0 +1,253 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+import hashlib
+import ssl
+from typing import Any, Callable
+from urllib.error import HTTPError, URLError
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+from urllib.request import Request, urlopen
+from xml.etree import ElementTree
+
+from app.core.config import Settings, get_settings
+from app.schemas.bathymetry import BathymetrySourceProbeRead
+
+
+class MdkBathymetryProbeService:
+    SOURCE_KEY = "mdk_bcp_bathymetry"
+    LIMITATION = (
+        "Deze probe leest alleen WCS GetCapabilities met strikte TLS-controle. "
+        "GeoIntel downloadt of activeert geen Noordzee-raster totdat endpoint, coverage-id, CRS, LAT, "
+        "nodata, resolutie, begrenzing en responslimieten live zijn gevalideerd."
+    )
+
+    @staticmethod
+    def _capabilities_url(configured_url: str) -> str:
+        parsed = urlsplit(configured_url.strip())
+        if parsed.scheme.lower() != "https" or not parsed.hostname:
+            raise ValueError("The MDK WCS probe requires an absolute HTTPS URL")
+        parameters = dict(parse_qsl(parsed.query, keep_blank_values=True))
+        parameters.update({"service": "WCS", "request": "GetCapabilities", "version": "1.0.0"})
+        return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, urlencode(parameters), ""))
+
+    @staticmethod
+    def _read_capabilities(
+        capabilities_url: str,
+        settings: Settings,
+        opener: Callable[..., Any] | None,
+    ) -> tuple[bytes, str]:
+        request = Request(
+            capabilities_url,
+            headers={
+                "Accept": "application/xml,text/xml;q=0.9,*/*;q=0.1",
+                "User-Agent": "GeoIntel/1.0 MDK-bathymetry-readiness-probe",
+            },
+        )
+        with (opener or urlopen)(request, timeout=settings.mdk_bathymetry_probe_timeout_seconds) as response:
+            limit = settings.mdk_bathymetry_probe_max_response_mb * 1024 * 1024
+            content = response.read(limit + 1)
+            if len(content) > limit:
+                raise ValueError("MDK WCS capabilities response exceeded the configured size limit")
+            content_type = str(response.headers.get("Content-Type") or "") if hasattr(response, "headers") else ""
+        return content, content_type
+
+    @staticmethod
+    def _local_name(tag: str) -> str:
+        return tag.rsplit("}", 1)[-1].casefold()
+
+    @staticmethod
+    def _parse_capabilities(content: bytes) -> dict[str, Any]:
+        root = ElementTree.fromstring(content)
+        root_name = MdkBathymetryProbeService._local_name(root.tag)
+        if "capabilities" not in root_name:
+            raise ValueError("MDK endpoint did not return a WCS capabilities document")
+
+        coverage_identifiers: set[str] = set()
+        advertised_formats: set[str] = set()
+        advertised_crs: set[str] = set()
+        for element in root.iter():
+            local_name = MdkBathymetryProbeService._local_name(element.tag)
+            text = (element.text or "").strip()
+            if local_name in {"coverageofferingbrief", "coverageoffering"}:
+                for child in element:
+                    if MdkBathymetryProbeService._local_name(child.tag) in {"name", "identifier"}:
+                        identifier = (child.text or "").strip()
+                        if identifier:
+                            coverage_identifiers.add(identifier)
+                            break
+            if local_name in {"format", "formats"} and text:
+                advertised_formats.add(text)
+            if local_name in {"requestresponsecrss", "requestcrss", "responsecrss", "nativecrss", "crs"} and text:
+                advertised_crs.add(text)
+            for attribute_value in element.attrib.values():
+                normalized = str(attribute_value).strip()
+                if "EPSG" in normalized.upper() or "CRS:" in normalized.upper():
+                    advertised_crs.add(normalized)
+
+        return {
+            "wcs_version": str(root.attrib.get("version") or "") or None,
+            "coverage_identifiers": sorted(coverage_identifiers),
+            "advertised_formats": sorted(advertised_formats),
+            "advertised_crs": sorted(advertised_crs),
+        }
+
+    @staticmethod
+    def _result(
+        *,
+        settings: Settings,
+        status: str,
+        checked_at: datetime,
+        message: str,
+        capabilities_url: str | None = None,
+        tls_verified: bool = False,
+        capabilities_reachable: bool = False,
+        response_sha256: str | None = None,
+        parsed: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        parsed = parsed or {}
+        return BathymetrySourceProbeRead(
+            source_key=MdkBathymetryProbeService.SOURCE_KEY,
+            status=status,
+            configured_url=settings.mdk_bathymetry_wcs_url,
+            capabilities_url=capabilities_url,
+            tls_verified=tls_verified,
+            capabilities_reachable=capabilities_reachable,
+            acquisition_supported=False,
+            wcs_version=parsed.get("wcs_version"),
+            coverage_identifiers=parsed.get("coverage_identifiers") or [],
+            advertised_formats=parsed.get("advertised_formats") or [],
+            advertised_crs=parsed.get("advertised_crs") or [],
+            response_sha256=response_sha256,
+            checked_at=checked_at,
+            message=message,
+            limitation_message=MdkBathymetryProbeService.LIMITATION,
+        ).model_dump(mode="json")
+
+    @staticmethod
+    def probe(
+        *,
+        settings: Settings | None = None,
+        opener: Callable[..., Any] | None = None,
+        checked_at: datetime | None = None,
+    ) -> dict[str, Any]:
+        resolved_settings = settings or get_settings()
+        now = checked_at or datetime.now(UTC)
+        if not resolved_settings.mdk_bathymetry_probe_enabled:
+            return MdkBathymetryProbeService._result(
+                settings=resolved_settings,
+                status="disabled",
+                checked_at=now,
+                message="MDK bathymetry readiness probing is disabled.",
+            )
+        try:
+            capabilities_url = MdkBathymetryProbeService._capabilities_url(
+                resolved_settings.mdk_bathymetry_wcs_url
+            )
+        except ValueError as exc:
+            return MdkBathymetryProbeService._result(
+                settings=resolved_settings,
+                status="invalid_configuration",
+                checked_at=now,
+                message=str(exc),
+            )
+
+        try:
+            content, content_type = MdkBathymetryProbeService._read_capabilities(
+                capabilities_url, resolved_settings, opener
+            )
+        except HTTPError as exc:
+            return MdkBathymetryProbeService._result(
+                settings=resolved_settings,
+                status="endpoint_unavailable",
+                checked_at=now,
+                capabilities_url=capabilities_url,
+                tls_verified=True,
+                message=f"MDK WCS GetCapabilities returned HTTP {exc.code}.",
+            )
+        except ssl.SSLCertVerificationError:
+            return MdkBathymetryProbeService._result(
+                settings=resolved_settings,
+                status="tls_error",
+                checked_at=now,
+                capabilities_url=capabilities_url,
+                message="MDK WCS TLS certificate validation failed; insecure fallback is prohibited.",
+            )
+        except URLError as exc:
+            reason = exc.reason
+            is_tls_error = isinstance(reason, (ssl.SSLError, ssl.CertificateError)) or "certificate" in str(
+                reason
+            ).casefold()
+            return MdkBathymetryProbeService._result(
+                settings=resolved_settings,
+                status="tls_error" if is_tls_error else "endpoint_unavailable",
+                checked_at=now,
+                capabilities_url=capabilities_url,
+                message=(
+                    "MDK WCS TLS certificate validation failed; insecure fallback is prohibited."
+                    if is_tls_error
+                    else "MDK WCS GetCapabilities could not be reached."
+                ),
+            )
+        except (TimeoutError, OSError) as exc:
+            return MdkBathymetryProbeService._result(
+                settings=resolved_settings,
+                status="endpoint_unavailable",
+                checked_at=now,
+                capabilities_url=capabilities_url,
+                message=f"MDK WCS GetCapabilities could not be reached ({type(exc).__name__}).",
+            )
+        except ValueError as exc:
+            return MdkBathymetryProbeService._result(
+                settings=resolved_settings,
+                status="invalid_capabilities",
+                checked_at=now,
+                capabilities_url=capabilities_url,
+                tls_verified=True,
+                capabilities_reachable=True,
+                message=str(exc),
+            )
+
+        response_sha256 = hashlib.sha256(content).hexdigest()
+        try:
+            parsed = MdkBathymetryProbeService._parse_capabilities(content)
+        except (ElementTree.ParseError, ValueError) as exc:
+            detail = " ".join(str(exc).split())
+            if content_type:
+                detail = f"{detail} Content-Type: {content_type}."
+            return MdkBathymetryProbeService._result(
+                settings=resolved_settings,
+                status="invalid_capabilities",
+                checked_at=now,
+                capabilities_url=capabilities_url,
+                tls_verified=True,
+                capabilities_reachable=True,
+                response_sha256=response_sha256,
+                message=detail,
+            )
+
+        if not parsed["coverage_identifiers"]:
+            return MdkBathymetryProbeService._result(
+                settings=resolved_settings,
+                status="invalid_capabilities",
+                checked_at=now,
+                capabilities_url=capabilities_url,
+                tls_verified=True,
+                capabilities_reachable=True,
+                response_sha256=response_sha256,
+                parsed=parsed,
+                message="MDK WCS capabilities are reachable but advertise no coverage identifier.",
+            )
+        return MdkBathymetryProbeService._result(
+            settings=resolved_settings,
+            status="reachable",
+            checked_at=now,
+            capabilities_url=capabilities_url,
+            tls_verified=True,
+            capabilities_reachable=True,
+            response_sha256=response_sha256,
+            parsed=parsed,
+            message=(
+                "MDK WCS capabilities are reachable with verified TLS. "
+                "Raster acquisition remains disabled pending bounded coverage validation."
+            ),
+        )