#!/usr/bin/env python3 """Die Wartezeit nach dem Zuendungs-Aus und der Nachlauf des Geraets. Warum es das gibt (03.09.2026): das Trip-Signal des FMM003 endet erst nach seinem "Ignition OFF Timeout" von 900 s - und der Schlaf-Timeout des Geraets stand auf denselben 900 s. Beide Uhren starten beim Zuendungs-Aus und fallen in derselben Sekunde. Am 02.09.2026 ist das zweimal beobachtet worden, mit entgegengesetztem Ausgang: die Fahrt um 16:54 verlor ihr Trip-Ende (danach 1 h 48 min ohne einen einzigen Datensatz), die um 21:43 bekam es um 21:58:16, genau 901 s spaeter. Ein Wettlauf, den unsere Seite nicht gewinnen kann. Deshalb loest jetzt die Zuendung aus, und die Wartezeit laeuft bei uns - auf einer Maschine, die nicht schlaeft. Geprueft wird genau das: dass eine zurueckkehrende Zuendung die Fahrt fortsetzt statt sie zu teilen, dass die Nachlaeufe des Geraets abgezogen werden, und dass ein spaet eintreffendes Aus keine Phantomfahrt erzeugt. Aufruf: python3 tests/fahrterkennung/test_zuendungspause.py (braucht das homeassistant-Paket, weil die Integration es importiert - also z. B. im Test-Container) """ import asyncio import datetime import os import sys import unittest _HIER = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.dirname(os.path.dirname(_HIER))) from custom_components.audi_dashboard import fahrterkennung as f # noqa: E402 T0 = datetime.datetime(2026, 9, 2, 16, 32, 18, tzinfo=datetime.UTC) AUS = datetime.datetime(2026, 9, 2, 16, 54, 21, tzinfo=datetime.UTC) # Zwei verschiedene Dinge, seit die Pausenregel in GERAETEZEIT rechnet: # # PAUSE_S ist das Fenster, in dem eine zurueckkehrende Zuendung noch dieselbe # Fahrt ist - gemessen an den Zeitstempeln des Geraets. Es muss realistisch # sein, sonst prueft der Test etwas anderes als der Betrieb tut. # # WARTEN ist die Wanduhrzeit, die der Test tatsaechlich absitzt. Sie darf kurz # sein; ob abgebrochen wird, haengt nicht an ihrer Laenge. PAUSE_S = 900 WARTEN = 0.15 class FakeWerte: def __init__(self): self.ZUENDUNG_SENSOR = "binary_sensor.zuendung" # Ohne Meldezeit-Sensor keine Geraetezeit und keine Klemme - die Tests # pruefen die Nachlaufrechnung, nicht die Zuordnung. self.MELDEZEIT_SENSOR = "" class FakeZuordnung: def __init__(self): self.werte = FakeWerte() class FakeAblage: def __init__(self, minuten=15, fahrten=None): self.profil = {"einstellungen": {"fahrten_pausenzeit_min": minuten}} self.fahrten = fahrten or [] async def profil_lesen(self): return self.profil async def fahrten_lesen(self): return self.fahrten class FakeKoordinator: def __init__(self, start_ts=T0, minuten=15, fahrten=None): self.fahrt_start_ts = start_ts self.fahrt_sperre = asyncio.Lock() self.zuordnung = FakeZuordnung() self.ablage = FakeAblage(minuten, fahrten) self.beendet = [] self._aufgabe = None self.ende_wartet = None def warte_ende_ab(self, koroutine, wartet=None): self.warte_ende_ab_abbrechen() self.ende_wartet = wartet self._aufgabe = asyncio.ensure_future(koroutine) def warte_ende_ab_abbrechen(self): if self._aufgabe is not None and not self._aufgabe.done(): self._aufgabe.cancel() self._aufgabe = None self.ende_wartet = None async def fahrt_start_setzen(self, ts): self.fahrt_start_ts = ts async def fahrzeugstatus_veroeffentlichen(self): pass async def ausklingen(self): """Der wartenden Aufgabe Zeit geben, fertig zu werden.""" if self._aufgabe is None: return try: await asyncio.wait_for(asyncio.shield(self._aufgabe), timeout=2) except (asyncio.CancelledError, asyncio.TimeoutError): pass class Basis(unittest.IsolatedAsyncioTestCase): def _pause_kurz(self): async def pausenzeit(k): return PAUSE_S return pausenzeit async def asyncSetUp(self): self._echt = (f._geraetezeit, f.fahrt_beenden, f._pausenzeit_s) async def geraetezeit(k, ereigniszeit, standard): return ereigniszeit or standard async def beenden(k, start_ts, ende_ts): k.beendet.append((start_ts, ende_ts)) await k.fahrt_start_setzen(None) async def nach_pause(k, start_ts, ende_ts, pause_s): # Wie das Original, nur mit kurzer Wanduhrzeit - die echte # Pausenzeit steckt weiterhin in der Entscheidung, nicht im Warten. await asyncio.sleep(WARTEN) await asyncio.shield(f._jetzt_schliessen(k, start_ts, ende_ts)) f._geraetezeit = geraetezeit f.fahrt_beenden = beenden f._pausenzeit_s = self._pause_kurz() self._echt_nach_pause = f._nach_pause_beenden f._nach_pause_beenden = nach_pause async def asyncTearDown(self): (f._geraetezeit, f.fahrt_beenden, f._pausenzeit_s) = self._echt f._nach_pause_beenden = self._echt_nach_pause class Wartezeit(Basis): async def test_aus_schliesst_nicht_sofort(self): k = FakeKoordinator() await f.zuendung_geaendert(k, "off", "on", AUS) self.assertEqual( k.beendet, [], "waehrend der Wartezeit darf nichts geschrieben sein" ) self.assertIsNotNone(k.fahrt_start_ts) await k.ausklingen() self.assertEqual(len(k.beendet), 1, "nach der Wartezeit muss die Fahrt enden") async def test_zuendung_kommt_zurueck_und_die_fahrt_laeuft_weiter(self): """Der eigentliche Zweck der Wartezeit.""" k = FakeKoordinator() await f.zuendung_geaendert(k, "off", "on", AUS) await asyncio.sleep(WARTEN / 3) await f.zuendung_geaendert(k, "on", "off", AUS + datetime.timedelta(seconds=20)) await asyncio.sleep(WARTEN * 2) self.assertEqual(k.beendet, [], "eine zurueckgekehrte Zuendung beendet nichts") self.assertEqual(k.fahrt_start_ts, T0, "es bleibt dieselbe Fahrt") async def test_start_stopp_zerlegt_die_fahrt_nicht(self): """Der RS4 stellt an der Ampel den Motor ab - aus einer Drehzahlquelle ist das nicht vom Abstellen zu unterscheiden.""" k = FakeKoordinator() t = AUS for _ in range(4): await f.zuendung_geaendert(k, "off", "on", t) await asyncio.sleep(WARTEN / 5) t += datetime.timedelta(seconds=45) await f.zuendung_geaendert(k, "on", "off", t) t += datetime.timedelta(seconds=200) self.assertEqual(k.beendet, []) self.assertEqual(k.fahrt_start_ts, T0) class Nachlauf(Basis): async def test_zuendungsweg_zieht_180_s_ab(self): k = FakeKoordinator() await f.zuendung_geaendert(k, "off", "on", AUS) await k.ausklingen() _, ende = k.beendet[0] self.assertEqual(ende, AUS - datetime.timedelta(seconds=180)) async def test_zu_kurz_wird_auf_den_beginn_geklemmt(self): k = FakeKoordinator(start_ts=AUS - datetime.timedelta(seconds=60)) await f.zuendung_geaendert(k, "off", "on", AUS) await k.ausklingen() start, ende = k.beendet[0] self.assertEqual(ende, start, "kuerzer als der Nachlauf heisst: nicht gefahren") class SpaetesAus(Basis): async def test_aus_zur_geschlossenen_fahrt_wird_verworfen(self): """Gepuffert eintreffendes Aus, dessen Geraetezeit in die schon geschlossene Fahrt faellt - es darf nichts ausloesen.""" ende = AUS - datetime.timedelta(seconds=180) k = FakeKoordinator( start_ts=None, fahrten=[{"ts_start": T0.isoformat(), "ts_end": ende.isoformat()}], ) await f.zuendung_geaendert(k, "off", "on", AUS) self.assertEqual(k.beendet, []) self.assertIsNone(k.fahrt_start_ts, "es darf keine Fahrt entstehen") async def test_aus_ohne_jede_fahrt_meldet_sich(self): k = FakeKoordinator(start_ts=None, fahrten=[]) with self.assertLogs(f._LOGGER, level="WARNING") as protokoll: await f.zuendung_geaendert(k, "off", "on", AUS) self.assertTrue( any("entgangen" in z for z in protokoll.output), "ein Aus ohne passende Fahrt gehoert ins Protokoll", ) class GepufferterStapel(Basis): """Der Fehler vom 04.09.2026: zwei Fahrten wurden zu einer. Das Geraet war fuenf Stunden ohne Verbindung. Als es sich meldete, kamen das Zuendungs-Aus der ersten Fahrt und das Zuendungs-Ein der zweiten in DERSELBEN Sekunde bei Home Assistant an (23:36:50). Ihre Geraetezeiten lagen 5 Stunden 12 auseinander: 18:23:32 und 23:35:43. Nach der Ankunft gerechnet sieht das aus wie eine Zuendung, die sofort zurueckkommt - also dieselbe Fahrt. Genau so wurde gerechnet, und dabei ist aus zwei Fahrten eine geworden: 18:10 bis 23:42, 5 Stunden 33, ohne Strecke. Die zweite Fahrt gab es gar nicht. Die Ankunftszeit spielt hier bewusst KEINE Rolle - beide Aufrufe folgen unmittelbar aufeinander, so wie sie es an dem Abend taten. """ async def test_zwei_fahrten_bleiben_zwei(self): beginn1 = datetime.datetime(2026, 9, 4, 16, 10, 27, tzinfo=datetime.UTC) aus1 = datetime.datetime(2026, 9, 4, 18, 23, 32, tzinfo=datetime.UTC) an2 = datetime.datetime(2026, 9, 4, 21, 35, 43, tzinfo=datetime.UTC) k = FakeKoordinator(start_ts=beginn1) await f.zuendung_geaendert(k, "off", "on", aus1) await f.zuendung_geaendert(k, "on", "off", an2) self.assertEqual( len(k.beendet), 1, "die erste Fahrt muss geschlossen worden sein" ) start, ende = k.beendet[0] self.assertEqual(start, beginn1) self.assertEqual( ende, aus1 - datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S), "geschlossen wird auf das Zuendungs-Aus minus Nachlauf", ) self.assertEqual( k.fahrt_start_ts, an2, "und die zweite Fahrt beginnt mit ihrem eigenen an" ) async def test_kurze_rueckkehr_bleibt_eine_fahrt(self): """Die Gegenprobe - sonst wuerde jeder Tankstopp die Fahrt zerreissen. Derselbe Ablauf, nur liegen die Geraetezeiten acht Minuten auseinander statt fuenf Stunden. """ beginn = datetime.datetime(2026, 9, 4, 16, 10, 27, tzinfo=datetime.UTC) aus = datetime.datetime(2026, 9, 4, 18, 23, 32, tzinfo=datetime.UTC) an = aus + datetime.timedelta(minutes=8) k = FakeKoordinator(start_ts=beginn) await f.zuendung_geaendert(k, "off", "on", aus) await f.zuendung_geaendert(k, "on", "off", an) self.assertEqual(k.beendet, [], "innerhalb der Pause wird nichts geschlossen") self.assertEqual(k.fahrt_start_ts, beginn, "es bleibt dieselbe Fahrt") async def test_genau_an_der_grenze_bleibt_es_eine_fahrt(self): """Die Pausenzeit gehoert noch zur Fahrt, nicht schon zur naechsten. Gemessen wird ab dem ECHTEN Ende (Zuendungs-Aus minus Nachlauf), nicht ab dem Zuendungssignal - sonst waere die Grenze um 180 s verschoben. """ beginn = datetime.datetime(2026, 9, 4, 16, 10, 27, tzinfo=datetime.UTC) aus = datetime.datetime(2026, 9, 4, 18, 23, 32, tzinfo=datetime.UTC) echtes_ende = aus - datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S) k = FakeKoordinator(start_ts=beginn) await f.zuendung_geaendert(k, "off", "on", aus) await f.zuendung_geaendert( k, "on", "off", echtes_ende + datetime.timedelta(seconds=PAUSE_S) ) self.assertEqual(k.beendet, [], "genau auf der Grenze noch dieselbe Fahrt") k2 = FakeKoordinator(start_ts=beginn) await f.zuendung_geaendert(k2, "off", "on", aus) await f.zuendung_geaendert( k2, "on", "off", echtes_ende + datetime.timedelta(seconds=PAUSE_S + 1) ) self.assertEqual(len(k2.beendet), 1, "eine Sekunde darueber sind es zwei") async def test_ohne_wartendes_ende_wird_nichts_zerschnitten(self): """Ein "an" ohne vorheriges "aus" darf keine Fahrt beenden. Fehlt das Aus ganz - weil es nie ankam -, waere ein Schnitt auf Verdacht schlimmer als eine zu lange Fahrt. Dann passiert nichts, wie bisher. """ beginn = datetime.datetime(2026, 9, 4, 16, 10, 27, tzinfo=datetime.UTC) k = FakeKoordinator(start_ts=beginn) await f.zuendung_geaendert( k, "on", "off", beginn + datetime.timedelta(hours=5) ) self.assertEqual(k.beendet, []) self.assertEqual(k.fahrt_start_ts, beginn) class Neuschreibung(Basis): """Home Assistant schreibt denselben Wert auch neu. Am 04.09.2026 gezaehlt: von 22 Zustandszeilen der Zuendung waren nur 6 echte Wechsel. Die uebrigen 16 waren Neuschreibungen - der Wert selbst hatte sich nicht geruehrt. Der Beobachter unterscheidet das nicht, er reicht neu und alt durch. Faellt so eine Neuschreibung von "aus" in die laufende Wartezeit, begann die Wartezeit von vorn - und bei genuegend vielen nie ein Ende. """ async def test_neuschreibung_verlaengert_die_wartezeit_nicht(self): k = FakeKoordinator() await f.zuendung_geaendert(k, "off", "on", AUS) wartet = k.ende_wartet self.assertIsNotNone(wartet, "nach dem Aus wartet eine Bestaetigung") # Dieselbe Meldung noch einmal, ohne Wertwechsel. await f.zuendung_geaendert(k, "off", "off", AUS + datetime.timedelta(seconds=30)) self.assertIs( k.ende_wartet, wartet, "es wartet weiterhin dieselbe Bestaetigung" ) await k.ausklingen() self.assertEqual(len(k.beendet), 1, "und sie kommt zum Zug") self.assertEqual( k.beendet[0][1], AUS - datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S), "mit dem Ende aus dem ERSTEN Aus, nicht dem der Neuschreibung", ) async def test_echtes_zweites_aus_zaehlt_weiterhin(self): """Die Gegenprobe: ein Wechsel darf die Bestaetigung sehr wohl ersetzen.""" k = FakeKoordinator() await f.zuendung_geaendert(k, "off", "on", AUS) erstes = k.ende_wartet await f.zuendung_geaendert(k, "on", "off", AUS + datetime.timedelta(seconds=30)) await f.zuendung_geaendert(k, "off", "on", AUS + datetime.timedelta(seconds=60)) self.assertIsNotNone(k.ende_wartet) self.assertNotEqual( k.ende_wartet, erstes, "ein echtes zweites Aus setzt ein neues Ende" ) async def test_neuschreibung_ohne_laufende_fahrt_geht_weiter_durch(self): """_spaetes_aus() braucht auch ein off -> off. Ein zweites, spaeter gepuffertes Aus traegt das Ende einer von Hand beendeten Fahrt nach. Ein Filter ganz oben haette genau das verschluckt - deshalb sitzt die Pruefung nur in Zweig 2. """ gesehen = [] async def spaetes_aus(k, ereigniszeit, jetzt): gesehen.append(ereigniszeit) echt = f._spaetes_aus f._spaetes_aus = spaetes_aus try: k = FakeKoordinator(start_ts=None) await f.zuendung_geaendert(k, "off", "off", AUS) self.assertEqual( len(gesehen), 1, "ohne laufende Fahrt muss das Aus ankommen" ) finally: f._spaetes_aus = echt class Einstellung(unittest.IsolatedAsyncioTestCase): """_pausenzeit_s selbst - hier ungepatcht.""" async def test_wert_aus_dem_profil(self): k = FakeKoordinator(minuten=25) self.assertEqual(await f._pausenzeit_s(k), 25 * 60) async def test_unsinn_faellt_auf_den_standard_zurueck(self): k = FakeKoordinator() k.ablage.profil = {"einstellungen": {"fahrten_pausenzeit_min": "viel"}} self.assertEqual(await f._pausenzeit_s(k), f.STANDARD_PAUSENZEIT_MIN * 60) async def test_obergrenze_greift(self): k = FakeKoordinator(minuten=9999) self.assertEqual(await f._pausenzeit_s(k), f.PAUSENZEIT_MAX_MIN * 60) async def test_null_ist_erlaubt(self): """Wer keine Pause will, bekommt keine - dann endet die Fahrt sofort.""" k = FakeKoordinator(minuten=0) self.assertEqual(await f._pausenzeit_s(k), 0) if __name__ == "__main__": unittest.main(verbosity=2)