audi-app/custom_components/audi_dashboard/tankerkennung.py

"""Automatische Tankerkennung über den Füllstandssensor.

Beobachtung: der Füllstand (in Prozent) springt beim Fahren nie nach oben, er
sinkt nur - jeder Anstieg ist also ein Tankvorgang. Schwelle nach Vorgabe: ab
+5 Liter ODER +9 Prozentpunkten Anstieg gilt als nachgetankt. Die
5-Liter-Vorgabe wird über fahrzeug.tankvolumen_liter aus dem Fahrzeugprofil in
Prozentpunkte umgerechnet, damit beide Angaben auf derselben Einheit
verglichen werden können; es gilt jeweils die empfindlichere (kleinere) der
beiden Schwellen.

Tiefststand-Tracking statt einfachem Vorher/Nachher-Vergleich: 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 Liter/Kosten/Station): 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

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

    await _automatisch_anlegen(k, anstieg)
    await k.tiefststand_setzen(aktuell)


async def _automatisch_anlegen(k: Koordinator, anstieg_pct: float) -> None:
    odometer_km = _als_zahl(zustand_oder_none(k.hass, k.zuordnung.werte.KM_SENSOR))
    tankvorgang = leerer_tankvorgang(
        datetime.datetime.now(datetime.UTC).isoformat(), "auto"
    )
    tankvorgang["odometer_km"] = odometer_km
    tankvorgang["distance_km"] = await k.ablage.distanz_seit_letzter_tankung(odometer_km)

    await k.ablage.tankvorgang_anhaengen(tankvorgang)
    await k.tankvorgaenge_veroeffentlichen()
    _LOGGER.info(
        "Tankvorgang %s automatisch erkannt (Füllstandsanstieg %s Prozentpunkte)",
        tankvorgang["tank_id"], round(anstieg_pct, 1),
    )