audi-app/custom_components/audi_dashboard/fahrterkennung.py

"""Fahrterkennung über den Zündungs-/ACC-Status (SPECIFICATION.md §7.1).

Start: die Zündungs-Entität wechselt auf "on". Ende: sie wechselt auf "off"
und kommt binnen der Pausenzeit nicht zurück.

Die Pausenregel läuft über eine abbrechbare Aufgabe: jede neue Änderung der
Zündung bricht eine noch wartende Ende-Bestätigung ab. Kommt die Zündung
während der Wartezeit zurück, wird die Fahrt dadurch nicht beendet - genau
die Pausenregel aus §7.1. (In der pyscript-Fassung leisteten task.unique() und
task.sleep() dasselbe.)

WAS HIER GEGENÜBER DER PYSCRIPT-FASSUNG BESSER IST: dort lebte der Beginn
einer laufenden Fahrt nur im Arbeitsspeicher. Ein HA-Neustart mitten in einer
Fahrt oder während der Wartezeit verlor diesen Zwischenstand - die Fahrt wurde
dann gar nicht aufgezeichnet. Das war als bekannte Schwäche vermerkt und ist
hier behoben: der Zwischenstand liegt in einem Store (siehe koordinator.py),
überlebt einen Neustart und wird beim Hochfahren wieder aufgenommen.
"""

from __future__ import annotations

import asyncio
import datetime
import logging
from typing import TYPE_CHECKING

from . import screening
from .ablage import neue_id
from .veroeffentlichung import zustand_oder_none
from .verlauf import verlauf_lesen, zeit_normalisiert

if TYPE_CHECKING:
    from .koordinator import Koordinator

_LOGGER = logging.getLogger(__name__)

STANDARD_PAUSENZEIT_MIN = 15


def leere_fahrt(start_ts: datetime.datetime, ende_ts: datetime.datetime, quelle: str) -> dict:
    """Das vollständige Fahrt-Schema mit allem, was noch nicht bekannt ist,
    auf None. Bewusst alle Felder ausgeschrieben: die Oberfläche darf sich auf
    ihre Existenz verlassen, und ein fehlendes Feld wäre schwerer zu finden
    als ein leeres."""
    return {
        "trip_id": neue_id("t"),
        "ts_start": start_ts.isoformat(),
        "ts_end": ende_ts.isoformat(),
        "duration_s": int((ende_ts - start_ts).total_seconds()),
        "distance_km": None,
        "km_quelle": None,
        "odo_start": None,
        "odo_end": None,
        "avg_speed_kmh": None,
        "verbrauch_l_100km": None,
        "start_lat": None,
        "start_lon": None,
        "end_lat": None,
        "end_lon": None,
        "start_address": None,
        "end_address": None,
        "art": "privat",
        "route": None,
        "pausen": [],
        "source": quelle,
        "status": "offen",
        "edited_fields": [],
    }


async def pausenzeit_sekunden(k: Koordinator) -> int:
    profil = await k.ablage.profil_lesen()
    if profil is None:
        return STANDARD_PAUSENZEIT_MIN * 60
    minuten = profil.get("einstellungen", {}).get(
        "fahrten_pausenzeit_min", STANDARD_PAUSENZEIT_MIN
    )
    return int(minuten) * 60


async def zuendung_geaendert(k: Koordinator, neu: str | None, alt: str | None) -> None:
    """Reagiert auf jede Zustandsänderung der Zündungs-Entität.

    "unavailable"/"unknown" ist kein Zündungszustand, sondern Funkstille des
    Trackers - beim FMM003 z. B. in Tiefgaragen oder Funklöchern, auch mitten
    in einer laufenden Fahrt. Würde das wie "aus" gewertet, risse jede solche
    Lücke eine einzelne Fahrt in zwei auseinander, sobald sie länger als die
    Pausenzeit dauert (der Tracker meldet sich zurück, die Zündung steht laut
    letztem echten Wert immer noch auf "on", also sieht es wie eine neue
    Fahrt aus). Ohne Signal ist deshalb "keine Aussage", nicht "aus": der
    Zustand bleibt unangetastet, bis wieder ein echter Wert kommt."""
    if neu not in ("on", "off"):
        return

    # Eine noch wartende Ende-Bestätigung aus einer vorherigen Änderung
    # abbrechen - das ist der Mechanismus hinter der Pausenregel.
    k.warte_ende_ab_abbrechen()

    an_jetzt = neu == "on"
    # Ob eine Fahrt läuft, entscheidet der eigene Zwischenstand
    # (fahrt_start_ts) - nicht der vorherige Rohwert der Entität. Der wäre bei
    # einer Funklücke "unavailable" statt "on" gewesen, obwohl die Fahrt die
    # ganze Zeit lief.
    fahrt_laeuft = k.fahrt_start_ts is not None

    if an_jetzt and not fahrt_laeuft:
        await k.fahrt_start_setzen(datetime.datetime.now(datetime.UTC))
        _LOGGER.info("Fahrt gestartet um %s", k.fahrt_start_ts)
        # Die Anzeige "fährt/steht" hängt an derselben Entität - sofort neu
        # veröffentlichen, statt bis zum nächsten 20-Sekunden-Takt zu warten.
        await k.fahrzeugstatus_veroeffentlichen()
        return

    if not an_jetzt and fahrt_laeuft:
        start_ts = k.fahrt_start_ts
        abbruch_ts = datetime.datetime.now(datetime.UTC)
        wartezeit = await pausenzeit_sekunden(k)
        k.warte_ende_ab(_ende_bestaetigen(k, start_ts, abbruch_ts, wartezeit))

    await k.fahrzeugstatus_veroeffentlichen()


async def _ende_bestaetigen(
    k: Koordinator, start_ts: datetime.datetime, ende_ts: datetime.datetime, wartezeit: int
) -> None:
    """Wartet die Pausenzeit ab und beendet die Fahrt dann.

    Kommt die Zündung vorher zurück, bricht zuendung_geaendert() genau diese
    Aufgabe ab und der Code hinter dem sleep läuft nie."""
    try:
        await asyncio.sleep(wartezeit)
    except asyncio.CancelledError:
        _LOGGER.debug("Fahrtende verworfen - Zündung kam innerhalb der Pausenzeit zurück")
        raise
    await fahrt_beenden(k, start_ts, ende_ts)


async def nach_neustart_fortsetzen(k: Koordinator) -> None:
    """Schließt eine Fahrt ab, die während eines HA-Ausfalls zu Ende ging.

    Der Zwischenstand einer laufenden Fahrt überlebt jetzt einen Neustart
    (Store im Koordinator). Damit entsteht aber ein neuer Fall, den es vorher
    nicht gab: ging die Zündung aus, WÄHREND Home Assistant unten war, hat
    niemand die Ende-Bestätigung gesehen - der Beobachter bekommt nach dem
    Hochfahren keinen Wechsel mehr, weil der längst passiert ist. Ohne diese
    Prüfung bliebe die Fahrt für immer offen, und die nächste echte Fahrt
    würde an ihren Beginn angehängt.

    Läuft bewusst erst, wenn Home Assistant vollständig gestartet ist: mitten
    im Einrichten hat die Datenquelle ihren Zustand oft noch nicht gemeldet,
    und ein "noch nicht da" ließe sich nicht von "aus" unterscheiden."""
    if k.fahrt_start_ts is None:
        return

    sensor = k.zuordnung.werte.ZUENDUNG_SENSOR
    if zustand_oder_none(k.hass, sensor) == "on":
        return  # fährt noch - der Beobachter übernimmt wie sonst auch

    ende_ts = await _letztes_lebenszeichen(k, sensor, k.fahrt_start_ts)
    _LOGGER.info(
        "Fahrt seit %s wurde während eines Ausfalls beendet - wird jetzt mit "
        "Ende %s abgeschlossen", k.fahrt_start_ts.isoformat(), ende_ts.isoformat(),
    )
    await fahrt_beenden(k, k.fahrt_start_ts, ende_ts)


async def _letztes_lebenszeichen(
    k: Koordinator, sensor: str, seit: datetime.datetime
) -> datetime.datetime:
    """Der zuletzt aufgezeichnete Zeitpunkt der Zündungs-Entität seit `seit`.

    Das ist der ehrlichste verfügbare Endzeitpunkt: bis dahin wissen wir, dass
    das Fahrzeug beobachtet wurde. "Jetzt" zu nehmen wäre falsch, wenn Home
    Assistant tagelang unten war - daraus würde eine Fahrt über Tage. Gibt es
    gar keine Aufzeichnung (Sensor nicht zugeordnet, recorder zu kurz), bleibt
    nur "jetzt"; das Screening trägt die Kilometerstände ohnehin nach."""
    jetzt = datetime.datetime.now(datetime.UTC)
    punkte = await verlauf_lesen(k.hass, sensor, seit, jetzt)
    return punkte[-1][0] if punkte else jetzt


async def fahrt_beenden(
    k: Koordinator, start_ts: datetime.datetime, ende_ts: datetime.datetime
) -> None:
    fahrt = leere_fahrt(start_ts, ende_ts, "ha")
    await k.ablage.fahrt_anhaengen(fahrt)
    await k.fahrt_start_setzen(None)
    await k.fahrten_veroeffentlichen()
    _LOGGER.info(
        "Fahrt %s beendet, Status 'offen', Screening angestoßen", fahrt["trip_id"]
    )
    await screening.durchfuehren(k)


# ------------------------------------------------------- Aus der Oberfläche

HANDFELDER = {"start_address", "end_address", "odo_start", "odo_end", "distance_km"}


def _handfelder(
    start_ort: str | None,
    ziel_ort: str | None,
    odo_start: float | None,
    odo_end: float | None,
    distanz: float | None,
) -> dict:
    """Die von Hand befüllbaren Felder auf ihre Datensatz-Namen abgebildet,
    ohne die leer gelassenen. Der Aufrufer schreibt die Schlüssel zusätzlich
    nach edited_fields - das schützt sie vor dem automatischen
    Kilometerstand-Screening, das sonst später darüberschreiben würde."""
    werte = {
        "start_address": start_ort or None,
        "end_address": ziel_ort or None,
        "odo_start": odo_start,
        "odo_end": odo_end,
        "distance_km": distanz,
    }
    felder = {k: v for k, v in werte.items() if v is not None}
    # Fehlt die Distanz, ergibt sie sich aus den beiden Kilometerständen -
    # dieselbe Rechnung wie im Screening.
    if "distance_km" not in felder and "odo_start" in felder and "odo_end" in felder:
        felder["distance_km"] = round(felder["odo_end"] - felder["odo_start"], 1)
    return felder


async def manuell_anlegen(k: Koordinator, daten: dict) -> None:
    """Manuelle Fahrterfassung aus der Oberfläche (§5.3), unabhängig von der
    automatischen Zündungs-Erkennung. Alles außer Start- und Endzeitpunkt ist
    optional und bleibt leer, wenn es das Screening später selbst ergänzen
    soll."""
    start_ts = zeit_normalisiert(daten["ts_start"])
    ende_ts = zeit_normalisiert(daten["ts_end"])
    hand = _handfelder(
        daten.get("start_ort"), daten.get("ziel_ort"),
        daten.get("odo_start"), daten.get("odo_end"), daten.get("distanz"),
    )

    fahrt = leere_fahrt(start_ts, ende_ts, "manual")
    fahrt["art"] = daten.get("art") or "privat"
    fahrt.update(hand)
    fahrt["edited_fields"] = sorted(hand)
    if fahrt["distance_km"] is not None:
        fahrt["km_quelle"] = "manuell"
        fahrt["status"] = "vollständig"

    await k.ablage.fahrt_anhaengen(fahrt)
    await k.fahrten_veroeffentlichen()
    await screening.durchfuehren(k)
    _LOGGER.info("Fahrt %s von Hand angelegt", fahrt["trip_id"])


async def aktualisieren(k: Koordinator, daten: dict) -> None:
    """Bearbeitet eine bestehende Fahrt (Einzelfahrt-Seite), egal ob sie
    automatisch erkannt oder von Hand angelegt wurde."""
    trip_id = daten.get("trip_id")
    if not trip_id:
        _LOGGER.warning("fahrt_aktualisieren ohne trip_id aufgerufen")
        return

    aenderungen = _handfelder(
        daten.get("start_ort"), daten.get("ziel_ort"),
        daten.get("odo_start"), daten.get("odo_end"), daten.get("distanz"),
    )
    if daten.get("art"):
        aenderungen["art"] = daten["art"]
    if daten.get("ts_start") and daten.get("ts_end"):
        start_ts = zeit_normalisiert(daten["ts_start"])
        ende_ts = zeit_normalisiert(daten["ts_end"])
        aenderungen["ts_start"] = start_ts.isoformat()
        aenderungen["ts_end"] = ende_ts.isoformat()
        aenderungen["duration_s"] = int((ende_ts - start_ts).total_seconds())
    if aenderungen.get("distance_km") is not None:
        aenderungen["km_quelle"] = "manuell"
        aenderungen["status"] = "vollständig"

    # Von Hand gesetzte Felder bleiben von Hand gesetzt: die Liste wächst mit,
    # damit ein späteres Screening sie nicht wieder überschreibt.
    passend = [f for f in await k.ablage.fahrten_lesen() if f.get("trip_id") == trip_id]
    bisher = set(passend[0].get("edited_fields", [])) if passend else set()
    aenderungen["edited_fields"] = sorted(bisher | (HANDFELDER & set(aenderungen)))

    if await k.ablage.fahrt_bearbeiten(trip_id, aenderungen):
        await k.fahrten_veroeffentlichen()
        _LOGGER.info("Fahrt %s aktualisiert", trip_id)
    else:
        _LOGGER.warning("Fahrt %s nicht gefunden", trip_id)


async def loeschen(k: Koordinator, trip_id: str | None) -> None:
    if not trip_id:
        _LOGGER.warning("fahrt_loeschen ohne trip_id aufgerufen")
        return
    if await k.ablage.fahrt_loeschen(trip_id):
        await k.fahrten_veroeffentlichen()
        _LOGGER.info("Fahrt %s gelöscht", trip_id)
    else:
        _LOGGER.warning("Fahrt %s nicht gefunden", trip_id)