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 1c11cb6a2f
commit 4605698b02
+82 -8
View File
@@ -32,6 +32,7 @@ from pathlib import Path
from typing import Optional
import subprocess
import urllib.parse
import urllib.request
import numpy as np
import sounddevice as sd
@@ -1193,7 +1194,7 @@ class ARIABridge:
(~1 km Raster) und im Hintergrund. Ergebnis landet in
self._last_place_name; der Standort-Praefix liest es dann nur ab."""
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:
return
if key == getattr(self, "_geocoded_key", None):
@@ -1208,11 +1209,14 @@ class ARIABridge:
pass
async def _do_reverse_geocode(self, lat: float, lon: float, key) -> None:
"""Nominatim-Reverse-Geocode (keyless, gratis). Setzt self._last_place_name.
Fehler/Timeout → still (Praefix faellt dann auf reine Koordinaten zurueck)."""
"""Nominatim-Reverse-Geocode (keyless, gratis). Baut einen moeglichst
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("/")
url = (f"{base}/reverse?lat={lat:.5f}&lon={lon:.5f}"
f"&format=jsonv2&zoom=10&accept-language=de")
url = (f"{base}/reverse?lat={lat:.5f}&lon={lon:.5f}&format=jsonv2"
f"&zoom=18&addressdetails=1&extratags=1&accept-language=de")
def _fetch():
try:
@@ -1230,15 +1234,85 @@ class ARIABridge:
if not isinstance(data, dict):
return
addr = data.get("address") or {}
place = (addr.get("city") or addr.get("town") or addr.get("village")
or addr.get("municipality") or addr.get("county") or "")
extra = data.get("extratags") 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 ""
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).
if name and getattr(self, "_geocoded_key", None) == key:
self._last_place_name = 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:
"""Brain-Endpoint POST /triggers/check-now anstossen."""
brain_url = os.environ.get("BRAIN_URL", "http://aria-brain:8080")