"""Datenzugriff für Fahrzeugprofil, Fahrten und Tankvorgänge (Bauauftrag §6). Importierbar aus anderen pyscript-Dateien mit `import profil`. Drei getrennte Bestände, wie in §6.1 festgelegt: - Fahrzeugprofil: eine JSON-Datei, alles Fahrzeugspezifische - Fahrten: JSON Lines, eine Zeile je Fahrt - Tankvorgänge: JSON Lines Drei Eigenheiten von pyscript, alle an einer echten Testinstanz beobachtet, nicht nur aus der Dokumentation übernommen: 1. Das eingebaute `open()` existiert in pyscript nicht (NameError) - aus Sicherheitsgründen bewusst nicht freigegeben. Der funktionierende Weg ist `task.executor(io.open, pfad, modus)`: `io.open` ist eine echte externe Funktion aus der Standardbibliothek, kein im pyscript-Ordner selbst definierter Code. 2. task.executor() akzeptiert generell nur solche echten externen Funktionen - eigene, im pyscript-Ordner definierte Funktionen weist es mit "pyscript functions can't be called from task.executor" zurück. Deshalb hier nur `io.open` selbst über den Executor, Lesen/Schreiben/Schließen auf dem zurückgegebenen Datei-Objekt direkt (das ist kein bare-name-Aufruf mehr, sondern ein Methodenaufruf auf einem bereits vorhandenen Objekt). 3. Variablen, die innerhalb eines `with ... as f:`-Blocks zugewiesen werden, waren danach außerhalb nicht mehr auffindbar (NameError), deshalb kein `with` - offen/lesen/schließen nacheinander. """ import io import json import os import uuid BASIS = "/config/audi_dashboard" PROFIL_PFAD = f"{BASIS}/fahrzeugprofil.json" FAHRTEN_PFAD = f"{BASIS}/fahrten.jsonl" TANKVORGAENGE_PFAD = f"{BASIS}/tankvorgaenge.jsonl" BATTERIEVERLAUF_PFAD = f"{BASIS}/batteriespannung.jsonl" BELEGE_ORDNER = f"{BASIS}/belege" # ---------------------------------------------------------------- Ordner --- def ordner_sicherstellen(): os.makedirs(BASIS, exist_ok=True) os.makedirs(BELEGE_ORDNER, exist_ok=True) # ----------------------------------------------------------- Fahrzeugprofil def profil_lesen(): """Liest das Fahrzeugprofil, oder None wenn es fehlt bzw. beschädigt ist. Ohne diese Prüfung reißt eine fehlende Datei (Installation unvollständig, siehe INSTALL.md Schritt 2) jeden Trigger und jeden Service mit, der das Profil braucht — bei laufenden Zeittriggern also im Minutentakt. Jeder Aufrufer muss den None-Fall abfangen.""" if not os.path.exists(PROFIL_PFAD): log.error( f"audi_dashboard: {PROFIL_PFAD} fehlt. Siehe INSTALL.md Schritt 2 — " "bis dahin bleiben alle Funktionen aus, die das Profil brauchen." ) return None f = task.executor(io.open, PROFIL_PFAD, "r") inhalt = f.read() f.close() try: return json.loads(inhalt) except ValueError as fehler: log.error( f"audi_dashboard: {PROFIL_PFAD} ist kein gültiges JSON ({fehler}). " "Letztes Backup aus audi_dashboard/backups/ zurückspielen." ) return None def profil_schreiben(profil): tmp = PROFIL_PFAD + ".tmp" text = json.dumps(profil, ensure_ascii=False, indent=2) f = task.executor(io.open, tmp, "w") f.write(text) f.close() os.replace(tmp, PROFIL_PFAD) # --------------------------------------------------------- JSON-Lines-Basis def _zeilen_lesen(pfad): if not os.path.exists(pfad): return [] f = task.executor(io.open, pfad, "r") inhalt = f.read() f.close() datensaetze = [] for zeile in inhalt.splitlines(): zeile = zeile.strip() if zeile: datensaetze.append(json.loads(zeile)) return datensaetze def _zeilen_schreiben(pfad, datensaetze): tmp = pfad + ".tmp" zeilen = [json.dumps(d, ensure_ascii=False) for d in datensaetze] text = "\n".join(zeilen) if zeilen: text += "\n" f = task.executor(io.open, tmp, "w") f.write(text) f.close() os.replace(tmp, pfad) def _zeile_anhaengen(pfad, datensatz): f = task.executor(io.open, pfad, "a") f.write(json.dumps(datensatz, ensure_ascii=False) + "\n") f.close() # ------------------------------------------------------------------ Fahrten def fahrten_lesen(): return _zeilen_lesen(FAHRTEN_PFAD) def fahrten_schreiben(fahrten): _zeilen_schreiben(FAHRTEN_PFAD, fahrten) def fahrt_anhaengen(fahrt): _zeile_anhaengen(FAHRTEN_PFAD, fahrt) def offene_fahrten(): """Alle Fahrten mit status == 'offen', ältere zuerst.""" fahrten = fahrten_lesen() offen = [f for f in fahrten if f.get("status") == "offen"] return sorted(offen, key=lambda f: f.get("ts_start", "")) def _datensatz_aktualisieren(zeilen, id_feld, id_wert, aenderungen, schreiben, schutz=True): """Ersetzt ausgewählte Felder eines Datensatzes anhand seiner ID und schreibt den gesamten Bestand neu. Manuell geänderte Felder (edited_fields, analog zu §7.7 Regel 3) werden dabei nie überschrieben. Gemeinsame Grundlage für Fahrten und Tankvorgänge - beide Archive funktionieren nach demselben Muster. schutz=False hebt genau diese Sperre auf - nötig für Eingaben aus der Oberfläche: edited_fields schützt gegen die automatische Ergänzung, nicht gegen den Menschen, der das Feld gerade selbst korrigiert.""" geaendert = False for d in zeilen: if d.get(id_feld) == id_wert: geschuetzt = set(d.get("edited_fields", [])) if schutz else set() for feld, wert in aenderungen.items(): if feld not in geschuetzt: d[feld] = wert geaendert = True break if geaendert: schreiben(zeilen) return geaendert def fahrt_aktualisieren(trip_id, aenderungen): return _datensatz_aktualisieren(fahrten_lesen(), "trip_id", trip_id, aenderungen, fahrten_schreiben) def fahrt_bearbeiten(trip_id, aenderungen): """Wie fahrt_aktualisieren(), aber für Eingaben aus der Oberfläche: eine von Hand gesetzte Angabe sticht auch dann, wenn dasselbe Feld schon einmal von Hand gesetzt wurde (siehe schutz-Parameter oben).""" return _datensatz_aktualisieren(fahrten_lesen(), "trip_id", trip_id, aenderungen, fahrten_schreiben, schutz=False) def fahrt_loeschen(trip_id): fahrten = fahrten_lesen() uebrig = [f for f in fahrten if f.get("trip_id") != trip_id] if len(uebrig) == len(fahrten): return False fahrten_schreiben(uebrig) return True # ------------------------------------------------------------ Tankvorgänge def tankvorgaenge_lesen(): return _zeilen_lesen(TANKVORGAENGE_PFAD) def tankvorgaenge_schreiben(tankvorgaenge): _zeilen_schreiben(TANKVORGAENGE_PFAD, tankvorgaenge) def tankvorgang_anhaengen(tankvorgang): _zeile_anhaengen(TANKVORGAENGE_PFAD, tankvorgang) def tankvorgang_aktualisieren(tank_id, aenderungen): return _datensatz_aktualisieren(tankvorgaenge_lesen(), "tank_id", tank_id, aenderungen, tankvorgaenge_schreiben) def tankvorgang_loeschen(tank_id): tankvorgaenge = tankvorgaenge_lesen() uebrig = [t for t in tankvorgaenge if t.get("tank_id") != tank_id] if len(uebrig) == len(tankvorgaenge): return False tankvorgaenge_schreiben(uebrig) return True def letzter_tankvorgang(): """Der zeitlich jüngste bereits erfasste Tankvorgang (nach ts), oder None, falls noch keiner existiert. Grundlage für die Gefahrene-Distanz-Berechnung beim Anlegen eines neuen Tankvorgangs (§5.5).""" tankvorgaenge = tankvorgaenge_lesen() if not tankvorgaenge: return None return max(tankvorgaenge, key=lambda t: t.get("ts") or "") def distanz_seit_letzter_tankung(aktueller_km): """Gefahrene Distanz seit dem vorherigen Tankvorgang, als Vorschlag für das gleichnamige Formularfeld (§5.5) - frei überschreibbar, genau wie odometer_km selbst. None, wenn kein Kilometerstand oder kein vorheriger Tankvorgang vorliegt (erster Eintrag überhaupt). Gemeinsame Grundlage für Beleg-Erfassung (belegverarbeitung.py) und automatische Tankerkennung (tankerkennung.py).""" if aktueller_km is None: return None letzter = letzter_tankvorgang() if not letzter or letzter.get("odometer_km") is None: return None return round(aktueller_km - letzter["odometer_km"], 1) def tankvorgang_nach_id(tank_id): for t in tankvorgaenge_lesen(): if t.get("tank_id") == tank_id: return t return None def tankvorgang_nach_receipt_key(receipt_key): """Für §7.7 Regel: derselbe Beleg (receipt_key, minutengenau) erzeugt keinen zweiten Datensatz.""" for t in tankvorgaenge_lesen(): if t.get("receipt_key") == receipt_key: return t return None # ------------------------------------------------------- Batteriespannung def batterieverlauf_lesen(): """Ein Eintrag pro Tag ({datum, min, min_ts, max, max_ts}), älteste zuerst - siehe batterieverlauf.py für die Aufzeichnungslogik. min_ts/ max_ts sind die Zeitstempel (ISO, UTC) der jeweiligen Einzelmessung, für die Datum/Uhrzeit-Anzeige beim Antippen des Diagrammpunkts im Frontend - der Punkt selbst zeigt nur den Minimalwert (siehe dortiger Kommentar, warum der aussagekräftig für die Entladung ist).""" return _zeilen_lesen(BATTERIEVERLAUF_PFAD) def batterieverlauf_tageswert_aktualisieren(datum, ts, spannung): """Trägt eine neue Messung in den Tageseintrag für `datum` ein: legt ihn beim ersten Wert des Tages an, erweitert sonst nur min/max samt dem Zeitstempel der jeweils neuen Extremmessung. Das Fahrzeug meldet die Spannung künftig mehrfach pro Stunde (aktive Fahrt) statt nur einmal täglich - der komplette Bestand wird deshalb bei jeder Messung neu geschrieben (wie bei den übrigen JSON-Lines-Beständen hier), was bei einem Eintrag pro Tag über Jahre hinweg unproblematisch bleibt.""" verlauf = batterieverlauf_lesen() for eintrag in verlauf: if eintrag.get("datum") == datum: if spannung < eintrag["min"]: eintrag["min"] = spannung eintrag["min_ts"] = ts if spannung > eintrag["max"]: eintrag["max"] = spannung eintrag["max_ts"] = ts break else: verlauf.append({"datum": datum, "min": spannung, "min_ts": ts, "max": spannung, "max_ts": ts}) _zeilen_schreiben(BATTERIEVERLAUF_PFAD, verlauf) # --------------------------------------------------------------------- IDs def neue_id(praefix): return f"{praefix}-{uuid.uuid4().hex[:12]}"