from __future__ import annotations
from dataclasses import dataclass, field
from enum import IntEnum
import numpy as np
from shapely.geometry.base import BaseGeometry
[docs]
class SurfaceClass(IntEnum):
"""Stable numeric classes shared by meshes, rasters, exports, and the viewer."""
GROUND = 0
VEGETATION = 1
PAVED = 2
WATER = 3
[docs]
@dataclass(frozen=True)
class TerrainSource:
provider: str
dataset: str
vertical_datum: str
requested_spacing_m: float
source_resolution_m: float | None
interpolation: str
source_url: str
license: str
attributes: dict[str, str | int | float | bool | None] = field(default_factory=dict)
[docs]
@dataclass(frozen=True)
class TerrainGrid:
"""A regular local ENU grid with elevations relative to the scene origin.
Rows run south-to-north and columns west-to-east. ``source_elevation_m``
preserves provider elevations sampled onto this grid. ``model_elevation_m`` is the physical RF
surface after recorded transformations, and ``local_elevation_m`` removes
its elevation sampled at ENU (0, 0).
"""
east_m: np.ndarray
north_m: np.ndarray
source_elevation_m: np.ndarray
model_elevation_m: np.ndarray
local_elevation_m: np.ndarray
source_origin_elevation_m: float
origin_elevation_m: float
source: TerrainSource
model_operations: tuple[dict[str, object], ...] = ()
def __post_init__(self) -> None:
expected = (self.north_m.size, self.east_m.size)
if self.source_elevation_m.shape != expected:
raise ValueError(f"source elevation shape must be {expected}")
if self.model_elevation_m.shape != expected:
raise ValueError(f"model elevation shape must be {expected}")
if self.local_elevation_m.shape != expected:
raise ValueError(f"local elevation shape must be {expected}")
if self.east_m.size < 2 or self.north_m.size < 2:
raise ValueError("terrain grids require at least two samples per axis")
if not np.all(np.diff(self.east_m) > 0) or not np.all(np.diff(self.north_m) > 0):
raise ValueError("terrain axes must be strictly increasing")
if not all(
np.all(np.isfinite(value))
for value in (
self.source_elevation_m,
self.model_elevation_m,
self.local_elevation_m,
self.source_origin_elevation_m,
self.origin_elevation_m,
)
):
raise ValueError("terrain contains missing or non-finite elevations")
if not np.allclose(
self.local_elevation_m,
(self.model_elevation_m.astype(np.float64) - self.origin_elevation_m).astype(
self.local_elevation_m.dtype
),
rtol=0.0,
atol=1e-4,
):
# Older persisted scenes subtracted the origin in float32. Accept
# that exact representation without changing local heights or PLYs.
legacy_matches = (
self.model_elevation_m.dtype == np.float32
and self.local_elevation_m.dtype == np.float32
and np.array_equal(
self.local_elevation_m,
self.model_elevation_m - np.float32(self.origin_elevation_m),
)
)
if not legacy_matches:
raise ValueError(
"local elevation must equal model elevation minus the model origin"
)
@property
def shape(self) -> tuple[int, int]:
return self.source_elevation_m.shape
@property
def cell_shape(self) -> tuple[int, int]:
return self.shape[0] - 1, self.shape[1] - 1
@property
def spacing_m(self) -> float:
return float(np.mean(np.diff(self.east_m)))
[docs]
def elevation_at(self, east_m: float, north_m: float) -> float:
"""Bilinearly sample local elevation without extrapolating the scene."""
return self._sample(self.local_elevation_m, east_m, north_m)
[docs]
def with_model_elevation(
self,
values: np.ndarray,
operation: dict[str, object],
) -> TerrainGrid:
"""Return a terrain with an auditable physical-surface transformation."""
model = np.asarray(values, dtype=np.float32)
origin = self._sample(model, 0.0, 0.0)
return TerrainGrid(
east_m=self.east_m,
north_m=self.north_m,
source_elevation_m=self.source_elevation_m,
model_elevation_m=model,
local_elevation_m=(model.astype(np.float64) - origin).astype(np.float32),
source_origin_elevation_m=self.source_origin_elevation_m,
origin_elevation_m=origin,
source=self.source,
model_operations=(*self.model_operations, operation),
)
def _sample(self, values: np.ndarray, east_m: float, north_m: float) -> float:
east = float(np.clip(east_m, self.east_m[0], self.east_m[-1]))
north = float(np.clip(north_m, self.north_m[0], self.north_m[-1]))
column = int(np.clip(np.searchsorted(self.east_m, east) - 1, 0, self.east_m.size - 2))
row = int(np.clip(np.searchsorted(self.north_m, north) - 1, 0, self.north_m.size - 2))
x0, x1 = self.east_m[column : column + 2]
y0, y1 = self.north_m[row : row + 2]
tx = (east - x0) / (x1 - x0)
ty = (north - y0) / (y1 - y0)
window = values[row : row + 2, column : column + 2]
south = window[0, 0] * (1 - tx) + window[0, 1] * tx
north_value = window[1, 0] * (1 - tx) + window[1, 1] * tx
return float(south * (1 - ty) + north_value * ty)
[docs]
@dataclass(frozen=True)
class FeatureSource:
property: str | None
dataset: str | None
record_id: str | None
update_time: str | None
license: str | None
[docs]
@dataclass(frozen=True)
class SurfaceFeature:
feature_id: str
surface_class: SurfaceClass
geometry: BaseGeometry
source_theme: str
source_subtype: str | None
source_class: str | None
source_surface: str | None
sources: tuple[FeatureSource, ...]
classification_basis: str
precedence: int
width_m: float | None = None
width_source: str | None = None
[docs]
@dataclass(frozen=True)
class MaterialProfile:
surface_class: SurfaceClass
sionna_material: str
basis: str
confidence: str
note: str
RF_MATERIAL_PROFILE_VERSION = 1
RF_MATERIAL_PROFILES = {
SurfaceClass.GROUND: MaterialProfile(
SurfaceClass.GROUND,
"itu_medium_dry_ground",
"ITU-R P.2040 material model",
"surrogate",
"Generic exposed ground; not an in-situ material measurement.",
),
SurfaceClass.VEGETATION: MaterialProfile(
SurfaceClass.VEGETATION,
"itu_wood",
"ITU-R P.2040 material model",
"surrogate",
(
"Vegetated surfaces use wood as a reproducible RF proxy; "
"foliage volume loss is not modeled."
),
),
SurfaceClass.PAVED: MaterialProfile(
SurfaceClass.PAVED,
"itu_concrete",
"ITU-R P.2040 material model",
"surrogate",
"Paved surfaces use concrete as a reproducible RF proxy, not a measured composition.",
),
SurfaceClass.WATER: MaterialProfile(
SurfaceClass.WATER,
"itu_wet_ground",
"ITU-R P.2040 material model",
"surrogate",
"Open water uses wet ground as a documented proxy pending a calibrated water model.",
),
}
[docs]
@dataclass(frozen=True)
class EnvironmentData:
terrain: TerrainGrid
surface_features: tuple[SurfaceFeature, ...]
surface_cells: np.ndarray
overture_release: str | None
surface_provenance: dict[str, object]
def __post_init__(self) -> None:
if self.surface_cells.shape != self.terrain.cell_shape:
raise ValueError("surface classes must align with terrain cells")
valid = {int(value) for value in SurfaceClass}
if not set(np.unique(self.surface_cells)).issubset(valid):
raise ValueError("surface grid contains an unknown class")
[docs]
def surface_class_at(self, east_m: float, north_m: float) -> SurfaceClass:
"""Return the semantic class of the cell containing one local point."""
column = int(
np.clip(
np.searchsorted(self.terrain.east_m, east_m, side="right") - 1,
0,
self.surface_cells.shape[1] - 1,
)
)
row = int(
np.clip(
np.searchsorted(self.terrain.north_m, north_m, side="right") - 1,
0,
self.surface_cells.shape[0] - 1,
)
)
return SurfaceClass(int(self.surface_cells[row, column]))
[docs]
def flat_environment(radius_m: float, spacing_m: float = 5.0) -> EnvironmentData:
"""Create an explicit flat test fixture; production providers never call this."""
sample_count = max(2, int(np.ceil(2 * radius_m / spacing_m)) + 1)
axis = np.linspace(-radius_m, radius_m, sample_count, dtype=np.float64)
values = np.zeros((sample_count, sample_count), dtype=np.float32)
source = TerrainSource(
provider="synthetic",
dataset="explicit flat test terrain",
vertical_datum="local ENU origin",
requested_spacing_m=float(2 * radius_m / (sample_count - 1)),
source_resolution_m=None,
interpolation="none",
source_url="",
license="not applicable",
)
terrain = TerrainGrid(
axis,
axis.copy(),
values,
values.copy(),
values.copy(),
0.0,
0.0,
source,
)
classes = np.full(terrain.cell_shape, SurfaceClass.GROUND, dtype=np.uint8)
return EnvironmentData(terrain, (), classes, None, {"provider": "synthetic"})
