audi-app/custom_components/audi_dashboard/geraet.py

"""Der letzte Datensatz des Dongles, so wie er in Home Assistant angekommen ist.

WOFUER
------
Fuer die Fehlersuche am Fahrzeug, nicht fuer den taeglichen Gebrauch: Steht das
Fahrzeug wirklich, oder ist nur die Meldung alt? Liefert das Geraet gerade
ueberhaupt CAN-Werte? Die Antwort steht in Home Assistants Zustaenden, aber
verteilt auf siebzig Entitaeten - hier zusammengefasst an einer Stelle, die
beide Oberflaechen ohne eigene Registerabfrage lesen koennen.

WAS "EIN DATENSATZ" IST
-----------------------
Die flespi-Integration setzt die Entitaeten EINES Geraetedatensatzes
nacheinander, aber innerhalb weniger Millisekunden (am 01.09.2026 gemessen:
Zuendung 07:35:23.633, zugehoerige Meldezeit 07:35:23.635 - siehe Abschnitt AY
der AGENTS.md). Der letzte Datensatz ist deshalb: alle Entitaeten des Geraets,
deren letzte Aktualisierung hoechstens FENSTER_S nach der juengsten liegt.

Das ist bewusst eine Naeherung ueber die ANKUNFTSzeit. Die Zeit des Geraets
steht daneben (aus MELDEZEIT_SENSOR), und genau der Abstand zwischen beiden ist
die Angabe, die bei gepufferten Datensaetzen interessiert - er war die Wurzel
des 3-km-Fahrt-Fehlers (Abschnitt CB).

WELCHES GERAET
--------------
Nicht geraten: genommen wird das Geraet der Entitaet, die MELDEZEIT_SENSOR
zugeordnet ist, sonst das des Fahrtsignals. Ist keines von beiden zugeordnet,
gibt es nichts zu zeigen - dann bleibt der Rueckgabewert None, statt
irgendein Geraet der Instanz zu waehlen.
"""

from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any

from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers import entity_registry as er

from . import verlauf

if TYPE_CHECKING:
    from .einstellungen import Sensorzuordnung

_LOGGER = logging.getLogger(__name__)

# Wie weit ein Wert hinter dem juengsten liegen darf, um noch zum selben
# Datensatz gezaehlt zu werden. Gemessen wurden Millisekunden; zwei Sekunden
# sind grosszuegig genug fuer eine langsame Instanz und immer noch weit von der
# kuerzesten Sendeperiode des Geraets entfernt (10 s beim Fahren).
FENSTER_S = 2.0

# Ein Datensatz des FMM003 traegt je nach Konfiguration einige Dutzend Werte.
# Die Obergrenze ist kein fachliches Limit, sondern verhindert, dass eine
# fehlerhafte Zuordnung die veroeffentlichten Attribute aufblaeht.
MAX_WERTE = 80


def _meldezeit(hass: HomeAssistant, werte: Sensorzuordnung) -> str | None:
    """Die Zeit, die das Geraet dem letzten Datensatz gegeben hat."""
    if not werte.MELDEZEIT_SENSOR:
        return None
    zustand = hass.states.get(werte.MELDEZEIT_SENSOR)
    if zustand is None:
        return None
    try:
        gemeldet = datetime.fromtimestamp(float(zustand.state), tz=timezone.utc)
    except (TypeError, ValueError, OSError, OverflowError):
        return None
    # Dieselbe Pruefung wie in der Fahrterkennung: eine Sensor-ID statt eines
    # Zeitstempels ergibt sonst einen Zeitpunkt kurz nach 1970 (Abschnitt BO).
    if not verlauf.geraetezeit_plausibel(gemeldet, datetime.now(timezone.utc)):
        return None
    return gemeldet.isoformat()


def _geraet_id(hass: HomeAssistant, werte: Sensorzuordnung) -> str | None:
    register = er.async_get(hass)
    for eid in (werte.MELDEZEIT_SENSOR, werte.ZUENDUNG_SENSOR):
        if not eid:
            continue
        eintrag = register.async_get(eid)
        if eintrag and eintrag.device_id:
            return eintrag.device_id
    return None


def letzter_datensatz(hass: HomeAssistant, werte: Sensorzuordnung) -> dict[str, Any] | None:
    """Die Werte der juengsten Meldung des Dongles - None, wenn keine da ist."""
    geraet_id = _geraet_id(hass, werte)
    if not geraet_id:
        return None

    register = er.async_get(hass)
    eintraege = er.async_entries_for_device(register, geraet_id, include_disabled_entities=False)
    zustaende = []
    for eintrag in eintraege:
        zustand = hass.states.get(eintrag.entity_id)
        # unavailable/unknown tragen keinen Wert - sie stehen fuer eine
        # Entitaet, die das Geraet in diesem Datensatz nicht gemeldet hat.
        if zustand is None or zustand.state in ("unavailable", "unknown"):
            continue
        zustaende.append(zustand)
    if not zustaende:
        return None

    juengste = max(z.last_updated for z in zustaende)
    grenze = juengste - timedelta(seconds=FENSTER_S)
    im_datensatz = sorted(
        (z for z in zustaende if z.last_updated >= grenze),
        key=lambda z: z.attributes.get("friendly_name") or z.entity_id,
    )

    geraete_register = dr.async_get(hass)
    geraet = geraete_register.async_get(geraet_id)
    name = (geraet.name if geraet else None) or geraet_id

    def kurz(voll: str) -> str:
        """Der Geraetename steht schon in der Kopfzeile - hier nur das Signal."""
        rest = voll[len(name):].strip() if voll.startswith(name) else voll
        return rest or voll

    return {
        "geraet": name,
        "angekommen": juengste.isoformat(),
        # Die Zeit, die das GERAET dem Datensatz gegeben hat. Der Abstand zur
        # Ankunft ist der Puffer-Rueckstand.
        "meldezeit": _meldezeit(hass, werte),
        "werte": [
            {
                "name": kurz(z.attributes.get("friendly_name") or z.entity_id),
                "wert": z.state,
                "einheit": z.attributes.get("unit_of_measurement"),
            }
            for z in im_datensatz[:MAX_WERTE]
        ],
        "gesamt": len(im_datensatz),
    }