feat(bridge): Standort-Präfix mit Straße+Hausnr, Straßen-Ref + km (Autobahn)

Erweitert den Reverse-Geocode: Nominatim zoom=18 → Straße + Hausnummer, PLZ,
Stadt, Bundesland ("Am Wunderburgpark 6, 26135 Oldenburg, Niedersachsen").
Bei Autobahn/Bundes-/Kreisstraße wird die Ref (A 28 / B 75 / K …, aus
extratags) vorangestellt + best-effort Kilometer via Overpass-Milestone
(highway=milestone, distance-Tag, naechster im Umkreis) → z.B.
"A 28 bei km 55,6, Oldenburg, Niedersachsen". Für Pannenhilfe & Co.

Cache-Raster von ~1km auf ~110m verfeinert (Straßen-Genauigkeit). Overpass
nur wenn's nach Straße/Autobahn aussieht (ref/Autobahn im Namen). Alles
best-effort mit stiller Fallback-Kette (Fehler → gröberer Ort → nur GPS).
NOMINATIM_URL + OVERPASS_URL env → self-hostbar.

Richtung ("Richtung Leer") bewusst weggelassen — nicht zuverlaessig aus
einem Punkt ableitbar (braucht Fahrtrichtung/Routing).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-11 22:06:15 +02:00
co-authored by Claude Opus 4.8
parent f1da2cc9af
commit 4576f85a45
+82 -8
View File
@@ -32,6 +32,7 @@ from pathlib import Path
from typing import Optional from typing import Optional
import subprocess import subprocess
import urllib.parse
import urllib.request import urllib.request
import numpy as np import numpy as np
import sounddevice as sd import sounddevice as sd
@@ -1193,7 +1194,7 @@ class ARIABridge:
(~1 km Raster) und im Hintergrund. Ergebnis landet in (~1 km Raster) und im Hintergrund. Ergebnis landet in
self._last_place_name; der Standort-Praefix liest es dann nur ab.""" self._last_place_name; der Standort-Praefix liest es dann nur ab."""
try: try:
key = (round(lat, 2), round(lon, 2)) # ~1.1 km — reicht fuer "Stadt" key = (round(lat, 3), round(lon, 3)) # ~110 m — fuer Strasse/Hausnr
except Exception: except Exception:
return return
if key == getattr(self, "_geocoded_key", None): if key == getattr(self, "_geocoded_key", None):
@@ -1208,11 +1209,14 @@ class ARIABridge:
pass pass
async def _do_reverse_geocode(self, lat: float, lon: float, key) -> None: async def _do_reverse_geocode(self, lat: float, lon: float, key) -> None:
"""Nominatim-Reverse-Geocode (keyless, gratis). Setzt self._last_place_name. """Nominatim-Reverse-Geocode (keyless, gratis). Baut einen moeglichst
Fehler/Timeout → still (Praefix faellt dann auf reine Koordinaten zurueck).""" genauen Ort: Strasse+Hausnummer, PLZ+Stadt, Bundesland; bei Autobahn/
Bundesstrasse zusaetzlich die Ref (A 28 / B 75) + best-effort Kilometer
(Overpass-Milestone). Setzt self._last_place_name. Fehler → still
(Praefix faellt auf reine Koordinaten zurueck)."""
base = os.environ.get("NOMINATIM_URL", "https://nominatim.openstreetmap.org").rstrip("/") base = os.environ.get("NOMINATIM_URL", "https://nominatim.openstreetmap.org").rstrip("/")
url = (f"{base}/reverse?lat={lat:.5f}&lon={lon:.5f}" url = (f"{base}/reverse?lat={lat:.5f}&lon={lon:.5f}&format=jsonv2"
f"&format=jsonv2&zoom=10&accept-language=de") f"&zoom=18&addressdetails=1&extratags=1&accept-language=de")
def _fetch(): def _fetch():
try: try:
@@ -1230,15 +1234,85 @@ class ARIABridge:
if not isinstance(data, dict): if not isinstance(data, dict):
return return
addr = data.get("address") or {} addr = data.get("address") or {}
place = (addr.get("city") or addr.get("town") or addr.get("village") extra = data.get("extratags") or {}
or addr.get("municipality") or addr.get("county") or "") road = addr.get("road") or ""
house = addr.get("house_number") or ""
city = (addr.get("city") or addr.get("town") or addr.get("village")
or addr.get("municipality") or addr.get("county") or "")
plz = addr.get("postcode") or ""
state = addr.get("state") or "" state = addr.get("state") or ""
name = f"{place}, {state}" if (place and state) else (place or data.get("name") or "") ref = (extra.get("ref") or "").strip() # z.B. "A 28", "B 75"
# Strasse + Hausnummer
street = (f"{road} {house}".strip()) if road else ""
# PLZ + Ort
citypart = (f"{plz} {city}".strip()) if plz else city
segs: list[str] = []
# Autobahn/Bundes-/Kreisstrasse? Ref + (best-effort) Kilometer voranstellen.
is_road = bool(ref) or any(w in road.lower()
for w in ("autobahn", "bundesstrasse", "bundesstraße",
"kreisstrasse", "kreisstraße"))
if is_road:
label = ref or road
km = await self._overpass_km(lat, lon)
if km:
label = f"{label} bei km {km}"
if label:
segs.append(label)
if street:
segs.append(street)
if citypart:
segs.append(citypart)
if state and (state not in citypart):
segs.append(state)
name = ", ".join(s for s in segs if s)
if not name:
name = data.get("name") or ""
# Nur uebernehmen wenn Stefan nicht schon weitergezogen ist (key aktuell). # Nur uebernehmen wenn Stefan nicht schon weitergezogen ist (key aktuell).
if name and getattr(self, "_geocoded_key", None) == key: if name and getattr(self, "_geocoded_key", None) == key:
self._last_place_name = name self._last_place_name = name
logger.info("[geo] %.5f,%.5f → %s", lat, lon, name) logger.info("[geo] %.5f,%.5f → %s", lat, lon, name)
async def _overpass_km(self, lat: float, lon: float) -> str:
"""Best-effort Autobahn/Strassen-Kilometer: naechsten OSM-Milestone
(highway=milestone mit distance-Tag) im Umkreis suchen. Leer wenn keiner
gefunden / Overpass nicht erreichbar. Milestones stehen i.d.R. alle 500 m,
also grob (fuer Pannenhilfe 'ca. km X' voellig ausreichend)."""
ov = os.environ.get("OVERPASS_URL", "https://overpass-api.de/api/interpreter")
query = (f"[out:json][timeout:8];"
f"node(around:900,{lat:.5f},{lon:.5f})[highway=milestone];out;")
def _fetch():
try:
body = urllib.parse.urlencode({"data": query}).encode("utf-8")
req = urllib.request.Request(ov, data=body, headers={
"User-Agent": "ARIA-Assistant/1.0 (personal voice assistant)",
})
with urllib.request.urlopen(req, timeout=12) as r:
return json.loads(r.read().decode("utf-8", "ignore"))
except Exception as exc:
logger.debug("[geo] overpass-km fehlgeschlagen: %s", exc)
return None
d = await asyncio.get_event_loop().run_in_executor(None, _fetch)
if not isinstance(d, dict):
return ""
best_dist = ""
best_sq = 1e18
for e in (d.get("elements") or []):
tags = e.get("tags") or {}
dist = tags.get("distance") or tags.get("milestone:distance") or ""
elat, elon = e.get("lat"), e.get("lon")
if not dist or elat is None or elon is None:
continue
sq = (float(elat) - lat) ** 2 + (float(elon) - lon) ** 2
if sq < best_sq:
best_sq = sq
best_dist = str(dist)
return best_dist.replace(".", ",") if best_dist else ""
async def _trigger_brain_check_now(self) -> None: async def _trigger_brain_check_now(self) -> None:
"""Brain-Endpoint POST /triggers/check-now anstossen.""" """Brain-Endpoint POST /triggers/check-now anstossen."""
brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080") brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")