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