audi-app/homeassistant/pyscript/modules/profil.py

"""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():
    f = task.executor(io.open, PROFIL_PFAD, "r")
    inhalt = f.read()
    f.close()
    return json.loads(inhalt)


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):
    """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."""
    geaendert = False
    for d in zeilen:
        if d.get(id_feld) == id_wert:
            geschuetzt = set(d.get("edited_fields", []))
            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_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]}"