from __future__ import annotations from collections.abc import Iterable from dataclasses import dataclass from typing import Any from uuid import UUID from geoalchemy2.shape import to_shape from shapely.geometry import box from shapely.ops import unary_union from sqlalchemy.orm import Session from app.core.errors import AppError from app.models import Area, Dataset, Project from app.services.dataset_consumption_gate_service import DatasetConsumptionGate from app.schemas.coverage import ( CoverageBBox, CoverageCatalogResponse, CoverageResolutionItem, CoverageResolveResponse, CoverageSourceContract, ) THEMES = ( "admin", "buildings", "roads", "surface_water", "land_cover_use", "nature", "population", "parcels", "soil", "elevation", "orthophoto", "flood_climate", "maritime_planning", "marine_environment", "bathymetry", ) ZONES = ( "belgium", "flanders", "wallonia", "brussels", "belgian_north_sea", "territorial_sea", "exclusive_economic_zone", "continental_shelf", ) STATUS_ORDER = ("unsupported", "not_configured", "partial", "operational") STATUS_RANK = {status: index for index, status in enumerate(STATUS_ORDER)} SCOPE_AREA_NAMES = { "belgium": "Belgium land", "flanders": "Flanders", "wallonia": "Wallonia", "brussels": "Brussels-Capital Region", "belgian_north_sea": "Belgian part of the North Sea", "territorial_sea": "Belgian territorial sea (0-12 nautical miles)", "exclusive_economic_zone": "Belgian exclusive economic zone beyond territorial sea", "continental_shelf": "Belgian continental shelf beyond territorial sea", } DETAIL_ZONES = ( "flanders", "wallonia", "brussels", "territorial_sea", "exclusive_economic_zone", "continental_shelf", ) @dataclass(frozen=True) class _SourceDefinition: contract: CoverageSourceContract materialized_layer_names: tuple[str, ...] = () materialized_source_names: tuple[str, ...] = () operational_themes: tuple[str, ...] = () def _contract( *, source_name: str, display_name: str, authority_level: str, coverage_zones: tuple[str, ...], themes: tuple[str, ...], native_layers: tuple[str, ...], geometry_types: tuple[str, ...], acquisition_mode: str, integration_status: str, source_url: str, attribution: str, license_note: str, limitation_message: str, materialized_layer_names: tuple[str, ...] = (), materialized_source_names: tuple[str, ...] = (), operational_themes: tuple[str, ...] = (), ) -> _SourceDefinition: return _SourceDefinition( contract=CoverageSourceContract( source_name=source_name, display_name=display_name, authority_level=authority_level, coverage_zones=list(coverage_zones), themes=list(themes), native_layers=list(native_layers), supported_geometry_types=list(geometry_types), acquisition_mode=acquisition_mode, integration_status=integration_status, source_url=source_url, attribution=attribution, license_note=license_note, limitation_message=limitation_message, ), materialized_layer_names=materialized_layer_names, materialized_source_names=materialized_source_names or (source_name,), operational_themes=operational_themes, ) SOURCE_DEFINITIONS = ( _contract( source_name="ngi_adminvector", display_name="NGI AdminVector", authority_level="authoritative", coverage_zones=("belgium", "flanders", "wallonia", "brussels", "belgian_north_sea"), themes=("admin",), native_layers=( "belgianterritory", "belgianmaritimezone", "region", "province", "municipality", ), geometry_types=("Polygon", "MultiPolygon"), acquisition_mode="operator_archive", integration_status="operational", source_url="https://www.geo.be/catalog/details/fb1e2993-2020-428c-9188-eb5f75e284b9", attribution="National Geographic Institute (NGI), AdminVector", license_note="CC BY 4.0", limitation_message="Administrative reference geometry; it does not provide thematic land content.", materialized_layer_names=( "belgium_land_boundary", "belgium_regions", "belgium_provinces", "belgium_municipalities", ), ), _contract( source_name="statbel", display_name="Statbel statistical sectors and population", authority_level="authoritative", coverage_zones=("belgium", "flanders", "wallonia", "brussels"), themes=("admin", "population"), native_layers=("statistical_sectors", "population_statistics"), geometry_types=("Polygon", "MultiPolygon", "Tabular"), acquisition_mode="operator_archive", integration_status="operational", source_url="https://statbel.fgov.be/en/open-data", attribution="Statbel", license_note="Consult the license of the selected Statbel release.", limitation_message=( "National editions require the governed plan-stage-review-apply operator; " "population in partially selected sectors is area-weighted." ), materialized_layer_names=("population",), operational_themes=("population",), ), _contract( source_name="digitaal_vlaanderen", display_name="Flemish authoritative services", authority_level="authoritative", coverage_zones=("flanders",), themes=( "buildings", "roads", "surface_water", "land_cover_use", "nature", "parcels", "soil", "elevation", "orthophoto", "flood_climate", ), native_layers=("GRB", "BWK", "DHMV", "OMWRGBMRVL", "OGRK", "Mercator"), geometry_types=("Point", "LineString", "Polygon", "MultiPolygon", "Raster"), acquisition_mode="bounded_api", integration_status="operational", source_url="https://www.vlaanderen.be/datavindplaats", attribution="Digitaal Vlaanderen and the authoritative Flemish source owners", license_note="Consult the license and attribution stored with each acquired dataset.", limitation_message="Operational only for bounded products implemented by GeoIntel and materialized in the project.", materialized_source_names=( "grb", "digitaal_vlaanderen_buildings_addresses_register", "digitaal_vlaanderen_dhmv", "digitaal_vlaanderen_orthophoto", "vmm_flood_hazard", "department_omgeving_thematic_raster", "inbo_bwk_natura2000", "dov_soil_map", "agentschap_landbouw_zeevisserij_agricultural_parcels", ), ), _contract( source_name="vmm_vha_bathymetry_profiles", display_name="VHA historische dwarsprofielen", authority_level="authoritative", coverage_zones=("flanders",), themes=("bathymetry",), native_layers=("digitale_atlas_profile_points",), geometry_types=("Point",), acquisition_mode="bounded_api", integration_status="operational", source_url="https://vha.waterinfo.be/arcgis/rest/services/digitale_atlas/MapServer/0", attribution="Vlaamse Milieumaatschappij (VMM), Vlaamse Hydrografische Atlas", license_note="Hergebruik volgens de voorwaarden van de Vlaamse overheid en de bronmetadata.", limitation_message=( "Historische puntmetingen met bronafhankelijke meetdatum en verticale referentie; " "geen continue actuele bodemkaart en zonder gelijktijdig waterpeil geen watervolume." ), materialized_source_names=("vmm_vha_bathymetry_profiles",), ), _contract( source_name="spw_geoportail", display_name="SPW Geoportail Wallonie", authority_level="authoritative", coverage_zones=("wallonia",), themes=( "buildings", "roads", "surface_water", "land_cover_use", "nature", "soil", "elevation", "orthophoto", "flood_climate", "bathymetry", ), native_layers=("PICC", "orthophotos", "MNT", "hydrography", "land_cover"), geometry_types=("Point", "LineString", "Polygon", "MultiPolygon", "Raster"), acquisition_mode="bounded_api", integration_status="operational", source_url="https://geoportail.wallonie.be/catalogue", attribution="Service public de Wallonie", license_note="Consult the license of each Geoportail Wallonie product.", limitation_message=( "Bounded PICC buildings, road axes and hydrography, the legally current flood-hazard polygons, " "operator-imported SPW bathymetry and bounded SPW MNT terrain are operational; other Walloon themes remain separately governed." ), materialized_source_names=("spw_picc", "spw_flood_hazard", "spw_walous_land_cover", "spw_bathymetry", "spw_terrain"), operational_themes=("buildings", "roads", "surface_water", "land_cover_use", "elevation", "flood_climate", "bathymetry"), ), _contract( source_name="urbis", display_name="UrbIS Brussels", authority_level="authoritative", coverage_zones=("brussels",), themes=("buildings", "roads", "surface_water", "land_cover_use", "parcels", "orthophoto"), native_layers=("parcels", "buildings", "roads", "hydrography", "orthophoto"), geometry_types=("Point", "LineString", "Polygon", "MultiPolygon", "Raster"), acquisition_mode="bounded_api", integration_status="operational", source_url="https://datastore.brussels", attribution="Brussels UrbIS", license_note="Consult the license of the selected UrbIS dataset.", limitation_message=( "Bounded UrbIS buildings, cadastral parcels, street axes and Land Cover blocks are operational. " "Permanent water uses the official WB block class; no separate hydrography network is inferred." ), materialized_source_names=("urbis",), operational_themes=("buildings", "parcels", "roads", "surface_water", "land_cover_use"), ), _contract( source_name="rbins_marine_reporting_units", display_name="RBINS marine reporting units", authority_level="authoritative", coverage_zones=( "belgian_north_sea", "territorial_sea", "exclusive_economic_zone", "continental_shelf", ), themes=("admin", "marine_environment"), native_layers=("marine_reporting_units_2024",), geometry_types=("Polygon", "MultiPolygon"), acquisition_mode="operator_wfs", integration_status="operational", source_url=( "https://metadata.naturalsciences.be/geonetwork/srv/api/records/" "29f40b0d-2a3e-49a8-870a-e9b4acd4d1e3" ), attribution="Royal Belgian Institute of Natural Sciences (RBINS), BMDC", license_note="Reuse conditions are retained from the source metadata with every persisted artifact.", limitation_message="The EEZ and continental shelf can share geometry while retaining different legal semantics.", materialized_layer_names=("marine_legal_scopes",), ), _contract( source_name="rbins_msp_2026", display_name="Belgian Marine Spatial Plan 2026-2034", authority_level="authoritative", coverage_zones=( "belgian_north_sea", "territorial_sea", "exclusive_economic_zone", "continental_shelf", ), themes=("maritime_planning", "marine_environment"), native_layers=("imsp26",), geometry_types=("Point", "LineString", "Polygon", "MultiPolygon"), acquisition_mode="operator_wfs", integration_status="operational", source_url="https://www.health.belgium.be/en/themes/environment/marine-environment/marine-spatial-plan", attribution="Belgian federal Marine Environment service and RBINS", license_note="Official source metadata and attribution are retained with the imported snapshot.", limitation_message="The dataset represents the legally current 2026-2034 plan, not live maritime activity.", materialized_layer_names=("marine_spatial_plan_2026",), ), _contract( source_name="mdk_bathymetry", display_name="MDK Belgian North Sea depth model", authority_level="authoritative", coverage_zones=( "belgian_north_sea", "territorial_sea", "exclusive_economic_zone", "continental_shelf", ), themes=("bathymetry",), native_layers=("depth_model_20m_lat",), geometry_types=("Raster",), acquisition_mode="catalog_only", integration_status="not_configured", source_url="https://www.vlaanderen.be/datavindplaats", attribution="Agentschap Maritieme Dienstverlening en Kust (MDK)", license_note="Consult the official product license before acquisition.", limitation_message=( "Bounded strict-TLS WCS acquisition is implemented but stays disabled until the operator enables it " "with a live-validated coverage id; no depths are synthesized." ), ), ) FLANDERS_THEME_DATASETS: dict[str, dict[str, tuple[str, ...]]] = { "buildings": { "grb": ("buildings",), "digitaal_vlaanderen_buildings_addresses_register": (), }, "roads": {"grb": ("roads",)}, "surface_water": {"grb": ("water",)}, "land_cover_use": { "department_omgeving_thematic_raster": (), "agentschap_landbouw_zeevisserij_agricultural_parcels": (), }, "nature": {"inbo_bwk_natura2000": ()}, "parcels": { "grb": ("parcels",), "agentschap_landbouw_zeevisserij_agricultural_parcels": (), }, "soil": {"dov_soil_map": ()}, "elevation": {"digitaal_vlaanderen_dhmv": ()}, "orthophoto": {"digitaal_vlaanderen_orthophoto": ()}, "flood_climate": {"vmm_flood_hazard": ()}, } REGIONAL_THEME_DATASETS: dict[str, dict[str, dict[str, tuple[str, ...]]]] = { "spw_geoportail": { "buildings": {"spw_picc": ("buildings",)}, "roads": {"spw_picc": ("roads",)}, "surface_water": {"spw_picc": ("water",)}, "land_cover_use": {"spw_walous_land_cover": ()}, "elevation": {"spw_terrain": ()}, "flood_climate": {"spw_flood_hazard": ("flood_hazard",)}, "bathymetry": {"spw_bathymetry": ()}, "orthophoto": {"spw_orthophoto": ()}, }, "urbis": { "buildings": {"urbis": ("buildings",)}, "parcels": {"urbis": ("parcels",)}, "roads": {"urbis": ("roads",)}, "surface_water": {"urbis": ("water",)}, "land_cover_use": {"urbis": ("space_occupation", "forest")}, "orthophoto": {"urbis_orthophoto": ()}, }, } class CoverageRegistryService: @staticmethod def catalog() -> CoverageCatalogResponse: return CoverageCatalogResponse( themes=list(THEMES), zones=list(ZONES), statuses=list(STATUS_ORDER), sources=[definition.contract for definition in SOURCE_DEFINITIONS], ) @staticmethod def normalize_themes(themes: Iterable[str]) -> list[str]: requested = list(dict.fromkeys(str(theme).strip().lower() for theme in themes if str(theme).strip())) invalid = sorted(set(requested) - set(THEMES)) if invalid: raise AppError( code="COVERAGE_THEME_UNSUPPORTED", message="One or more coverage themes are unsupported", status_code=422, details={"unsupported_themes": invalid, "supported_themes": list(THEMES)}, ) return requested or list(THEMES) @staticmethod def _geometry(value: Any): if value is None: return None return value if hasattr(value, "__geo_interface__") else to_shape(value) @staticmethod def _intersected_zones(areas: list[Area], selection) -> tuple[list[str], bool]: geometries: dict[str, Any] = {} by_name = {area.name: area for area in areas} for zone, area_name in SCOPE_AREA_NAMES.items(): area = by_name.get(area_name) geometry = CoverageRegistryService._geometry(area.geometry) if area else None if geometry is not None and not geometry.is_empty: geometries[zone] = geometry detail_intersections = [ zone for zone in DETAIL_ZONES if zone in geometries and geometries[zone].intersects(selection) ] zones = detail_intersections if not any(zone in zones for zone in ("flanders", "wallonia", "brussels")): if "belgium" in geometries and geometries["belgium"].intersects(selection): zones = ["belgium", *zones] if not any(zone in zones for zone in ("territorial_sea", "exclusive_economic_zone", "continental_shelf")): if "belgian_north_sea" in geometries and geometries["belgian_north_sea"].intersects(selection): zones = [*zones, "belgian_north_sea"] intersected_geometries = [geometries[zone].intersection(selection) for zone in zones if zone in geometries] covered = unary_union(intersected_geometries) if intersected_geometries else None outside = covered is None or covered.is_empty or not covered.covers(selection) return zones, outside @staticmethod def _matching_datasets( datasets: list[Dataset], definition: _SourceDefinition, theme: str, zone: str, selection: Any, ) -> tuple[list[Dataset], bool]: if definition.operational_themes and theme not in definition.operational_themes: return [], False matches: list[Dataset] = [] bounded_scopes: list[Any] = [] zone_scoped_materialization = False for dataset in datasets: if dataset.status != "ready" or dataset.source_name not in definition.materialized_source_names: continue # A source-name claim alone must not cause an unsafe artifact to # appear as operational authoritative coverage. if not DatasetConsumptionGate.eligible_for_authoritative_coverage(dataset): continue layer_names = definition.materialized_layer_names if definition.contract.source_name == "digitaal_vlaanderen": theme_sources = FLANDERS_THEME_DATASETS.get(theme, {}) if dataset.source_name not in theme_sources: continue layer_names = theme_sources[dataset.source_name] elif definition.contract.source_name in REGIONAL_THEME_DATASETS: theme_sources = REGIONAL_THEME_DATASETS[definition.contract.source_name].get(theme, {}) if dataset.source_name not in theme_sources: continue layer_names = theme_sources[dataset.source_name] metadata = dataset.source_metadata if isinstance(dataset.source_metadata, dict) else {} coverage_zones = metadata.get("coverage_zones") or metadata.get("coverage_zone") or [] if isinstance(coverage_zones, str): coverage_zones = [coverage_zones] acquired_bbox = metadata.get("bbox_epsg4326") if ( definition.contract.acquisition_mode == "bounded_api" and isinstance(acquired_bbox, list) and len(acquired_bbox) == 4 ): try: acquired_scope = box(*(float(value) for value in acquired_bbox)) except (TypeError, ValueError): continue if not acquired_scope.is_valid or not acquired_scope.intersects(selection): continue bounded_scopes.append(acquired_scope) elif definition.contract.acquisition_mode == "bounded_api" and coverage_zones: zone_scoped_materialization = zone in coverage_zones or "belgium" in coverage_zones layer_matches = not layer_names or dataset.reference_layer_name in layer_names zone_matches = not coverage_zones or zone in coverage_zones or "belgium" in coverage_zones if layer_matches and zone_matches: matches.append(dataset) if not matches: return [], False fully_covered = True if definition.contract.acquisition_mode == "bounded_api": fully_covered = zone_scoped_materialization or (bool(bounded_scopes) and unary_union(bounded_scopes).covers(selection)) return matches, fully_covered @staticmethod def _resolve_item( *, zone: str, theme: str, datasets: list[Dataset], selection: Any, ) -> CoverageResolutionItem: definitions = [ definition for definition in SOURCE_DEFINITIONS if zone in definition.contract.coverage_zones and theme in definition.contract.themes ] if not definitions: return CoverageResolutionItem( zone=zone, theme=theme, status="unsupported", source_names=[], materialized_dataset_ids=[], limitation_message="No audited source contract supports this theme in the selected zone.", ) materialized: list[Dataset] = [] evidence: list[dict[str, Any]] = [] source_statuses: list[str] = [] limitations: list[str] = [] for definition in definitions: matches, fully_covered = CoverageRegistryService._matching_datasets( datasets, definition, theme, zone, selection, ) materialized.extend(matches) for dataset in matches: metadata = dataset.source_metadata if isinstance(getattr(dataset, "source_metadata", None), dict) else {} observed_at = getattr(dataset, "observed_at", None) published_at = metadata.get("published_at") or metadata.get("publication_date") or metadata.get("published_on") evidence.append({ "dataset_id": dataset.id, "source_name": str(dataset.source_name or definition.contract.source_name), "authority_level": definition.contract.authority_level, "source_version": getattr(dataset, "source_version", None), "observed_at": observed_at.isoformat() if hasattr(observed_at, "isoformat") else (str(observed_at) if observed_at else None), "published_at": str(published_at) if published_at else None, "crs": getattr(dataset, "crs", None) or metadata.get("source_crs"), "resolution": getattr(dataset, "resolution_json", None), "coverage_bbox_epsg4326": metadata.get("bbox_epsg4326"), "attribution": metadata.get("attribution") or definition.contract.attribution, "license_note": metadata.get("license_note") or definition.contract.license_note, "checksum_sha256": getattr(dataset, "checksum_sha256", None), }) if matches and fully_covered: source_statuses.append("operational") elif matches: source_statuses.append("partial") elif ( definition.contract.integration_status == "operational" and (not definition.operational_themes or theme in definition.operational_themes) ): source_statuses.append("partial") else: source_statuses.append( "not_configured" if definition.contract.integration_status == "operational" else definition.contract.integration_status ) limitations.append(definition.contract.limitation_message) best_status = max(source_statuses, key=STATUS_RANK.__getitem__) return CoverageResolutionItem( zone=zone, theme=theme, status=best_status, source_names=[definition.contract.source_name for definition in definitions], materialized_dataset_ids=list(dict.fromkeys(dataset.id for dataset in materialized)), evidence=list({str(item["dataset_id"]): item for item in evidence}.values()), limitation_message=" ".join(dict.fromkeys(limitations)), ) @staticmethod def resolve( db: Session, project_id: UUID, bbox: CoverageBBox, themes: Iterable[str], ) -> CoverageResolveResponse: if not db.get(Project, project_id): raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) requested_themes = CoverageRegistryService.normalize_themes(themes) selection = box(bbox.minx, bbox.miny, bbox.maxx, bbox.maxy) areas = db.query(Area).filter(Area.project_id == project_id).all() datasets = db.query(Dataset).filter(Dataset.project_id == project_id).all() zones, outside_supported_scope = CoverageRegistryService._intersected_zones(areas, selection) if not zones: return CoverageResolveResponse( project_id=project_id, bbox=bbox, requested_themes=requested_themes, intersected_zones=[], outside_supported_scope=True, items=[], warnings=["The selection does not intersect a persisted Belgium or Belgian North Sea scope."], ) items = [ CoverageRegistryService._resolve_item( zone=zone, theme=theme, datasets=datasets, selection=selection, ) for zone in zones for theme in requested_themes ] warnings = [] if outside_supported_scope: warnings.append("Part of the selection lies outside the persisted Belgium and Belgian North Sea scopes.") if len(zones) > 1: warnings.append( "The selection crosses coverage zones; results remain split and only semantically compatible metrics may be merged." ) return CoverageResolveResponse( project_id=project_id, bbox=bbox, requested_themes=requested_themes, intersected_zones=zones, outside_supported_scope=outside_supported_scope, items=items, warnings=warnings, )