audi-app/custom_components/audi_dashboard/screening.py

"""Zweistufiger Fahrtabschluss: offene Fahrten nachträglich mit Start- und
Endkilometerstand vervollständigen (SPECIFICATION.md §7.2).

Der Kilometerstand kommt laut Datenquelle nicht sicher mit Fahrtende, sondern
teils erst mit Beginn oder während der nächsten Fahrt. Statt auf einen festen
Zeitpunkt zu warten, wird deshalb der aufgezeichnete Verlauf des
Kilometerstand-Sensors gelesen und daraus der Stand zu Fahrtbeginn bzw. -ende
bestimmt - mit wert_bei(), also dem zuletzt DAVOR gemeldeten Wert.

Bis zum 31.08.2026 stand hier naechster_wert(): der zeitlich nächstgelegene
Wert, gleich ob davor oder danach. Für einen Zählerstand ist das falsch, und
wert_bei()s eigener Docstring sagt es wörtlich - der Kilometerstand bei
Fahrtbeginn ist der zuletzt gemeldete, nicht der nächste, der schon Strecke
enthält. Der Rückblick (historienimport.py) hat immer wert_bei() benutzt;
dieselbe Fahrt wurde also je nach Weg unterschiedlich bewertet - genau das,
was die gemeinsame Herkunft von UNPLAUSIBLE_KMH und MINDESTDAUER_S in
verlauf.py ausschließt.

Läuft nach jedem Fahrtende und zusätzlich bei jeder Änderung des
Kilometerstand-Sensors - unabhängig vom Fahrtende-Ereignis selbst, eben weil
der Wert auch erst mit der nächsten Fahrt eintreffen kann.

VORBEHALT: Der GPS-Fallback (km_quelle="gps") ist weiterhin nicht umgesetzt,
da die Adressauflösung ein eigenes Arbeitspaket ist. Fahrten ohne passenden
Kilometerstand bleiben "offen" statt auf GPS-Distanz auszuweichen.
"""

from __future__ import annotations

import datetime
import logging
from typing import TYPE_CHECKING

from .verlauf import (
    UNPLAUSIBLE_KMH,
    durchschnitt_kmh,
    hoechstwert_im_fenster,
    strecke_waehlen,
    zaehlerstrecke,
    naechster_wert,
    route_aus_verlauf,
    verbrauch_aus_literstaenden,
    verlauf_lesen,
    wert_bei,
)

if TYPE_CHECKING:
    from .koordinator import Koordinator

_LOGGER = logging.getLogger(__name__)

# Wie weit vor/zurück gesucht wird, ausgehend von Fahrtbeginn bzw. -ende.
# 48h deckt auch länger stehende Fahrzeuge ab, ohne bei jedem Screening die
# komplette Historie zu laden.
SUCHFENSTER = datetime.timedelta(hours=48)


def _als_zeit(wert: str | None) -> datetime.datetime | None:
    if not wert:
        return None
    try:
        return datetime.datetime.fromisoformat(wert)
    except ValueError:
        return None


def _vollstaendig(aenderungen: dict, fahrt: dict) -> dict:
    """Ergänzt Strecke, Quelle und Status, sobald beide Kilometerstände da
    sind. Eine Fahrt ist genau dann "vollständig", wenn ihre Strecke feststeht
    - "offen" heißt nicht "unterwegs", sondern "Daten fehlen noch".

    Verwirft eine physikalisch unplausible Durchschnittsgeschwindigkeit
    (UNPLAUSIBLE_KMH) statt sie anzuzeigen, statt der bislang unbehandelten
    Lücke gegenüber historienimport.py: eine falsch abgegrenzte Fahrt (z. B.
    durch ein chatterndes Zündungssignal) kann sonst einen Kilometerstand-
    Sprung wie "22 km in 67 s" als echte Strecke zeigen. Sowohl odo_start als
    auch odo_end werden dann verworfen (nicht nur die Strecke) - die Fahrt
    bleibt "offen" und wird beim nächsten Screening erneut versucht, statt
    einen unmöglichen Wert dauerhaft zu speichern."""
    if fahrt.get("odo_start") is not None and fahrt.get("odo_end") is not None:
        distanz = round(fahrt["odo_end"] - fahrt["odo_start"], 1)
        if distanz < 0:
            # Ein Zähler läuft nicht rückwärts - der Wert stammt also nicht
            # aus derselben Quelle wie sein Gegenstück. Während der Testphase
            # wandert der Dongle zwischen Fahrzeugen; am 31.08.2026 meldete der
            # CAN-Kilometerstand deshalb 21325 zwischen 209177 und 209267.
            #
            # Die Prüfung vergleicht Anfang und Ende DERSELBEN Fahrt und steht
            # dem Fahrzeugwechsel nicht im Weg: jedes Fahrzeug zählt innerhalb
            # seiner eigenen Fahrt aufwärts. Sie greift nur, wenn die beiden
            # Enden aus verschiedenen Fahrzeugen stammen.
            #
            # Warum an der Strecke und nicht an der Geschwindigkeit: die
            # Plausibilitätsgrenze darunter prüft nur nach oben, und
            # durchschnitt_kmh() liefert bei Dauer 0 None - eine Fahrt mit null
            # Sekunden Dauer käme also an einer Geschwindigkeitsprüfung vorbei.
            #
            # historienimport.py verlangt seit jeher odo_end >= odo_start.
            _LOGGER.warning(
                "Fahrt %s: Kilometerstand verworfen - Rückwärtssprung (%s -> %s, "
                "%s km), vermutlich ein Fahrzeugwechsel des Geräts.",
                fahrt.get("trip_id"), fahrt["odo_start"], fahrt["odo_end"], distanz,
            )
            aenderungen["odo_start"] = None
            aenderungen["odo_end"] = None
            return aenderungen
        geschwindigkeit = durchschnitt_kmh(distanz, fahrt.get("duration_s"))
        if geschwindigkeit is not None and geschwindigkeit > UNPLAUSIBLE_KMH:
            _LOGGER.warning(
                "Fahrt %s: Kilometerstand verworfen - unmögliche %s km/h im "
                "Schnitt (%s km in %s s), vermutlich ein Ausreißer im "
                "Verlauf oder eine falsch abgegrenzte Fahrt.",
                fahrt.get("trip_id"), geschwindigkeit, distanz, fahrt.get("duration_s"),
            )
            aenderungen["odo_start"] = None
            aenderungen["odo_end"] = None
            return aenderungen
        aenderungen["distance_km"] = distanz
        aenderungen["km_quelle"] = "odometer"
        aenderungen["status"] = "vollständig"
    return aenderungen


async def durchfuehren(k: Koordinator) -> None:
    """Versucht, alle offenen Fahrten zu vervollständigen, und trägt allen
    Fahrten ohne Standortdaten Start-/Zielposition sowie eine Streckenlinie
    nach - unabhängig vom Kilometerstand-Status, da beide Datenquellen
    unabhängig voneinander verspätet eintreffen können."""
    km_sensor = k.zuordnung.werte.KM_SENSOR
    fahrten = await k.ablage.fahrten_lesen()
    if not fahrten:
        return

    offene = [f for f in fahrten if f.get("status") == "offen"]
    if km_sensor:
        offene.sort(key=lambda f: f.get("ts_start", ""))
        for fahrt in offene:
            await _fahrt_screenen(k, km_sensor, fahrt)

    lat_sensor = k.zuordnung.werte.STANDORT_LAT_SENSOR
    lon_sensor = k.zuordnung.werte.STANDORT_LON_SENSOR
    if lat_sensor and lon_sensor:
        ohne_position = [
            f for f in fahrten
            if f.get("start_lat") is None or f.get("end_lat") is None or f.get("route") is None
        ]
        for fahrt in ohne_position:
            await _position_screenen(k, lat_sensor, lon_sensor, fahrt)

    # Durchschnitt braucht keinen Sensor - er folgt aus Strecke und Dauer,
    # die beide schon dastehen. Deshalb vor dem Geschwindigkeitssensor und
    # unabhaengig davon, ob einer zugeordnet ist.
    for fahrt in fahrten:
        if fahrt.get("avg_speed_kmh") is not None:
            continue
        schnitt = durchschnitt_kmh(fahrt.get("distance_km"), fahrt.get("duration_s"))
        if schnitt is None:
            continue
        await k.ablage.fahrt_aktualisieren(
            fahrt["trip_id"], {"avg_speed_kmh": schnitt}
        )
        fahrt["avg_speed_kmh"] = schnitt

    # Erst nach den Kilometerstaenden: die Verfeinerung braucht die grobe
    # Strecke als Anker und laeuft nur auf Fahrten, die sie schon haben.
    gnss_sensor = k.zuordnung.werte.GNSS_KM_SENSOR
    if gnss_sensor:
        for fahrt in fahrten:
            if fahrt.get("km_quelle") == "odometer":
                await _gnss_verfeinern(k, gnss_sensor, fahrt)

    tempo_sensor = k.zuordnung.werte.GESCHWINDIGKEIT_SENSOR
    if tempo_sensor:
        ohne_vmax = [f for f in fahrten if f.get("vmax_kmh") is None]
        for fahrt in ohne_vmax:
            await _vmax_screenen(k, tempo_sensor, fahrt)

    tank_liter_sensor = k.zuordnung.werte.TANK_LITER_SENSOR
    if tank_liter_sensor:
        ohne_verbrauch = [
            f for f in fahrten
            if f.get("verbrauch_l_100km") is None and f.get("distance_km")
        ]
        for fahrt in ohne_verbrauch:
            await _verbrauch_screenen(k, tank_liter_sensor, fahrt)

    await _lueckenlos_verketten(k)
    await k.fahrten_veroeffentlichen()


async def _fahrt_screenen(k: Koordinator, km_sensor: str, fahrt: dict) -> None:
    start = _als_zeit(fahrt.get("ts_start"))
    ende = _als_zeit(fahrt.get("ts_end"))
    if start is None or ende is None:
        _LOGGER.warning(
            "Fahrt %s hat keinen lesbaren Zeitraum - Screening übersprungen",
            fahrt.get("trip_id"),
        )
        return

    # Ein Abruf für beide Zeitpunkte statt zwei: die Fenster überlappen sich
    # bei einer normalen Fahrt ohnehin fast vollständig.
    punkte = await verlauf_lesen(k.hass, km_sensor, start - SUCHFENSTER, ende + SUCHFENSTER)
    if not punkte:
        return

    # wert_bei() statt naechster_wert(): der zuletzt DAVOR gemeldete Stand, nicht
    # der zeitlich nächstgelegene. Siehe Modulkopf - für einen Zähler ist das die
    # richtige Wahl, und der Rückblick in historienimport.py macht es genauso.
    # Steht davor nichts, nimmt wert_bei() den ersten Wert danach.
    geaendert = False
    if fahrt.get("odo_end") is None:
        wert = wert_bei(punkte, ende)
        if wert is not None:
            fahrt["odo_end"] = wert
            geaendert = True
    if fahrt.get("odo_start") is None:
        wert = wert_bei(punkte, 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")}
        aenderungen = _vollstaendig(aenderungen, fahrt)
        await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], aenderungen)
        # Das In-Memory-Objekt sofort mitziehen (nicht nur die Ablage) - sonst
        # sehen spätere Schritte innerhalb desselben durchfuehren()-Laufs
        # (z. B. _verbrauch_screenen(), das eine bekannte distance_km
        # braucht) noch den alten Stand, obwohl die Strecke gerade erst in
        # diesem Durchlauf ermittelt wurde. _position_screenen() macht das
        # bereits genauso.
        fahrt.update(aenderungen)


async def _gnss_verfeinern(k: Koordinator, gnss_sensor: str, fahrt: dict) -> None:
    """Ersetzt die auf ganze Kilometer gerundete Strecke durch die metergenaue
    des Geraets - aber nur, wenn beide zueinander passen.

    Der Kilometerstand des Fahrzeugs bleibt der Anker: er ist der Tacho, er
    driftet nicht, und er ist die Zahl, die der Fahrer im Cockpit sieht. Nur
    loest er eben ganze Kilometer auf. Der GNSS-Zaehler des Geraets loest auf
    einen Meter auf, kann aber driften und Empfang verlieren.

    Deshalb keine Ersetzung, sondern eine Pruefung: die feine Zahl gilt, solange
    sie innerhalb der Rundungsunschaerfe des groben Ankers liegt
    (GNSS_TOLERANZ_KM). Liegt sie darueber, hat der Anker recht und die feine
    Zahl wird verworfen - dann fehlte Empfang, oder der Zaehler ist gedriftet.

    Laeuft nur auf Fahrten mit km_quelle == \"odometer\", also genau einmal je
    Fahrt: danach steht dort \"odometer+gnss\" (oder es blieb bei \"odometer\",
    wenn nichts Brauchbares vorlag, und ein spaeterer Lauf versucht es erneut,
    sobald mehr Verlauf da ist)."""
    start = _als_zeit(fahrt.get("ts_start"))
    ende = _als_zeit(fahrt.get("ts_end"))
    grob = fahrt.get("distance_km")
    if start is None or ende is None or grob is None:
        return

    # Etwas vor dem Fahrtbeginn mitlesen, damit zaehlerstrecke() ihren Anker
    # findet - den letzten Datensatz am oder vor dem Beginn.
    punkte = await verlauf_lesen(k.hass, gnss_sensor, start - SUCHFENSTER, ende)
    fein = zaehlerstrecke(punkte, start, ende)
    gewaehlt, genommen = strecke_waehlen(grob, fein)
    if not genommen:
        if fein is not None:
            _LOGGER.info(
                "Fahrt %s: GNSS-Strecke %s km passt nicht zum Kilometerstand "
                "(%s km) - der Fahrzeugwert bleibt stehen",
                fahrt.get("trip_id"), fein, grob,
            )
        return

    aenderungen = {"distance_km": gewaehlt, "km_quelle": "odometer+gnss"}
    await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], aenderungen)
    fahrt.update(aenderungen)
    _LOGGER.info(
        "Fahrt %s: Strecke auf %s km verfeinert (Kilometerstand sagte %s km)",
        fahrt.get("trip_id"), gewaehlt, grob,
    )


async def _position_screenen(
    k: Koordinator, lat_sensor: str, lon_sensor: str, fahrt: dict
) -> None:
    """Trägt Start-/Zielposition und eine echte Streckenlinie nach, sobald der
    GPS-Verlauf sie hergibt - unabhängig vom Kilometerstand-Status.

    Dieselbe "Fahrt gehört der Vergangenheit an" wie bei den Kilometerständen:
    kein Live-Zustand zum Fahrtbeginn/-ende wird herangezogen (der wäre bei
    einer Funklücke des Trackers unzuverlässig, siehe fahrterkennung.py), nur
    der aufgezeichnete Verlauf im Fahrtfenster."""
    start = _als_zeit(fahrt.get("ts_start"))
    ende = _als_zeit(fahrt.get("ts_end"))
    if start is None or ende is None:
        return

    lat_verlauf = await verlauf_lesen(k.hass, lat_sensor, start - SUCHFENSTER, ende + SUCHFENSTER)
    lon_verlauf = await verlauf_lesen(k.hass, lon_sensor, start - SUCHFENSTER, ende + SUCHFENSTER)
    if not lat_verlauf or not lon_verlauf:
        return

    aenderungen: dict = {}
    if fahrt.get("start_lat") is None:
        lat = naechster_wert(start, lat_verlauf)
        lon = naechster_wert(start, lon_verlauf)
        if lat is not None and lon is not None:
            aenderungen["start_lat"] = lat
            aenderungen["start_lon"] = lon
    if fahrt.get("end_lat") is None:
        lat = naechster_wert(ende, lat_verlauf)
        lon = naechster_wert(ende, lon_verlauf)
        if lat is not None and lon is not None:
            aenderungen["end_lat"] = lat
            aenderungen["end_lon"] = lon
    if fahrt.get("route") is None:
        route = route_aus_verlauf(lat_verlauf, lon_verlauf, start, ende)
        if route is not None:
            aenderungen["route"] = route

    if aenderungen:
        await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], aenderungen)
        fahrt.update(aenderungen)


async def _verbrauch_screenen(
    k: Koordinator, tank_liter_sensor: str, fahrt: dict
) -> None:
    """Trägt den Verbrauch (l/100 km) nach, sobald die Distanz feststeht und
    der Tankfüllstand (Liter) das hergibt - siehe verbrauch_aus_literstaenden()
    für die Näherung selbst und ihre Begrenzung."""
    start = _als_zeit(fahrt.get("ts_start"))
    ende = _als_zeit(fahrt.get("ts_end"))
    if start is None or ende is None:
        return

    verlauf = await verlauf_lesen(k.hass, tank_liter_sensor, start - SUCHFENSTER, ende + SUCHFENSTER)
    if not verlauf:
        return

    # wert_bei() aus demselben Grund wie beim Kilometerstand (siehe Modulkopf):
    # der Tankstand zu Fahrtbeginn ist der zuletzt DAVOR gemeldete, nicht der
    # zeitlich nächstgelegene - der läge womöglich schon in der Fahrt und hätte
    # einen Teil des Verbrauchs vorweggenommen. historienimport.py liest ihn
    # seit jeher so.
    verbrauch = verbrauch_aus_literstaenden(
        wert_bei(verlauf, start), wert_bei(verlauf, ende), fahrt.get("distance_km")
    )
    if verbrauch is not None:
        await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], {"verbrauch_l_100km": verbrauch})
        fahrt["verbrauch_l_100km"] = verbrauch


async def _vmax_screenen(k: Koordinator, tempo_sensor: str, fahrt: dict) -> None:
    """Trägt die Höchstgeschwindigkeit der Fahrt nach.

    Anders als beim Kilometerstand wird hier nicht der nächstgelegene Wert
    gesucht, sondern der größte innerhalb des Fahrtfensters - und ohne
    Suchfenster drumherum, weil ein Wert von vor oder nach der Fahrt nicht
    zu ihr gehört.

    Unplausible Werte werden verworfen statt angezeigt: dieselbe Grenze wie
    beim Durchschnitt (UNPLAUSIBLE_KMH), denn ein Sensorsprung sieht in
    beiden Fällen gleich aus."""
    start = _als_zeit(fahrt.get("ts_start"))
    ende = _als_zeit(fahrt.get("ts_end"))
    if start is None or ende is None:
        return

    verlauf = await verlauf_lesen(k.hass, tempo_sensor, start, ende)
    if not verlauf:
        return

    vmax = hoechstwert_im_fenster(verlauf, start, ende)
    if vmax is None or vmax <= 0 or vmax > UNPLAUSIBLE_KMH:
        return

    await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], {"vmax_kmh": vmax})
    fahrt["vmax_kmh"] = vmax


async def _lueckenlos_verketten(k: Koordinator) -> None:
    """Fahrt-Ende(N) == Fahrt-Start(N+1), solange dazwischen nicht gefahren
    wurde.

    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
    Ereignisse. Das ist der Normalfall, aber keine Garantie."""
    fahrten = sorted(await k.ablage.fahrten_lesen(), key=lambda f: f.get("ts_start", ""))
    geaendert: set[str] = set()

    for aktuelle, naechste in zip(fahrten, fahrten[1:]):
        if aktuelle.get("odo_end") is not None and naechste.get("odo_start") is None:
            naechste["odo_start"] = aktuelle["odo_end"]
            geaendert.add(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.add(aktuelle["trip_id"])

    for fahrt in fahrten:
        if fahrt["trip_id"] not in geaendert:
            continue
        aenderungen = {"odo_start": fahrt.get("odo_start"), "odo_end": fahrt.get("odo_end")}
        await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], _vollstaendig(aenderungen, fahrt))