"""Fahrterkennung über den Zündungs-/ACC-Status des FMM003 (§7.1). Start: die Zündungs-Entität (einstellungen.ZUENDUNG_SENSOR) wechselt auf "on". Ende: sie wechselt auf "off" und kommt binnen der Pausenzeit nicht zurück. 2026-08-12: ersetzt die frühere WLAN-basierte Erkennung (Vergleich des vom iPhone gemeldeten verbundenen WLANs gegen den im Fahrzeugprofil hinterlegten Namen) - die Zündung ist ein direkteres, von jeder Companion-App-Eigenheit unabhängiges Signal. Die Pausenregel wird über task.unique() umgesetzt: jede neue Änderung der Zündungs-Entität killt eine noch laufende, wartende Bestätigung des vorherigen Ereignisses. Kommt die Zündung während der Wartezeit zurück, wird dadurch die wartende Ende-Bestätigung automatisch abgebrochen, ohne dass die Fahrt beendet wird - das entspricht der Pausenregel aus §7.1. Die Entity-IDs stehen zentral in pyscript/modules/einstellungen.py - dort vor der Installation einmalig anpassen (siehe INSTALL.md), oder über das Setup-Menü in der Oberfläche zuordnen (siehe entitaeten.py; Änderungen dort wirken für DIESEN Trigger allerdings erst nach einem Neustart, da @state_trigger die Entity-ID einmalig beim Modul-Laden einliest). EINSCHRÄNKUNG: der Zustand einer laufenden Fahrt (_fahrt_start_ts) lebt nur im Arbeitsspeicher dieses Skripts. Ein HA-Neustart mitten in einer Fahrt oder während der Wartezeit verliert diesen Zwischenstand - die Fahrt würde dann nicht aufgezeichnet. Das ist ein bekannter Punkt für eine spätere Härtung (z. B. Zwischenstand in einer input_datetime-Helper-Entität sichern), hier bewusst nicht mitgebaut, um die Kernlogik nicht zu verkomplizieren, bevor sie sich im echten Betrieb bewährt hat. """ import datetime import einstellungen import fahrtabschluss_logik import frontend_veroeffentlichung import profil _fahrt_start_ts = None def _pausenzeit_sekunden(): p = profil.profil_lesen() if p is None: return 15 * 60 return p.get("einstellungen", {}).get("fahrten_pausenzeit_min", 15) * 60 @time_trigger("startup") def beim_start(): profil.ordner_sicherstellen() # Ohne konfigurierte Zündungs-Entität keinen Trigger registrieren - eine # leere Entity-ID als @state_trigger-Ausdruck ist ungetestetes Verhalten # (siehe zustand_oder_none()-Kommentar in frontend_veroeffentlichung.py zur # Grundhaltung "kein geratener Wert"), hier stattdessen ganz weggelassen. if einstellungen.ZUENDUNG_SENSOR: @state_trigger(f"{einstellungen.ZUENDUNG_SENSOR}") def zuendung_geaendert(value=None, old_value=None): # Killt eine noch laufende Ende-Bestätigung aus einer vorherigen # Änderung - das ist der Mechanismus hinter der Pausenregel. task.unique("audi_dashboard_fahrt_ende_pruefung") global _fahrt_start_ts an_jetzt = (value == "on") an_vorher = (old_value == "on") if an_jetzt and not an_vorher: if _fahrt_start_ts is None: _fahrt_start_ts = datetime.datetime.now(datetime.timezone.utc) log.info(f"audi_dashboard: Fahrt gestartet um {_fahrt_start_ts.isoformat()}") return if an_vorher and not an_jetzt and _fahrt_start_ts is not None: start_ts = _fahrt_start_ts abbruch_ts = datetime.datetime.now(datetime.timezone.utc) wartezeit_s = _pausenzeit_sekunden() # task.unique() oben markiert diese Ausführung als die aktuell # gültige. Kommt die Zündung vor Ablauf der Wartezeit zurück, # ruft der nächste Trigger erneut task.unique() auf und killt # genau diesen task.sleep() - der Code danach läuft dann nie. task.sleep(wartezeit_s) _fahrt_beenden(start_ts, abbruch_ts) def _fahrt_beenden(start_ts, ende_ts): global _fahrt_start_ts dauer_s = int((ende_ts - start_ts).total_seconds()) fahrt = { "trip_id": profil.neue_id("t"), "ts_start": start_ts.isoformat(), "ts_end": ende_ts.isoformat(), "duration_s": dauer_s, "distance_km": None, "km_quelle": None, "odo_start": None, "odo_end": None, "avg_speed_kmh": 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": "ha", "status": "offen", "edited_fields": [], } profil.fahrt_anhaengen(fahrt) frontend_veroeffentlichung.fahrten_veroeffentlichen() log.info(f"audi_dashboard: Fahrt {fahrt['trip_id']} beendet, Status 'offen', Screening angestoßen") _fahrt_start_ts = None fahrtabschluss_logik.screening_durchfuehren(einstellungen.KM_SENSOR) def _handfelder(start_ort, ziel_ort, odo_start, odo_end, distanz): """Die von Hand befüllbaren Felder einer Fahrt 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 (§7.2), 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, } return {k: v for k, v in werte.items() if v is not None} def _strecke_ergaenzen(felder): """Fehlt die Distanz, ergibt sie sich aus den beiden Kilometerständen - dieselbe Rechnung wie im Screening (fahrtabschluss_logik).""" 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 @service def audi_dashboard_fahrt_manuell_anlegen(ts_start=None, ts_end=None, art="privat", start_ort=None, ziel_ort=None, odo_start=None, odo_end=None, distanz=None): """Manuelle Fahrterfassung aus der Oberfläche (§5.3), unabhängig von der automatischen Zündungs-Erkennung. Dieselben Felder wie beim Bearbeiten einer bestehenden Fahrt (siehe audi_dashboard_fahrt_aktualisieren); alles außer Start- und Endzeitpunkt ist optional und bleibt leer, wenn es das Screening später selbst ergänzen soll. Aufruf als pyscript.audi_dashboard_fahrt_manuell_anlegen.""" start_ts = datetime.datetime.fromisoformat(ts_start) ende_ts = datetime.datetime.fromisoformat(ts_end) dauer_s = int((ende_ts - start_ts).total_seconds()) hand = _strecke_ergaenzen(_handfelder(start_ort, ziel_ort, odo_start, odo_end, distanz)) fahrt = { "trip_id": profil.neue_id("t"), "ts_start": start_ts.isoformat(), "ts_end": ende_ts.isoformat(), "duration_s": dauer_s, "distance_km": None, "km_quelle": None, "odo_start": None, "odo_end": None, "avg_speed_kmh": None, "start_lat": None, "start_lon": None, "end_lat": None, "end_lon": None, "start_address": None, "end_address": None, "art": art, "route": None, "pausen": [], "source": "manual", "status": "offen", "edited_fields": [], } fahrt.update(hand) fahrt["edited_fields"] = sorted(hand) if fahrt["distance_km"] is not None: fahrt["km_quelle"] = "manuell" fahrt["status"] = "vollständig" profil.fahrt_anhaengen(fahrt) frontend_veroeffentlichung.fahrten_veroeffentlichen() fahrtabschluss_logik.screening_durchfuehren(einstellungen.KM_SENSOR) @service def audi_dashboard_fahrt_aktualisieren(trip_id=None, ts_start=None, ts_end=None, art=None, start_ort=None, ziel_ort=None, odo_start=None, odo_end=None, distanz=None): """Bearbeitet eine bestehende Fahrt (Einzelfahrt-Seite), egal ob sie automatisch erkannt oder von Hand angelegt wurde. Aufruf als pyscript.audi_dashboard_fahrt_aktualisieren.""" if not trip_id: log.warning("audi_dashboard: fahrt_aktualisieren ohne trip_id aufgerufen") return aenderungen = _strecke_ergaenzen(_handfelder(start_ort, ziel_ort, odo_start, odo_end, distanz)) if art: aenderungen["art"] = art if ts_start and ts_end: start_ts = datetime.datetime.fromisoformat(ts_start) ende_ts = datetime.datetime.fromisoformat(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. # Bewusst eine Listen-Abstraktion statt next(): pyscript kann # keine Generator-Ausdrücke auswerten ("not implemented ast # ast_generatorexp"), Listen-Abstraktionen dagegen schon. passend = [f for f in profil.fahrten_lesen() if f.get("trip_id") == trip_id] handfelder = {"start_address", "end_address", "odo_start", "odo_end", "distance_km"} bisher = set(passend[0].get("edited_fields", [])) if passend else set() aenderungen["edited_fields"] = sorted(bisher | (handfelder & set(aenderungen))) if profil.fahrt_bearbeiten(trip_id, aenderungen): frontend_veroeffentlichung.fahrten_veroeffentlichen() log.info(f"audi_dashboard: Fahrt {trip_id} aktualisiert") else: log.warning(f"audi_dashboard: Fahrt {trip_id} nicht gefunden") @service def audi_dashboard_fahrt_loeschen(trip_id=None): """Löscht eine Fahrt unwiderruflich (Löschfunktion in der Oberfläche, §Löschfunktion). Aufruf als pyscript.audi_dashboard_fahrt_loeschen.""" if not trip_id: log.warning("audi_dashboard: fahrt_loeschen ohne trip_id aufgerufen") return if profil.fahrt_loeschen(trip_id): frontend_veroeffentlichung.fahrten_veroeffentlichen() log.info(f"audi_dashboard: Fahrt {trip_id} gelöscht") else: log.warning(f"audi_dashboard: Fahrt {trip_id} nicht gefunden")