added land

api/app/core/landcover.py

@@ -0,0 +1,158 @@
+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))