audi-app/custom_components/audi_dashboard/verlauf.py

"""Lesezugriff auf den aufgezeichneten Zustandsverlauf (recorder).

Über Home Assistants eigene recorder-API
(`homeassistant.components.recorder.history.get_significant_states`), nicht
über direkten SQL-Zugriff auf home-assistant_v2.db: das Datenbankschema des
recorders ist HA-intern und ändert sich zwischen Versionen, die Funktion
dagegen ist die von HA selbst benutzte und stabile Schnittstelle.

`significant_changes_only=False` ist wichtig: bei numerischen Sensoren
(Kilometerstand, Tankfüllstand) liefert der Standardmodus nur "auffällige"
Änderungen und verschluckt genau die kleinen Schritte, aus denen sich
Fahrstrecke und Tankvorgänge zusammensetzen.

WAS HIER GEGENÜBER DER PYSCRIPT-FASSUNG WEGFÄLLT: das Fahrtabschluss-
Screening las den Verlauf früher über die HTTP-REST-API (/api/history/period)
und brauchte dafür ein langlebiges Zugriffstoken in
audi_dashboard/ha_token.txt - ein eigener Installationsschritt, der bei jeder
Neuinstallation vergessen werden konnte und dessen Fehlen sich nur als
Warnung im Protokoll zeigte (Fahrten blieben dann stumm ohne Strecke). Als
echte Integration liest die App den Verlauf direkt; Token und
Installationsschritt entfallen ersatzlos.

Die recorder-Abfrage läuft über den Executor: ein Datenbankzugriff hat im
Event-Loop nichts verloren.
"""

from __future__ import annotations

import datetime
import logging

from homeassistant.components.recorder import get_instance, history
from homeassistant.core import HomeAssistant

_LOGGER = logging.getLogger(__name__)

Verlaufspunkt = tuple[datetime.datetime, str]


def _rohverlauf(
    hass: HomeAssistant, entity_id: str, start: datetime.datetime, ende: datetime.datetime
) -> list[Verlaufspunkt]:
    roh = history.get_significant_states(
        hass, start, ende, [entity_id], None, True, False
    )
    ergebnis: list[Verlaufspunkt] = []
    for zustand in roh.get(entity_id) or []:
        # "unknown"/"unavailable" bedeuten "keine Meldung", nicht "Wert 0" -
        # würden sie durchgereicht, ergäbe ein Ausfall der Datenquelle eine
        # Fahrt mit absurder Kilometerdifferenz.
        if zustand.state in ("unknown", "unavailable", None, ""):
            continue
        ergebnis.append((zustand.last_updated, zustand.state))
    ergebnis.sort(key=lambda p: p[0])
    return ergebnis


async def verlauf_lesen(
    hass: HomeAssistant,
    entity_id: str | None,
    start: datetime.datetime,
    ende: datetime.datetime,
) -> list[Verlaufspunkt]:
    """Zustandsverlauf einer Entität als aufsteigende Liste von
    (Zeitpunkt, Rohwert). Leere Liste, wenn die Entität nicht zugeordnet ist,
    im Zeitraum nichts vorliegt oder der recorder nicht erreichbar ist."""
    if not entity_id:
        return []
    try:
        return await get_instance(hass).async_add_executor_job(
            _rohverlauf, hass, entity_id, start, ende
        )
    except Exception:  # noqa: BLE001 - der Verlauf ist Beiwerk, nie der Kern
        # Ohne Verlauf bleibt eine Fahrt ohne Strecke bzw. der Import leer -
        # beides ist verkraftbar. Den aufrufenden Ablauf deswegen abzubrechen
        # wäre es nicht.
        _LOGGER.warning("Verlauf von %s nicht lesbar", entity_id, exc_info=True)
        return []


def zeit_normalisiert(wert: object) -> datetime.datetime | None:
    """Akzeptiert ISO-Zeit mit oder ohne Zeitzone und normalisiert auf UTC.

    Ohne Zeitzone gilt die lokale Zeit von Home Assistant - die Oberfläche
    schickt sowohl bei manueller Eingabe (Fahrt/Tankvorgang von Hand anlegen
    oder bearbeiten, fahrterkennung.py/belege.py) als auch beim rückwirkenden
    Import (historienimport.py) Ortszeit, nie UTC. Ohne diese Normalisierung
    bliebe ein von Hand erfasster Zeitstempel naiv (kein tzinfo) - und
    historienimport.py's eigene Dublettenprüfung (_ueberschneidet() für
    Fahrten, bekannte_zeiten für Tankvorgänge) überspringt jeden Datensatz mit
    naivem Zeitstempel stillschweigend, statt ihn auf Überschneidung zu
    prüfen. Ein von Hand angelegter oder bearbeiteter Datensatz wäre dadurch
    nie vor einem doppelten Import geschützt - genau die Lücke, die diese
    gemeinsame Funktion schließt, indem jeder Schreibpfad dieselbe
    Normalisierung durchläuft."""
    if not wert:
        return None
    ts = datetime.datetime.fromisoformat(str(wert))
    if ts.tzinfo is None:
        ts = ts.astimezone()
    return ts.astimezone(datetime.UTC)


def zahl(wert: object) -> float | None:
    try:
        return float(wert)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return None


# Kein Auto dieser Art erreicht diesen Schnitt. Eine errechnete
# Durchschnittsgeschwindigkeit darüber ist kein Beleg für eine schnelle
# Fahrt, sondern für einen fehlerhaften Kilometerstand-Sprung am Fahrtrand
# (z. B. ein Ausreißer im Verlauf oder ein knapp daneben liegender
# wert_bei()/naechster_wert()-Treffer über eine Lücke hinweg). Gemeinsam für
# den Import (historienimport.py) und die Live-Vervollständigung
# (screening.py) - beide dürfen dieselbe Strecke nie unterschiedlich
# bewerten.
UNPLAUSIBLE_KMH = 300


def durchschnitt_kmh(distanz_km: float | None, dauer_s: float | None) -> float | None:
    """km/h aus Strecke und Dauer, oder None wenn eine der beiden fehlt oder
    die Dauer nicht positiv ist."""
    if distanz_km is None or not dauer_s or dauer_s <= 0:
        return None
    return round(distanz_km / (dauer_s / 3600.0), 1)


def wert_bei(
    verlauf: list[Verlaufspunkt], zeitpunkt: datetime.datetime
) -> float | None:
    """Der zuletzt vor `zeitpunkt` gemeldete Zahlenwert, sonst der erste
    danach, sonst None.

    "Zuletzt davor" ist die richtige Wahl für einen Zählerstand: der
    Kilometerstand bei Fahrtbeginn ist der, der zuletzt gemeldet wurde, nicht
    der nächste (der schon Strecke enthält)."""
    davor = None
    for ts, wert in verlauf:
        gezahlt = zahl(wert)
        if gezahlt is None:
            continue
        if ts <= zeitpunkt:
            davor = gezahlt
        else:
            return davor if davor is not None else gezahlt
    return davor


def route_aus_verlauf(
    lat_verlauf: list[Verlaufspunkt],
    lon_verlauf: list[Verlaufspunkt],
    start: datetime.datetime,
    ende: datetime.datetime,
) -> list[list[float]] | None:
    """Baut eine chronologische Liste von [lat, lon]-Punkten für die
    Fahrt-Detailkarte, aus zwei getrennt geführten Sensor-Verläufen (die
    FMM003 liefert Breiten-/Längengrad als zwei unabhängige Sensoren, siehe
    STANDORT_LAT_SENSOR/STANDORT_LON_SENSOR). Jeder Breitengrad-Punkt im
    Fenster [start, ende] bekommt den zeitlich nächstgelegenen Längengrad-Wert
    zugeordnet - beide Sensoren aktualisieren unabhängig voneinander, exakt
    gleiche Zeitstempel sind nicht garantiert. None bei weniger als zwei
    Punkten (keine sinnvolle Linie)."""
    punkte: list[list[float]] = []
    for ts, wert in lat_verlauf:
        if not (start <= ts <= ende):
            continue
        lat = zahl(wert)
        if lat is None:
            continue
        lon = naechster_wert(ts, lon_verlauf)
        if lon is None:
            continue
        punkte.append([lat, lon])
    return punkte if len(punkte) >= 2 else None


def verbrauch_aus_literstaenden(
    start_liter: float | None, end_liter: float | None, distance_km: float | None
) -> float | None:
    """l/100km aus der Differenz zweier Literstände (TANK_LITER_SENSOR) und
    der gefahrenen Strecke - eine Näherung, kein vom Fahrzeug selbst für
    diese eine Fahrt gemeldeter Wert (ein solcher OBD-Sensor steht als
    offener Punkt in AGENTS.md; taucht einer auf, ersetzt er diese Näherung).

    None, wenn einer der Werte fehlt, die Strecke nicht positiv ist, oder der
    Literstand nicht gefallen ist (z. B. ein Tankvorgang während der Fahrt,
    oder unveränderter Füllstand) - in beiden Fällen ist der Quotient nicht
    aussagekräftig."""
    if start_liter is None or end_liter is None or not distance_km or distance_km <= 0:
        return None
    verbraucht = start_liter - end_liter
    if verbraucht <= 0:
        return None
    return round(verbraucht / distance_km * 100, 1)


def naechster_wert(
    zielzeit: datetime.datetime, verlauf: list[Verlaufspunkt]
) -> float | None:
    """Der Wert, dessen Zeitstempel am nächsten an `zielzeit` liegt.

    Anders als wert_bei() ist hier bewusst egal, ob der Wert davor oder danach
    liegt: der Kilometerstand kommt laut Datenquelle nicht sicher mit
    Fahrtende, sondern teils erst mit Beginn der nächsten Fahrt."""
    bester_wert = None
    beste_diff = None
    for ts, wert in verlauf:
        gezahlt = zahl(wert)
        if gezahlt is None:
            continue
        diff = abs((ts - zielzeit).total_seconds())
        if beste_diff is None or diff < beste_diff:
            bester_wert = gezahlt
            beste_diff = diff
    return bester_wert


# Wie lange eine Nebenmessung (z. B. Außentemperatur) höchstens von der
# Hauptmessung entfernt liegen darf, um noch als "zur selben Zeit" zu gelten -
# vom Nutzer auf 300s (5 Minuten) festgelegt, nach ursprünglich 60s. Der
# Wechsel ist an echten FMM003-Daten begründet, nicht geschätzt: das Gerät
# sendet Außentemperatur deutlich seltener als Spannung (an einem
# beobachteten Fahrtfenster 851 Spannungs- gegen 187 Temperaturmeldungen in
# 48h) - selbst bei laufender Fahrt und ohne jede Verbindungsstörung wurden
# reale Lücken bis 593s zwischen zwei Temperaturmeldungen gemessen. 60s hätte
# genau diesen völlig normalen Fall verworfen. Physikalisch unproblematisch:
# die Außentemperatur ändert sich über wenige Minuten kaum, anders als z. B.
# eine GPS-Position oder die Spannung selbst - und die Live-Prüfung
# (batterie.py's pruefen()) läuft ohnehin nur alle 5 Minuten, verlangt hier
# also keine feinere Genauigkeit, als sie selbst liefert. Ein Verbindungs-
# abbruch des Geräts (beobachtet: 45 Minuten, beide Sensoren gleichzeitig
# stumm) bleibt mit 300s weiterhin sicher außerhalb der Grenze und liefert
# bewusst KEINEN Wert statt eines beliebig alten.
NEBENWERT_MAX_ABSTAND_S = 300


def wert_in_naehe(
    zielzeit: datetime.datetime, verlauf: list[Verlaufspunkt], max_abstand_s: float = NEBENWERT_MAX_ABSTAND_S
) -> float | None:
    """Wie naechster_wert(), aber None statt eines beliebig weit entfernten
    Werts, wenn selbst der zeitlich nächste mehr als max_abstand_s von
    zielzeit entfernt liegt."""
    bester_wert = None
    beste_diff = None
    for ts, wert in verlauf:
        gezahlt = zahl(wert)
        if gezahlt is None:
            continue
        diff = abs((ts - zielzeit).total_seconds())
        if beste_diff is None or diff < beste_diff:
            bester_wert = gezahlt
            beste_diff = diff
    if beste_diff is None or beste_diff > max_abstand_s:
        return None
    return bester_wert