audi-app/custom_components/audi_dashboard/historienimport.py

"""Nachträglicher Import vergangener Zeiträume aus dem Home-Assistant-Verlauf
(Einstellungen -> "Daten importieren aus Home Assistant").

WOZU
----
Fahrterkennung, Tankerkennung und Batterieverlauf arbeiten alle nur ab dem
Moment, in dem sie laufen. Alles, was das Fahrzeug gemeldet hat, BEVOR die App
lief (oder während Home Assistant aus war, oder bevor ein Sensor überhaupt
zugeordnet war), taucht in den Beständen der App deshalb nie auf - obwohl der
recorder es längst aufgezeichnet hat. Dieser Import schließt genau diese
Lücke: er liest denselben Verlauf, den die Live-Erkennung sonst in Echtzeit
sieht, und leitet daraus rückwirkend dieselben Datensätze ab.

GRENZE, DIE MAN KENNEN MUSS
---------------------------
Weiter zurück als der recorder aufbewahrt, geht es nicht - was dort gelöscht
ist, ist endgültig weg. Home Assistant löscht standardmäßig nach 10 Tagen;
recorder_snippet.yaml hebt das auf ein Jahr an. Der Import meldet deshalb im
Ergebnis mit, ab wann im gewählten Zeitraum überhaupt Daten vorlagen
(`ab_wann_daten`), damit ein leeres Ergebnis nicht wie ein Fehler aussieht.

DOPPELTE DATENSÄTZE
-------------------
Der Import ist absichtlich mehrfach ausführbar (überlappende Zeiträume,
zweiter Versuch nach einem Abbruch): jede erzeugte Fahrt wird gegen die
bereits vorhandenen geprüft und übersprungen, wenn sich ihr Zeitraum mit einer
bestehenden Fahrt überschneidet - egal ob die live erkannt, von Hand angelegt
oder aus einem früheren Import stammt. Tankvorgänge werden über ein
Zeitfenster (TANK_DUBLETTE_MIN) entdoppelt, Batteriewerte über den Tag als
Schlüssel (dort führt die Ablage min/max ohnehin zusammen, statt Zeilen zu
vervielfachen).

Erzeugte Datensätze tragen `source: "import"` - dieselbe Rolle wie "ha" (live
erkannt), "manual" (von Hand) und "auto" (Tankerkennung), damit später
nachvollziehbar bleibt, woher ein Eintrag stammt.
"""

from __future__ import annotations

import datetime
import logging
from typing import TYPE_CHECKING

from .batterie import SPANNUNG_MIN_V
from .fahrterkennung import leere_fahrt
from .tankerkennung import LITER_SCHWELLE, leerer_tankvorgang, schwelle_prozent
from .verlauf import (
    durchschnitt_kmh,
    hoechstwert_im_fenster,
    route_aus_verlauf,
    UNPLAUSIBLE_KMH,
    verbrauch_aus_literstaenden,
    verlauf_lesen,
    Verlaufspunkt,
    wert_bei,
    wert_in_naehe,
    zahl,
    zeit_normalisiert,
)

if TYPE_CHECKING:
    from .koordinator import Koordinator

_LOGGER = logging.getLogger(__name__)

# Zwei Tankvorgänge innerhalb dieser Spanne gelten als derselbe - schützt
# gegen Dubletten, wenn derselbe Zeitraum zweimal importiert wird oder sich
# Import und Live-Erkennung am Rand überschneiden.
TANK_DUBLETTE_MIN = 90

# Fahrten, die kürzer sind, sind Zündung-an-ohne-Fahrt (Radio, Tür öffnen mit
# Zündung, Diagnose) - die Live-Erkennung legt sie zwar an, im Rückblick
# fluten sie den Bestand aber mit Nulleinträgen. Bewusst konservativ.
MINDESTDAUER_S = 60

# UNPLAUSIBLE_KMH selbst lebt in verlauf.py - gemeinsam mit screening.py, das
# dieselbe Prüfung seit 2026.8.27 auch für die Live-Vervollständigung nutzt
# (siehe dortiges _vollstaendig()). Die Strecke wird dann verworfen statt
# eine physikalisch unmögliche Fahrt anzuzeigen - die Fahrt bleibt "offen",
# genau wie bei fehlendem Kilometerstand.


als_zeit = zeit_normalisiert  # lokaler Name beibehalten, Funktion lebt jetzt in verlauf.py (siehe dort)


def _fahrtfenster(
    zuendung_verlauf: list[Verlaufspunkt],
) -> list[tuple[datetime.datetime, datetime.datetime]]:
    """Aus dem Zündungsverlauf die Zeiträume, in denen gefahren wurde.

    Jeder zusammenhängende "on"-Abschnitt ist eine Fahrt. Bis 2026-08-31
    wurden benachbarte Abschnitte hier zusätzlich verschmolzen, wenn ihre
    Lücke kürzer war als eine Pausenzeit - das bildete die damalige
    Live-Erkennung nach. Beide sind entfallen, weil der FMM003 die Regel
    selbst anwendet ("Ignition OFF Timeout" = 900 s): was im Verlauf als
    getrennte Abschnitte ankommt, hat das Gerät bereits als getrennte
    Fahrten beurteilt."""
    roh: list[tuple[datetime.datetime, datetime.datetime]] = []
    offen: datetime.datetime | None = None
    for ts, wert in zuendung_verlauf:
        an = str(wert).lower() in ("on", "true", "1")
        if an and offen is None:
            offen = ts
        elif not an and offen is not None:
            roh.append((offen, ts))
            offen = None
    # Ein am Ende des Zeitraums noch offener Abschnitt wird verworfen: die
    # Fahrt ist zu diesem Zeitpunkt noch nicht beendet, ihr Ende läge hinter
    # dem gewählten Fenster. Sie beim Fensterende abzuschneiden würde eine
    # Fahrt mit erfundener Endzeit erzeugen.

    return roh


def _ueberschneidet(
    start: datetime.datetime, ende: datetime.datetime, bestehende: list[dict]
) -> bool:
    """True, wenn sich [start, ende] mit einer bereits erfassten Fahrt
    überschneidet. Verhindert Dubletten beim wiederholten Import.

    Läuft über zeit_normalisiert() statt über ein rohes fromisoformat(): eine
    von Hand angelegte oder bearbeitete Fahrt (fahrterkennung.py) wird seit
    dieser Funktion ebenfalls über zeit_normalisiert() gespeichert und trägt
    deshalb einen zeitzonenbewussten Zeitstempel - vorher wurde ein naiver
    Zeitstempel hier stillschweigend übersprungen (kein Vergleich, keine
    Erkennung), sodass ein von Hand angelegter Datensatz nie gegen einen
    späteren Import geschützt war. Ältere, bereits gespeicherte naive
    Zeitstempel werden hier trotzdem korrekt behandelt: zeit_normalisiert()
    interpretiert sie als lokale Zeit, genau wie beim Schreiben."""
    for f in bestehende:
        try:
            f_start = zeit_normalisiert(f.get("ts_start"))
            f_ende = zeit_normalisiert(f.get("ts_end"))
        except (TypeError, ValueError):
            continue
        if f_start is None or f_ende is None:
            continue
        if start < f_ende and f_start < ende:
            return True
    return False


def _vmax(
    verlauf: list[Verlaufspunkt],
    start: datetime.datetime,
    ende: datetime.datetime,
) -> float | None:
    """Höchstgeschwindigkeit im Fahrtfenster, unplausible Werte verworfen -
    dieselbe Grenze und dieselbe Begründung wie beim Durchschnitt oben und
    wie im Live-Weg (screening._vmax_screenen())."""
    wert = hoechstwert_im_fenster(verlauf, start, ende)
    if wert is None or wert <= 0 or wert > UNPLAUSIBLE_KMH:
        return None
    return wert


async def _fahrten_importieren(k: Koordinator, verlaeufe: dict) -> dict:
    """Fahrten aus dem Zündungsverlauf, mit Kilometerstand und Start-/
    Zielkoordinaten aus den übrigen Verläufen ergänzt."""
    fenster = _fahrtfenster(verlaeufe["zuendung"])
    if not fenster:
        return {"angelegt": 0, "uebersprungen": 0, "zu_kurz": 0}

    bestehende = await k.ablage.fahrten_lesen()
    km_verlauf = verlaeufe["km"]
    lat_verlauf = verlaeufe["lat"]
    lon_verlauf = verlaeufe["lon"]
    tank_liter_verlauf = verlaeufe["tank_liter"]
    tempo_verlauf = verlaeufe["tempo"]

    angelegt = uebersprungen = zu_kurz = 0
    neue: list[dict] = []

    for f_start, f_ende in fenster:
        dauer_s = int((f_ende - f_start).total_seconds())
        if dauer_s < MINDESTDAUER_S:
            zu_kurz += 1
            continue
        if _ueberschneidet(f_start, f_ende, bestehende):
            uebersprungen += 1
            continue

        odo_start = wert_bei(km_verlauf, f_start)
        odo_end = wert_bei(km_verlauf, f_ende)
        distanz = None
        if odo_start is not None and odo_end is not None and odo_end >= odo_start:
            distanz = round(odo_end - odo_start, 1)

        durchschnitt = durchschnitt_kmh(distanz, dauer_s)
        if durchschnitt is not None and durchschnitt > UNPLAUSIBLE_KMH:
            _LOGGER.warning(
                "Kilometerstand für Fahrt %s bis %s verworfen: unmögliche %s km/h "
                "im Schnitt (%s km in %s s) - vermutlich ein Ausreißer im Verlauf. "
                "Die Fahrt wird trotzdem angelegt, bleibt aber ohne Strecke.",
                f_start.isoformat(), f_ende.isoformat(), durchschnitt, distanz, dauer_s,
            )
            distanz = None
            odo_start = odo_end = None
            durchschnitt = None

        fahrt = leere_fahrt(f_start, f_ende, "import")
        fahrt.update({
            "distance_km": distanz,
            "km_quelle": "sensor" if distanz is not None else None,
            "odo_start": odo_start,
            "odo_end": odo_end,
            "avg_speed_kmh": durchschnitt,
            # Der größte gemeldete Wert im Fahrtfenster, nicht die
            # tatsächliche Spitze - siehe hoechstwert_im_fenster().
            "vmax_kmh": _vmax(tempo_verlauf, f_start, f_ende),
            "start_lat": wert_bei(lat_verlauf, f_start),
            "start_lon": wert_bei(lon_verlauf, f_start),
            "end_lat": wert_bei(lat_verlauf, f_ende),
            "end_lon": wert_bei(lon_verlauf, f_ende),
            "route": route_aus_verlauf(lat_verlauf, lon_verlauf, f_start, f_ende),
            "verbrauch_l_100km": verbrauch_aus_literstaenden(
                wert_bei(tank_liter_verlauf, f_start), wert_bei(tank_liter_verlauf, f_ende), distanz
            ),
            "status": "vollständig" if distanz is not None else "offen",
        })
        neue.append(fahrt)
        bestehende.append(fahrt)
        angelegt += 1

    await k.ablage.fahrten_ergaenzen(neue)
    return {"angelegt": angelegt, "uebersprungen": uebersprungen, "zu_kurz": zu_kurz}


async def _tankvorgaenge_importieren(k: Koordinator, verlaeufe: dict) -> dict:
    """Tankvorgänge aus dem Füllstands- und/oder Tankvolumenverlauf - dieselbe
    Tiefststand-Logik wie in der Live-Erkennung (tankerkennung.py): jeder
    Anstieg über die Schwelle gegen den zuletzt gesehenen Tiefststand ist ein
    Tankvorgang, nicht jeder Anstieg gegen den unmittelbar vorherigen Wert.

    Ist ein Litersensor zugeordnet, übernimmt der für den Zeitraum, den seine
    eigene Verlaufshistorie abdeckt - genauer, und der Anstieg dient gleich
    als grobe Vorbelegung für die getankte Menge. Für den Teil DAVOR (der
    Litersensor wurde z. B. erst später zugeordnet, oder der recorder hält
    seine Historie nicht so weit zurück wie die des Prozentsensors) läuft
    zusätzlich der Prozentsensor - sonst würde ein Import über einen
    Zeitraum, der vor die Liter-Historie zurückreicht, die dort tatsächlich
    vorhandenen älteren Tankvorgänge stillschweigend verlieren, obwohl der
    Prozentsensor sie längst hätte erkennen können. Beide Läufe teilen sich
    `bekannte_zeiten`/`neue`, damit derselbe physische Tankvorgang nicht
    doppelt entsteht, falls sich die beiden Verläufe zeitlich überschneiden.

    distance_km bevorzugt ebenso TANK_DISTANZ_SENSOR (nächstliegender Wert
    vor dem erkannten Tankvorgang) vor der eigenen Kilometerstand-Subtraktion
    (_distanz_zum_vorherigen()) - dieselbe Präferenz wie im Live-Pfad
    (tankerkennung.distanz_seit_tankung())."""
    prozent_verlauf = verlaeufe["tank"]
    liter_verlauf = verlaeufe["tank_liter"] if k.zuordnung.werte.TANK_LITER_SENSOR else []
    distanz_verlauf = verlaeufe["tank_distanz"] if k.zuordnung.werte.TANK_DISTANZ_SENSOR else []
    if not prozent_verlauf and not liter_verlauf:
        return {"angelegt": 0, "uebersprungen": 0}

    km_verlauf = verlaeufe["km"]
    bestehende = await k.ablage.tankvorgaenge_lesen()
    fenster_s = TANK_DUBLETTE_MIN * 60

    # zeit_normalisiert() statt rohem fromisoformat(): siehe die identische
    # Begründung in _ueberschneidet() oben - ein von Hand angelegter oder
    # bearbeiteter Tankvorgang (belege.py) trägt seither ebenfalls einen
    # zeitzonenbewussten Zeitstempel und wurde vorher hier stillschweigend
    # aus der Dublettenprüfung ausgeschlossen.
    bekannte_zeiten: list[datetime.datetime] = []
    for t in bestehende:
        try:
            ts = zeit_normalisiert(t.get("ts"))
        except (TypeError, ValueError):
            continue
        if ts is not None:
            bekannte_zeiten.append(ts)

    angelegt = uebersprungen = 0
    neue: list[dict] = []

    def _verlauf_abarbeiten(verlauf: list, schwelle: float, mit_liter: bool) -> None:
        nonlocal angelegt, uebersprungen
        tiefststand: float | None = None
        for ts, wert in verlauf:
            aktuell = zahl(wert)
            if aktuell is None:
                continue
            if tiefststand is None or aktuell <= tiefststand:
                tiefststand = aktuell
                continue
            anstieg = aktuell - tiefststand
            if anstieg < schwelle:
                continue

            if any(abs((bekannt - ts).total_seconds()) < fenster_s for bekannt in bekannte_zeiten):
                uebersprungen += 1
                tiefststand = aktuell
                continue

            odometer_km = wert_bei(km_verlauf, ts)
            tankvorgang = leerer_tankvorgang(ts.isoformat(), "import")
            tankvorgang["odometer_km"] = odometer_km
            if mit_liter:
                tankvorgang["liters"] = round(anstieg, 1)
            distanz_sensor = wert_bei(distanz_verlauf, ts) if distanz_verlauf else None
            tankvorgang["distance_km"] = (
                distanz_sensor if distanz_sensor is not None
                else _distanz_zum_vorherigen(ts.isoformat(), odometer_km, neue + bestehende)
            )
            neue.append(tankvorgang)
            bekannte_zeiten.append(ts)
            angelegt += 1
            tiefststand = aktuell

    if liter_verlauf:
        grenze = liter_verlauf[0][0]
        aelterer_teil = [p for p in prozent_verlauf if p[0] < grenze]
        if aelterer_teil:
            _verlauf_abarbeiten(aelterer_teil, schwelle_prozent(await k.ablage.profil_lesen()), False)
        _verlauf_abarbeiten(liter_verlauf, LITER_SCHWELLE, True)
    else:
        _verlauf_abarbeiten(prozent_verlauf, schwelle_prozent(await k.ablage.profil_lesen()), False)

    await k.ablage.tankvorgaenge_ergaenzen(neue)
    return {"angelegt": angelegt, "uebersprungen": uebersprungen}


def _distanz_zum_vorherigen(
    eigene_ts: str, odometer_km: float | None, kandidaten: list[dict]
) -> float | None:
    """"Gefahren seit der letzten Tankung" heißt: seit der letzten Tankung VOR
    dieser hier - nicht seit der zeitlich jüngsten überhaupt.

    Beim Import eines vergangenen Zeitraums liegen im Bestand regelmäßig
    bereits neuere Tankvorgänge; die als Bezug zu nehmen ergäbe eine negative
    Strecke (und damit, nach der Prüfung unten, gar keine)."""
    if odometer_km is None:
        return None
    vorheriger: tuple[str, float] | None = None
    for t in kandidaten:
        t_ts = t.get("ts") or ""
        if t.get("odometer_km") is None or not t_ts or t_ts >= eigene_ts:
            continue
        if vorheriger is None or t_ts > vorheriger[0]:
            vorheriger = (t_ts, t["odometer_km"])
    if vorheriger is None:
        return None
    distanz = round(odometer_km - vorheriger[1], 1)
    return None if distanz < 0 else distanz


async def _batterie_importieren(k: Koordinator, verlaeufe: dict) -> dict:
    """Tagesminimum/-maximum der 12V-Spannung je Tag des Zeitraums.

    Die Ablage führt bestehende und neue Werte pro Tag zusammen (min bleibt
    min, max bleibt max), deshalb braucht es hier keine eigene
    Dubletten-Prüfung: ein zweiter Import desselben Zeitraums verändert die
    Einträge nicht mehr - das gilt auch für die Außentemperatur (siehe unten),
    ein erneuter Import über einen bereits importierten Zeitraum ergänzt sie
    dort nachträglich, ohne etwas doppelt zu schreiben.

    Trägt bei jedem Tagesminimum auch die Außentemperatur nach, sofern
    AUSSENTEMP_SENSOR zugeordnet ist - Parität zur Live-Erfassung
    (batterie.py's pruefen()), die das seit ihrer Einführung bereits tut. Bis
    zu dieser Korrektur fehlte das hier: verlaeufe["aussentemp"] wurde gar
    nicht erst gelesen, ein Import füllte min_temp_c also nie, selbst wenn die
    Temperaturhistorie im recorder für denselben Zeitraum längst vorlag.

    wert_in_naehe() statt naechster_wert(): der Temperatursensor dieses
    Testbestands meldet teils erst nach 30+ Stunden erneut - ohne Zeitgrenze
    würde ein beliebig alter Temperaturwert der frischen Spannungsmessung
    zugeschlagen. Siehe NEBENWERT_MAX_ABSTAND_S (verlauf.py) für die vom
    Nutzer festgelegte Grenze."""
    verlauf = verlaeufe["batterie"]
    if not verlauf:
        return {"tage": 0}
    aussentemp_verlauf = verlaeufe.get("aussentemp") or []

    tage: dict[str, dict] = {}
    for ts, wert in verlauf:
        spannung = zahl(wert)
        # Unter SPANNUNG_MIN_V ist es kein echter Messwert, sondern ein
        # Sensor-/Verbindungsfehler (siehe batterie.py) - live wie
        # rückwirkend gleich behandelt, sonst landete ein Ausreißer aus dem
        # Verlauf als Tagesminimum, wo die Live-Erfassung ihn nie hätte
        # aufzeichnen können.
        if spannung is None or spannung < SPANNUNG_MIN_V:
            continue
        tag = ts.date().isoformat()
        eintrag = tage.get(tag)
        if eintrag is None:
            tage[tag] = {"min": spannung, "min_ts": ts, "max": spannung, "max_ts": ts}
            continue
        if spannung < eintrag["min"]:
            eintrag["min"] = spannung
            eintrag["min_ts"] = ts
        if spannung > eintrag["max"]:
            eintrag["max"] = spannung
            eintrag["max_ts"] = ts

    for tag in sorted(tage):
        werte = tage[tag]
        aussentemp = (
            wert_in_naehe(werte["min_ts"], aussentemp_verlauf) if aussentemp_verlauf else None
        )
        await k.ablage.batterieverlauf_tageswert_aktualisieren(
            tag, werte["min_ts"].isoformat(), werte["min"], aussentemp
        )
        await k.ablage.batterieverlauf_tageswert_aktualisieren(
            tag, werte["max_ts"].isoformat(), werte["max"]
        )

    return {"tage": len(tage)}


async def importieren(k: Koordinator, start: object, ende: object) -> None:
    """Liest den Verlauf im gewählten Zeitraum und leitet daraus Fahrten,
    Tankvorgänge und Batteriewerte ab.

    Der Fortschritt geht über die Entität sensor.audi_dashboard_import_status
    an die Oberfläche - ein Dienstaufruf kehrt sofort zurück und kann kein
    Ergebnis mitbringen."""
    await k.ablage.ordner_sicherstellen()

    try:
        von = als_zeit(start)
        bis = als_zeit(ende)
    except ValueError as fehler:
        _LOGGER.error("Import mit unlesbarem Zeitraum aufgerufen (%s)", fehler)
        await k.import_status_veroeffentlichen("fehler", {"meldung": "Zeitraum nicht lesbar"})
        return

    if von is None or bis is None or von >= bis:
        _LOGGER.warning("Import ohne gültigen Zeitraum aufgerufen")
        await k.import_status_veroeffentlichen("fehler", {
            "meldung": "Bitte einen Zeitraum wählen, dessen Ende nach dem Start liegt."
        })
        return

    await k.import_status_veroeffentlichen(
        "laeuft", {"von": von.isoformat(), "bis": bis.isoformat()}
    )
    _LOGGER.info("Import gestartet für %s bis %s", von.isoformat(), bis.isoformat())

    werte = k.zuordnung.werte
    verlaeufe = {
        "zuendung": await verlauf_lesen(k.hass, werte.ZUENDUNG_SENSOR, von, bis),
        "km": await verlauf_lesen(k.hass, werte.KM_SENSOR, von, bis),
        "tank": await verlauf_lesen(k.hass, werte.TANK_SENSOR, von, bis),
        "tank_liter": await verlauf_lesen(k.hass, werte.TANK_LITER_SENSOR, von, bis),
        "tank_distanz": await verlauf_lesen(k.hass, werte.TANK_DISTANZ_SENSOR, von, bis),
        "batterie": await verlauf_lesen(k.hass, werte.BATTERIE_SENSOR, von, bis),
        "aussentemp": await verlauf_lesen(k.hass, werte.AUSSENTEMP_SENSOR, von, bis),
        "lat": await verlauf_lesen(k.hass, werte.STANDORT_LAT_SENSOR, von, bis),
        "lon": await verlauf_lesen(k.hass, werte.STANDORT_LON_SENSOR, von, bis),
        "tempo": await verlauf_lesen(k.hass, werte.GESCHWINDIGKEIT_SENSOR, von, bis),
    }

    # Frühester Zeitpunkt, zu dem im gewählten Fenster überhaupt etwas
    # aufgezeichnet war - damit ein leeres Ergebnis erklärbar wird
    # ("recorder reicht nur bis ...") statt wie ein Fehler auszusehen.
    frueheste = None
    for punkte in verlaeufe.values():
        if punkte and (frueheste is None or punkte[0][0] < frueheste):
            frueheste = punkte[0][0]

    fahrten = await _fahrten_importieren(k, verlaeufe)
    tank = await _tankvorgaenge_importieren(k, verlaeufe)
    batterie = await _batterie_importieren(k, verlaeufe)

    await k.fahrten_veroeffentlichen()
    await k.tankvorgaenge_veroeffentlichen()
    await k.batterieverlauf_veroeffentlichen()

    ergebnis = {
        "von": von.isoformat(),
        "bis": bis.isoformat(),
        "ab_wann_daten": frueheste.isoformat() if frueheste else None,
        "fahrten_angelegt": fahrten["angelegt"],
        "fahrten_uebersprungen": fahrten["uebersprungen"],
        "fahrten_zu_kurz": fahrten["zu_kurz"],
        "tankvorgaenge_angelegt": tank["angelegt"],
        "tankvorgaenge_uebersprungen": tank["uebersprungen"],
        "batterie_tage": batterie["tage"],
    }
    await k.import_status_veroeffentlichen("fertig", ergebnis)
    _LOGGER.info("Import abgeschlossen - %s", ergebnis)