"""Der Verbrauch einer Fahrt, korrigiert an den Tankbelegen. DAS PROBLEM. Der Verbrauch einer einzelnen Fahrt entsteht aus der Differenz zweier Tankstaende. Der Geber liefert zwar Liter (das Steuergeraet hat seine Kennlinie schon angewandt), aber er rundet, und er ist stark gedaempft - sonst zappelte die Anzeige im Cockpit bei jeder Kurve. Auf einer Fahrt von sieben Kilometern faellt dieser Verzug anteilig weit staerker ins Gewicht als auf dreihundert. DIE RECHNUNG, so wie der Eigentuemer sie ohnehin von Hand macht: Faktor = getankte Liter laut Beleg / Summe der Liter aus den Einzelfahrten Die Annahme dahinter ist seine: die Ungenauigkeit des Gebers trifft jede Fahrt gleichermassen, ist also ein Faktor und kein Summand. Beide Seiten decken denselben Zeitraum ab; was der Faktor korrigiert, ist genau der Unterschied zwischen dem, was der Geber gemeldet hat, und dem, was wirklich in den Tank gelaufen ist. WARUM NUR VON VOLLTANKUNG ZU VOLLTANKUNG. Nur dann gilt "eingefuellte Liter = seit dem letzten Mal verbrauchte Liter": eine Betankung, die den Tank wieder bis zum selben Punkt fuellt, ersetzt genau das, was seither entnommen wurde. Nach einer Teilbetankung waere der Zaehler eine beliebige Zahl und der Faktor falsch. Als voll gilt ein Tankvorgang, wenn der Stand danach mindestens VOLL_ANTEIL des Tankvolumens erreicht (aus dem Fahrzeugprofil). DIE LITER MUESSEN AUS DEM BELEG STAMMEN, nicht aus dem Geber. Ein automatisch erkannter Tankvorgang traegt in `liters` den vom Geber gemessenen Anstieg - den gegen die ebenfalls vom Geber gerechneten Fahrten zu stellen, waere ein Zirkelschluss und ergaebe immer ungefaehr 1,0. Gezaehlt wird deshalb nur, was aus einem Beleg geparst oder von Hand eingetragen wurde. STANDLAUF IM STILLSTAND stoert die Rechnung nicht: laeuft der Motor bei eingeschalteter Zuendung, bleibt die Fahrt offen, der Verbrauch steht also in der Summe der Einzelfahrten UND in der getankten Menge und kuerzt sich heraus. Die betroffene Fahrt zeigt dann zu Recht einen hohen Verbrauch. Verzerrend ist nur Verbrauch ohne offene Fahrt - Standheizung bei Zuendung aus; der fehlt im Nenner und verteilt sich ueber den Faktor auf alle Fahrten des Zeitraums. Der Eigentuemer hat das am 01.09.2026 ausdruecklich so gewollt. WAS GEPRUEFT WIRD, UND WARUM NICHT DER FAKTOR SELBST. Die beiden Tore oben sichern den ZAEHLER: die Liter stammen aus einem Beleg, und voll zu voll heisst, dass sie dem Verbrauch entsprechen. Verwundbar bleibt der NENNER - fehlt im Intervall eine Fahrt (Dongle offline, HA unten, Erkennung ausgesetzt), ist die Summe zu klein und der Faktor zu gross. Er multipliziert dann jede Fahrt des Zeitraums mit. Eine feste Ober- und Untergrenze am Faktor (erst 0,7 bis 1,5) traf das Symptom, nicht die Ursache - und haette einen echten, groesseren Geberfehler mit abgeklemmt. Geprueft wird deshalb, ob der Nenner ueberhaupt vollstaendig ist: die Summe der Fahrtstrecken gegen die Tacho-Spanne desselben Intervalls. Deckung = Summe der Fahrtstrecken / (Kilometerstand bis - Kilometerstand von) Das Mass ist belegt: beim Umbau des Radzaehlers am 01.09.2026 lag es in den sauberen Bereichen bei 99 bis 103 %. Liegt es dort, ist der Faktor vertrauenswuerdig - egal wie er ausfaellt, ohne obere oder untere Klemme. Liegt es daneben, beschreibt der Faktor fehlende Daten und nicht den Geber, und das Intervall wird uebersprungen und protokolliert. WAS GESPEICHERT WIRD. verbrauch_roh_l_100km ist die reine Messung und bleibt unangetastet, damit die Rechnung wiederholbar ist. verbrauch_faktor ist der angewandte Faktor. verbrauch_l_100km ist das Produkt aus beiden - der Wert, den die Oberflaechen anzeigen; die brauchen von alldem nichts zu wissen. """ from __future__ import annotations import datetime import logging from typing import TYPE_CHECKING from .tankerkennung import STANDARD_TANKVOLUMEN_LITER from .verlauf import verlauf_lesen, wert_bei if TYPE_CHECKING: from .koordinator import Koordinator _LOGGER = logging.getLogger(__name__) # Wie vollstaendig die erkannten Fahrten die tatsaechlich gefahrene Strecke # des Intervalls abdecken muessen. Der Wert sollte per Definition bei 1,0 # liegen; eine Abweichung in BEIDE Richtungen heisst, dass die Fahrtmenge # nicht zur Wirklichkeit passt - zu wenig (fehlende Fahrten) ebenso wie zu # viel (doppelte oder falsch vermessene Fahrten, im Altbestand real # vorgekommen: 256 %). DECKUNG_MIN = 0.90 DECKUNG_MAX = 1.10 # Wie weit der Kilometerstand eines Tankvorgangs vom Tacho zu seinem eigenen # Zeitstempel abweichen darf. Der Tacho springt in ganzen Kilometern, und # zwischen Zapfsaeule und dem Datensatz, den das Geraet danach schickt, liegen # selten mehr als ein paar - siehe _datum_passt(). ODOMETER_TOLERANZ_KM = 5.0 # Ab welchem Anteil des Tankvolumens ein Tankvorgang als Volltankung gilt. # 95 % laesst Luft fuer die Ungenauigkeit des Gebers und fuer den Umstand, dass # "voll" an der Zapfsaeule nicht auf den Liter reproduzierbar ist. VOLL_ANTEIL = 0.95 # Der gedaempfte Tankstand steht nach dem Tanken nicht sofort. Eine Minute ist # die Vorgabe des Eigentuemers (01.09.2026). # # Zu kurz zu greifen ist hier die harmlose Richtung: liest die Pruefung einen # noch nicht ganz nachgezogenen Stand, faellt er unter die 95-%-Grenze und das # Intervall gilt als "nicht voll" - es entsteht dann kein Faktor. Zu lang zu # warten waere die gefaehrliche Richtung, denn nach der Abfahrt sinkt der Stand # wieder und eine echte Volltankung saehe nachtraeglich unvoll aus. SETZZEIT = datetime.timedelta(minutes=1) # Wie weit ueber die Intervallgrenzen hinaus der Verlauf geholt wird, damit # wert_bei() links vom Fensterrand einen Anker findet. PUFFER = datetime.timedelta(hours=6) def _zeit(wert: str | None) -> datetime.datetime | None: if not wert: return None try: gelesen = datetime.datetime.fromisoformat(str(wert)) except ValueError: return None if gelesen.tzinfo is None: gelesen = gelesen.replace(tzinfo=datetime.UTC) return gelesen def beleg_liter(tankvorgang: dict) -> float | None: """Die getankte Menge, aber nur wenn sie NICHT vom Geber stammt. Ein automatisch erkannter Tankvorgang traegt in `liters` den gemessenen Anstieg des Tankstands. Den gegen die ebenfalls aus dem Tankstand gerechneten Fahrten zu stellen, ergaebe zwangslaeufig ungefaehr 1,0 - der Faktor koennte gar nichts korrigieren. """ liter = tankvorgang.get("liters") if liter is None: return None aus_beleg = bool(tankvorgang.get("receipt_key")) von_hand = "liters" in (tankvorgang.get("edited_fields") or []) manuell = tankvorgang.get("source") == "manual" if not (aus_beleg or von_hand or manuell): return None try: return float(liter) except (TypeError, ValueError): return None def _liter_der_fahrt(fahrt: dict) -> float | None: """Was diese Fahrt laut Messung verbraucht hat, in Litern.""" roh = fahrt.get("verbrauch_roh_l_100km") strecke = fahrt.get("distance_km") if roh is None or not strecke or strecke <= 0: return None return float(roh) / 100.0 * float(strecke) async def _tacho_spanne( k: Koordinator, km_sensor: str, von: datetime.datetime, bis: datetime.datetime, ) -> tuple[float, float] | None: """Die Tachostaende zu Beginn und Ende des Intervalls, oder None. Zwei Dinge haengen daran: die Deckungspruefung (passt die Fahrtmenge zur gefahrenen Strecke?) und die Datumspruefung (passt der eingetragene Zeitpunkt eines Tankvorgangs zu seinem eigenen Kilometerstand?). Der Kilometerstand ist der Tacho des Fahrzeugs: er zaehlt jeden Kilometer, auch die einer nicht erkannten Fahrt. Genau deshalb taugt er als Mass dafuer, ob die Fahrtmenge vollstaendig ist.""" verlauf = await verlauf_lesen(k.hass, km_sensor, von - PUFFER, bis + PUFFER) if not verlauf: return None odo_von = wert_bei(verlauf, von) odo_bis = wert_bei(verlauf, bis) if odo_von is None or odo_bis is None: return None if odo_bis - odo_von <= 0: # Kein Fortschritt oder ein Rueckwaertssprung (Fahrzeugwechsel) - der # Tacho taugt hier nicht als Mass. return None return odo_von, odo_bis def _datum_passt(tankvorgang: dict, tacho: float) -> bool: """Passt der eingetragene Zeitpunkt zum Kilometerstand des Tankvorgangs? Die Fahrten werden ueber die ZEIT zugeordnet, nicht ueber den Tacho. Traegt jemand die Tankung von Dienstag am Freitag mit "jetzt" ein, wandern die Liter auf die Fahrten von Mittwoch bis Freitag - und die Deckungspruefung merkt es nicht, weil sie den Tacho zu denselben verschobenen Zeitpunkten liest: Zaehler und Nenner sind konsistent verschoben. Der Tankvorgang traegt aber einen zweiten, unabhaengigen Anker: seinen eigenen Kilometerstand. Weicht der stark vom Tacho zum eingetragenen Zeitpunkt ab, passen Datum und Kilometerstand nicht zusammen. Ist kein Kilometerstand hinterlegt (der Nutzer kannte ihn nicht, siehe verlauf.als_kilometerstand()), gibt es nichts zu vergleichen - dann gilt der Zeitpunkt, wie er eingetragen wurde. Ein automatisch erkannter Tankvorgang besteht die Pruefung ohnehin immer: sein Kilometerstand und sein Zeitstempel stammen aus demselben Datensatz.""" eigener = tankvorgang.get("odometer_km") if eigener is None: return True return abs(float(eigener) - tacho) <= ODOMETER_TOLERANZ_KM async def _ist_voll(k: Koordinator, tank_sensor: str, zeitpunkt: datetime.datetime, volumen: float) -> bool: """Stand der Tank nach dieser Betankung (nahezu) voll? Gemessen SETZZEIT nach dem Zeitpunkt des Tankvorgangs - siehe dort, warum die Zahl lieber knapp als grosszuegig ist.""" verlauf = await verlauf_lesen( k.hass, tank_sensor, zeitpunkt - PUFFER, zeitpunkt + SETZZEIT + PUFFER ) if not verlauf: return False stand = wert_bei(verlauf, zeitpunkt + SETZZEIT) return stand is not None and stand >= volumen * VOLL_ANTEIL async def anwenden(k: Koordinator) -> None: """Rechnet jedes noch unbewertete Voll-zu-Voll-Intervall durch. Bewusst nur die unbewerteten: das Screening laeuft bei jeder Aenderung des Kilometerstands, und jedes Intervall braucht eigene Abfragen des Tankverlaufs. Ohne diese Einschraenkung liefen bei dreizehn Betankungen dieselben Abfragen bei jedem Fahrtende erneut. """ tank_sensor = k.zuordnung.werte.TANK_LITER_SENSOR if not tank_sensor: return km_sensor = k.zuordnung.werte.KM_SENSOR if not km_sensor: # Ohne Tacho laesst sich nicht pruefen, ob die Fahrtmenge vollstaendig # ist - siehe Modulkopf. Lieber kein Faktor als ein ungepruefter. _LOGGER.debug("Kein Kilometerstand-Sensor zugeordnet - keine Verbrauchskorrektur") return profil = await k.ablage.profil_lesen() volumen = float( (profil or {}).get("fahrzeug", {}).get("tankvolumen_liter") or STANDARD_TANKVOLUMEN_LITER ) tankvorgaenge = [t for t in await k.ablage.tankvorgaenge_lesen() if _zeit(t.get("ts"))] if len(tankvorgaenge) < 2: return tankvorgaenge.sort(key=lambda t: t["ts"]) fahrten = await k.ablage.fahrten_lesen() voll_bekannt: dict[str, bool] = {} for vorher, nachher in zip(tankvorgaenge, tankvorgaenge[1:]): von = _zeit(vorher.get("ts")) bis = _zeit(nachher.get("ts")) if von is None or bis is None or bis <= von: continue # Der Beleg der SPAETEREN Betankung nennt, was seit der frueheren # verbraucht wurde - er ist der Zaehler. getankt = beleg_liter(nachher) if getankt is None or getankt <= 0: continue # Nur Fahrten, die GANZ im Intervall liegen. Eine Fahrt, die ueber # eine Betankung hinweggeht, laesst sich keinem der beiden Intervalle # zuordnen - sie bleibt ohne Faktor, was richtig ist. im_fenster = [] for f in fahrten: start = _zeit(f.get("ts_start")) ende = _zeit(f.get("ts_end")) if start is None or ende is None: continue if start >= von and ende <= bis: im_fenster.append(f) mit_messung = [f for f in im_fenster if _liter_der_fahrt(f) is not None] if not mit_messung: continue if all(f.get("verbrauch_faktor") is not None for f in mit_messung): continue # Deckt die Fahrtmenge die tatsaechlich gefahrene Strecke ab? Das ist # die Pruefung des NENNERS und kommt vor der teuren Voll-Pruefung, weil # eine Luecke in den Fahrten der haeufigere Fall ist. strecke_fahrten = sum( float(f["distance_km"]) for f in im_fenster if f.get("distance_km") ) spanne = await _tacho_spanne(k, km_sensor, von, bis) if spanne is None: _LOGGER.debug( "Tachospanne fuer %s bis %s nicht bestimmbar - keine Korrektur", von.isoformat(), bis.isoformat(), ) continue odo_von, odo_bis = spanne # Passt das eingetragene Datum zum Kilometerstand? Nur bei Handeingabe # kann es auseinanderfallen - siehe _datum_passt(). schief = [ (v, t) for v, t in ((vorher, odo_von), (nachher, odo_bis)) if not _datum_passt(v, t) ] if schief: vorgang, tacho = schief[0] _LOGGER.warning( "Tankvorgang %s nennt %s km, der Tacho stand zu seinem Zeitpunkt " "(%s) aber bei %s km - Datum und Kilometerstand passen nicht " "zusammen, deshalb keine Korrektur", vorgang.get("tank_id"), vorgang.get("odometer_km"), vorgang.get("ts"), round(tacho, 1), ) continue deckung = strecke_fahrten / (odo_bis - odo_von) if not DECKUNG_MIN <= deckung <= DECKUNG_MAX: _LOGGER.warning( "Die erkannten Fahrten decken %.0f %% der gefahrenen Strecke zwischen " "%s und %s ab (%.1f km aus %d Fahrten) - ausserhalb von %.0f-%.0f %%. " "Der Verbrauchsfaktor wuerde fehlende Daten beschreiben und nicht den " "Geber, deshalb keine Korrektur", deckung * 100, von.isoformat(), bis.isoformat(), strecke_fahrten, len(im_fenster), DECKUNG_MIN * 100, DECKUNG_MAX * 100, ) continue # Erst jetzt der teure Teil: beide Enden muessen Volltankungen sein. for vorgang, zeitpunkt in ((vorher, von), (nachher, bis)): kennung = str(vorgang.get("tank_id")) if kennung not in voll_bekannt: voll_bekannt[kennung] = await _ist_voll(k, tank_sensor, zeitpunkt, volumen) if not (voll_bekannt[str(vorher.get("tank_id"))] and voll_bekannt[str(nachher.get("tank_id"))]): _LOGGER.debug( "Kein Voll-zu-Voll-Intervall zwischen %s und %s - keine Korrektur", von.isoformat(), bis.isoformat(), ) continue gerechnet = sum(_liter_der_fahrt(f) or 0.0 for f in mit_messung) if gerechnet <= 0: continue # Kein Ober- oder Unterwert am Faktor: der Nenner ist oben geprueft, # der Zaehler durch Beleg und Volltankung. Was jetzt herauskommt, ist # die Messung. gerundet = round(getankt / gerechnet, 4) for f in mit_messung: roh = float(f["verbrauch_roh_l_100km"]) aenderungen = { "verbrauch_faktor": gerundet, "verbrauch_l_100km": round(roh * gerundet, 1), } await k.ablage.fahrt_aktualisieren(f["trip_id"], aenderungen) f.update(aenderungen) _LOGGER.info( "Verbrauchsfaktor %.3f auf %d Fahrten zwischen %s und %s angewandt " "(%.1f l laut Beleg, %.1f l gerechnet, Streckendeckung %.0f %%)", gerundet, len(mit_messung), von.isoformat(), bis.isoformat(), getankt, gerechnet, deckung * 100, )