"""Built-in building sources and deterministic, explicitly versioned routing."""
from __future__ import annotations
from openworld_radio_twin.config import Settings
from openworld_radio_twin.providers import berlin, boston
from openworld_radio_twin.providers.base import BuildingQueryResult
from openworld_radio_twin.providers.berlin import BerlinProvider
from openworld_radio_twin.providers.osm import OsmProvider
from openworld_radio_twin.providers.overture import OvertureBuildingProvider
from openworld_radio_twin.providers.registry import (
BuildingSource,
BuildingSourceRegistry,
in_berlin,
in_boston,
in_netherlands,
)
from openworld_radio_twin.providers.three_d_bag import ATTRIBUTION, LICENSE_URL, ThreeDBagProvider
ROUTING_POLICY_VERSION = "regional-nl-berlin-boston-v2"
GLOBAL_IMPLEMENTATION_VERSION = "overture-osm-v1"
[docs]
class OsmBuildingSource:
def __init__(self, settings, client):
self.provider = OsmProvider(settings, client)
[docs]
async def buildings(self, latitude, longitude, radius_m, limit):
return BuildingQueryResult(
buildings=await self.provider.buildings(latitude, longitude, radius_m, limit=limit),
provider="OpenStreetMap Overpass",
)
[docs]
def default_building_sources() -> BuildingSourceRegistry:
registry = BuildingSourceRegistry(
(
BuildingSource(
"boston",
"Boston BPDA OBJ",
"1",
lambda settings, client: boston.BostonProvider(settings, client),
auto_covers=in_boston,
auto_priority=100,
attribution=boston.ATTRIBUTION,
license_url=boston.LICENSE_URL,
configuration_identity=boston.configuration_identity,
),
BuildingSource(
"berlin",
"Berlin LoD2",
"1",
lambda settings, client: BerlinProvider(client),
auto_covers=in_berlin,
auto_priority=100,
attribution=berlin.ATTRIBUTION,
license_url=berlin.LICENSE_URL,
),
BuildingSource(
"3dbag",
"3DBAG LoD2.2",
"2",
lambda settings, client: ThreeDBagProvider(client),
covers=in_netherlands,
auto_priority=100,
attribution=ATTRIBUTION,
license_url=LICENSE_URL,
),
BuildingSource(
"overture",
"Overture Maps",
"1",
lambda settings, client: OvertureBuildingProvider(settings),
),
BuildingSource("osm", "OpenStreetMap", "1", OsmBuildingSource),
)
)
registry.load_installed()
return registry
[docs]
def resolve_building_source(
requested: str,
latitude: float,
longitude: float,
settings: Settings,
registry: BuildingSourceRegistry,
) -> dict[str, object]:
"""Resolve before cache lookup; explicit selections never fall back silently."""
resolved = requested
if requested == "auto":
source = registry.automatic(
latitude, longitude, allowed=settings.automatic_building_sources
)
resolved = source.identifier if source is not None else "global"
if resolved in {"global", "none"}:
version = GLOBAL_IMPLEMENTATION_VERSION if resolved == "global" else "1"
else:
source = registry.get(resolved)
if not source.covers(latitude, longitude):
raise ValueError(f"Building source '{resolved}' does not cover the query centre")
version = source.version
identity = {
"requested": requested,
"resolved": resolved,
"implementation_version": version,
"routing_policy": ROUTING_POLICY_VERSION,
}
if resolved not in {"global", "none"} and source.configuration_identity is not None:
identity["configuration"] = source.configuration_identity(settings)
return identity
