JensandClaude Opus 5
7351993fee
An ArcGIS layer without supportsPagination accepts resultOffset and ignores it, answering every page with the first one. The VHA profile reader advanced its offset by the page length and stopped at the announced count, so for a count that is a multiple of the page size it collected N copies of page one — and its completeness check, len(features) == candidate_count, passed. Four announced records became four stored records, two of them duplicates, filed under an official provenance. That is the substitution bounded acquisition exists to prevent, arriving through the front door. The reader now refuses a record it already collected. It fails rather than silently dropping the duplicate: a provider that cannot page is a provider whose count proves nothing, so a smaller-but-clean result would still be unverifiable. Its watercourse-name loop was worse — a bare `while True` that ended only when the provider stopped setting exceededTransferLimit, with names deduplicated by code so a stuck provider produced no visible change while the requests continued. It now refuses a repeated page body, and both loops have the page budget the sibling readers already had. Those siblings turned out to be fine. GRB and official vector already refuse a repeated page URL, bound the page count, and deduplicate on feature identity — but none of it had a test, so none of it was known to work. Exercised now, including the case where distinct URLs defeat the loop check and the budget is the only backstop. A duplicate across two genuinely different pages is kept once rather than failing, because a cursor over a changing table produces that legitimately. Also: _bash_path fell back to the raw path whenever wslpath failed, except on timeout, which propagated and reddened the suite when starting WSL took more than ten seconds under load. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
959 lines
45 KiB
Python
959 lines
45 KiB
Python
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.services.outbound_request_guard import guarded_opener
|
|
from app.models import Area, Dataset, DatasetVersion, Project
|
|
from app.schemas.bathymetry import (
|
|
BathymetryPartitionFinalizeRequest,
|
|
BathymetryPartitionFinalizationResult,
|
|
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": "probe_only",
|
|
"acquisition_supported": False,
|
|
"configured": False,
|
|
"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": (
|
|
"Alleen een read-only GetCapabilities-probe is beschikbaar. Rasteracquisitie blijft uit totdat "
|
|
"WCS, maritieme begrenzing, tegels, LAT-semantiek en servercertificaten live 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": ["operator_archive", "bounded_raster", "arcgis_map_service"],
|
|
"vertical_reference": "mDNG",
|
|
"horizontal_crs": "EPSG:3812; visualisatieservice kan EPSG:31370 aanbieden",
|
|
"native_resolution": "0,5 m",
|
|
"integration_status": "operational",
|
|
"acquisition_supported": True,
|
|
"configured": True,
|
|
"service_url": "https://geoservices.wallonie.be/arcgis/rest/services/EAU/BATHY/MapServer",
|
|
"catalog_url": "https://geoportail.wallonie.be/catalogue/0a544b42-0b30-4c8e-85e7-38149b99eae0.html",
|
|
"attribution": "Service public de Wallonie",
|
|
"license_note": "CC BY 4.0 volgens de officiële Geoportail-metadata.",
|
|
"limitation_message": (
|
|
"De gepinde officiële release kan begrensd als raster worden geïmporteerd via de operator. "
|
|
"Dekking verschilt per vaarweg; de waarden zijn bodemhoogtes in mDNG uit 2019-2022, "
|
|
"zonder stilzwijgende datumconversie of afleiding van actuele waterdiepte."
|
|
),
|
|
},
|
|
{
|
|
"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(settings=None) -> list[dict[str, Any]]:
|
|
from app.core.config import get_settings
|
|
|
|
resolved_settings = settings or get_settings()
|
|
items: list[dict[str, Any]] = []
|
|
for source in BathymetryProfileAcquisitionService._SOURCES:
|
|
item = dict(source)
|
|
if item["key"] == "mdk_bcp_bathymetry":
|
|
mdk_configured = bool(
|
|
resolved_settings.mdk_bathymetry_acquisition_enabled
|
|
and (resolved_settings.mdk_bathymetry_coverage_id or "").strip()
|
|
)
|
|
item["acquisition_supported"] = True
|
|
item["configured"] = mdk_configured
|
|
if mdk_configured:
|
|
item["integration_status"] = "operational"
|
|
item["limitation_message"] = (
|
|
"Begrensde WCS-acquisitie is expliciet ingeschakeld en draait alleen wanneer de "
|
|
"live readiness-probe bereikbaar is en het geconfigureerde coverage-id door de "
|
|
"capabilities wordt geadverteerd. Dieptes blijven LAT-gerefereerd; watervolume "
|
|
"blijft zonder compatibel wateroppervlak niet ondersteund."
|
|
)
|
|
else:
|
|
item["limitation_message"] = (
|
|
"Begrensde WCS-acquisitie bestaat maar staat uit. Zet "
|
|
"MDK_BATHYMETRY_ACQUISITION_ENABLED=true en configureer MDK_BATHYMETRY_COVERAGE_ID "
|
|
"pas nadat de readiness-probe live 'reachable' rapporteert. Er wordt nooit "
|
|
"onbeveiligd of ongevalideerd gedownload."
|
|
)
|
|
items.append(item)
|
|
return [BathymetrySourceRead(**item).model_dump() for item in items]
|
|
|
|
@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 guarded_opener(url))(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 _unseen_records(
|
|
page_features: list[Any],
|
|
seen_object_ids: set[str],
|
|
) -> list[dict[str, Any]]:
|
|
"""Every page must bring records the earlier pages did not.
|
|
|
|
An ArcGIS layer without ``supportsPagination`` accepts ``resultOffset``
|
|
and ignores it, answering every page with the first one. Advancing the
|
|
offset by the page length still reaches the announced count, so the
|
|
completeness check below passed while the dataset held N copies of page
|
|
one — a silent substitution of the source data, which is the one thing
|
|
bounded acquisition exists to prevent.
|
|
"""
|
|
|
|
fresh: list[dict[str, Any]] = []
|
|
for item in page_features:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
attributes = item.get("attributes")
|
|
object_id = attributes.get("OBJECTID") if isinstance(attributes, dict) else None
|
|
if object_id is None:
|
|
raise AppError(
|
|
code="BATHYMETRY_PROVIDER_INVALID_RESPONSE",
|
|
message="VHA profile record has no OBJECTID, so pagination cannot be verified",
|
|
status_code=502,
|
|
)
|
|
key = str(object_id)
|
|
if key in seen_object_ids:
|
|
raise AppError(
|
|
code="BATHYMETRY_PROVIDER_UNSTABLE_PAGINATION",
|
|
message="VHA profile pagination repeated a record; the layer is not honouring resultOffset",
|
|
details={"object_id": key},
|
|
status_code=502,
|
|
)
|
|
seen_object_ids.add(key)
|
|
fresh.append(item)
|
|
return fresh
|
|
|
|
@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]
|
|
seen_object_ids: set[str] = set()
|
|
offset = 0
|
|
while offset < candidate_count:
|
|
if len(request_urls) > settings.bathymetry_profiles_max_pages:
|
|
raise AppError(
|
|
code="BATHYMETRY_SCOPE_TOO_LARGE",
|
|
message="VHA profile pagination exceeded the configured page limit; acquire smaller area partitions",
|
|
details={"max_pages": settings.bathymetry_profiles_max_pages},
|
|
status_code=422,
|
|
)
|
|
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,
|
|
)
|
|
response_hashes.append(page_sha)
|
|
request_urls.append(page_url)
|
|
if not page_features:
|
|
break
|
|
features.extend(
|
|
BathymetryProfileAcquisitionService._unseen_records(page_features, seen_object_ids)
|
|
)
|
|
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
|
|
seen_page_hashes: set[str] = set()
|
|
while True:
|
|
if len(seen_page_hashes) >= settings.bathymetry_profiles_max_pages:
|
|
raise AppError(
|
|
code="BATHYMETRY_SCOPE_TOO_LARGE",
|
|
message="VHA watercourse pagination exceeded the configured page limit",
|
|
details={"max_pages": settings.bathymetry_profiles_max_pages},
|
|
status_code=422,
|
|
)
|
|
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,
|
|
)
|
|
if response_sha in seen_page_hashes:
|
|
# The names themselves deduplicate by code, so a stuck
|
|
# provider produced no visible change while the loop, which
|
|
# ended only on exceededTransferLimit, kept requesting.
|
|
raise AppError(
|
|
code="BATHYMETRY_PROVIDER_UNSTABLE_PAGINATION",
|
|
message="VHA watercourse pagination returned the same page again",
|
|
status_code=502,
|
|
)
|
|
seen_page_hashes.add(response_sha)
|
|
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]],
|
|
*,
|
|
partition_properties: dict[str, Any] | None = 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",
|
|
**(partition_properties or {}),
|
|
},
|
|
"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 _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 (
|
|
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
|
|
)
|
|
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,
|
|
"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,
|
|
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(
|
|
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",
|
|
"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": {
|
|
"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)
|
|
|
|
@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")
|