159 lines
4.5 KiB
Python
159 lines
4.5 KiB
Python
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))
|