audi-app/custom_components/audi_dashboard/geokodierung.py

"""Koordinaten zu Ortsnamen - einmal im Backend, nicht in jedem Geraet.

Bis zum 01.09.2026 loeste jede Oberflaeche die Fahrtkoordinaten selbst auf und
legte das Ergebnis in ihren eigenen Zwischenspeicher (localStorage im Panel,
derselbe Mechanismus in der App). Drei Folgen, alle drei unnoetig:

* Jedes Geraet fragte dieselben Koordinaten erneut ab. Bei zwei Telefonen und
  dem Panel also dreimal - fuer einen Ort, der sich nie wieder aendert.
* Ein frisch eingerichtetes Geraet zeigte fuer Monate alte Fahrten wieder
  "unbekannt", bis jemand jede einzelne Fahrt geoeffnet hatte.
* Ein geleerter Browserspeicher warf alles weg.

Jetzt steht der Ort in der Fahrt selbst. Er wird genau einmal aufgeloest und
danach von jedem Geraet gelesen wie jedes andere Feld auch.

NOMINATIMS NUTZUNGSBEDINGUNGEN sind der Grund fuer fast jede Entscheidung
hier: hoechstens eine Anfrage je Sekunde, kein Vorratsabruf, ein
aussagekraeftiger User-Agent, und Ergebnisse sind zu speichern statt erneut
abzurufen. Daher die Sperre zwischen zwei Anfragen, daher die Obergrenze je
Durchlauf, und daher der Zwischenspeicher auf der Platte.
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import time

from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession

_LOGGER = logging.getLogger(__name__)

DIENST = "https://nominatim.openstreetmap.org/reverse"

# Nominatim bittet um hoechstens eine Anfrage je Sekunde. 1,1 s laesst Luft
# fuer Ungenauigkeiten der Uhr, ohne spuerbar laenger zu dauern.
ABSTAND_S = 1.1

# Wie viele Orte ein einzelner Screening-Durchlauf hoechstens aufloest. Ein
# Rueckblick ueber ein Jahr braechte sonst Hunderte Anfragen am Stueck - genau
# der Vorratsabruf, um dessen Unterlassung Nominatim bittet. Der Rest kommt
# beim naechsten Durchlauf dran; die laufen haeufig genug.
PRO_DURCHLAUF = 8

# Drei Nachkommastellen sind rund 110 m. Feiner aufzuloesen braechte fuer eine
# Anschrift nichts und wuerde den Zwischenspeicher unbrauchbar machen: jede
# Fahrt endete auf einem eigenen Schluessel. Dieselbe Rundung wie bisher in
# beiden Oberflaechen, damit deren Altbestand weiter passt.
STELLEN = 3

ZEITSPERRE_S = 8

# Ein benannter Ort sagt mehr als die naechstgelegene Hausnummer - aber nur,
# wenn er wirklich ein Ziel ist. Ohne diese Pruefung verdraengte jedes
# zufaellig benannte Gebaeude die Anschrift.
BENANNTE_ARTEN = {
    "parking",
    "parking_space",
    "parking_entrance",
    "bicycle_parking",
    "motorcycle_parking",
    "rest_area",
    "services",
    "fuel",
    "charging_station",
}


def schluessel(lat: float, lon: float) -> str:
    return f"{round(lat, STELLEN):.{STELLEN}f},{round(lon, STELLEN):.{STELLEN}f}"


def anschrift_bauen(daten: dict) -> tuple[str | None, str | None]:
    """(Anschrift, Ort) aus Nominatims Antwort.

    Das Format ist bewusst dasselbe, das die Oberflaechen bisher selbst
    gebaut haben - "<Strasse> <Nr>, <PLZ> <Ort>" -, damit bereits
    gespeicherte Anschriften und neue nicht unterschiedlich aussehen.

    Der Ort wird SEPARAT zurueckgegeben, nicht spaeter aus der Anschrift
    geschnitten: faellt Nominatim auf display_name zurueck, stuende hinter dem
    letzten Komma das Land, und in der Fahrtenliste stuende dann
    "Deutschland -> Deutschland".
    """
    a = daten.get("address") or {}
    stadt = (
        a.get("city")
        or a.get("town")
        or a.get("village")
        or a.get("municipality")
        or a.get("suburb")
        or ""
    ).strip()

    name = (daten.get("name") or "").strip()
    art = daten.get("type") or daten.get("category") or ""
    if name and art in BENANNTE_ARTEN:
        vorn = name
    else:
        strasse = (a.get("road") or a.get("pedestrian") or "").strip()
        nummer = (a.get("house_number") or "").strip()
        vorn = " ".join(t for t in (strasse, nummer) if t)

    plz = (a.get("postcode") or "").strip()
    hinten = " ".join(t for t in (plz, stadt) if t)
    anschrift = ", ".join(t for t in (vorn, hinten) if t)

    if not anschrift:
        anschrift = (daten.get("display_name") or "").strip()

    return (anschrift or None), (stadt or None)


class Geokodierung:
    """Loest Koordinaten auf und merkt sich jedes Ergebnis auf der Platte."""

    def __init__(self, hass: HomeAssistant, basis: str) -> None:
        self._hass = hass
        self._pfad = os.path.join(basis, "orte.json")
        self._speicher: dict[str, dict] | None = None
        self._sperre = asyncio.Lock()
        self._zuletzt = 0.0
        self.rest = PRO_DURCHLAUF

    def durchlauf_beginnen(self) -> None:
        """Setzt das Anfragenbudget fuer einen Screening-Lauf zurueck."""
        self.rest = PRO_DURCHLAUF

    # ------------------------------------------------------ Zwischenspeicher

    def _laden(self) -> dict:
        if not os.path.exists(self._pfad):
            return {}
        try:
            with open(self._pfad, encoding="utf-8") as datei:
                inhalt = json.load(datei)
            return inhalt if isinstance(inhalt, dict) else {}
        except (OSError, ValueError):
            # Eine beschaedigte Datei darf das Screening nicht anhalten - im
            # schlimmsten Fall wird noch einmal abgefragt.
            _LOGGER.warning("Ortsspeicher unlesbar, wird neu aufgebaut: %s", self._pfad)
            return {}

    def _sichern(self, inhalt: dict) -> None:
        tmp = self._pfad + ".tmp"
        with open(tmp, "w", encoding="utf-8") as datei:
            json.dump(inhalt, datei, ensure_ascii=False, indent=1)
        os.replace(tmp, self._pfad)

    async def _speicher_holen(self) -> dict:
        if self._speicher is None:
            self._speicher = await self._hass.async_add_executor_job(self._laden)
        return self._speicher

    # ---------------------------------------------------------- Aufloesung

    async def aufloesen(self, lat: float | None, lon: float | None) -> dict | None:
        """{"anschrift": ..., "stadt": ...} oder None.

        None heisst "noch nicht bekannt und diesmal nicht abgefragt" - entweder
        weil das Budget aufgebraucht ist oder weil der Abruf scheiterte. Beides
        wird beim naechsten Durchlauf erneut versucht.

        Ein Ergebnis OHNE Anschrift ({"anschrift": None}) heisst dagegen "dort
        ist wirklich nichts" und wird gemerkt, nicht erneut abgefragt.
        """
        if lat is None or lon is None:
            return None

        try:
            key = schluessel(float(lat), float(lon))
        except (TypeError, ValueError):
            return None

        speicher = await self._speicher_holen()
        if key in speicher:
            return speicher[key]

        if self.rest <= 0:
            return None

        async with self._sperre:
            # Erneut pruefen: waehrend des Wartens kann ein anderer Aufruf
            # dieselbe Koordinate bereits aufgeloest haben.
            if key in speicher:
                return speicher[key]
            abstand = time.monotonic() - self._zuletzt
            if abstand < ABSTAND_S:
                await asyncio.sleep(ABSTAND_S - abstand)
            ergebnis = await self._abfragen(float(lat), float(lon))
            self._zuletzt = time.monotonic()
            self.rest -= 1

        if ergebnis is None:
            # Netz- oder Serverfehler: NICHT merken. Sonst stuende ein
            # voruebergehend gedrosselter Abruf fuer immer als "kein Ort hier".
            return None

        speicher[key] = ergebnis
        await self._hass.async_add_executor_job(self._sichern, dict(speicher))
        return ergebnis

    async def _abfragen(self, lat: float, lon: float) -> dict | None:
        session = async_get_clientsession(self._hass)
        try:
            antwort = await session.get(
                DIENST,
                params={
                    "lat": f"{lat}",
                    "lon": f"{lon}",
                    "format": "jsonv2",
                    "zoom": "18",
                    "addressdetails": "1",
                },
                headers={
                    # Nominatim verlangt eine identifizierbare Anwendung und
                    # sperrt anonyme Aufrufer.
                    "User-Agent": "audi-dashboard (Home Assistant Integration)",
                    "Accept-Language": "de",
                },
                timeout=ZEITSPERRE_S,
            )
            if antwort.status != 200:
                _LOGGER.debug("Nominatim antwortete mit %s", antwort.status)
                return None
            daten = await antwort.json(content_type=None)
        except Exception as fehler:  # Netzfehler, Zeitueberschreitung, kaputtes JSON
            _LOGGER.debug("Ortsabfrage fehlgeschlagen: %s", fehler)
            return None

        if not isinstance(daten, dict) or daten.get("error"):
            return {"anschrift": None, "stadt": None}
        anschrift, stadt = anschrift_bauen(daten)
        return {"anschrift": anschrift, "stadt": stadt}