from dataclasses import replace
import httpx
import numpy as np
from shapely.geometry import Polygon, box
from openworld_radio_twin.config import Settings
from openworld_radio_twin.environment import EnvironmentData, SurfaceClass
from openworld_radio_twin.geodesy import LocalFrame
from openworld_radio_twin.models import BuildingFeature, SearchResult
from openworld_radio_twin.providers.base import BuildingProviderError, BuildingQueryResult
from openworld_radio_twin.providers.building_sources import (
default_building_sources,
resolve_building_source,
)
from openworld_radio_twin.providers.osm import OsmProvider
from openworld_radio_twin.providers.overture import OvertureBuildingProvider
from openworld_radio_twin.providers.registry import BuildingSourceRegistry
from openworld_radio_twin.providers.surfaces import (
OvertureSurfaceProvider,
apply_water_surface_model,
)
from openworld_radio_twin.providers.terrain import (
TerrainProvider,
TerrainQuery,
flat_terrain,
validate_terrestrial_surface,
)
[docs]
class MapDataProvider:
"""Coordinates search and building providers with one explicit fallback policy."""
def __init__(
self,
settings: Settings,
client: httpx.AsyncClient,
*,
building_sources: BuildingSourceRegistry | None = None,
) -> None:
self.settings = settings
self.client = client
self.building_sources = building_sources or default_building_sources()
self._building_providers = {}
self.osm = OsmProvider(settings, client)
self.overture = OvertureBuildingProvider(settings)
self.terrain = TerrainProvider(client)
self.surfaces = OvertureSurfaceProvider(settings)
[docs]
async def search(self, query: str, limit: int = 5) -> list[SearchResult]:
return await self.osm.search(query, limit)
[docs]
async def buildings(
self,
latitude: float,
longitude: float,
radius_m: int,
limit: int,
source: str = "auto",
) -> BuildingQueryResult:
selection = resolve_building_source(
source, latitude, longitude, self.settings, self.building_sources
)
resolved = selection["resolved"]
if resolved == "none":
result = BuildingQueryResult(buildings=[], provider="disabled")
elif resolved == "global":
result = await self._global_buildings(latitude, longitude, radius_m, limit)
else:
if resolved not in self._building_providers:
descriptor = self.building_sources.get(resolved)
self._building_providers[resolved] = descriptor.factory(self.settings, self.client)
try:
result = await self._building_providers[resolved].buildings(
latitude, longitude, radius_m, limit
)
except httpx.HTTPError as exc:
raise BuildingProviderError(f"Building source '{resolved}' failed: {exc}") from exc
result = _apply_scene_bounds(result, longitude, latitude, radius_m)
return replace(result, source_selection=selection)
async def _global_buildings(
self,
latitude: float,
longitude: float,
radius_m: float,
limit: int,
) -> BuildingQueryResult:
"""Preserve the original worldwide Overture-to-OSM fallback policy."""
try:
primary = await self.overture.buildings(latitude, longitude, radius_m, limit)
if primary.buildings:
return _apply_scene_bounds(primary, longitude, latitude, radius_m)
fallback_reason = "Overture returned no buildings for the query area."
except BuildingProviderError as exc:
fallback_reason = str(exc)
try:
buildings = await self.osm.buildings(
latitude,
longitude,
radius_m,
limit=limit,
)
except httpx.HTTPError as exc:
raise BuildingProviderError(
f"Both building providers failed; Overture: {fallback_reason}; OSM: {exc}"
) from exc
return _apply_scene_bounds(
BuildingQueryResult(
buildings=buildings,
provider="OpenStreetMap Overpass",
warnings=[f"Using OSM fallback because {fallback_reason}"],
),
longitude,
latitude,
radius_m,
)
[docs]
async def environment(
self,
latitude: float,
longitude: float,
radius_m: int,
spacing_m: float,
include_surfaces: bool,
terrain_mode: str = "elevation",
) -> tuple[EnvironmentData, list[str]]:
"""Return the terrain grid and, when requested, the surface classes on it.
``terrain_mode="elevation"`` samples the DEM providers. ``terrain_mode="flat"``
synthesizes a planar grid at local z=0 without querying any elevation
source, so surface classes can still be assigned per cell; water bodies
then lie in the plane and are represented by their material only.
"""
query = TerrainQuery(longitude, latitude, radius_m, spacing_m)
if terrain_mode == "flat":
terrain, warnings = flat_terrain(query), []
elif terrain_mode == "elevation":
terrain, warnings = await self.terrain.terrain(
query,
allow_bathymetry=include_surfaces,
)
else:
raise ValueError(f"Unsupported terrain mode for an environment grid: {terrain_mode}")
if include_surfaces:
surface_result = await self.surfaces.surfaces(
longitude,
latitude,
radius_m,
terrain,
)
features = surface_result.features
cells = surface_result.cells
if terrain_mode == "elevation":
terrain = apply_water_surface_model(terrain, features, cells)
release = surface_result.release
provenance = surface_result.provenance
provenance["terrain_model_operations"] = list(terrain.model_operations)
if terrain_mode == "flat":
provenance["water_surface_model"] = (
"flat ground: water cells lie in the z=0 plane and differ by material only"
)
warnings.extend(surface_result.warnings)
else:
features = ()
cells = np.full(terrain.cell_shape, SurfaceClass.GROUND, dtype=np.uint8)
release = None
provenance = {"provider": "disabled"}
validate_terrestrial_surface(terrain)
return (
EnvironmentData(
terrain=terrain,
surface_features=features,
surface_cells=cells,
overture_release=release,
surface_provenance=provenance,
),
warnings,
)
def _apply_scene_bounds(
result: BuildingQueryResult,
origin_longitude: float,
origin_latitude: float,
radius_m: float,
) -> BuildingQueryResult:
"""Keep complete footprints within the square ENU scene domain."""
frame = LocalFrame.at(origin_longitude, origin_latitude)
domain = box(-radius_m, -radius_m, radius_m, radius_m)
selected: list[BuildingFeature] = []
for building in result.buildings:
exterior = [
frame.to_local(longitude, latitude)[:2] for longitude, latitude in building.coordinates
]
holes = [
[frame.to_local(longitude, latitude)[:2] for longitude, latitude in ring]
for ring in building.holes
]
if domain.covers(Polygon(exterior, holes)):
selected.append(building)
excluded = len(result.buildings) - len(selected)
warnings = list(result.warnings)
if excluded:
warnings.append(
f"Excluded {excluded} building footprints crossing the square scene boundary."
)
return replace(
result,
buildings=selected,
warnings=warnings,
boundary_excluded_count=result.boundary_excluded_count + excluded,
)
