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 nach dem Wert durchsucht, dessen Zeitstempel am
nächsten am Fahrtbeginn bzw. -ende liegt.

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 naechster_wert, route_aus_verlauf, verlauf_lesen

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

    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

    geaendert = False
    if fahrt.get("odo_end") is None:
        wert = naechster_wert(ende, punkte)
        if wert is not None:
            fahrt["odo_end"] = wert
            geaendert = True
    if fahrt.get("odo_start") is None:
        wert = naechster_wert(start, punkte)
        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")}
        await k.ablage.fahrt_aktualisieren(fahrt["trip_id"], _vollstaendig(aenderungen, fahrt))


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