from __future__ import annotations from pathlib import Path from types import SimpleNamespace from uuid import uuid4 import numpy as np from fastapi.testclient import TestClient from pyproj import Transformer import rasterio from rasterio.transform import from_origin from app.core.config import Settings from app.db.session import get_db from app.main import app from app.models import Dataset, Job, Project from app.schemas.dhmv import TerrainSelectionRequest from app.schemas.spw_terrain import SpwTerrainAcquireRequest from app.services.dataset_service import DatasetService from app.services.spw_terrain_service import SpwTerrainService from app.services.terrain_analysis_service import TerrainAnalysisService class FakeQuery: def filter(self, *_args): return self def order_by(self, *_args): return self def first(self): return None class FakeSession: def __init__(self, project, dataset=None): self.project = project self.dataset = dataset self.added = [] def get(self, model, row_id): if model is Project and row_id == self.project.id: return self.project if model is Dataset and self.dataset is not None and row_id == self.dataset.id: return self.dataset return next( ( item for item in self.added if isinstance(item, model) and item.id == row_id ), None, ) def query(self, _model): return FakeQuery() def add(self, row): self.added.append(row) def commit(self): return None def rollback(self): return None def refresh(self, row): return row def settings(source_dir: Path) -> Settings: return Settings( _env_file=None, SPW_TERRAIN_SOURCE_DIR=str(source_dir), SPW_TERRAIN_ANALYSIS_RESOLUTION_M=5, SPW_TERRAIN_MAX_SIDE_M=20_000, SPW_TERRAIN_MAX_PIXELS=1_000_000, DHMV_MAX_PIXELS=1_000_000, ) def make_source(path: Path) -> list[float]: to_3812 = Transformer.from_crs("EPSG:4326", "EPSG:3812", always_xy=True) to_4326 = Transformer.from_crs("EPSG:3812", "EPSG:4326", always_xy=True) x, y = to_3812.transform(4.85, 50.45) values = np.linspace(100.0, 125.0, 40_000, dtype="float32").reshape(200, 200) with rasterio.open( path, "w", driver="GTiff", width=200, height=200, count=1, dtype="float32", crs="EPSG:3812", transform=from_origin(x, y + 200, 1, 1), nodata=-9999.0, ) as target: target.write(values, 1) min_lon, min_lat = to_4326.transform(x, y) max_lon, max_lat = to_4326.transform(x + 200, y + 200) return [min_lon, min_lat, max_lon, max_lat] def write_source_checksum(source_dir: Path) -> str: digest = "a" * 64 (source_dir / SpwTerrainService.SOURCE_SHA256_FILENAME).write_text( f"{digest} {SpwTerrainService.SOURCE_FILENAME}\n", encoding="ascii" ) return digest def test_spw_terrain_registry_reports_real_source_state(tmp_path: Path) -> None: before = SpwTerrainService.list_products(settings=settings(tmp_path))[0] assert before["status"] == "source_not_provisioned" make_source(tmp_path / SpwTerrainService.SOURCE_FILENAME) without_checksum = SpwTerrainService.list_products(settings=settings(tmp_path))[0] assert without_checksum["configured"] is False write_source_checksum(tmp_path) after = SpwTerrainService.list_products(settings=settings(tmp_path))[0] assert after["configured"] is True assert after["coverage_zones"] == ["wallonia"] assert after["source_crs"] == "EPSG:3812" assert after["vertical_reference"].endswith("(EPSG:5710)") def test_spw_terrain_acquisition_persists_bounded_dng_raster_and_provenance( tmp_path: Path, monkeypatch ) -> None: bbox = make_source(tmp_path / SpwTerrainService.SOURCE_FILENAME) source_digest = write_source_checksum(tmp_path) project = Project(id=uuid4(), name="Belgium") captured = {} def persist(_db, **kwargs): captured.update(kwargs) return SimpleNamespace(id=uuid4()) monkeypatch.setattr(DatasetService, "import_raster_bytes", persist) result = SpwTerrainService.acquire( FakeSession(project), project.id, SpwTerrainAcquireRequest( bbox={ "min_x": bbox[0], "min_y": bbox[1], "max_x": bbox[2], "max_y": bbox[3], "crs": "EPSG:4326", }, resolution_m=5, force_refresh=True, ), settings=settings(tmp_path), ) assert result["provider"] == "spw_terrain" assert result["resolution_m"] == 5 assert result["valid_pixel_count"] > 0 assert captured["source_metadata"]["vertical_unit_label"] == "m DNG" assert captured["source_metadata"]["coverage_zones"] == ["wallonia"] assert captured["provenance_metadata"]["resampling"] == "bilinear" assert captured["provenance_metadata"]["source_sha256"] == source_digest assert captured["temporal_series_key"].startswith("spw:terrain:mnt:") assert captured["temporal_granularity"] == "period" assert captured["valid_from"].date().isoformat() == "2021-02-19" temporal = DatasetService._validate_temporal_metadata( temporal_series_key=captured["temporal_series_key"], observed_at=captured["observed_at"], valid_from=captured["valid_from"], valid_to=captured["valid_to"], temporal_granularity=captured["temporal_granularity"], source_version=captured["source_version"], ) assert temporal["temporal_granularity"] == "period" with rasterio.MemoryFile(captured["content"]) as memory: with memory.open() as derived: assert derived.crs.to_epsg() == 3812 assert derived.res == (5.0, 5.0) assert derived.nodata == -9999.0 def test_terrain_analysis_preserves_spw_vertical_datum_and_limitations( tmp_path: Path, ) -> None: bbox = make_source(tmp_path / "derived.tif") project = Project(id=uuid4(), name="Belgium") dataset = Dataset( id=uuid4(), project_id=project.id, name="derived.tif", dataset_type="raster", source="SPW MNT", source_name="spw_terrain", source_metadata={ "product_key": SpwTerrainService.PRODUCT_KEY, "surface_model": "terrain", "vertical_reference": SpwTerrainService.VERTICAL_REFERENCE, "vertical_unit_label": "m DNG", "limitation_message": SpwTerrainService.LIMITATION, }, storage_path=str(tmp_path / "derived.tif"), status="ready", ) result = TerrainAnalysisService.analyze( FakeSession(project, dataset), project.id, dataset.id, TerrainSelectionRequest( bbox={ "min_x": bbox[0], "min_y": bbox[1], "max_x": bbox[2], "max_y": bbox[3], "crs": "EPSG:4326", }, ), settings=settings(tmp_path), ) assert result["vertical_reference"] == SpwTerrainService.VERTICAL_REFERENCE assert result["summary"]["metric_unit"] == "m DNG" assert result["summary"]["metrics"][0]["metric_unit"] == "m DNG" assert result["limitation_message"] == SpwTerrainService.LIMITATION assert result["unsupported_metrics"] == ["water_depth_m", "water_volume_m3"] def test_spw_terrain_routes_use_canonical_envelopes(monkeypatch) -> None: project = Project(id=uuid4(), name="Belgium") db = FakeSession(project) monkeypatch.setattr( SpwTerrainService, "acquire", lambda *_args, **_kwargs: { "output_dataset_id": str(uuid4()), "provider": SpwTerrainService.PROVIDER, }, ) monkeypatch.setattr( SpwTerrainService, "list_products", lambda *_args, **_kwargs: [ { "key": SpwTerrainService.PRODUCT_KEY, "display_name": SpwTerrainService.DISPLAY_NAME, "surface_model": "terrain", "source_filename": SpwTerrainService.SOURCE_FILENAME, "native_resolution_m": 1, "analysis_resolution_m": 5, "source_crs": "EPSG:3812", "vertical_reference": SpwTerrainService.VERTICAL_REFERENCE, "acquisition_period": SpwTerrainService.ACQUISITION_PERIOD, "catalog_url": SpwTerrainService.CATALOG_URL, "attribution": SpwTerrainService.ATTRIBUTION, "license_note": SpwTerrainService.LICENSE_NOTE, "limitation_message": SpwTerrainService.LIMITATION, "coverage_zones": ["wallonia"], "configured": True, "status": "configured", } ], ) app.dependency_overrides[get_db] = lambda: db try: client = TestClient(app) products = client.get( f"/api/v1/projects/{project.id}/datasets/spw-terrain/products" ) acquisition = client.post( f"/api/v1/projects/{project.id}/datasets/spw-terrain/acquire", json={ "bbox": { "min_x": 4.8, "min_y": 50.4, "max_x": 4.9, "max_y": 50.5, "crs": "EPSG:4326", }, "product_key": SpwTerrainService.PRODUCT_KEY, }, ) finally: app.dependency_overrides.clear() assert products.status_code == 200 and products.json()["data"]["total"] == 1 assert ( acquisition.status_code == 200 and acquisition.json()["data"]["job_type"] == "raster.spw-terrain.acquire" ) assert any(isinstance(item, Job) for item in db.added)