audi-app/custom_components/audi_dashboard/veroeffentlichung.py

"""Baut die Nutzlasten, die als Entitäts-Attribut `daten` an die Oberfläche
gehen.

Das Frontend liest die Werte über das ganz normale hass.states statt über
einen Dienst mit Rückgabewert: Zustände lesen ist ein seit Jahren stabiler,
einfacher Weg, während das Verhalten response-fähiger Dienste aus Sicht des
Browsers nicht durchgängig dokumentiert ist.

Hier stehen nur die reinen Rechenfunktionen - wer sie wann veröffentlicht,
entscheidet der Koordinator (koordinator.py).

GRÖSSENGRENZE, DIE JETZT KEINE MEHR IST: Zustands-Attribute sind in Home
Assistant nicht für beliebig große Datenmengen gedacht; ab ~16 KB warnt der
Recorder und schreibt sie trotzdem in die Datenbank. Die Fahrten- und
Tankvorgänge-Archive überschreiten das nach ein paar Jahren zwangsläufig. In
der pyscript-Fassung war das eine bekannte, offene Flanke. Diese Integration
markiert das Attribut `daten` als nicht aufzuzeichnen (siehe
_unrecorded_attributes in sensor.py) - die Nutzlast geht weiterhin an jede
Oberfläche, landet aber nie in der Datenbank.
"""

from __future__ import annotations

import datetime

from homeassistant.core import HomeAssistant

from .einstellungen import POSITIONEN, Sensorzuordnung

# Letzte plausible Batteriespannung, für die "Zustand"-Kachel, wenn die
# aktuelle Live-Messung gerade unplausibel ist (Sensor kurzzeitig offline,
# Dongle getrennt o.ä. - erwartungsgemäß bei ausgeschaltetem Fahrzeug immer)
# - der Nutzer will dort die zuletzt gemessene Spannung sehen statt
# "unbekannt", ohne Ablaufdatum. Im RAM statt in einem Store wie andere
# Laufzeitzustände dieses Projekts (siehe die bekannten Lücken zu
# Fahrtstart/Tankstand) - aber koordinator.starten() lädt diesen Wert bei
# jedem Neustart einmalig aus der gespeicherten Historie vor
# (spannung_cache_vorladen()), bevor die erste Veröffentlichung läuft, damit
# ein Neustart NICHT mehr kurz "unbekannt" zeigt.
_letzte_gueltige_spannung: dict[str, float | None] = {"wert": None}


def spannung_cache_vorladen(verlauf: list[dict]) -> None:
    """Lädt die "Zustand"-Kachel-Fallback-Spannung beim Start aus der
    gespeicherten Historie (batteriespannung.jsonl) vor, statt bis zur
    nächsten echten Live-Messung nach jedem Neustart erst "unbekannt" zu
    zeigen.

    Der Dongle ist bei ausgeschaltetem Fahrzeug erwartungsgemäß immer
    offline (Nutzerangabe) - "die zuletzt gemessene Spannung" soll deshalb
    auch über einen Neustart hinweg sichtbar bleiben, und zwar ohne
    Ablaufdatum: hier gibt es bewusst KEINE Prüfung, wie alt der Wert ist -
    ein alter, aber echter Messwert ist der Kachel immer noch lieber als
    "unbekannt".

    Aber nur innerhalb der vereinbarten Grenze [SPANNUNG_MIN_V,
    AGM_RUHE_MAX_V] (10,0-13,0 V, batterie.py): min/max werden roh
    gespeichert (siehe batterie.py's Moduldocstring) und können deshalb auch
    Generatorspannung enthalten (z. B. ein Tagesmaximum von 15,4 V bei
    laufendem Alternator) - so ein Wert soll nie als "Batteriespannung"
    auf der Zustand-Kachel landen, auch nicht als letzter bekannter Wert.
    Durchsucht deshalb die GESAMTE Historie (nicht nur den letzten
    Tageseintrag) nach dem zeitlich jüngsten min- oder max-Wert, der
    innerhalb der Grenze liegt - ein Tag, an dem das Fahrzeug nie im
    Ruhezustand beobachtet wurde (beide Werte über der Grenze), wird dabei
    übersprungen, der davor gemessene gültige Wert bleibt der Fallback."""
    from .batterie import AGM_RUHE_MAX_V, SPANNUNG_MIN_V

    kandidaten: list[tuple[str, float]] = []
    for eintrag in verlauf:
        for wert_feld, ts_feld in (("min", "min_ts"), ("max", "max_ts")):
            wert = eintrag.get(wert_feld)
            ts = eintrag.get(ts_feld)
            if wert is None or not ts:
                continue
            if SPANNUNG_MIN_V <= wert <= AGM_RUHE_MAX_V:
                kandidaten.append((ts, wert))
    if not kandidaten:
        return
    kandidaten.sort(key=lambda p: p[0])
    _letzte_gueltige_spannung["wert"] = kandidaten[-1][1]


def zustand_oder_none(hass: HomeAssistant, entity_id: str | None) -> str | None:
    """Sicherer Zustandszugriff.

    Eine nicht (mehr) existierende Entität ist hier der Normalfall, solange
    Rollen im Setup-Menü unbelegt sind, und bleibt auch danach relevant: fällt
    eine Datenquelle aus, soll die Oberfläche das zeigen, nicht an einem
    Fehler hängen bleiben."""
    if not entity_id:
        return None
    zustand = hass.states.get(entity_id)
    if zustand is None or zustand.state in (None, "", "unknown", "unavailable"):
        return None
    return zustand.state


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


def _abs_zahl(wert: object) -> float | None:
    zahl = _zu_zahl(wert)
    return None if zahl is None else abs(zahl)


def _zu_bool(wert: object) -> bool | None:
    """binary_sensor-Zustand als echtes True/False, None bei fehlender Meldung.

    None ist hier ausdrücklich kein "nein": ohne zugeordneten Sensor weiß die
    App schlicht nicht, ob gefahren wird - und muss das anzeigen dürfen, statt
    "steht" zu behaupten."""
    if wert is None:
        return None
    return str(wert).lower() in ("on", "true", "1", "open", "yes")


def _standort(hass: HomeAssistant, werte: Sensorzuordnung) -> dict:
    """Live-GPS-Position - liest STANDORT_LAT_SENSOR/STANDORT_LON_SENSOR, zwei
    eigene sensor-Entities für Breiten-/Längengrad (flespi liefert Koordinaten
    so, nicht als Attribute einer einzelnen device_tracker-Entity).

    Fehlt eine der beiden Entity-IDs oder ist der Zustand (noch) nicht
    verfügbar, liefert diese Funktion durchgehend None statt eines geratenen
    Werts."""
    leer = {"lat": None, "lon": None, "genauigkeit_m": None, "zeit": None}

    lat_id, lon_id = werte.STANDORT_LAT_SENSOR, werte.STANDORT_LON_SENSOR
    if not lat_id or not lon_id:
        return leer
    lat_zustand = hass.states.get(lat_id)
    lon_zustand = hass.states.get(lon_id)
    if lat_zustand is None or lon_zustand is None:
        return leer
    lat = _zu_zahl(zustand_oder_none(hass, lat_id))
    lon = _zu_zahl(zustand_oder_none(hass, lon_id))
    if lat is None or lon is None:
        return leer
    return {
        "lat": lat,
        "lon": lon,
        "genauigkeit_m": None,
        "zeit": lat_zustand.last_updated.isoformat(),
    }


def _sicherheitscheck(hass: HomeAssistant, werte: Sensorzuordnung) -> list[dict]:
    """Die einzeln geprüften Punkte hinter "Sicher abgestellt", fürs Frontend -
    gruppiert über das Feld "gruppe" in vier Sammelzeilen (Fahrzeug verriegelt
    / Türen und Klappen geschlossen / Fenster und Dach geschlossen / Licht
    ausgeschaltet), die die Oberfläche durch Gruppieren dieser flachen Liste
    berechnet, statt hier schon vier separate Listen zu bauen - eine flache
    Liste bleibt die einfachere, additive Änderung gegenüber dem, was das
    Frontend vorher schon kannte.

    "ok" ist None, wenn der Sensor fehlt oder nicht verfügbar ist - genau
    daraus leitet sich auch die zusammengefasste Kennzahl unten ab, damit
    beide nie auseinanderlaufen können. Alle binary_sensor-device_classes
    hier (door/window/opening/lock/light) folgen HAs eigener Konvention "off
    = der gute Zustand" (zu/verriegelt/kein Licht) - ein einziges Prüfmuster
    für alle Zeilen, kein Sonderfall je Art."""
    eintraege: list[dict] = []
    for pos, sensor in zip(POSITIONEN, werte.TUER_SENSOREN):
        w = zustand_oder_none(hass, sensor)
        eintraege.append({"label": f"Tür {pos}", "ok": None if w is None else w == "off", "gruppe": "tueren_klappen"})
    for pos, sensor in zip(POSITIONEN, werte.FENSTER_SENSOREN):
        w = zustand_oder_none(hass, sensor)
        eintraege.append({"label": f"Fenster {pos}", "ok": None if w is None else w == "off", "gruppe": "fenster_dach"})
    # "Kofferraum" statt "Heckklappe": alltagssprachlicher Name für dieselbe
    # Öffnung beim Avant/Kombi - das Sensorfeld HECKKLAPPE_SENSOR selbst
    # bleibt unverändert, um eine bestehende Zuordnung nicht zu brechen.
    for label, sensor in (
        ("Kofferraum", werte.HECKKLAPPE_SENSOR),
        ("Motorhaube", werte.HAUBE_SENSOR),
    ):
        w = zustand_oder_none(hass, sensor)
        eintraege.append({"label": label, "ok": None if w is None else w == "off", "gruppe": "tueren_klappen"})
    # Nur die Fahrertür: die Zentralverriegelung schließt alle Türen gemeinsam,
    # ihr Zustand gilt fürs ganze Fahrzeug (siehe TUERSCHLOSS_SENSOR).
    w = zustand_oder_none(hass, werte.TUERSCHLOSS_SENSOR)
    eintraege.append({"label": "Türschloss", "ok": None if w is None else w == "off", "gruppe": "verriegelt"})
    w = zustand_oder_none(hass, werte.DACH_SENSOR)
    eintraege.append({"label": "Dach", "ok": None if w is None else w == "off", "gruppe": "fenster_dach"})
    w = zustand_oder_none(hass, werte.LICHT_SENSOR)
    eintraege.append({"label": "Standlicht", "ok": None if w is None else w == "off", "gruppe": "licht"})
    return eintraege


def fahrzeugstatus(
    hass: HomeAssistant,
    werte: Sensorzuordnung,
    geparkt_seit: datetime.datetime | None = None,
) -> dict:
    """Bündelt die live aus Home Assistant gelesenen Fahrzeugwerte
    frontend-freundlich, damit die Oberfläche keine Entity-IDs kennen muss."""
    sicherheitscheck = _sicherheitscheck(hass, werte)
    # "Sicher abgestellt": erst wenn WIRKLICH jeder einzeln geprüfte Punkt
    # zu/verriegelt ist, gilt das Fahrzeug als gesichert; fehlt auch nur eine
    # Meldung, ist der Status unbekannt statt geraten.
    if any(e["ok"] is None for e in sicherheitscheck):
        gesichert = None
    else:
        gesichert = all(e["ok"] for e in sicherheitscheck)

    standort = _standort(hass, werte)

    # Verzögerter Import (statt oben am Dateikopf): batterie.py importiert
    # seinerseits zustand_oder_none aus diesem Modul - ein Import auf
    # Modulebene würde bei bestimmter Ladereihenfolge zu einem
    # Circular-Import-Fehler führen. Zum Zeitpunkt dieses Funktionsaufrufs
    # sind beide Module immer schon vollständig geladen.
    from .batterie import SPANNUNG_MIN_V

    spannung = _zu_zahl(zustand_oder_none(hass, werte.BATTERIE_SENSOR))
    if spannung is not None and spannung < SPANNUNG_MIN_V:
        # Dieselbe Plausibilitätsgrenze wie beim Aufzeichnen (batterie.py) -
        # sonst zeigt die "Zustand"-Kachel einen Wert, der im Messwertverlauf
        # nie auftaucht, weil er dort schon verworfen wird.
        spannung = None
    if spannung is not None:
        _letzte_gueltige_spannung["wert"] = spannung
    else:
        # Nutzerwunsch: statt "unbekannt" die zuletzt gemessene Spannung
        # zeigen, wenn der Sensor gerade nichts Plausibles liefert (z. B.
        # Dongle offline, meldet dann oft 0 V) - None bleibt nur, solange
        # noch nie ein gültiger Wert beobachtet wurde.
        spannung = _letzte_gueltige_spannung["wert"]

    return {
        "km": _zu_zahl(zustand_oder_none(hass, werte.KM_SENSOR)),
        "tankprozent": _zu_zahl(zustand_oder_none(hass, werte.TANK_SENSOR)),
        "reichweite_km": _zu_zahl(zustand_oder_none(hass, werte.RANGE_SENSOR)),
        "batteriespannung": spannung,
        "gesichert": gesichert,
        "sicherheitscheck": sicherheitscheck,
        # Fährt das Fahrzeug gerade? Kommt aus derselben Zündungs-Entität, die
        # auch die Fahrterkennung als maßgebliches Signal für Fahrtbeginn und
        # -ende nimmt - damit sagen Anzeige und Erfassung zwangsläufig
        # dasselbe. Vorher wurde dieser Zustand im Frontend aus dem Status der
        # Fahrten abgeleitet ("offen" = fährt). Das war falsch: "offen" heißt
        # unvollständige Daten, nicht "unterwegs" - eine Fahrt ohne
        # Kilometerstand blieb dauerhaft "offen" und das Fahrzeug damit
        # dauerhaft "fahrend". None bedeutet: kein Zündungssensor zugeordnet.
        "zuendung": _zu_bool(zustand_oder_none(hass, werte.ZUENDUNG_SENSOR)),
        # Seit wann das Fahrzeug steht - der Zeitpunkt, zu dem die Zündung
        # zuletzt ausging, in der Zeit des Geräts. None heißt ausdrücklich
        # "unbekannt" und ist von den Oberflächen auch so anzuzeigen: die
        # frühere Herleitung aus der jüngsten Fahrt der Liste sah zwar immer
        # nach einer Antwort aus, war aber eine über die Fahrterkennung und
        # nicht über das Parken (siehe koordinator._zuendung_angezeigt).
        "geparkt_seit": geparkt_seit.isoformat() if geparkt_seit else None,
        "standort_lat": standort["lat"],
        "standort_lon": standort["lon"],
        "standort_genauigkeit_m": standort["genauigkeit_m"],
        "standort_zeit": standort["zeit"],
        # Vom Fahrzeug selbst gemeldete Service-Fälligkeit (ergänzt die
        # App-eigene Servicebuch-Berechnung) - die Streckensensoren liefern
        # negative Restkilometer-Werte, hier deshalb der Betrag.
        "oelwechsel_faellig_ts": zustand_oder_none(hass, werte.NAECHSTER_OELWECHSEL_SENSOR),
        "oelwechsel_faellig_km": _abs_zahl(zustand_oder_none(hass, werte.OELWECHSEL_STRECKE_SENSOR)),
        "inspektion_faellig_ts": zustand_oder_none(hass, werte.NAECHSTE_INSPEKTION_SENSOR),
        "inspektion_faellig_km": _abs_zahl(zustand_oder_none(hass, werte.INSPEKTION_STRECKE_SENSOR)),
    }