from __future__ import annotations from datetime import datetime, timezone import importlib.util 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.thematic_raster import ( ThematicRasterAcquireRequest, ThematicRasterSelectionRequest, ) from app.schemas.temporal import TemporalComparisonRequest from app.services.dataset_service import DatasetService from app.services.temporal_analysis_service import TemporalAnalysisService from app.services.walous_land_cover_service import WalousLandCoverService def load_provisioner(): path = ( Path(__file__).resolve().parents[2] / "scripts" / "provision_walous_sources.py" ) spec = importlib.util.spec_from_file_location( "walous_source_provisioner_test", path ) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module 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 match = next( ( item for item in self.added if isinstance(item, model) and item.id == row_id ), None, ) if match is not None: return match return 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 make_source( path: Path, *, dtype: str = "uint8", nodata: int = 255, class_codes: list[int] | None = None, ) -> tuple[list[float], np.ndarray]: 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) transform = from_origin(x, y + 100, 1, 1) class_codes = class_codes or [1, 2, 3, 4, 5, 6, 7, 8, 9, 80, 90] values = np.empty((100, len(class_codes) * 20), dtype=dtype) for index, class_code in enumerate(class_codes): values[:, index * 20 : (index + 1) * 20] = class_code with rasterio.open( path, "w", driver="GTiff", width=values.shape[1], height=100, count=1, dtype=dtype, crs="EPSG:3812", transform=transform, nodata=nodata, ) as target: target.write(values, 1) min_lon, min_lat = to_4326.transform(x, y) max_lon, max_lat = to_4326.transform(x + values.shape[1], y + values.shape[0]) return [min_lon, min_lat, max_lon, max_lat], values def settings(source_dir: Path) -> Settings: return Settings( _env_file=None, WALOUS_SOURCE_DIR=str(source_dir), WALOUS_ANALYSIS_RESOLUTION_M=10, WALOUS_MAX_SIDE_M=60_000, WALOUS_MAX_PIXELS=1_000_000, ) def test_walous_registry_reports_real_provisioning_state(tmp_path: Path) -> None: before = { item["key"]: item for item in WalousLandCoverService.list_products(settings=settings(tmp_path)) } assert before["walous_land_cover_2023"]["status"] == "source_not_provisioned" make_source(tmp_path / "walous_land_cover_2023_3812.tif") after = { item["key"]: item for item in WalousLandCoverService.list_products(settings=settings(tmp_path)) } assert after["walous_land_cover_2023"]["configured"] is True assert after["walous_land_cover_2023"]["source_crs"] == "EPSG:3812" assert after["walous_land_cover_2023"]["native_resolution_m"] == 1.0 assert after["walous_land_cover_2023"]["analysis_resolution_m"] == 10.0 assert after["walous_land_cover_2023"]["coverage_zones"] == ["wallonia"] assert after["walous_land_cover_2023"]["included_source_values"] == [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 80, 90, ] assert after["walous_land_cover_2023"]["source_value_unit"] == "walous_class_code" def test_walous_provisioner_accepts_official_non_contiguous_class_codes( tmp_path: Path, ) -> None: source_path = tmp_path / "walous_land_cover_2023_3812.tif" make_source(source_path) validation = load_provisioner().validate_raster(source_path) assert validation["sample_classes"] == [1, 2, 3, 4, 5, 6, 7, 8, 9, 80, 90] def test_walous_2018_registry_and_provisioner_accept_official_stacked_classes( tmp_path: Path, ) -> None: source_codes = sorted(WalousLandCoverService.WALOUS_2018_CLASS_CROSSWALK) source_path = tmp_path / "walous_land_cover_2018_3812.tif" make_source(source_path, class_codes=source_codes) validation = load_provisioner().validate_raster(source_path) registry = { item["key"]: item for item in WalousLandCoverService.list_products(settings=settings(tmp_path)) } assert validation["sample_classes"] == source_codes assert validation["implicit_source_nodata_values"] == [0] assert registry["walous_land_cover_2018"]["configured"] is True assert registry["walous_land_cover_2018"]["observation_year"] == 2018 assert "crosswalk" in registry["walous_land_cover_2018"]["limitation_message"] def test_walous_2018_acquisition_normalizes_stacked_classes_with_explicit_provenance( tmp_path: Path, monkeypatch ) -> None: source_codes = sorted(WalousLandCoverService.WALOUS_2018_CLASS_CROSSWALK) bbox, _values = make_source( tmp_path / "walous_land_cover_2018_3812.tif", class_codes=source_codes, ) 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 = WalousLandCoverService.acquire( FakeSession(project), project.id, ThematicRasterAcquireRequest( bbox={ "min_x": bbox[0], "min_y": bbox[1], "max_x": bbox[2], "max_y": bbox[3], "crs": "EPSG:4326", }, product_key="walous_land_cover_2018", force_refresh=True, ), settings=settings(tmp_path), ) assert result["observation_year"] == 2018 assert captured["source_metadata"]["classes_present"] == [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 80, 90, ] assert captured["source_metadata"]["source_classes_present"] == source_codes assert captured["source_metadata"]["class_crosswalk"][62] == 2 assert captured["source_metadata"]["class_crosswalk"][0] == 255 assert ( captured["source_metadata"]["attribution"] == "Service public de Wallonie (SPW), UCLouvain, ULB, ISSeP" ) assert captured["observed_at"].date().isoformat() == "2018-12-31" def test_walous_acquisition_reads_real_classes_and_persists_provenance( tmp_path: Path, monkeypatch ) -> None: bbox, _values = make_source(tmp_path / "walous_land_cover_2023_3812.tif") project = Project(id=uuid4(), name="Belgium") db = FakeSession(project) captured = {} output_id = uuid4() def persist(_db, **kwargs): captured.update(kwargs) return SimpleNamespace(id=output_id) monkeypatch.setattr(DatasetService, "import_raster_bytes", persist) result = WalousLandCoverService.acquire( db, project.id, ThematicRasterAcquireRequest( bbox={ "min_x": bbox[0], "min_y": bbox[1], "max_x": bbox[2], "max_y": bbox[3], "crs": "EPSG:4326", }, product_key="walous_land_cover_2023", force_refresh=True, ), settings=settings(tmp_path), ) assert result["output_dataset_id"] == str(output_id) assert result["resolution_m"] == 10 assert captured["source_name"] == "spw_walous_land_cover" assert captured["source_metadata"]["classes_present"] == [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 80, 90, ] assert captured["provenance_metadata"]["resampling"] == "nearest" assert captured["temporal_series_key"].startswith("spw:walous:land-cover:") assert captured["observed_at"].date().isoformat() == "2023-06-25" assert captured["valid_from"].date().isoformat() == "2023-05-27" assert captured["valid_to"] == captured["observed_at"] def test_walous_acquisition_accepts_official_signed_int8_nodata( tmp_path: Path, monkeypatch ) -> None: bbox, _values = make_source( tmp_path / "walous_land_cover_2023_3812.tif", dtype="int8", nodata=-128, ) project = Project(id=uuid4(), name="Belgium") output_id = uuid4() captured = {} def persist(_db, **kwargs): captured.update(kwargs) return SimpleNamespace(id=output_id) monkeypatch.setattr(DatasetService, "import_raster_bytes", persist) result = WalousLandCoverService.acquire( FakeSession(project), project.id, ThematicRasterAcquireRequest( bbox={ "min_x": bbox[0], "min_y": bbox[1], "max_x": bbox[2], "max_y": bbox[3], "crs": "EPSG:4326", }, product_key="walous_land_cover_2023", force_refresh=True, ), settings=settings(tmp_path), ) assert result["output_dataset_id"] == str(output_id) assert captured["source_metadata"]["classes_present"] == [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 80, 90, ] with rasterio.MemoryFile(captured["content"]) as memory: with memory.open() as derived: assert derived.dtypes == ("uint8",) assert derived.nodata == 255 def test_walous_analysis_returns_semantic_area_metrics( tmp_path: Path, monkeypatch ) -> None: bbox, _values = make_source(tmp_path / "walous_land_cover_2023_3812.tif") project = Project(id=uuid4(), name="Belgium") output_id = uuid4() captured = {} def persist(_db, **kwargs): captured.update(kwargs) return SimpleNamespace(id=output_id) monkeypatch.setattr(DatasetService, "import_raster_bytes", persist) db = FakeSession(project) payload = ThematicRasterAcquireRequest( bbox={ "min_x": bbox[0], "min_y": bbox[1], "max_x": bbox[2], "max_y": bbox[3], "crs": "EPSG:4326", }, product_key="walous_land_cover_2023", force_refresh=True, ) WalousLandCoverService.acquire(db, project.id, payload, settings=settings(tmp_path)) persisted_path = tmp_path / "derived.tif" persisted_path.write_bytes(captured["content"]) dataset = Dataset( id=output_id, project_id=project.id, name="derived.tif", dataset_type="raster", source="SPW WALOUS", source_name="spw_walous_land_cover", source_metadata=captured["source_metadata"], provenance_metadata=captured["provenance_metadata"], storage_path=str(persisted_path), status="ready", ) db.dataset = dataset result = WalousLandCoverService.analyze( db, project.id, output_id, ThematicRasterSelectionRequest(bbox=payload.bbox), ) metrics = { item["metric_key"]: item["metric_value"] for item in result["summary"]["metrics"] } assert result["metric_kind"] == "categorical_area" assert metrics["land_cover_observed_area_ha"] > 0 assert metrics["forest_cover_area_ha"] > 0 assert metrics["surface_water_area_ha"] > 0 assert metrics["artificial_cover_area_ha"] > 0 assert metrics["annual_herbaceous_cover_area_ha"] > 0 assert metrics["permanent_herbaceous_cover_area_ha"] > 0 assert metrics["bare_soil_area_ha"] > 0 assert "water_volume" in result["unsupported_metrics"] def test_walous_render_png_uses_governed_class_colours(tmp_path: Path) -> None: bbox, _values = make_source(tmp_path / "walous_land_cover_2023_3812.tif") project = Project(id=uuid4(), name="Belgium") dataset = Dataset( id=uuid4(), project_id=project.id, name="walous.tif", dataset_type="raster", source="SPW WALOUS", source_name="spw_walous_land_cover", source_metadata={ "product_key": "walous_land_cover_2023", "bbox_epsg4326": bbox, }, storage_path=str(tmp_path / "walous_land_cover_2023_3812.tif"), status="ready", ) db = FakeSession(project, dataset) rendered = WalousLandCoverService.render_png(db, project.id, dataset.id) assert rendered.startswith(b"\x89PNG\r\n\x1a\n") def test_walous_temporal_comparison_reuses_persisted_raster_metrics( monkeypatch, ) -> None: project_id = uuid4() earlier = Dataset( id=uuid4(), project_id=project_id, name="walous-2020.tif", dataset_type="raster", source="SPW WALOUS", source_name="spw_walous_land_cover", temporal_series_key="spw:walous:land-cover:selection", observed_at=datetime(2020, 12, 31, 23, 59, 59, tzinfo=timezone.utc), source_version="WAL_OCS_IA__2020", ) later = Dataset( id=uuid4(), project_id=project_id, name="walous-2023.tif", dataset_type="raster", source="SPW WALOUS", source_name="spw_walous_land_cover", temporal_series_key=earlier.temporal_series_key, observed_at=datetime(2023, 12, 31, 23, 59, 59, tzinfo=timezone.utc), source_version="WAL_OCS_IA__2023", ) rows = {earlier.id: earlier, later.id: later} class TemporalSession: def get(self, model, row_id): return rows.get(row_id) if model is Dataset else None def analyze(_db, _project_id, dataset_id, _payload): value = 4.0 if dataset_id == earlier.id else 5.5 return { "summary": { "metric_label": "Gekarteerde landbedekking", "metric_value": value, "metric_unit": "ha", "aggregation_method": "nearest_resampled_cells_times_cell_area", "primary_metric_key": "land_cover_observed_area_ha", "metrics": [ { "metric_key": "land_cover_observed_area_ha", "metric_label": "Gekarteerde landbedekking", "metric_value": value, "metric_unit": "ha", "aggregation_method": "nearest_resampled_cells_times_cell_area", "is_estimate": True, } ], }, "limitation_message": ( "2018 crosswalk and methodology limitation." if dataset_id == earlier.id else "2023 edition accuracy limitation." ), } monkeypatch.setattr(WalousLandCoverService, "analyze", analyze) payload = TemporalComparisonRequest( earlier_dataset_id=earlier.id, later_dataset_id=later.id, bbox={ "min_x": 4.8, "min_y": 50.4, "max_x": 4.9, "max_y": 50.5, "crs": "EPSG:4326", }, ) result = TemporalAnalysisService.compare( TemporalSession(), project_id=project_id, payload=payload ) assert result.metric.earlier_value == 4.0 assert result.metric.later_value == 5.5 assert result.metric.absolute_change == 1.5 assert result.object_changes.available is False assert "2018 crosswalk and methodology limitation." in result.warnings assert "2023 edition accuracy limitation." in result.warnings def test_walous_api_routes_use_canonical_envelopes(monkeypatch) -> None: project = Project(id=uuid4(), name="Belgium") dataset_id = uuid4() db = FakeSession(project) monkeypatch.setattr( WalousLandCoverService, "acquire", lambda *_args, **_kwargs: { "output_dataset_id": str(dataset_id), "provider": WalousLandCoverService.PROVIDER, }, ) monkeypatch.setattr( WalousLandCoverService, "analyze", lambda *_args, **_kwargs: { "dataset_id": str(dataset_id), "product_key": "walous_land_cover_2023", "theme": "land_cover_use", "metric_kind": "categorical_area", "selection_bbox": { "min_x": 4.8, "min_y": 50.4, "max_x": 4.9, "max_y": 50.5, "crs": "EPSG:4326", }, "selected_cell_count": 100, "valid_cell_count": 100, "coverage_ratio": 1.0, "resolution_m": 10.0, "observation_year": 2023, "summary": { "metric_label": "Gekarteerde landbedekking", "metric_value": 1.0, "metric_unit": "ha", "aggregation_method": "nearest_resampled_cells_times_cell_area", "primary_metric_key": "land_cover_observed_area_ha", "metrics": [], }, "unsupported_metrics": ["water_volume"], "limitation_message": "Cell-based estimate.", "generated_at": "2026-07-22T00:00:00Z", }, ) app.dependency_overrides[get_db] = lambda: db try: client = TestClient(app) products = client.get(f"/api/v1/projects/{project.id}/datasets/walous/products") acquisition = client.post( f"/api/v1/projects/{project.id}/datasets/walous/acquire", json={ "bbox": { "min_x": 4.8, "min_y": 50.4, "max_x": 4.9, "max_y": 50.5, "crs": "EPSG:4326", }, "product_key": "walous_land_cover_2023", }, ) selection = client.post( f"/api/v1/projects/{project.id}/datasets/{dataset_id}/raster/walous/select", json={ "bbox": { "min_x": 4.8, "min_y": 50.4, "max_x": 4.9, "max_y": 50.5, "crs": "EPSG:4326", } }, ) finally: app.dependency_overrides.clear() assert products.status_code == 200 and set(products.json()) == {"data"} assert products.json()["data"]["total"] == 3 assert acquisition.status_code == 200 and set(acquisition.json()) == {"data"} assert acquisition.json()["data"]["job_type"] == "raster.walous.acquire" assert ( selection.status_code == 200 and selection.json()["data"]["theme"] == "land_cover_use" ) assert any(isinstance(item, Job) for item in db.added)