import math
from enum import Enum
from typing import Literal
import numpy as np
from pydantic import BaseModel, ConfigDict, Field, computed_field, model_validator
from openworld_radio_twin.geodesy import LocalFrame
[docs]
class EngineName(str, Enum):
preview = "preview"
sionna = "sionna"
MAX_PREVIEW_GRID_CELLS = 20_000
MAX_SIONNA_GRID_CELLS = 2_250_000
MAX_TERRAIN_GRID_CELLS = 250_000
[docs]
class Coordinate(BaseModel):
latitude: float = Field(ge=-90, le=90)
longitude: float = Field(ge=-180, le=180)
altitude_m: float = Field(
default=25.0,
ge=0,
le=1000,
description="Transmitter height above the local ground plane in meters (AGL)",
)
[docs]
class TransmitterConfig(BaseModel):
position: Coordinate
frequency_ghz: float = Field(default=3.5, ge=0.1, le=100)
power_dbm: float = Field(default=30.0, ge=-20, le=80)
azimuth_deg: float = Field(default=0.0, ge=0, lt=360)
downtilt_deg: float = Field(default=6.0, ge=-30, le=90)
antenna_pattern: str = Field(default="sector", pattern="^(sector|isotropic)$")
SIONNA_PROPAGATION_PARAMETERS = (
"los",
"specular_reflection",
"diffuse_reflection",
"refraction",
"diffraction",
"edge_diffraction",
"diffraction_lit_region",
)
[docs]
class SionnaEngineConfig(BaseModel):
"""Runtime and geometry controls for the built-in Sionna RT adapter."""
model_config = ConfigDict(extra="forbid")
device: Literal["auto", "cuda", "cpu"] = "auto"
include_ground: bool = True
include_vegetation: bool = True
include_paved: bool = True
include_water: bool = True
include_building_roofs: bool = True
include_building_walls: bool = True
# Names and defaults follow Sionna RT 2.0.1 RadioMapSolver.__call__.
los: bool = True
specular_reflection: bool = True
diffuse_reflection: bool = False
refraction: bool = True
diffraction: bool = False
edge_diffraction: bool = False
diffraction_lit_region: bool = True
[docs]
class SceneRequest(BaseModel):
"""Geometry inputs of a compiled scene, keyed like ``openworld_radio_twin.rt.load_scene``."""
latitude: float = Field(ge=-90, le=90)
longitude: float = Field(ge=-180, le=180)
radius_m: float = Field(default=750, ge=100, le=5000)
include_buildings: bool = True
building_source: str = Field(default="auto", pattern=r"^(?:[a-z][a-z0-9_-]*|3dbag)$")
include_terrain: bool = False
material_profile: Literal["itu", "uniform"] = "itu"
terrain_resolution_m: float = Field(
default=5.0,
ge=1.0,
le=100.0,
description="ENU terrain mesh spacing; distinct from source DEM resolution",
)
[docs]
@model_validator(mode="after")
def limit_terrain_grid(self) -> "SceneRequest":
terrain_side = math.ceil(2 * self.radius_m / self.terrain_resolution_m)
if self.include_terrain and terrain_side**2 > MAX_TERRAIN_GRID_CELLS:
raise ValueError(
f"Requested {terrain_side}x{terrain_side} terrain grid exceeds the "
f"{MAX_TERRAIN_GRID_CELLS:,}-cell viewer and scene limit; "
"increase terrain resolution"
)
return self
@property
def buildings_mode(self) -> Literal["auto", "none"]:
return "auto" if self.include_buildings else "none"
@property
def terrain_mode(self) -> Literal["elevation", "flat"]:
return "elevation" if self.include_terrain else "flat"
[docs]
class SceneCenter(BaseModel):
"""Explicit scene origin; the compiled square is centred here rather than on the first TX."""
latitude: float = Field(ge=-90, le=90)
longitude: float = Field(ge=-180, le=180)
[docs]
class SimulationRequest(BaseModel):
transmitters: list[TransmitterConfig] = Field(min_length=1, max_length=8)
scene_center: SceneCenter | None = Field(
default=None,
description=(
"Centre of the compiled scene square. Defaults to the first transmitter, "
"which is what dataset generation always uses."
),
)
engine: str = Field(default=EngineName.preview.value, pattern=r"^[a-z][a-z0-9-]*$")
engine_config: dict[str, str | int | float | bool] = Field(default_factory=dict)
radius_m: float = Field(default=750, ge=100, le=5000)
resolution_m: float = Field(default=30, ge=0.1, le=200)
receiver_height_m: float = Field(
default=1.5,
ge=0.1,
le=100,
description="Receiver height above the local ground plane in meters (AGL)",
)
max_depth: int = Field(default=3, ge=0, le=8)
samples_per_tx: int = Field(default=1_000_000, ge=10_000, le=100_000_000)
seed: int = Field(default=42, ge=0, le=4_294_967_295)
association_metric: str = Field(default="sinr", pattern="^(path_gain|rss|sinr)$")
include_buildings: bool = True
building_source: str = Field(default="auto", pattern=r"^(?:[a-z][a-z0-9_-]*|3dbag)$")
include_terrain: bool = False
material_profile: Literal["itu", "uniform"] = "itu"
terrain_resolution_m: float = Field(
default=5.0,
ge=1.0,
le=100.0,
description="ENU terrain mesh spacing; distinct from source DEM resolution",
)
[docs]
@model_validator(mode="after")
def limit_grid_size(self) -> "SimulationRequest":
terrain_side = math.ceil(2 * self.radius_m / self.terrain_resolution_m)
if self.include_terrain and terrain_side**2 > MAX_TERRAIN_GRID_CELLS:
raise ValueError(
f"Requested {terrain_side}x{terrain_side} terrain grid exceeds the "
f"{MAX_TERRAIN_GRID_CELLS:,}-cell viewer and scene limit; "
"increase terrain resolution"
)
side = math.ceil(2 * self.radius_m / self.resolution_m)
maximum = (
MAX_PREVIEW_GRID_CELLS
if self.engine == EngineName.preview.value
else MAX_SIONNA_GRID_CELLS
)
if side**2 > maximum:
raise ValueError(
f"Requested {side}x{side} coverage grid exceeds the "
f"{maximum:,}-cell {self.engine} limit; increase resolution"
)
origin = self.scene_origin
frame = LocalFrame.at(origin.longitude, origin.latitude)
for index, transmitter in enumerate(self.transmitters, start=1):
east, north, _ = frame.to_local(
transmitter.position.longitude, transmitter.position.latitude
)
if max(abs(east), abs(north)) > self.radius_m + 1e-6:
raise ValueError(
f"Transmitter {index} lies outside the {2 * self.radius_m:g} m scene square"
)
if self.engine == EngineName.sionna.value:
self.engine_config = SionnaEngineConfig.model_validate(self.engine_config).model_dump()
frequencies = {transmitter.frequency_ghz for transmitter in self.transmitters}
patterns = {transmitter.antenna_pattern for transmitter in self.transmitters}
if any(not 1 <= frequency <= 10 for frequency in frequencies):
raise ValueError("The default Sionna ITU ground model is valid from 1 to 10 GHz")
if len(frequencies) != 1:
raise ValueError("Sionna transmitters must use one shared scene frequency")
if len(patterns) != 1:
raise ValueError("Sionna transmitters must use one shared antenna pattern")
return self
@property
def include_surface_materials(self) -> bool:
"""Return whether semantic surface classes are required by this request.
Classes are assigned on the ground grid whether that grid follows the DEM
or is a flat plane; only the uniform profile omits them.
"""
return self.material_profile == "itu"
@property
def include_environment_grid(self) -> bool:
"""Return whether a gridded ground (DEM or flat) must be acquired for this request.
The planar preview engine ignores materials, so a flat scene only needs the
environment grid when Sionna evaluates it.
"""
if self.include_terrain:
return True
return self.include_surface_materials and self.engine == EngineName.sionna.value
@property
def reference_transmitter(self) -> TransmitterConfig:
"""First TX; it sets the shared antenna preset and, without ``scene_center``, the origin."""
return self.transmitters[0]
@property
def scene_origin(self) -> SceneCenter:
"""Local ENU origin and coverage-grid centre of the compiled scene."""
if self.scene_center is not None:
return self.scene_center
position = self.reference_transmitter.position
return SceneCenter(latitude=position.latitude, longitude=position.longitude)
@property
def scene_request(self) -> SceneRequest:
"""Geometry of the compiled scene this simulation runs in."""
origin = self.scene_origin
return SceneRequest(
latitude=origin.latitude,
longitude=origin.longitude,
radius_m=self.radius_m,
include_buildings=self.include_buildings,
building_source=self.building_source,
include_terrain=self.include_terrain,
material_profile=self.material_profile,
terrain_resolution_m=self.terrain_resolution_m,
)
[docs]
class BuildingGeometry(BaseModel):
"""Source-resolved triangle shell, before scene placement.
Horizontal coordinates use ``crs``; vertex Z and ``ground_elevation_m``
use the explicitly named vertical datum, never implicit WGS84 heights.
"""
crs: str
vertical_datum: str
lod: str
ground_elevation_m: float = Field(allow_inf_nan=False)
vertices: list[tuple[float, float, float]] = Field(min_length=3)
triangles: list[tuple[int, int, int]] = Field(min_length=1)
surface_types: list[str]
[docs]
@model_validator(mode="after")
def validate_mesh(self) -> "BuildingGeometry":
vertices = np.asarray(self.vertices, dtype=np.float64).reshape(-1, 3)
if not np.isfinite(vertices).all():
raise ValueError("Building geometry coordinates must be finite")
if len(self.surface_types) != len(self.triangles):
raise ValueError("Each building triangle needs a surface type")
triangles = np.asarray(self.triangles, dtype=np.int64).reshape(-1, 3)
if (
(triangles < 0).any()
or (triangles >= len(self.vertices)).any()
or (triangles[:, 0] == triangles[:, 1]).any()
or (triangles[:, 1] == triangles[:, 2]).any()
or (triangles[:, 0] == triangles[:, 2]).any()
):
raise ValueError("Building geometry has an invalid triangle index")
return self
[docs]
class BuildingFeature(BaseModel):
feature_id: str
source: str
feature_type: str = Field(default="building", pattern="^(building|building_part)$")
source_dataset_id: str | None = None
parent_building_id: str | None = None
source_record_id: str | None = None
source_release: str | None = None
name: str | None = None
height_m: float = Field(
gt=0,
le=1000,
description="Vertical extent from the feature's lowest to highest point in meters",
)
min_height_m: float = Field(
default=0,
ge=0,
le=1000,
description="Height AGL at which the feature begins",
)
height_source: str = "unspecified"
height_uncertainty_m: float | None = None
coordinates: list[list[float]]
holes: list[list[list[float]]] = Field(default_factory=list)
geometry: BuildingGeometry | None = None
source_attributes: dict[str, object] = Field(default_factory=dict)
@computed_field
@property
def roof_height_agl_m(self) -> float:
return self.min_height_m + self.height_m
[docs]
class GridMapping(BaseModel):
horizontal_datum: str = "WGS84"
local_axes: str = "east,north,up"
origin_longitude: float
origin_latitude: float
first_sample_east_m: float
first_sample_north_m: float
column_step_m: float
row_step_m: float
sample_layout: str
corner_coordinates: list[list[float]]
[docs]
class RadioLayer(BaseModel):
label: str
unit: str
values: list[float | None] | None = None
minimum: float
maximum: float
display_minimum: float
display_maximum: float
reduction: str = "best_transmitter_per_cell"
[docs]
class CoverageGrid(BaseModel):
crs: str = "EPSG:4326"
no_data_value: None = None
receiver_height_agl_m: float
grid_mapping: GridMapping
west: float
south: float
east: float
north: float
width: int
height: int
layers: dict[str, RadioLayer]
association: list[int] | None = None
building_mask: list[bool] | None = None
building_heights_m: list[float | None] | None = None
building_cell_count: int = 0
[docs]
@model_validator(mode="after")
def validate_aligned_grids(self) -> "CoverageGrid":
arrays = [self.association, self.building_mask, self.building_heights_m]
arrays.extend(layer.values for layer in self.layers.values())
if all(value is None for value in arrays):
return self
if any(value is None for value in arrays):
raise ValueError("Coverage cell arrays must be either all present or all omitted")
assert self.association is not None
assert self.building_mask is not None
assert self.building_heights_m is not None
expected = self.width * self.height
aligned = {
"association": len(self.association),
"building_mask": len(self.building_mask),
"building_heights_m": len(self.building_heights_m),
}
for name, layer in self.layers.items():
assert layer.values is not None
aligned[f"layers.{name}"] = len(layer.values)
for name, actual in aligned.items():
if actual != expected:
raise ValueError(f"{name} has {actual} cells; expected {expected}")
if any(
mask != (height is not None)
for mask, height in zip(self.building_mask, self.building_heights_m, strict=True)
):
raise ValueError("building_mask and building_heights_m disagree")
return self
[docs]
class NativeBuildingMesh(BaseModel):
feature_id: str
vertices: list[tuple[float, float, float]]
triangles: list[tuple[int, int, int]]
surface_types: list[str]
lod: str
placement_rule: str
[docs]
class BuildingGeometryContext(BaseModel):
"""Solver-coordinate native shells, with no visual offset or vertical scaling."""
origin_longitude: float
origin_latitude: float
coordinate_system: str = "WGS84 topocentric ENU"
meshes: list[NativeBuildingMesh]
[docs]
class SimulationResponse(BaseModel):
simulation_id: str
engine: str
elapsed_ms: int
coverage: CoverageGrid
buildings: list[BuildingFeature]
building_geometry: BuildingGeometryContext | None = None
warnings: list[str] = Field(default_factory=list)
engine_details: dict[str, str | int | float | bool] = Field(default_factory=dict)
data_provenance: dict[str, str | int | float | bool] = Field(default_factory=dict)
environment: "EnvironmentContext | None" = None
scene: "SceneResponse | None" = None
[docs]
class SceneBounds(BaseModel):
origin_longitude: float
origin_latitude: float
radius_m: float
wgs84_corners: list[list[float]]
[docs]
class SceneResponse(BaseModel):
"""A compiled, cached scene as the browser displays it.
``scene_id`` is the cache directory name shared with ``openworld_radio_twin.rt.load_scene``
for the same geometry inputs; the buildings and environment are read from that directory.
"""
scene_id: str
cached: bool
specification: SceneRequest
bounds: SceneBounds
buildings: list[BuildingFeature]
building_geometry: BuildingGeometryContext | None = None
environment: "EnvironmentContext | None" = None
warnings: list[str] = Field(default_factory=list)
data_provenance: dict[str, str | int | float | bool] = Field(default_factory=dict)
[docs]
class TerrainContext(BaseModel):
width: int
height: int
east_m: list[float]
north_m: list[float]
local_elevation_m: list[float]
origin_elevation_m: float
minimum_local_elevation_m: float
maximum_local_elevation_m: float
model_operations: list[dict[str, object]]
source: dict[str, object]
[docs]
class SurfaceContext(BaseModel):
width: int
height: int
classes: list[int]
legend: dict[int, str]
material_profile_version: int
material_profiles: dict[str, dict[str, str]]
provenance: dict[str, object]
[docs]
class EnvironmentContext(BaseModel):
terrain: TerrainContext
surfaces: SurfaceContext
[docs]
class BuildingContextResponse(BaseModel):
buildings: list[BuildingFeature]
building_geometry: BuildingGeometryContext | None = None
warnings: list[str] = Field(default_factory=list)
data_provenance: dict[str, str | int | float | bool] = Field(default_factory=dict)
[docs]
class SearchResult(BaseModel):
display_name: str
latitude: float
longitude: float
kind: str | None = None
