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) -> 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 = [1, 2, 3, 4, 5, 6, 7, 8, 9, 80, 90] values = np.empty((100, len(class_codes) * 20), dtype="uint8") 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="uint8", crs="EPSG:3812", transform=transform, nodata=255, ) 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_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_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": "Cell-based estimate.", } 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 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"] == 2 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)