import asyncio
import math
import time
from dataclasses import dataclass
import httpx
from openworld_radio_twin.config import Settings
from openworld_radio_twin.geodesy import LocalFrame
from openworld_radio_twin.models import BuildingFeature, SearchResult
[docs]
@dataclass
class CacheEntry:
expires_at: float
value: object
[docs]
class OsmProvider:
"""Fetches only the active area from public OSM services and caches it in memory."""
def __init__(self, settings: Settings, client: httpx.AsyncClient) -> None:
self.settings = settings
self.client = client
self._cache: dict[str, CacheEntry] = {}
self._search_lock = asyncio.Lock()
self._last_search_at = 0.0
def _cached(self, key: str) -> object | None:
entry = self._cache.get(key)
if entry and entry.expires_at > time.monotonic():
return entry.value
self._cache.pop(key, None)
return None
def _store(self, key: str, value: object) -> object:
self._cache[key] = CacheEntry(
expires_at=time.monotonic() + self.settings.cache_ttl_seconds,
value=value,
)
return value
[docs]
async def search(self, query: str, limit: int = 5) -> list[SearchResult]:
key = f"search:{query.casefold()}:{limit}"
cached = self._cached(key)
if cached is not None:
return cached # type: ignore[return-value]
async with self._search_lock:
wait_for = 1.0 - (time.monotonic() - self._last_search_at)
if wait_for > 0:
await asyncio.sleep(wait_for)
response = await self.client.get(
self.settings.nominatim_url,
params={"q": query, "format": "jsonv2", "limit": limit},
headers={"User-Agent": self.settings.user_agent},
timeout=12,
)
self._last_search_at = time.monotonic()
response.raise_for_status()
results = [
SearchResult(
display_name=item["display_name"],
latitude=float(item["lat"]),
longitude=float(item["lon"]),
kind=item.get("type"),
)
for item in response.json()
]
return self._store(key, results) # type: ignore[return-value]
[docs]
async def buildings(
self,
latitude: float,
longitude: float,
radius_m: int,
limit: int | None = None,
) -> list[BuildingFeature]:
result_limit = limit or self.settings.max_buildings
rounded = (round(latitude, 7), round(longitude, 7), radius_m, result_limit)
key = f"buildings:{rounded}"
cached = self._cached(key)
if cached is not None:
return cached # type: ignore[return-value]
# The scene is the ENU square of half-width radius_m; query its geographic
# bounding box so the corners are covered, as the other providers do.
west, south, east, north = self._bbox(longitude, latitude, radius_m)
query = f"""
[out:json][timeout:20];
(
way["building"]({south},{west},{north},{east});
relation["building"]({south},{west},{north},{east});
);
out tags geom {result_limit};
"""
response = await self.client.post(
self.settings.overpass_url,
data={"data": query},
headers={"User-Agent": self.settings.user_agent},
timeout=30,
)
response.raise_for_status()
features: list[BuildingFeature] = []
for element in response.json().get("elements", []):
tags = element.get("tags", {})
height_m, height_source, height_uncertainty_m = self._building_height_details(tags)
geometries = self._element_geometries(element)
for osm_id, geometry in geometries:
if len(geometry) < 3:
continue
coordinates = [[float(p["lon"]), float(p["lat"])] for p in geometry]
if coordinates[0] != coordinates[-1]:
coordinates.append(coordinates[0])
features.append(
BuildingFeature(
feature_id=f"osm:{osm_id}",
source="OpenStreetMap",
source_record_id=str(osm_id),
name=tags.get("name"),
height_m=height_m,
height_source=height_source,
height_uncertainty_m=height_uncertainty_m,
coordinates=coordinates,
)
)
if len(features) >= result_limit:
break
if len(features) >= result_limit:
break
return self._store(key, features) # type: ignore[return-value]
@staticmethod
def _bbox(
longitude: float,
latitude: float,
radius_m: float,
) -> tuple[float, float, float, float]:
frame = LocalFrame.at(longitude, latitude)
corners = frame.corners(-radius_m, -radius_m, radius_m, radius_m)
longitudes = [point[0] for point in corners]
latitudes = [point[1] for point in corners]
return min(longitudes), min(latitudes), max(longitudes), max(latitudes)
@classmethod
def _element_geometries(
cls, element: dict[str, object]
) -> list[tuple[int | str, list[dict[str, float]]]]:
if element.get("type") != "relation":
return [(element.get("id", "unknown"), element.get("geometry") or [])] # type: ignore[list-item]
relation_id = element.get("id", "unknown")
segments = [
member.get("geometry") or []
for member in element.get("members", []) # type: ignore[union-attr]
if member.get("type") == "way" and member.get("role", "") in {"", "outer"}
]
rings = cls._merge_outer_rings(segments)
return [(f"{relation_id}:{index}", ring) for index, ring in enumerate(rings)]
@staticmethod
def _merge_outer_rings(
segments: list[list[dict[str, float]]],
) -> list[list[dict[str, float]]]:
remaining = [segment[:] for segment in segments if len(segment) >= 2]
rings: list[list[dict[str, float]]] = []
def endpoint(point: dict[str, float]) -> tuple[float, float]:
return float(point["lon"]), float(point["lat"])
while remaining:
ring = remaining.pop(0)
while endpoint(ring[0]) != endpoint(ring[-1]):
tail = endpoint(ring[-1])
match_index = next(
(
index
for index, segment in enumerate(remaining)
if endpoint(segment[0]) == tail or endpoint(segment[-1]) == tail
),
None,
)
if match_index is None:
break
segment = remaining.pop(match_index)
if endpoint(segment[-1]) == tail:
segment.reverse()
ring.extend(segment[1:])
if len(ring) >= 4 and endpoint(ring[0]) == endpoint(ring[-1]):
rings.append(ring)
return rings
@staticmethod
def _building_height(tags: dict[str, str]) -> float:
return OsmProvider._building_height_details(tags)[0]
@staticmethod
def _building_height_details(tags: dict[str, str]) -> tuple[float, str, float | None]:
raw_height = tags.get("height", "").lower().strip()
try:
is_feet = raw_height.endswith(("ft", "feet", "foot", "'"))
suffixes = ("meters", "metres", "meter", "metre", "feet", "foot", "ft", "m", "'")
for suffix in suffixes:
if raw_height.endswith(suffix):
raw_height = raw_height[: -len(suffix)].strip()
break
value = float(raw_height) * (0.3048 if is_feet else 1.0)
if math.isfinite(value) and 2 <= value <= 500:
return value, "osm_height", None
except ValueError:
pass
try:
levels = float(tags.get("building:levels", ""))
if math.isfinite(levels) and 0 < levels <= 150:
return max(3.0, levels * 3.2), "estimated_from_osm_levels", 1.6
except ValueError:
pass
return 10.0, "default_assumption", 5.0
