Station name as "brand, street, city", grey box on all editable fields
The "grey box around the icons" report turned out not to be a defect: the box on "10.000 km"/"1 Jahr"/"15" is the wanted pattern, and every value the user can change should carry it. All .feld inputs and selects now get the grey iOS fill; inputs with their own visual language (checkbox/radio/range/file) keep the transparent base rule, so the switches are unaffected. Fuel stations are now displayed as "Shell, Pascalstr. 8, Ingolstadt" instead of the bare operator name. New _marke()/_ist_strasse()/_tankstelle() helpers are shared by both parser paths so Shell and non-Shell receipts format the same. The brand is only matched against the receipt header - searching the whole text would let "Total" hit a totals line. Missing parts are dropped instead of leaving empty comma slots, and without a known brand the operator name takes its place. station_address still carries the full street and postcode. test_bekannte_stationen updated to the new format; suite stays green. installationspaket/ is versioned from now on (user request). It contains no real vehicle data - only the example profile with empty FIN/plate placeholders. Keep it in sync whenever pyscript/ or www/ changes. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,196 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user