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 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