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


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