audi-app/custom_components/audi_dashboard/verbrauchskorrektur.py

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