5cef596a3c
Right-click (long-press on touch) opens a What's here? HUD panel: reverse geocode via GET /api/place (honest UA, 1 req/s, 60s/500-key cache) and lists already-loaded cameras, aircraft, vessels, fires, and alerts within 5 km.
99 lines
3 KiB
Python
99 lines
3 KiB
Python
"""Nominatim reverse-geocode proxy for the map place dossier.
|
|
|
|
Browser clients cannot set an identifying User-Agent, and Nominatim typically
|
|
blocks CORS — so the HUD calls GET /api/place instead of talking to OSM
|
|
directly. Cache 60s / 500 keys; never exceed 1 req/s upstream.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
|
|
import httpx
|
|
from cachetools import TTLCache
|
|
|
|
from config import NOMINATIM_MIN_INTERVAL, NOMINATIM_URL, OSINT_USER_AGENT
|
|
|
|
_NOMINATIM = NOMINATIM_URL.rstrip("/")
|
|
|
|
place_cache: TTLCache = TTLCache(maxsize=500, ttl=60)
|
|
|
|
_lock = asyncio.Lock()
|
|
_last_req = 0.0
|
|
|
|
_ADDR_KEEP = (
|
|
"house_number", "road", "neighbourhood", "suburb", "city", "town",
|
|
"village", "hamlet", "county", "state", "postcode", "country", "country_code",
|
|
)
|
|
|
|
|
|
def cache_key(lat: float, lon: float) -> str:
|
|
return f"{lat:.4f},{lon:.4f}"
|
|
|
|
|
|
def slim_place(lat: float, lon: float, data: dict | None) -> dict:
|
|
data = data or {}
|
|
raw_addr = data.get("address")
|
|
addr_in: dict = raw_addr if isinstance(raw_addr, dict) else {}
|
|
address = {k: addr_in[k] for k in _ADDR_KEEP if addr_in.get(k)}
|
|
err = data.get("error")
|
|
display = None if err else (data.get("display_name") or None)
|
|
name = None if err else (data.get("name") or address.get("city")
|
|
or address.get("town") or address.get("village") or None)
|
|
return {
|
|
"lat": lat,
|
|
"lon": lon,
|
|
"display_name": display,
|
|
"name": name,
|
|
"address": address,
|
|
"osm_type": None if err else data.get("osm_type"),
|
|
"osm_id": None if err else data.get("osm_id"),
|
|
"attribution": "© OpenStreetMap contributors",
|
|
}
|
|
|
|
|
|
async def reverse_geocode(lat: float, lon: float) -> dict:
|
|
"""Reverse-geocode a point. Cache hits skip Nominatim entirely."""
|
|
if not (-90.0 <= lat <= 90.0 and -180.0 <= lon <= 180.0):
|
|
raise ValueError("lat/lon out of range")
|
|
key = cache_key(lat, lon)
|
|
qlat, qlon = (float(p) for p in key.split(","))
|
|
async with _lock:
|
|
hit = place_cache.get(key)
|
|
if hit is not None:
|
|
return hit
|
|
global _last_req
|
|
wait = _last_req + NOMINATIM_MIN_INTERVAL - time.monotonic()
|
|
if wait > 0:
|
|
await asyncio.sleep(wait)
|
|
body = await _fetch_nominatim(qlat, qlon)
|
|
_last_req = time.monotonic()
|
|
place_cache[key] = body
|
|
return body
|
|
|
|
|
|
async def _fetch_nominatim(lat: float, lon: float) -> dict:
|
|
headers = {
|
|
"User-Agent": OSINT_USER_AGENT,
|
|
"Accept": "application/json",
|
|
}
|
|
url = f"{_NOMINATIM}/reverse"
|
|
params = {
|
|
"lat": f"{lat:.6f}",
|
|
"lon": f"{lon:.6f}",
|
|
"format": "jsonv2",
|
|
"addressdetails": "1",
|
|
"zoom": "18",
|
|
}
|
|
async with _http_client(timeout=10.0, follow_redirects=True) as client:
|
|
r = await client.get(url, params=params, headers=headers)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
if not isinstance(data, dict):
|
|
data = {}
|
|
return slim_place(lat, lon, data)
|
|
|
|
|
|
def _http_client(**kwargs):
|
|
return httpx.AsyncClient(**kwargs)
|