audi-app/homeassistant/pyscript/modules/fahrtabschluss_logik.py

"""Screening-Logik für den zweistufigen Fahrtabschluss (Bauauftrag §7.2).

Importierbar aus anderen pyscript-Dateien mit `import fahrtabschluss_logik`.

Der Kilometerstand kommt laut Datenquelle nicht sicher mit Fahrtende, sondern
teils erst mit Beginn oder während der nächsten Fahrt (§10 Punkt 4). Statt auf
einen festen Zeitpunkt zu warten, wird deshalb die Recorder-Historie des
Kilometerstand-Sensors nach dem Wert durchsucht, dessen Zeitstempel am
nächsten am Verbindungsabbruch liegt.

Zugriff auf die Historie läuft über die HA-REST-API (/api/history/period),
nicht über einen internen pyscript-Mechanismus - dafür gibt es in der
pyscript-Referenz keinen dokumentierten Weg. Das erfordert ein Long-Lived
Access Token, siehe README.

WICHTIGER VORBEHALT: Der GPS-Fallback aus §7.2 (km_quelle="gps") ist hier noch
NICHT umgesetzt, da die Erfassung von Start-/Endposition (device_tracker,
Adressauflösung) ein eigenes Arbeitspaket ist. Fahrten ohne passenden
Kilometerstand bleiben aktuell dauerhaft "offen" statt auf GPS-Distanz
auszuweichen. Siehe die Restarbeiten im Abschlussbericht.
"""

import datetime
import io
import json
import urllib.parse
import urllib.request

import frontend_veroeffentlichung
import profil

BASIS = "/config/audi_dashboard"
TOKEN_PFAD = f"{BASIS}/ha_token.txt"
HA_BASIS_URL = "http://localhost:8123"

# Wie weit vor/zurück in der Historie gesucht wird, ausgehend vom
# Verbindungsabbruch. 48h deckt auch länger stehende Fahrzeuge ab, ohne bei
# jedem Screening die komplette Historie zu laden.
SUCHFENSTER_STUNDEN = 48


# ---------------------------------------------------------- HA-REST-Zugriff

def _token_lesen():
    # io.open über task.executor statt bloßem open() - siehe Kopfkommentar
    # in profil.py.
    f = task.executor(io.open, TOKEN_PFAD, "r")
    inhalt = f.read()
    f.close()
    return inhalt.strip()


def historie_abfragen(entity_id, start_iso, end_iso):
    """urllib.request.urlopen läuft direkt, nicht über task.executor - anders
    als ein bloßes open() ist das ein modulqualifizierter Aufruf einer echten
    externen Funktion, der an der Testinstanz ohne Fehler durchlief. Blockiert
    kurz, läuft nur beim Fahrtabschluss-Screening, nicht bei jedem Tick."""
    token = _token_lesen()
    start_enc = urllib.parse.quote(start_iso)
    end_enc = urllib.parse.quote(end_iso)
    url = (
        f"{HA_BASIS_URL}/api/history/period/{start_enc}"
        f"?filter_entity_id={entity_id}&end_time={end_enc}"
    )
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
    resp = urllib.request.urlopen(req, timeout=10)
    rohtext = resp.read().decode("utf-8")
    resp.close()
    daten = json.loads(rohtext)
    return daten[0] if daten else []


# ------------------------------------------------------------- Kernlogik

def naechster_wert(zielzeit, punkte):
    """Wert aus der Historie, dessen Zeitstempel am nächsten an zielzeit
    liegt. Gibt (wert, zeitdifferenz_s) zurück, oder (None, None). Öffentlich,
    weil auch belegverarbeitung.py sie für den Tankfüllstand nutzt."""
    beste_wert = None
    beste_diff = None
    for p in punkte:
        try:
            wert = float(p["state"])
        except (ValueError, TypeError, KeyError):
            continue
        try:
            zeit = datetime.datetime.fromisoformat(
                p["last_changed"].replace("Z", "+00:00")
            )
        except (ValueError, KeyError):
            continue
        diff = abs((zeit - zielzeit).total_seconds())
        if beste_diff is None or diff < beste_diff:
            beste_wert = wert
            beste_diff = diff
    return beste_wert, beste_diff


def _zeitpunkt_screenen(km_sensor, zielzeit):
    start = (zielzeit - datetime.timedelta(hours=SUCHFENSTER_STUNDEN)).isoformat()
    ende = (zielzeit + datetime.timedelta(hours=SUCHFENSTER_STUNDEN)).isoformat()
    try:
        punkte = historie_abfragen(km_sensor, start, ende)
    except Exception as exc:
        # Fehlt z. B. der Token (INSTALL.md Schritt 5) oder ist die API kurz
        # nicht erreichbar, bleibt die Fahrt einfach offen statt den ganzen
        # Screening-Lauf für alle anderen offenen Fahrten abzubrechen.
        log.warning(f"audi_dashboard: Kilometerstand-Screening fehlgeschlagen: {exc}")
        return None
    wert, _ = naechster_wert(zielzeit, punkte)
    return wert


def screening_durchfuehren(km_sensor):
    """Versucht, offene Fahrten mit Start-/Endkilometer zu vervollständigen.
    Wird sowohl nach Ablauf der Pausenzeit (§7.1) als auch bei jeder Änderung
    des Kilometerstand-Sensors aufgerufen."""
    fahrten = profil.fahrten_lesen()
    offene = [f for f in fahrten if f.get("status") == "offen"]
    if not offene:
        return

    offene.sort(key=lambda f: f.get("ts_start", ""))

    for fahrt in offene:
        geaendert = False

        if fahrt.get("odo_end") is None:
            ts_end = datetime.datetime.fromisoformat(fahrt["ts_end"])
            wert = _zeitpunkt_screenen(km_sensor, ts_end)
            if wert is not None:
                fahrt["odo_end"] = wert
                geaendert = True

        if fahrt.get("odo_start") is None:
            ts_start = datetime.datetime.fromisoformat(fahrt["ts_start"])
            wert = _zeitpunkt_screenen(km_sensor, ts_start)
            if wert is not None:
                fahrt["odo_start"] = wert
                geaendert = True

        if geaendert:
            aenderungen = {"odo_start": fahrt.get("odo_start"), "odo_end": fahrt.get("odo_end")}
            if fahrt.get("odo_start") is not None and fahrt.get("odo_end") is not None:
                aenderungen["distance_km"] = round(fahrt["odo_end"] - fahrt["odo_start"], 1)
                aenderungen["km_quelle"] = "odometer"
                aenderungen["status"] = "vollständig"
            profil.fahrt_aktualisieren(fahrt["trip_id"], aenderungen)

    _lueckenlos_verketten()
    frontend_veroeffentlichung.fahrten_veroeffentlichen()


def _lueckenlos_verketten():
    """Fahrt-Ende(N) == Fahrt-Start(N+1), solange dazwischen nicht gefahren
    wurde (§7.2). Zwischen zwei in der Liste direkt aufeinanderfolgenden
    Fahrten steht das Fahrzeug nur - der Kilometerstand ändert sich dabei
    nicht, unabhängig davon, wie lange die Standzeit dauert. Ein bestätigter
    Wert wird deshalb an den direkten Nachbarn weitergereicht, auch wenn
    dessen eigenes Screening noch keinen Treffer hatte.

    Annahme: die Fahrten-Liste enthält keine Lücken durch verpasste
    Verbindungsereignisse (z. B. durch einen HA-Neustart mitten in einer
    Fahrt). Das ist der Normalfall, aber keine Garantie - siehe
    Abschlussbericht."""
    fahrten = sorted(profil.fahrten_lesen(), key=lambda f: f.get("ts_start", ""))
    geaendert_ids = []

    for i in range(len(fahrten) - 1):
        aktuelle, naechste = fahrten[i], fahrten[i + 1]

        if aktuelle.get("odo_end") is not None and naechste.get("odo_start") is None:
            naechste["odo_start"] = aktuelle["odo_end"]
            geaendert_ids.append(naechste["trip_id"])
        elif naechste.get("odo_start") is not None and aktuelle.get("odo_end") is None:
            aktuelle["odo_end"] = naechste["odo_start"]
            geaendert_ids.append(aktuelle["trip_id"])

    for fahrt in fahrten:
        if fahrt["trip_id"] not in geaendert_ids:
            continue
        aenderungen = {"odo_start": fahrt.get("odo_start"), "odo_end": fahrt.get("odo_end")}
        if fahrt.get("odo_start") is not None and fahrt.get("odo_end") is not None:
            aenderungen["distance_km"] = round(fahrt["odo_end"] - fahrt["odo_start"], 1)
            aenderungen["km_quelle"] = "odometer"
            aenderungen["status"] = "vollständig"
        profil.fahrt_aktualisieren(fahrt["trip_id"], aenderungen)