"""Trusted, installed building providers shared by library, batch and HTTP APIs."""
from __future__ import annotations
import json
import re
from collections.abc import Callable
from dataclasses import dataclass
from functools import lru_cache
from importlib.metadata import entry_points
from importlib.resources import files
from typing import Protocol
import httpx
from shapely.geometry import Point, shape
from shapely.ops import unary_union
from shapely.prepared import prep
from openworld_radio_twin.config import DEFAULT_AUTOMATIC_BUILDING_SOURCES, Settings
from openworld_radio_twin.providers.base import BuildingQueryResult
ENTRY_POINT_GROUP = "openworld_radio_twin.building_sources"
[docs]
class BuildingProvider(Protocol):
self,
latitude: float,
longitude: float,
radius_m: float,
limit: int,
) -> BuildingQueryResult: ...
[docs]
def worldwide(latitude: float, longitude: float) -> bool:
return -90 <= latitude <= 90 and -180 <= longitude <= 180
@lru_cache(maxsize=1)
def _netherlands_boundary():
document = json.loads(
files("openworld_radio_twin.providers").joinpath("data/netherlands.geojson").read_text()
)
features = document["features"]
if len(features) != 1 or features[0]["properties"]["identificatie"] != "LND6030":
raise ValueError("Packaged Netherlands boundary is not the expected PDOK landgebied")
return shape(features[0]["geometry"])
[docs]
def in_netherlands(latitude: float, longitude: float) -> bool:
"""Classify the query centre using the official, unsimplified country polygon."""
if not (50 <= latitude <= 54 and 3 <= longitude <= 8):
return False
return bool(_netherlands_boundary().covers(Point(longitude, latitude)))
@lru_cache(maxsize=2)
def _city_boundary(city: str):
document = json.loads(
files("openworld_radio_twin.providers").joinpath(f"data/{city}.geojson").read_text()
)
features = document["features"]
if not features:
raise ValueError(f"Packaged {city} routing boundary is empty")
polygons = [shape(feature["geometry"]) for feature in features]
if any(not polygon.is_valid for polygon in polygons):
raise ValueError(f"Packaged {city} routing boundary contains invalid polygons")
boundary = unary_union(polygons)
if (
boundary.is_empty
or not boundary.is_valid
or boundary.geom_type not in {"Polygon", "MultiPolygon"}
):
raise ValueError(f"Packaged {city} routing boundary is invalid")
return boundary.bounds, prep(boundary)
def _in_city(city: str, latitude: float, longitude: float) -> bool:
if not worldwide(latitude, longitude):
return False
(west, south, east, north), boundary = _city_boundary(city)
return (
west <= longitude <= east
and south <= latitude <= north
and boundary.covers(Point(longitude, latitude))
)
[docs]
def in_boston(latitude: float, longitude: float) -> bool:
"""Boston municipal routing boundary, excluding neighbouring municipalities."""
return bool(_in_city("boston", latitude, longitude))
[docs]
def in_berlin(latitude: float, longitude: float) -> bool:
"""Berlin state routing boundary, excluding the surrounding Brandenburg area."""
return bool(_in_city("berlin", latitude, longitude))
[docs]
@dataclass(frozen=True)
class BuildingSource:
"""Provider identity and factory; auto-selection is explicitly opt-in.
``covers`` receives (latitude, longitude). Higher ``auto_priority`` wins;
None leaves a source available only by explicit selection. Version changes
must accompany changes in source interpretation or geometry reconstruction.
``auto_covers`` can restrict default routing without narrowing explicit
acquisition, whose tile/archive coverage is still checked by the provider.
"""
identifier: str
title: str
version: str
factory: Callable[[Settings, httpx.AsyncClient], BuildingProvider]
covers: Callable[[float, float], bool] = worldwide
auto_priority: int | None = None
attribution: str = ""
license_url: str = ""
configuration_identity: Callable[[Settings], dict[str, str]] | None = None
auto_covers: Callable[[float, float], bool] | None = None
def __post_init__(self):
if not re.fullmatch(r"[a-z][a-z0-9_-]*|3dbag", self.identifier):
raise ValueError("Building source identifiers must be lowercase slugs")
if self.identifier in {"auto", "global", "none"}:
raise ValueError(f"Reserved building source identifier: {self.identifier}")
if not self.version:
raise ValueError("A building source requires an implementation version")
[docs]
class BuildingSourceRegistry:
def __init__(self, sources: tuple[BuildingSource, ...] = ()) -> None:
self._sources: dict[str, BuildingSource] = {}
for source in sources:
self.register(source)
[docs]
def register(self, source: BuildingSource) -> None:
if not isinstance(source, BuildingSource):
raise TypeError("Building source plugins must return a BuildingSource")
if source.identifier in self._sources:
raise ValueError(f"Duplicate building source: {source.identifier}")
self._sources[source.identifier] = source
[docs]
def load_installed(self) -> None:
"""Load explicitly installed entry points, never a request-supplied import path."""
for entry in sorted(entry_points(group=ENTRY_POINT_GROUP), key=lambda item: item.name):
source = entry.load()()
if source.identifier != entry.name:
raise ValueError("Building source entry-point name must match its identifier")
self.register(source)
[docs]
def get(self, identifier: str) -> BuildingSource:
try:
return self._sources[identifier]
except KeyError:
raise ValueError(f"Unknown building source: {identifier}") from None
[docs]
def automatic(
self,
latitude: float,
longitude: float,
*,
allowed: tuple[str, ...] = DEFAULT_AUTOMATIC_BUILDING_SOURCES,
) -> BuildingSource | None:
for identifier in allowed:
self.get(identifier) # Report misspelled policy instead of changing the source.
candidates = [
source
for source in self._sources.values()
if source.identifier in allowed
and source.auto_priority is not None
and source.covers(latitude, longitude)
and (source.auto_covers is None or source.auto_covers(latitude, longitude))
]
return max(
candidates, key=lambda source: (source.auto_priority, source.identifier), default=None
)
[docs]
def describe(self) -> list[dict[str, object]]:
return [
{
"id": source.identifier,
"title": source.title,
"version": source.version,
"automatic": source.auto_priority is not None,
"attribution": source.attribution,
"license_url": source.license_url,
}
for source in self._sources.values()
]
