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 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 MINDESTDAUER_S, verlauf_lesen, zeit_normalisiert

if TYPE_CHECKING:
    from .koordinator import Koordinator

_LOGGER = logging.getLogger(__name__)

# Die frühere Pausenregel (STANDARD_PAUSENZEIT_MIN = 15) ist am 2026-08-31
# entfallen. Der FMM003 wendet sie selbst an - in seiner Konfiguration unter
# "Trip  Odometer" steht "Ignition OFF Timeout" auf 900 Sekunden. Eine
# zweite Wartezeit hier oben drauf hätte daraus 30 Minuten gemacht.
#
# Zurückholen, falls die Erkennung je wieder ohne das Gerät auskommen muss:
# eine abbrechbare Wartezeit vor fahrt_beenden() legen (k.warte_ende_ab /
# k.warte_ende_ab_abbrechen im Koordinator sind dafür noch da) und die
# Einstellung fahrten_pausenzeit_min in beiden Oberflächen wieder anbieten.


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,
        "vmax_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 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:
        # Ohne Wartezeit: der FMM003 bringt die Pausenregel selbst mit
        # (Trip  Odometer -> "Ignition OFF Timeout" = 900 s). Was hier
        # ankommt, ist bereits eine beendete Fahrt - noch einmal 15 Minuten
        # zu warten haette daraus eine halbe Stunde gemacht und zwei
        # aufeinanderfolgende Fahrten verschmolzen, die das Geraet sauber
        # getrennt hatte.
        await fahrt_beenden(k, k.fahrt_start_ts, datetime.datetime.now(datetime.UTC))

    await k.fahrzeugstatus_veroeffentlichen()



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:
    dauer_s = int((ende_ts - start_ts).total_seconds())
    if dauer_s < MINDESTDAUER_S:
        # Zündung an, aber nicht gefahren - oder ein prellender
        # Zündungseingang. Der Import verwirft solche Fenster seit jeher;
        # hier taten wir es nicht, und der Bestand füllte sich mit
        # Nulleinträgen (siehe MINDESTDAUER_S in verlauf.py). Der laufende
        # Start wird trotzdem zurückgesetzt, sonst hinge er für immer.
        await k.fahrt_start_setzen(None)
        _LOGGER.info(
            "Zündung war nur %s s an - keine Fahrt angelegt (Grenze %s s)",
            dauer_s,
            MINDESTDAUER_S,
        )
        return
    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)