from __future__ import annotations from dataclasses import dataclass from pathlib import Path import numpy as np from app.core.geo import PathPoint from app.core.raster_sampling import RasterNotConfiguredError, sample_along class LandcoverNotConfiguredError(RasterNotConfiguredError): """Raised when landcover rasters are missing or incomplete.""" ESA_WORLDCOVER_CLASSES: dict[int, str] = { 10: "tree_cover", 20: "shrubland", 30: "grassland", 40: "cropland", 50: "built_up", 60: "bare_sparse_vegetation", 70: "snow_ice", 80: "water", 90: "herbaceous_wetland", 95: "mangroves", 100: "moss_lichen", } @dataclass(frozen=True) class LandcoverPoint: distance_m: float class_name: str forest_type: str | None canopy_height_m: float | None @dataclass(frozen=True) class LandcoverPathSegment: from_m: float to_m: float class_name: str forest_type: str | None canopy_height_m: float | None @dataclass(frozen=True) class LandcoverPath: segments: list[LandcoverPathSegment] vegetation_depth_m: float def worldcover_class(value: float) -> str: if np.isnan(value): return "unknown" return ESA_WORLDCOVER_CLASSES.get(int(round(value)), "unknown") def forest_type_for_class(class_name: str) -> str | None: if class_name in {"tree_cover", "mangroves"}: return "unknown" return None def landcover_path( points: list[PathPoint], landcover_path_dir: str | Path, canopy_path_dir: str | Path | None = None, ) -> LandcoverPath: if len(points) < 2: raise ValueError("points must contain at least two samples") try: landcover_values = sample_along(points, landcover_path_dir, "landcover") except RasterNotConfiguredError as exc: raise LandcoverNotConfiguredError(str(exc)) from exc canopy_values = _canopy_values(points, canopy_path_dir) classified_points = [ LandcoverPoint( distance_m=point.distance_m, class_name=worldcover_class(landcover_values[index]), forest_type=forest_type_for_class(worldcover_class(landcover_values[index])), canopy_height_m=_canopy_height(canopy_values[index]), ) for index, point in enumerate(points) ] return _segments_from_points(classified_points) def _canopy_values(points: list[PathPoint], canopy_path_dir: str | Path | None) -> np.ndarray: if canopy_path_dir is None: return np.full(len(points), np.nan, dtype=float) try: return sample_along(points, canopy_path_dir, "canopy", require_all=False) except RasterNotConfiguredError: return np.full(len(points), np.nan, dtype=float) def _canopy_height(value: float) -> float | None: if np.isnan(value) or value < 0: return None return float(value) def _segments_from_points(points: list[LandcoverPoint]) -> LandcoverPath: segments: list[LandcoverPathSegment] = [] vegetation_depth_m = 0.0 start = points[0] canopy_values: list[float] = [] for index in range(len(points) - 1): current = points[index] next_point = points[index + 1] if current.canopy_height_m is not None: canopy_values.append(current.canopy_height_m) if current.class_name in {"tree_cover", "mangroves"}: vegetation_depth_m += next_point.distance_m - current.distance_m same_segment = ( next_point.class_name == start.class_name and next_point.forest_type == start.forest_type ) if not same_segment: segments.append( LandcoverPathSegment( from_m=start.distance_m, to_m=next_point.distance_m, class_name=start.class_name, forest_type=start.forest_type, canopy_height_m=_mean_canopy(canopy_values), ) ) start = next_point canopy_values = [] last = points[-1] if last.canopy_height_m is not None: canopy_values.append(last.canopy_height_m) segments.append( LandcoverPathSegment( from_m=start.distance_m, to_m=last.distance_m, class_name=start.class_name, forest_type=start.forest_type, canopy_height_m=_mean_canopy(canopy_values), ) ) return LandcoverPath(segments=segments, vegetation_depth_m=vegetation_depth_m) def _mean_canopy(values: list[float]) -> float | None: if not values: return None return float(np.mean(values))