audi-app/custom_components/audi_dashboard/tankerkennung.py

"""Automatische Tankerkennung über den Füllstandssensor - oder, wenn
zugeordnet, direkt über einen Tankvolumen-Sensor in Litern.

Beobachtung: der Füllstand springt beim Fahren nie nach oben, er sinkt nur -
jeder Anstieg ist also ein Tankvorgang. Zwei Datenquellen dafür, in
absteigender Genauigkeit:

- TANK_LITER_SENSOR (optional, z. B. vom CAN): Anstieg direkt in Litern,
  Schwelle LITER_SCHWELLE. Braucht keine Umrechnung und liefert den Anstieg
  selbst gleich als Vorbelegung für die getankte Menge mit.
- TANK_SENSOR (Prozent, siehe fuellstand_geaendert()): Schwelle nach Vorgabe
  ab +5 Liter ODER +9 Prozentpunkten Anstieg, wobei die 5-Liter-Vorgabe über
  fahrzeug.tankvolumen_liter aus dem Fahrzeugprofil in Prozentpunkte
  umgerechnet wird (beide Angaben so auf derselben Einheit); es gilt jeweils
  die empfindlichere (kleinere) der beiden Schwellen.

Ist der Litersensor zugeordnet, übernimmt er die Erkennung vollständig -
beide gleichzeitig auszuwerten würde denselben Tankvorgang zweimal anlegen.
Ohne ihn bleibt es beim Prozentsensor wie bisher.

Tiefststand-Tracking statt einfachem Vorher/Nachher-Vergleich, für beide
Quellen gleichermaßen: die Datenquelle liefert einen Tankvorgang oft in
mehreren kleinen Schritten (40% -> 45% -> 60%), von denen keiner allein die
Schwelle überschreiten muss. Deshalb wird der zuletzt bekannte Tiefststand
gespeichert und der Anstieg dagegen gemessen. Nach Anlage eines Tankvorgangs
wird der Tiefststand auf den aktuellen Wert zurückgesetzt, damit derselbe
Vorgang nicht mehrfach Datensätze erzeugt.

Der so angelegte Tankvorgang ist bewusst ein unvollständiger Platzhalter
(status "unvollständig", ohne Kosten/Station, Liter nur als grobe Schätzung
vom Litersensor oder ganz ohne bei reiner Prozent-Erkennung): erwartet wird
ein automatisch erkannter Tankvorgang mit Zeitstempel, Kilometerstand und
gefahrener Distanz, den der Nutzer später per Beleg-Nachtrag vervollständigt.

Auch hier gilt die Härtung aus fahrterkennung.py: der Tiefststand lebt nicht
mehr nur im Arbeitsspeicher, sondern im Store des Koordinators und überlebt
damit einen Neustart.
"""

from __future__ import annotations

import datetime
import logging
from typing import TYPE_CHECKING

from .ablage import neue_id
from .veroeffentlichung import zustand_oder_none
from .verlauf import geraetezeit

if TYPE_CHECKING:
    from .koordinator import Koordinator

_LOGGER = logging.getLogger(__name__)

LITER_SCHWELLE = 5
PROZENT_SCHWELLE = 9
STANDARD_TANKVOLUMEN_LITER = 58  # Rückfall, falls im Profil nicht hinterlegt


def _als_zahl(wert: object) -> float | None:
    try:
        return float(wert)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return None


def schwelle_prozent(profil: dict | None) -> float:
    """Die empfindlichere der beiden Schwellen, in Prozentpunkten."""
    if profil is None:
        tankvolumen = STANDARD_TANKVOLUMEN_LITER
    else:
        tankvolumen = (
            profil.get("fahrzeug", {}).get("tankvolumen_liter") or STANDARD_TANKVOLUMEN_LITER
        )
    return min((LITER_SCHWELLE / tankvolumen) * 100, PROZENT_SCHWELLE)


def leerer_tankvorgang(ts: str, quelle: str) -> dict:
    """Das vollständige Tankvorgang-Schema, alles Unbekannte auf None."""
    return {
        "tank_id": neue_id("f"),
        "receipt_key": None,
        "ts": ts,
        "liters": None,
        "fuel_total_eur": None,
        "price_per_l": None,
        "discount": None,
        "station_name": None,
        "odometer_km": None,
        "distance_km": None,
        "fuel_type": None,
        "source": quelle,
        "status": "unvollständig",
        "receipt_file": None,
        "edited_fields": [],
    }


async def fuellstand_geaendert(
    k: Koordinator, neu: str | None, ereigniszeit: datetime.datetime | None = None
) -> None:
    aktuell = _als_zahl(neu)
    if aktuell is None:
        return

    tiefststand = k.tiefststand_pct
    if tiefststand is None or aktuell <= tiefststand:
        await k.tiefststand_setzen(aktuell)
        return

    anstieg = aktuell - tiefststand
    if anstieg < schwelle_prozent(await k.ablage.profil_lesen()):
        return

    # Ist ein direkter Litersensor zugeordnet, übernimmt der die Erkennung
    # (siehe tankvolumen_geaendert) - genauer, weil ohne den Umweg über das im
    # Fahrzeugprofil hinterlegte Tankvolumen. Beide gleichzeitig auswerten
    # würde denselben Tankvorgang zweimal anlegen.
    if not k.zuordnung.werte.TANK_LITER_SENSOR:
        await _automatisch_anlegen(
            k, f"Füllstandsanstieg {round(anstieg, 1)} Prozentpunkte", ereigniszeit=ereigniszeit
        )
    await k.tiefststand_setzen(aktuell)


async def tankvolumen_geaendert(
    k: Koordinator, neu: str | None, ereigniszeit: datetime.datetime | None = None
) -> None:
    """Wie fuellstand_geaendert(), aber direkt in Litern statt in Prozent -
    braucht deshalb keine Umrechnung über das Fahrzeugprofil und liefert mit
    dem gemessenen Anstieg selbst eine sinnvolle Vorbelegung für die getankte
    Menge (siehe leerer_tankvorgang() - "liters" bleibt sonst None, bis der
    Nutzer sie von Hand einträgt oder ein Beleg sie liefert)."""
    aktuell = _als_zahl(neu)
    if aktuell is None:
        return

    tiefststand = k.tiefststand_liter
    if tiefststand is None or aktuell <= tiefststand:
        await k.tiefststand_liter_setzen(aktuell)
        return

    anstieg = aktuell - tiefststand
    if anstieg < LITER_SCHWELLE:
        return

    await _automatisch_anlegen(
        k,
        f"Tankvolumenanstieg {round(anstieg, 1)} l",
        liter_schaetzung=anstieg,
        ereigniszeit=ereigniszeit,
    )
    await k.tiefststand_liter_setzen(aktuell)


async def distanz_seit_tankung(k: Koordinator, odometer_km: float | None) -> float | None:
    """Bevorzugt TANK_DISTANZ_SENSOR (vom Fahrzeug selbst berechnet), fällt
    ohne ihn auf die eigene Kilometerstand-Subtraktion zurück - dieselbe
    "Sensor bevorzugt, Berechnung als Rückfall"-Präferenz wie bei
    TANK_LITER_SENSOR gegenüber TANK_SENSOR oben."""
    sensor = _als_zahl(zustand_oder_none(k.hass, k.zuordnung.werte.TANK_DISTANZ_SENSOR))
    if sensor is not None:
        return sensor
    return await k.ablage.distanz_seit_letzter_tankung(odometer_km)


async def _automatisch_anlegen(
    k: Koordinator,
    anlass: str,
    liter_schaetzung: float | None = None,
    ereigniszeit: datetime.datetime | None = None,
) -> None:
    odometer_km = _als_zahl(zustand_oder_none(k.hass, k.zuordnung.werte.KM_SENSOR))
    # Der Zeitstempel des Geräts, nicht die Ankunft in Home Assistant -
    # dieselbe Begründung wie bei Fahrtbeginn und -ende, siehe
    # verlauf.geraetezeit(). Ein Tankvorgang, der aus einem gepufferten
    # Datensatz erkannt wird, trug bisher die Uhrzeit des Auftauchens; bei
    # 43200 s Aufzeichnungstakt im Stand konnten das Stunden sein.
    ts = await geraetezeit(
        k.hass,
        k.zuordnung.werte.MELDEZEIT_SENSOR,
        ereigniszeit,
        datetime.datetime.now(datetime.UTC),
    )
    tankvorgang = leerer_tankvorgang(ts.isoformat(), "auto")
    tankvorgang["odometer_km"] = odometer_km
    tankvorgang["distance_km"] = await distanz_seit_tankung(k, odometer_km)
    if liter_schaetzung is not None:
        # Grobe Vorbelegung, keine feststehende Menge - der Status bleibt
        # "unvollständig" (siehe leerer_tankvorgang), bis Preis/Tankstelle
        # dazukommen. Der Nutzer korrigiert sie bei Bedarf selbst, ein
        # nachgetragener Beleg überschreibt sie ohnehin.
        tankvorgang["liters"] = round(liter_schaetzung, 1)

    await k.ablage.tankvorgang_anhaengen(tankvorgang)
    await k.tankvorgaenge_veroeffentlichen()
    _LOGGER.info(
        "Tankvorgang %s automatisch erkannt (%s)", tankvorgang["tank_id"], anlass
    )