audi-app/homeassistant/pyscript/historienimport.py

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

WOZU
----
Fahrterkennung (fahrterkennung.py), Tankerkennung (tankerkennung.py) und
Batterieverlauf (batterieverlauf.py) arbeiten alle nur ab dem Moment, in dem
sie laufen: sie hängen an @state_trigger bzw. an einem 5-Minuten-Takt. 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 von Home
Assistant es längst aufgezeichnet hat.

Dieser Import schließt genau diese Lücke: er liest denselben Verlauf, den die
Live-Trigger sonst in Echtzeit sehen, 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.

WIE DER VERLAUF GELESEN WIRD
----------------------------
Über Home Assistants eigene recorder-API
(`homeassistant.components.recorder.history.get_significant_states`), nicht
über direkten SQL-Zugriff auf home-assistant_v2.db. Der Umweg über die API
ist Absicht: das Datenbankschema des recorders ist HA-intern und ändert sich
zwischen Versionen, die Funktion dagegen ist die von HA selbst benutzte und
stabile Schnittstelle.

Voraussetzung dafür ist `hass_is_global: true` im pyscript-Block der
configuration.yaml (siehe configuration_snippet.yaml) - ohne das existiert
der `hass`-Name hier nicht. Der Aufruf läuft über task.executor(), weil es
eine echte externe Funktion ist und ein Datenbankzugriff nichts im
Event-Loop verloren hat (dieselbe pyscript-Einschränkung wie bei io.open in
profil.py, siehe dortiger Kopfkommentar).

`significant_changes_only=False` ist wichtig: bei numerischen Sensoren
(Kilometerstand, Tankfüllstand) liefert der Standardmodus nur "auffällige"
Änderungen und verschluckt genau die kleinen Schritte, aus denen sich
Fahrstrecke und Tankvorgänge zusammensetzen.

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 profil.batterieverlauf_tageswert_aktualisieren() 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.
"""

import datetime

from homeassistant.components.recorder.history import get_significant_states

import einstellungen
import frontend_veroeffentlichung
import profil

# Ein Anstieg des Tankfüllstands gilt ab denselben Schwellen als Tankvorgang
# wie in der Live-Erkennung - bewusst dieselben Zahlen wie in
# tankerkennung.py, damit ein importierter Zeitraum dieselben Vorgänge
# erzeugt, die die Live-Erkennung erzeugt hätte.
LITER_SCHWELLE = 5
PROZENT_SCHWELLE = 9
STANDARD_TANKVOLUMEN_LITER = 58

# 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


def _status(zustand, daten=None):
    """Fortschritt/Ergebnis für die Oberfläche. Gleiches Muster wie
    _status_veroeffentlichen() in updateverwaltung.py."""
    state.set(
        "pyscript.audi_dashboard_import_status",
        zustand,
        new_attributes={"daten": daten or {}},
    )


def _als_zeit(wert):
    """Akzeptiert, was die Oberfläche schickt: ISO mit oder ohne Zeitzone.
    Ohne Zeitzone gilt die lokale Zeit von Home Assistant - der Nutzer wählt
    im Formular schließlich Ortszeit, keine UTC."""
    if not wert:
        return None
    ts = datetime.datetime.fromisoformat(str(wert))
    if ts.tzinfo is None:
        ts = ts.astimezone()
    return ts.astimezone(datetime.timezone.utc)


def _zahl(wert):
    try:
        return float(wert)
    except (TypeError, ValueError):
        return None


def _verlauf(entity_id, start, ende):
    """Zustandsverlauf einer Entität als Liste von (Zeitpunkt, Rohwert),
    aufsteigend. Leere Liste, wenn die Entität nicht zugeordnet ist oder im
    Zeitraum nichts vorliegt.

    Die Zustände "unknown"/"unavailable" werden verworfen: sie bedeuten
    "keine Meldung", nicht "Wert 0" - würden sie durchgereicht, ergäbe ein
    Ausfall der Integration eine Fahrt mit absurder Kilometerdifferenz."""
    if not entity_id:
        return []
    roh = task.executor(
        get_significant_states, hass, start, ende, [entity_id], None, True, False
    )
    reihen = roh.get(entity_id) or []
    ergebnis = []
    for s in reihen:
        if s.state in ("unknown", "unavailable", None, ""):
            continue
        ergebnis.append((s.last_updated, s.state))
    ergebnis.sort(key=lambda p: p[0])
    return ergebnis


def _wert_bei(verlauf, zeitpunkt):
    """Der zuletzt vor `zeitpunkt` gemeldete Zahlenwert, sonst der erste
    danach, sonst None. "Zuletzt davor" ist die richtige Wahl für einen
    Zählerstand: der Kilometerstand bei Fahrtbeginn ist der, der zuletzt
    gemeldet wurde, nicht der nächste (der schon Strecke enthält)."""
    davor = None
    for ts, wert in verlauf:
        zahl = _zahl(wert)
        if zahl is None:
            continue
        if ts <= zeitpunkt:
            davor = zahl
        else:
            return davor if davor is not None else zahl
    return davor


def _pausenzeit_sekunden():
    p = profil.profil_lesen()
    if p is None:
        return 15 * 60
    return p.get("einstellungen", {}).get("fahrten_pausenzeit_min", 15) * 60


def _fahrtfenster(zuendung_verlauf, pausenzeit_s):
    """Aus dem Zündungsverlauf die Zeiträume, in denen gefahren wurde.

    Zwei Schritte, die zusammen die Pausenregel aus §7.1 nachbilden:
    erst jeden zusammenhängenden "on"-Abschnitt sammeln, dann benachbarte
    Abschnitte verschmelzen, deren Lücke kürzer als die Pausenzeit ist. Genau
    das tut die Live-Erkennung über task.unique() + task.sleep(), nur eben
    im Nachhinein und ohne Warten."""
    roh = []
    offen = 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.

    if not roh:
        return []

    verschmolzen = [roh[0]]
    for start, ende in roh[1:]:
        vorheriger_start, vorheriges_ende = verschmolzen[-1]
        if (start - vorheriges_ende).total_seconds() < pausenzeit_s:
            verschmolzen[-1] = (vorheriger_start, ende)
        else:
            verschmolzen.append((start, ende))
    return verschmolzen


def _ueberschneidet(start, ende, bestehende):
    """True, wenn sich [start, ende] mit einer bereits erfassten Fahrt
    überschneidet. Verhindert Dubletten beim wiederholten Import."""
    for f in bestehende:
        try:
            f_start = datetime.datetime.fromisoformat(f.get("ts_start"))
            f_ende = datetime.datetime.fromisoformat(f.get("ts_end"))
        except (TypeError, ValueError):
            continue
        if f_start.tzinfo is None or f_ende.tzinfo is None:
            continue
        if start < f_ende and f_start < ende:
            return True
    return False


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

    bestehende = profil.fahrten_lesen()
    km_verlauf = verlaeufe["km"]
    lat_verlauf = verlaeufe["lat"]
    lon_verlauf = verlaeufe["lon"]

    angelegt = 0
    uebersprungen = 0
    zu_kurz = 0
    neue = []

    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 = None
        if distanz is not None and dauer_s > 0:
            durchschnitt = round(distanz / (dauer_s / 3600.0), 1)

        fahrt = {
            "trip_id": profil.neue_id("t"),
            "ts_start": f_start.isoformat(),
            "ts_end": f_ende.isoformat(),
            "duration_s": dauer_s,
            "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,
            "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),
            "start_address": None,
            "end_address": None,
            "art": "privat",
            "route": None,
            "pausen": [],
            "source": "import",
            "status": "vollständig" if distanz is not None else "offen",
            "edited_fields": [],
        }
        neue.append(fahrt)
        bestehende.append(fahrt)
        angelegt += 1

    # Alle neuen Fahrten in einem Rutsch anhängen statt je Fahrt einmal die
    # Datei zu öffnen - bei einem Jahr Verlauf sind das sonst hunderte
    # Einzelschreibvorgänge.
    if neue:
        alle = profil.fahrten_lesen() + neue
        alle.sort(key=lambda f: f.get("ts_start") or "")
        profil.fahrten_schreiben(alle)

    return {"angelegt": angelegt, "uebersprungen": uebersprungen, "zu_kurz": zu_kurz}


def _schwelle_prozent():
    p = profil.profil_lesen()
    if p is None:
        tankvolumen = STANDARD_TANKVOLUMEN_LITER
    else:
        tankvolumen = p.get("fahrzeug", {}).get("tankvolumen_liter") or STANDARD_TANKVOLUMEN_LITER
    return min((LITER_SCHWELLE / tankvolumen) * 100, PROZENT_SCHWELLE)


def _tankvorgaenge_importieren(verlaeufe):
    """Tankvorgänge aus dem Füllstandsverlauf - dieselbe Tiefststand-Logik
    wie tankerkennung.py (siehe dortiger Kopfkommentar): jeder Anstieg über
    die Schwelle gegen den zuletzt gesehenen Tiefststand ist ein Tankvorgang,
    nicht jeder Anstieg gegen den unmittelbar vorherigen Wert."""
    verlauf = verlaeufe["tank"]
    if not verlauf:
        return {"angelegt": 0, "uebersprungen": 0}

    schwelle = _schwelle_prozent()
    km_verlauf = verlaeufe["km"]
    bestehende = profil.tankvorgaenge_lesen()
    fenster = datetime.timedelta(minutes=TANK_DUBLETTE_MIN)

    bekannte_zeiten = []
    for t in bestehende:
        try:
            ts = datetime.datetime.fromisoformat(t.get("ts"))
        except (TypeError, ValueError):
            continue
        if ts.tzinfo is not None:
            bekannte_zeiten.append(ts)

    angelegt = 0
    uebersprungen = 0
    neue = []
    tiefststand = None

    for ts, wert in verlauf:
        aktuell = _zahl(wert)
        if aktuell is None:
            continue
        if tiefststand is None or aktuell <= tiefststand:
            tiefststand = aktuell
            continue
        if aktuell - tiefststand < schwelle:
            continue

        dublette = False
        for bekannt in bekannte_zeiten:
            if abs((bekannt - ts).total_seconds()) < fenster.total_seconds():
                dublette = True
                break
        if dublette:
            uebersprungen += 1
            tiefststand = aktuell
            continue

        odometer_km = _wert_bei(km_verlauf, ts)
        # "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).
        eigene_ts = ts.isoformat()
        vorheriger = None
        for t in neue + bestehende:
            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"])
        distanz = None
        if odometer_km is not None and vorheriger is not None:
            distanz = round(odometer_km - vorheriger[1], 1)
            if distanz < 0:
                distanz = None

        tankvorgang = {
            "tank_id": profil.neue_id("f"),
            "receipt_key": None,
            "ts": ts.isoformat(),
            "liters": None,
            "fuel_total_eur": None,
            "price_per_l": None,
            "discount": None,
            "station_name": None,
            "odometer_km": odometer_km,
            "distance_km": distanz,
            "fuel_type": None,
            "source": "import",
            "status": "unvollständig",
            "receipt_file": None,
            "edited_fields": [],
        }
        neue.append(tankvorgang)
        bekannte_zeiten.append(ts)
        angelegt += 1
        tiefststand = aktuell

    if neue:
        alle = profil.tankvorgaenge_lesen() + neue
        alle.sort(key=lambda t: t.get("ts") or "")
        profil.tankvorgaenge_schreiben(alle)

    return {"angelegt": angelegt, "uebersprungen": uebersprungen}


def _batterie_importieren(verlaeufe):
    """Tagesminimum/-maximum der 12V-Spannung je Tag des Zeitraums.

    profil.batterieverlauf_tageswert_aktualisieren() 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."""
    verlauf = verlaeufe["batterie"]
    if not verlauf:
        return {"tage": 0}

    tage = {}
    for ts, wert in verlauf:
        spannung = _zahl(wert)
        if spannung is None:
            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]
        profil.batterieverlauf_tageswert_aktualisieren(
            tag, werte["min_ts"].isoformat(), werte["min"]
        )
        profil.batterieverlauf_tageswert_aktualisieren(
            tag, werte["max_ts"].isoformat(), werte["max"]
        )

    return {"tage": len(tage)}


@service
def audi_dashboard_historie_importieren(start=None, ende=None):
    """Liest den Home-Assistant-Verlauf im gewählten Zeitraum und leitet
    daraus Fahrten, Tankvorgänge und Batteriewerte ab.

    start/ende sind ISO-Zeitstempel aus der Oberfläche (Ortszeit ohne
    Zeitzone ist zulässig). Aufruf als
    pyscript.audi_dashboard_historie_importieren."""
    profil.ordner_sicherstellen()

    try:
        von = _als_zeit(start)
        bis = _als_zeit(ende)
    except ValueError as fehler:
        log.error(f"audi_dashboard: Import mit unlesbarem Zeitraum aufgerufen ({fehler})")
        _status("fehler", {"meldung": "Zeitraum nicht lesbar"})
        return

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

    _status("laeuft", {"von": von.isoformat(), "bis": bis.isoformat()})
    log.info(f"audi_dashboard: Import gestartet für {von.isoformat()} bis {bis.isoformat()}")

    try:
        verlaeufe = {
            "zuendung": _verlauf(einstellungen.ZUENDUNG_SENSOR, von, bis),
            "km": _verlauf(einstellungen.KM_SENSOR, von, bis),
            "tank": _verlauf(einstellungen.TANK_SENSOR, von, bis),
            "batterie": _verlauf(einstellungen.BATTERIE_SENSOR, von, bis),
            "lat": _verlauf(einstellungen.STANDORT_LAT_SENSOR, von, bis),
            "lon": _verlauf(einstellungen.STANDORT_LON_SENSOR, von, bis),
        }
    except Exception as fehler:
        log.error(f"audi_dashboard: Verlauf nicht lesbar ({type(fehler).__name__}: {fehler})")
        _status("fehler", {"meldung": (
            "Der Verlauf konnte nicht gelesen werden. Steht hass_is_global: true "
            "im pyscript-Block der configuration.yaml?"
        )})
        return

    # 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 name in verlaeufe:
        if verlaeufe[name]:
            erster = verlaeufe[name][0][0]
            if frueheste is None or erster < frueheste:
                frueheste = erster

    fahrten = _fahrten_importieren(von, bis, verlaeufe)
    tank = _tankvorgaenge_importieren(verlaeufe)
    batterie = _batterie_importieren(verlaeufe)

    frontend_veroeffentlichung.fahrten_veroeffentlichen()
    frontend_veroeffentlichung.tankvorgaenge_veroeffentlichen()
    frontend_veroeffentlichung.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"],
    }
    _status("fertig", ergebnis)
    log.info(f"audi_dashboard: Import abgeschlossen - {ergebnis}")