"""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): """Fragt die Verlaufsdaten des Kilometerstand-Sensors über die REST-API ab. Beide blockierenden Schritte laufen über task.executor. Ein früherer Stand rief urlopen() direkt auf - das lief jahrelang durch, bis Home Assistant anfing, blockierende Aufrufe im Ereignisschleifen-Thread zu erkennen und abzubrechen (real beobachtet ab 2026.8: "Caught blocking call to putrequest ... inside the event loop"). Seitdem schlug jedes Screening fehl, und manuell angelegte Fahrten blieben ohne Strecke - sichtbar nur als Warnung im Protokoll, weil der Aufrufer den Fehler bewusst abfängt. task.executor braucht echte externe Funktionen; urllib.request.urlopen und die read-Methode der Antwort sind genau das (dieselbe Einschränkung wie bei io.open in profil.py).""" 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 = task.executor(urllib.request.urlopen, req, timeout=10) rohdaten = task.executor(resp.read) resp.close() daten = json.loads(rohdaten.decode("utf-8")) 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)