"""Zweistufiger Fahrtabschluss: offene Fahrten nachträglich mit Start- und Endkilometerstand vervollständigen (SPECIFICATION.md §7.2). Der Kilometerstand kommt laut Datenquelle nicht sicher mit Fahrtende, sondern teils erst mit Beginn oder während der nächsten Fahrt. Statt auf einen festen Zeitpunkt zu warten, wird deshalb der aufgezeichnete Verlauf des Kilometerstand-Sensors gelesen und daraus der Stand zu Fahrtbeginn bzw. -ende bestimmt - mit wert_bei(), also dem zuletzt DAVOR gemeldeten Wert. Bis zum 31.08.2026 stand hier naechster_wert(): der zeitlich nächstgelegene Wert, gleich ob davor oder danach. Für einen Zählerstand ist das falsch, und wert_bei()s eigener Docstring sagt es wörtlich - der Kilometerstand bei Fahrtbeginn ist der zuletzt gemeldete, nicht der nächste, der schon Strecke enthält. Der Rückblick (historienimport.py) hat immer wert_bei() benutzt; dieselbe Fahrt wurde also je nach Weg unterschiedlich bewertet - genau das, was die gemeinsame Herkunft von UNPLAUSIBLE_KMH und MINDESTDAUER_S in verlauf.py ausschließt. Läuft nach jedem Fahrtende und zusätzlich bei jeder Änderung des Kilometerstand-Sensors - unabhängig vom Fahrtende-Ereignis selbst, eben weil der Wert auch erst mit der nächsten Fahrt eintreffen kann. Dazu einmal nach dem Start von Home Assistant, und danach so lange erneut, wie ein Lauf mehr Ortsauflösungen bräuchte, als sein Budget hergibt (_nachfassen_planen). VORBEHALT: Der GPS-Fallback (km_quelle="gps") ist weiterhin nicht umgesetzt, da die Adressauflösung ein eigenes Arbeitspaket ist. Fahrten ohne passenden Kilometerstand bleiben "offen" statt auf GPS-Distanz auszuweichen. """ from __future__ import annotations import datetime import logging from typing import TYPE_CHECKING from homeassistant.helpers.event import async_call_later from . import reifen, verbrauchskorrektur from .verlauf import ( UNPLAUSIBLE_KMH, durchschnitt_kmh, hoechstwert_im_fenster, strecke_waehlen, zaehlerstrecke, naechster_wert, route_aus_verlauf, verbrauch_aus_literstaenden, verlauf_lesen, wert_ab, aufzeichnungsluecke_km, wert_bei, ) if TYPE_CHECKING: from .koordinator import Koordinator _LOGGER = logging.getLogger(__name__) # Wie lange nach einem Lauf gewartet wird, dessen Ortsbudget nicht gereicht # hat. Siehe _nachfassen_planen(). NACHFASSEN_S = 60 # Wie weit vor/zurück gesucht wird, ausgehend von Fahrtbeginn bzw. -ende. # 48h deckt auch länger stehende Fahrzeuge ab, ohne bei jedem Screening die # komplette Historie zu laden. SUCHFENSTER = datetime.timedelta(hours=48) def _als_zeit(wert: str | None) -> datetime.datetime | None: if not wert: return None try: return datetime.datetime.fromisoformat(wert) except ValueError: return None def _vollstaendig(aenderungen: dict, fahrt: dict) -> dict: """Ergänzt Strecke, Quelle und Status, sobald beide Kilometerstände da sind. Eine Fahrt ist genau dann "vollständig", wenn ihre Strecke feststeht - "offen" heißt nicht "unterwegs", sondern "Daten fehlen noch". Verwirft eine physikalisch unplausible Durchschnittsgeschwindigkeit (UNPLAUSIBLE_KMH) statt sie anzuzeigen, statt der bislang unbehandelten Lücke gegenüber historienimport.py: eine falsch abgegrenzte Fahrt (z. B. durch ein chatterndes Zündungssignal) kann sonst einen Kilometerstand- Sprung wie "22 km in 67 s" als echte Strecke zeigen. Sowohl odo_start als auch odo_end werden dann verworfen (nicht nur die Strecke) - die Fahrt bleibt "offen" und wird beim nächsten Screening erneut versucht, statt einen unmöglichen Wert dauerhaft zu speichern.""" if fahrt.get("odo_start") is not None and fahrt.get("odo_end") is not None: distanz = round(fahrt["odo_end"] - fahrt["odo_start"], 1) if distanz < 0: # Ein Zähler läuft nicht rückwärts - der Wert stammt also nicht # aus derselben Quelle wie sein Gegenstück. Während der Testphase # wandert der Dongle zwischen Fahrzeugen; am 31.08.2026 meldete der # CAN-Kilometerstand deshalb 21325 zwischen 209177 und 209267. # # Die Prüfung vergleicht Anfang und Ende DERSELBEN Fahrt und steht # dem Fahrzeugwechsel nicht im Weg: jedes Fahrzeug zählt innerhalb # seiner eigenen Fahrt aufwärts. Sie greift nur, wenn die beiden # Enden aus verschiedenen Fahrzeugen stammen. # # Warum an der Strecke und nicht an der Geschwindigkeit: die # Plausibilitätsgrenze darunter prüft nur nach oben, und # durchschnitt_kmh() liefert bei Dauer 0 None - eine Fahrt mit null # Sekunden Dauer käme also an einer Geschwindigkeitsprüfung vorbei. # # historienimport.py verlangt seit jeher odo_end >= odo_start. _LOGGER.warning( "Fahrt %s: Kilometerstand verworfen - Rückwärtssprung (%s -> %s, " "%s km), vermutlich ein Fahrzeugwechsel des Geräts.", fahrt.get("trip_id"), fahrt["odo_start"], fahrt["odo_end"], distanz, ) aenderungen["odo_start"] = None aenderungen["odo_end"] = None return aenderungen # Nur der aufgezeichnete Teil darf gegen die Dauer geprueft werden. # Die verschlafenen Kilometer sind zwar gefahren worden, aber in einer # Zeit, die NICHT in duration_s steckt - der Beginn liegt ja erst dort, # wo das Geraet aufgewacht ist. Beides zusammen ergaebe eine # Phantomgeschwindigkeit; am 06.09.2026 hat genau das eine Fahrt aus # der Testphase verworfen (siehe _aufzeichnungsluecken_schliessen). gemessen = round(distanz - (fahrt.get("luecke_km") or 0), 1) geschwindigkeit = durchschnitt_kmh(gemessen, fahrt.get("duration_s")) if geschwindigkeit is not None and geschwindigkeit > UNPLAUSIBLE_KMH: _LOGGER.warning( "Fahrt %s: Kilometerstand verworfen - unmögliche %s km/h im " "Schnitt (%s km in %s s), vermutlich ein Ausreißer im " "Verlauf oder eine falsch abgegrenzte Fahrt.", fahrt.get("trip_id"), geschwindigkeit, gemessen, fahrt.get("duration_s"), ) aenderungen["odo_start"] = None aenderungen["odo_end"] = None return aenderungen aenderungen["distance_km"] = distanz aenderungen["km_quelle"] = "odometer" aenderungen["status"] = "vollständig" return aenderungen async def durchfuehren(k: Koordinator) -> None: """Versucht, alle offenen Fahrten zu vervollständigen, und trägt allen Fahrten ohne Standortdaten Start-/Zielposition sowie eine Streckenlinie nach - unabhängig vom Kilometerstand-Status, da beide Datenquellen unabhängig voneinander verspätet eintreffen können.""" km_sensor = k.zuordnung.werte.KM_SENSOR fahrten = await k.ablage.fahrten_lesen() if not fahrten: return offene = [f for f in fahrten if f.get("status") == "offen"] if km_sensor: offene.sort(key=lambda f: f.get("ts_start", "")) for fahrt in offene: await _fahrt_screenen(k, km_sensor, fahrt) lat_sensor = k.zuordnung.werte.STANDORT_LAT_SENSOR lon_sensor = k.zuordnung.werte.STANDORT_LON_SENSOR if lat_sensor and lon_sensor: ohne_position = [ f for f in fahrten if f.get("start_lat") is None or f.get("end_lat") is None or f.get("route") is None ] for fahrt in ohne_position: await _position_screenen(k, lat_sensor, lon_sensor, fahrt) # Erst NACH der Positionsermittlung: eine Fahrt, die gerade eben ihre # Koordinaten bekommen hat, soll im selben Durchlauf auch ihren Ort # bekommen und nicht erst beim naechsten. k.geokodierung.durchlauf_beginnen() ohne_ort = [ f for f in fahrten if (f.get("start_lat") is not None and not f.get("start_stadt")) or (f.get("end_lat") is not None and not f.get("end_stadt")) ] # Neueste zuerst: die sieht der Nutzer als naechstes. Ist das Budget # aufgebraucht, warten die alten auf den naechsten Durchlauf. ohne_ort.sort(key=lambda f: f.get("ts_start") or "", reverse=True) for fahrt in ohne_ort: if k.geokodierung.rest <= 0: break await _orte_screenen(k, fahrt) # Durchschnitt braucht keinen Sensor - er folgt aus Strecke und Dauer, # die beide schon dastehen. Deshalb vor dem Geschwindigkeitssensor und # unabhaengig davon, ob einer zugeordnet ist. for fahrt in fahrten: if fahrt.get("luecke_km"): # Strecke und Dauer decken hier verschiedene Zeitraeume ab: die # Strecke enthaelt die verschlafenen Kilometer, die Dauer nicht. # Ein Quotient daraus waere eine Zahl ohne Bedeutung - die Fahrt # vom 05.09.2026 kaeme auf 92 km/h fuer eine 33-km/h-Fahrt. # "Liegt nicht vor" ist die einzige ehrliche Angabe. if fahrt.get("avg_speed_kmh") is not None: await k.ablage.fahrt_aktualisieren( fahrt["trip_id"], {"avg_speed_kmh": None} ) fahrt["avg_speed_kmh"] = None continue if fahrt.get("avg_speed_kmh") is not None: continue schnitt = durchschnitt_kmh(fahrt.get("distance_km"), fahrt.get("duration_s")) if schnitt is None: continue await k.ablage.fahrt_aktualisieren( fahrt["trip_id"], {"avg_speed_kmh": schnitt} ) fahrt["avg_speed_kmh"] = schnitt # Vor der Verfeinerung: eine Fahrt mit Aufzeichnungsluecke bekommt hier # ihren richtigen Startkilometerstand, und die Verfeinerung darf sie # danach nicht mehr anfassen. await _aufzeichnungsluecken_schliessen(k, fahrten) # Erst nach den Kilometerstaenden: die Verfeinerung braucht die grobe # Strecke als Anker und laeuft nur auf Fahrten, die sie schon haben. gnss_sensor = k.zuordnung.werte.GNSS_KM_SENSOR if gnss_sensor: for fahrt in fahrten: if fahrt.get("km_quelle") == "odometer": await _gnss_verfeinern(k, gnss_sensor, fahrt) tempo_sensor = k.zuordnung.werte.GESCHWINDIGKEIT_SENSOR if tempo_sensor: ohne_vmax = [f for f in fahrten if f.get("vmax_kmh") is None] for fahrt in ohne_vmax: await _vmax_screenen(k, tempo_sensor, fahrt) tank_liter_sensor = k.zuordnung.werte.TANK_LITER_SENSOR if tank_liter_sensor: ohne_verbrauch = [ f for f in fahrten if f.get("verbrauch_roh_l_100km") is None and f.get("distance_km") ] for fahrt in ohne_verbrauch: await _verbrauch_screenen(k, tank_liter_sensor, fahrt) # Zuletzt: die Eichung braucht die Rohwerte aller Fahrten eines # Betankungsintervalls und ist damit auf alle Schritte davor angewiesen. await verbrauchskorrektur.anwenden(k) # Die Laufleistung der Raeder ist die Summe der Fahrten - sie muss also # nachziehen, sobald eine Strecke dazukommt oder sich aendert. await reifen.nachtragen(k) await _lueckenlos_verketten(k) await k.fahrten_veroeffentlichen() _nachfassen_planen(k) def _nachfassen_planen(k: Koordinator) -> None: """Plant einen zweiten Lauf, wenn dieser hier nicht gereicht hat. Das Ortsbudget sind acht Anfragen je Lauf (PRO_DURCHLAUF in geokodierung.py, damit Nominatims Bitte um keinen Vorratsabruf gewahrt bleibt). Eine einzelne Fahrt braucht zwei davon - im Normalbetrieb ist das Budget also nie das Ende. Aufgebraucht wird es genau in zwei Lagen: nach einem Rueckblick-Import aus Home Assistant, der viele Fahrten auf einmal anlegt, und einmalig nach einem Update, das ein neues Feld einfuehrt. Genau dafuer und nur dafuer ist dieses Nachfassen da - kein Dauertakt. Bedingung ist BEIDES: das Budget ist leer (es liegt also noch Arbeit an) UND es ist in diesem Lauf mindestens ein Ort dazugekommen (die Gegenstelle antwortet also). Kommt ein Lauf mit uebrigem Budget durch, ist alles aufgeloest und es wird nichts mehr eingeplant - die Kette endet von selbst. Antwortet Nominatim gerade nicht, endet sie ebenfalls, statt im Minutentakt gegen eine Wand zu laufen; der naechste natuerliche Ausloeser (Fahrtende, Kilometerstand, Neustart) versucht es wieder. """ if k.screening_nachlauf is not None: return if k.geokodierung.rest > 0 or k.geokodierung.neu <= 0: return async def _spaeter(_jetzt) -> None: k.screening_nachlauf = None await durchfuehren(k) _LOGGER.debug( "Ortsbudget aufgebraucht, %s Orte neu - weiterer Lauf in %s s", k.geokodierung.neu, NACHFASSEN_S, ) k.screening_nachlauf = async_call_later(k.hass, NACHFASSEN_S, _spaeter) async def _fahrt_screenen(k: Koordinator, km_sensor: str, fahrt: dict) -> None: start = _als_zeit(fahrt.get("ts_start")) ende = _als_zeit(fahrt.get("ts_end")) if start is None or ende is None: _LOGGER.warning( "Fahrt %s hat keinen lesbaren Zeitraum - Screening übersprungen", fahrt.get("trip_id"), ) return # Ein Abruf für beide Zeitpunkte statt zwei: die Fenster überlappen sich # bei einer normalen Fahrt ohnehin fast vollständig. punkte = await verlauf_lesen( k.hass, km_sensor, start - SUCHFENSTER, ende + SUCHFENSTER, k.zuordnung.werte.MELDEZEIT_SENSOR ) if not punkte: return # wert_bei() statt naechster_wert(): der zuletzt DAVOR gemeldete Stand, nicht # der zeitlich nächstgelegene. Siehe Modulkopf - für einen Zähler ist das die # richtige Wahl, und der Rückblick in historienimport.py macht es genauso. # Steht davor nichts, nimmt wert_bei() den ersten Wert danach. geaendert = False if fahrt.get("odo_end") is None: wert = wert_bei(punkte, ende) if wert is not None: fahrt["odo_end"] = wert geaendert = True if fahrt.get("odo_start") is None: wert = wert_bei(punkte, start) if wert is not None: fahrt["odo_start"] = wert geaendert = True if geaendert: aenderungen = {"odo_start": fahrt.get("odo_start"), "odo_end": fahrt.get("odo_end")} aenderungen = _vollstaendig(aenderungen, fahrt) await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], aenderungen) # Das In-Memory-Objekt sofort mitziehen (nicht nur die Ablage) - sonst # sehen spätere Schritte innerhalb desselben durchfuehren()-Laufs # (z. B. _verbrauch_screenen(), das eine bekannte distance_km # braucht) noch den alten Stand, obwohl die Strecke gerade erst in # diesem Durchlauf ermittelt wurde. _position_screenen() macht das # bereits genauso. fahrt.update(aenderungen) async def _gnss_verfeinern(k: Koordinator, gnss_sensor: str, fahrt: dict) -> None: """Ersetzt die auf ganze Kilometer gerundete Strecke durch die metergenaue des Geraets - aber nur, wenn beide zueinander passen. Der Kilometerstand des Fahrzeugs bleibt der Anker: er ist der Tacho, er driftet nicht, und er ist die Zahl, die der Fahrer im Cockpit sieht. Nur loest er eben ganze Kilometer auf. Der GNSS-Zaehler des Geraets loest auf einen Meter auf, kann aber driften und Empfang verlieren. Deshalb keine Ersetzung, sondern eine Pruefung: die feine Zahl gilt, solange sie innerhalb der Rundungsunschaerfe des groben Ankers liegt (GNSS_TOLERANZ_KM). Liegt sie darueber, hat der Anker recht und die feine Zahl wird verworfen - dann fehlte Empfang, oder der Zaehler ist gedriftet. Laeuft nur auf Fahrten mit km_quelle == \"odometer\", also genau einmal je Fahrt: danach steht dort \"odometer+gnss\" (oder es blieb bei \"odometer\", wenn nichts Brauchbares vorlag, und ein spaeterer Lauf versucht es erneut, sobald mehr Verlauf da ist).""" start = _als_zeit(fahrt.get("ts_start")) ende = _als_zeit(fahrt.get("ts_end")) grob = fahrt.get("distance_km") if start is None or ende is None or grob is None: return # Etwas vor dem Fahrtbeginn mitlesen, damit zaehlerstrecke() ihren Anker # findet - den letzten Datensatz am oder vor dem Beginn. punkte = await verlauf_lesen( k.hass, gnss_sensor, start - SUCHFENSTER, ende, k.zuordnung.werte.MELDEZEIT_SENSOR ) fein = zaehlerstrecke(punkte, start, ende) # Bei einer Aufzeichnungsluecke hat der GNSS-Zaehler nur den Teil gesehen, # in dem das Geraet wach war - seine Zahl waere allein also systematisch zu # klein, und bei einer kleinen Luecke laege sie sogar noch innerhalb von # GNSS_TOLERANZ_KM und kassierte die gerade erst geholte Strecke wieder. # # Deshalb keine Sperre, sondern eine Zerlegung: der verschlafene Anfang in # ganzen Kilometern aus dem Tacho, der aufgezeichnete Rest metergenau aus # dem Zaehler. Die Unschaerfe bleibt damit im verschlafenen Stueck statt in # der ganzen Fahrt (Eigentuemer am 06.09.2026: "das macht die gesamte # Nutzbarkeit des Fahrtenbuchs sehr grob"). Die Summe wird unten wie sonst # auch gegen den Tacho geprueft - passt sie nicht, gilt weiter der Tacho. luecke = fahrt.get("luecke_km") if fein is not None and luecke: fein = round(fein + luecke, 3) gewaehlt, genommen = strecke_waehlen(grob, fein) if not genommen: if fein is not None: _LOGGER.info( "Fahrt %s: GNSS-Strecke %s km passt nicht zum Kilometerstand " "(%s km) - der Fahrzeugwert bleibt stehen", fahrt.get("trip_id"), fein, grob, ) return aenderungen = {"distance_km": gewaehlt, "km_quelle": "odometer+gnss"} await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], aenderungen) fahrt.update(aenderungen) _LOGGER.info( "Fahrt %s: Strecke auf %s km verfeinert (Kilometerstand sagte %s km)%s", fahrt.get("trip_id"), gewaehlt, grob, f", davon {luecke} km unaufgezeichnet" if luecke else "", ) async def _position_screenen( k: Koordinator, lat_sensor: str, lon_sensor: str, fahrt: dict ) -> None: """Trägt Start-/Zielposition und eine echte Streckenlinie nach, sobald der GPS-Verlauf sie hergibt - unabhängig vom Kilometerstand-Status. Dieselbe "Fahrt gehört der Vergangenheit an" wie bei den Kilometerständen: kein Live-Zustand zum Fahrtbeginn/-ende wird herangezogen (der wäre bei einer Funklücke des Trackers unzuverlässig, siehe fahrterkennung.py), nur der aufgezeichnete Verlauf im Fahrtfenster.""" start = _als_zeit(fahrt.get("ts_start")) ende = _als_zeit(fahrt.get("ts_end")) if start is None or ende is None: return lat_verlauf = await verlauf_lesen( k.hass, lat_sensor, start - SUCHFENSTER, ende + SUCHFENSTER, k.zuordnung.werte.MELDEZEIT_SENSOR ) lon_verlauf = await verlauf_lesen( k.hass, lon_sensor, start - SUCHFENSTER, ende + SUCHFENSTER, k.zuordnung.werte.MELDEZEIT_SENSOR ) if not lat_verlauf or not lon_verlauf: return aenderungen: dict = {} if fahrt.get("start_lat") is None: lat = naechster_wert(start, lat_verlauf) lon = naechster_wert(start, lon_verlauf) if lat is not None and lon is not None: aenderungen["start_lat"] = lat aenderungen["start_lon"] = lon if fahrt.get("end_lat") is None: lat = naechster_wert(ende, lat_verlauf) lon = naechster_wert(ende, lon_verlauf) if lat is not None and lon is not None: aenderungen["end_lat"] = lat aenderungen["end_lon"] = lon if fahrt.get("route") is None: route = route_aus_verlauf(lat_verlauf, lon_verlauf, start, ende) if route is not None: aenderungen["route"] = route if aenderungen: await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], aenderungen) fahrt.update(aenderungen) async def _orte_screenen(k: Koordinator, fahrt: dict) -> None: """Traegt Start- und Zielort nach, sobald die Koordinaten dastehen. Der Ort gehoert zu den Fahrtdaten, nicht in den Zwischenspeicher jedes einzelnen Geraets - Begruendung im Kopf von geokodierung.py. start_address/end_address stehen in fahrterkennung.HANDFELDER: eine von Hand eingetragene Anschrift ("Zuhause") darf hier nie ueberschrieben werden. Zwei Sperren dagegen, absichtlich beide: hier wird nur ein LEERES Feld gefuellt, und fahrt_aktualisieren() laesst ohnehin jedes Feld in edited_fields unberuehrt. Der Ortsname (start_stadt) wird auch dann gesetzt, wenn die Anschrift von Hand stammt - er beschreibt die Koordinate, nicht die Eingabe, und die Liste braucht ihn zum Ausrichten der Spalten.""" enden = ( ("start", fahrt.get("start_lat"), fahrt.get("start_lon")), ("end", fahrt.get("end_lat"), fahrt.get("end_lon")), ) aenderungen: dict = {} for seite, lat, lon in enden: if lat is None or lon is None: continue if fahrt.get(f"{seite}_address") and fahrt.get(f"{seite}_stadt"): continue ort = await k.geokodierung.aufloesen(lat, lon) if ort is None: continue if not fahrt.get(f"{seite}_address") and ort.get("anschrift"): aenderungen[f"{seite}_address"] = ort["anschrift"] if not fahrt.get(f"{seite}_stadt") and ort.get("stadt"): aenderungen[f"{seite}_stadt"] = ort["stadt"] if aenderungen: await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], aenderungen) fahrt.update(aenderungen) async def _verbrauch_screenen( k: Koordinator, tank_liter_sensor: str, fahrt: dict ) -> None: """Trägt den Verbrauch (l/100 km) nach, sobald die Distanz feststeht und der Tankfüllstand (Liter) das hergibt - siehe verbrauch_aus_literstaenden() für die Näherung selbst und ihre Begrenzung.""" start = _als_zeit(fahrt.get("ts_start")) ende = _als_zeit(fahrt.get("ts_end")) if start is None or ende is None: return verlauf = await verlauf_lesen( k.hass, tank_liter_sensor, start - SUCHFENSTER, ende + SUCHFENSTER, k.zuordnung.werte.MELDEZEIT_SENSOR ) if not verlauf: return # Anders als beim Kilometerstand: fuer den Tankstand gilt der erste Wert AB # dem Fahrtbeginn, nicht der letzte davor. Ein Fuellstand ist keine # Zaehlung, er driftet im Stand - siehe wert_ab() in verlauf.py fuer die # Messung, die das gezeigt hat. Das Ende bleibt bei wert_bei(). verbrauch = verbrauch_aus_literstaenden( wert_ab(verlauf, start), wert_bei(verlauf, ende), fahrt.get("distance_km") ) if verbrauch is None: return # Der Rohwert wird immer geschrieben, der Anzeigewert nur, solange noch # kein Faktor vorliegt - sonst ueberschriebe das Screening eine bereits # an den Betankungen geeichte Zahl mit der ungeeichten. aenderungen = {"verbrauch_roh_l_100km": verbrauch} faktor = fahrt.get("verbrauch_faktor") aenderungen["verbrauch_l_100km"] = ( round(verbrauch * float(faktor), 1) if faktor else verbrauch ) await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], aenderungen) fahrt.update(aenderungen) async def _vmax_screenen(k: Koordinator, tempo_sensor: str, fahrt: dict) -> None: """Trägt die Höchstgeschwindigkeit der Fahrt nach. Anders als beim Kilometerstand wird hier nicht der nächstgelegene Wert gesucht, sondern der größte innerhalb des Fahrtfensters - und ohne Suchfenster drumherum, weil ein Wert von vor oder nach der Fahrt nicht zu ihr gehört. Unplausible Werte werden verworfen statt angezeigt: dieselbe Grenze wie beim Durchschnitt (UNPLAUSIBLE_KMH), denn ein Sensorsprung sieht in beiden Fällen gleich aus.""" start = _als_zeit(fahrt.get("ts_start")) ende = _als_zeit(fahrt.get("ts_end")) if start is None or ende is None: return verlauf = await verlauf_lesen(k.hass, tempo_sensor, start, ende, k.zuordnung.werte.MELDEZEIT_SENSOR) if not verlauf: return vmax = hoechstwert_im_fenster(verlauf, start, ende) if vmax is None or vmax <= 0 or vmax > UNPLAUSIBLE_KMH: return await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], {"vmax_kmh": vmax}) fahrt["vmax_kmh"] = vmax # Wie viele Kilometer eine Aufzeichnungsluecke hoechstens haben darf, um noch # als solche zu gelten. # # Die Luecke entsteht, waehrend das Geraet nach dem Motorstart noch schlaeft. # Gemessen ueber alle Aufwachvorgaenge vom 18.08. bis 05.09.2026 dauert das # zwischen 28 s und 1170 s; die laengste gemessene Verzoegerung sind also # 19,5 Minuten, bei Autobahntempo rund 42 km. 50 km lassen das durch und # sperren alles, was keine Weckverzoegerung mehr sein kann. # # Ohne diese Grenze griff die Regel am 06.09.2026 sofort daneben: sie fand # 1078 km zwischen zwei Fahrten vom 27.08. (dort fehlen schlicht Fahrten in der # Liste) und 187831 km am 29.08. (dort sprang der Kilometerstand zwischen den # beiden Zahlenbaendern des Geraets, siehe Abschnitt zum Fahrzeugwechsel in # _vollstaendig). Beide Male ist die Differenz echt, aber sie gehoert nicht an # den Anfang der naechsten Fahrt. LUECKE_MAX_KM = 50.0 async def _aufzeichnungsluecken_schliessen( k: Koordinator, fahrten: list[dict] ) -> None: """Holt die Kilometer zurueck, die das Geraet am Anfang einer Fahrt verschlafen hat. Das Screening nimmt als Startkilometerstand den zuletzt VOR dem Fahrtbeginn gemeldeten Wert (wert_bei). Beginnt die Fahrt aber erst in dem Moment, in dem sich das Geraet zurueckmeldet, ist dieser Wert schon der Stand NACH den verschlafenen Kilometern - am 05.09.2026 stand dort 21353 statt 21347, und die Fahrt haette 1 km statt 7 km bekommen. Der ehrliche Anker ist der Endkilometerstand der Vorgaengerin: dazwischen stand das Fahrzeug. Die Strecke stimmt danach auf ganze Kilometer genau - mehr gibt der CAN-Tacho nicht her, und die Route der verschlafenen Minuten ist unwiederbringlich weg. luecke_km haelt fest, wie viel davon betroffen ist, damit die Oberflaeche es zeigen kann und die GNSS-Verfeinerung die Finger davon laesst.""" sortiert = sorted(fahrten, key=lambda f: f.get("ts_start") or "") for vorherige, fahrt in zip(sortiert, sortiert[1:]): fehlt = aufzeichnungsluecke_km( vorherige.get("odo_end"), fahrt.get("odo_start") ) if fehlt is None: continue if fehlt > LUECKE_MAX_KM: _LOGGER.info( "Fahrt %s: %s km Unterschied zur Vorgaengerin - zu viel fuer " "eine Weckverzoegerung (Grenze %s km), bleibt unangetastet", fahrt.get("trip_id"), fehlt, LUECKE_MAX_KM, ) continue # Der Anker ist nur so gut wie das ENDE der Vorgaengerin. Wurde die von # Hand beendet, steht ihr Endkilometerstand auf einem vorlaeufigen Ende # - dann koennen die Kilometer auch ihr gehoeren und nur falsch # zugeschlagen sein. Die Gesamtleistung stimmt in beiden Faellen, die # Aufteilung zwischen zwei Fahrten nicht (Einwand des Eigentuemers, # 06.09.2026). Deshalb wird es vermerkt statt still verrechnet. alt_start = fahrt.get("odo_start") alt_quelle = fahrt.get("km_quelle") fahrt.update({ "luecke_km": fehlt, "luecke_unsicher": bool(vorherige.get("ende_vorlaeufig")), "odo_start": vorherige["odo_end"], "km_quelle": "odometer", }) aenderungen = _vollstaendig( { "luecke_km": fehlt, "luecke_unsicher": fahrt["luecke_unsicher"], "odo_start": fahrt["odo_start"], "odo_end": fahrt.get("odo_end"), }, fahrt, ) # Letzte Sicherung: _vollstaendig() verwirft beide Kilometerstaende, # wenn die Strecke daraus unplausibel wird (Rueckwaertssprung oder # unmoegliche Durchschnittsgeschwindigkeit). Passiert das durch UNSERE # Korrektur, war die Annahme falsch, dass die beiden Fahrten wirklich # aufeinanderfolgen - dann darf sie den Bestand nicht schlechter # zuruecklassen, als sie ihn vorgefunden hat. Am 06.09.2026 genau so # eingetreten, an einer Fahrt aus der Testphase vom 27.08. if aenderungen.get("odo_start") is None: fahrt["odo_start"] = alt_start fahrt["km_quelle"] = alt_quelle fahrt["luecke_km"] = None fahrt["luecke_unsicher"] = False _LOGGER.info( "Fahrt %s: %s km Unterschied zur Vorgaengerin ergaeben eine " "unplausible Strecke - Korrektur verworfen, Bestand bleibt", fahrt.get("trip_id"), fehlt, ) continue await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], aenderungen) fahrt.update(aenderungen) _LOGGER.info( "Fahrt %s: %s km am Anfang nicht aufgezeichnet - Startkilometer " "von der Vorgaengerin uebernommen (%s statt %s)", fahrt.get("trip_id"), fehlt, vorherige["odo_end"], round(vorherige["odo_end"] + fehlt, 1), ) async def _lueckenlos_verketten(k: Koordinator) -> None: """Fahrt-Ende(N) == Fahrt-Start(N+1), solange dazwischen nicht gefahren wurde. Zwischen zwei in der Liste direkt aufeinanderfolgenden Fahrten steht das Fahrzeug nur - der Kilometerstand ändert sich dabei nicht, unabhängig davon, wie lange die Standzeit dauert. Ein bestätigter Wert wird deshalb an den direkten Nachbarn weitergereicht, auch wenn dessen eigenes Screening noch keinen Treffer hatte. Annahme: die Fahrten-Liste enthält keine Lücken durch verpasste Ereignisse. Das ist der Normalfall, aber keine Garantie.""" fahrten = sorted(await k.ablage.fahrten_lesen(), key=lambda f: f.get("ts_start", "")) geaendert: set[str] = set() for aktuelle, naechste in zip(fahrten, fahrten[1:]): if aktuelle.get("odo_end") is not None and naechste.get("odo_start") is None: naechste["odo_start"] = aktuelle["odo_end"] geaendert.add(naechste["trip_id"]) elif naechste.get("odo_start") is not None and aktuelle.get("odo_end") is None: aktuelle["odo_end"] = naechste["odo_start"] geaendert.add(aktuelle["trip_id"]) for fahrt in fahrten: if fahrt["trip_id"] not in geaendert: continue aenderungen = {"odo_start": fahrt.get("odo_start"), "odo_end": fahrt.get("odo_end")} await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], _vollstaendig(aenderungen, fahrt))