paul/audi-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2a4337e030b15e08fec70e05adf24dd04deec4de
audi-app/custom_components/audi_dashboard/dienste.py
T
tobias 462970251e Meldung verschwindet durch das Neuladen, zu dem sie auffordert 
Vorgabe des Eigentuemers: klarer Zustand statt einer Meldung, die auch nach
dem Neuladen stehenbleibt.

Panel: nach einem Update ohne Neustart wird der Sidebar-Eintrag mit der neuen
Versionsnummer neu angemeldet (koordinator.panel_neu_anmelden) - ohne das
traegt die Modul-URL weiter die alte Nummer. Das Panel vergleicht jetzt die
Version aus seiner Modul-URL mit der von der Platte gelesenen: sind sie
gleich, laeuft es bereits auf der installierten Fassung und die Meldung
entfaellt. Sie verschwindet damit genau durch das Neuladen, zu dem sie
auffordert.

App: fuer ein Update ohne Neustart erscheint gar nichts mehr. Geaendert wurden
Dateien des Panels; die App hat hier nichts zu tun. Damit gilt durchgehend:
keine Meldung = alles in Ordnung, Neustart-Knopf = handeln.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-05 15:26:19 +02:00

483 lines
20 KiB
Python
Raw Blame History

"""Die Dienste, die beide Oberflächen aufrufen.
Aus `pyscript.audi_dashboard_<name>` ist `audi_dashboard.<name>` geworden -
der doppelte Präfix war nur nötig, weil alle pyscript-Dienste sich eine Domain
teilten. Die Namen selbst und die Feldnamen sind unverändert; beide
Oberflächen führen dieselbe Liste in je einer zentralen Tabelle.
Die Schemata sind bewusst großzügig: die Oberfläche schickt Zahlenfelder je
nach Formularzustand mal als Zahl, mal gar nicht. Ein zu enges Schema würde
einen Dienstaufruf ablehnen, und ein abgelehnter Aufruf ist in der Oberfläche
unsichtbar - er landet nur in der Browser-Konsole. Was inhaltlich nicht passt,
weisen deshalb die Fachfunktionen mit einer verständlichen Protokollzeile ab,
nicht die Schema-Prüfung.
"""
from __future__ import annotations
import datetime
import json
import logging
from collections.abc import Callable, Coroutine
from typing import Any
import voluptuous as vol
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import config_validation as cv
from . import (
aktualisierung,
batterie,
belege,
bilder,
csv_import,
fahrterkennung,
flespi,
historienimport,
identitaet,
reifen,
screening,
sicherung,
zugaenge,
)
from .const import (
BILDER_ORDNER,
CONF_GITEA_TOKEN,
D_BACKUP_JETZT,
D_BACKUP_WIEDERHERSTELLEN,
D_BATTERIEVERLAUF_LOESCHEN,
D_BELEG_HOCHLADEN,
D_BILD_HOCHLADEN,
D_BILD_LOESCHEN,
D_CSV_IMPORTIEREN,
D_ENTITAETEN_SCHREIBEN,
D_FAHRT_AKTUALISIEREN,
D_FAHRT_JETZT_BEENDEN,
D_DONGLE_LESEN,
D_ZUGANG_SETZEN,
D_FAHRT_LOESCHEN,
D_FAHRT_MANUELL_ANLEGEN,
D_HISTORIE_IMPORTIEREN,
D_JETZT_AKTUALISIEREN,
D_PROFIL_SCHREIBEN,
D_REIFEN_ARCHIV_AKTUALISIEREN,
D_REIFEN_ARCHIV_LOESCHEN,
D_REIFEN_ARCHIVIEREN,
D_REIFEN_KM_SETZEN,
D_REIFEN_WECHSELN,
D_SCREENING_JETZT,
D_TANKVORGANG_AKTUALISIEREN,
D_TANKVORGANG_LOESCHEN,
D_TANKVORGANG_MANUELL,
D_UPDATE_INSTALLIEREN,
D_UPDATE_PRUEFEN,
DOMAIN,
)
from .koordinator import Koordinator
_LOGGER = logging.getLogger(__name__)
# Ein JSON-Feld darf als Zeichenkette oder schon ausgepackt ankommen: das
# Panel schickt JSON.stringify(...), ein Aufruf aus einer Automation heraus
# reicht dagegen naheliegenderweise die Struktur selbst durch.
_JSON = vol.Any(cv.string, dict, list)
_ZAHL = vol.Any(vol.Coerce(float), None)
_TEXT = vol.Any(cv.string, None)
_FAHRT_FELDER = {
vol.Optional("art"): _TEXT,
vol.Optional("start_ort"): _TEXT,
vol.Optional("ziel_ort"): _TEXT,
vol.Optional("odo_start"): _ZAHL,
vol.Optional("odo_end"): _ZAHL,
vol.Optional("distanz"): _ZAHL,
}
_TANK_FELDER = {
vol.Optional("ts"): _TEXT,
vol.Optional("liter"): _ZAHL,
vol.Optional("kosten"): _ZAHL,
vol.Optional("km"): _ZAHL,
vol.Optional("ersparnis"): _ZAHL,
vol.Optional("station"): _TEXT,
vol.Optional("distanz"): _ZAHL,
vol.Optional("kraftstoff"): _TEXT,
vol.Optional("receipt_key"): _TEXT,
vol.Optional("receipt_file"): _TEXT,
}
_REIFEN_ARCHIV_FELDER = {
vol.Optional("ersetzt_am"): _TEXT,
vol.Optional("km"): _ZAHL,
vol.Optional("marke"): _TEXT,
vol.Optional("modell"): _TEXT,
vol.Optional("dot"): _TEXT,
vol.Optional("mass"): _TEXT,
vol.Optional("druck_vorne"): _TEXT,
vol.Optional("druck_hinten"): _TEXT,
vol.Optional("kommentar"): _TEXT,
}
SCHEMATA: dict[str, vol.Schema] = {
D_JETZT_AKTUALISIEREN: vol.Schema({}),
D_SCREENING_JETZT: vol.Schema({}),
D_BACKUP_JETZT: vol.Schema({}),
D_PROFIL_SCHREIBEN: vol.Schema({vol.Required("profil_json"): _JSON}),
D_ENTITAETEN_SCHREIBEN: vol.Schema({vol.Required("zuordnung_json"): _JSON}),
D_FAHRT_MANUELL_ANLEGEN: vol.Schema({
vol.Required("ts_start"): cv.string,
vol.Required("ts_end"): cv.string,
**_FAHRT_FELDER,
}),
D_FAHRT_AKTUALISIEREN: vol.Schema({
vol.Required("trip_id"): cv.string,
vol.Optional("ts_start"): _TEXT,
vol.Optional("ts_end"): _TEXT,
**_FAHRT_FELDER,
}),
D_FAHRT_LOESCHEN: vol.Schema({vol.Required("trip_id"): cv.string}),
D_FAHRT_JETZT_BEENDEN: vol.Schema({}),
D_DONGLE_LESEN: vol.Schema({}),
D_ZUGANG_SETZEN: vol.Schema({
vol.Required("dienst"): cv.string,
# Leerer Text loescht - ein eigener Loesch-Dienst waere ein zweiter Weg
# zu derselben Wirkung.
vol.Optional("token", default=""): cv.string,
}),
D_BELEG_HOCHLADEN: vol.Schema({
vol.Required("pdf_base64"): cv.string,
vol.Optional("dateiname"): _TEXT,
vol.Optional("tank_id"): _TEXT,
}),
D_TANKVORGANG_MANUELL: vol.Schema(dict(_TANK_FELDER)),
D_TANKVORGANG_AKTUALISIEREN: vol.Schema({
vol.Required("tank_id"): cv.string, **_TANK_FELDER
}),
D_TANKVORGANG_LOESCHEN: vol.Schema({vol.Required("tank_id"): cv.string}),
D_BILD_HOCHLADEN: vol.Schema({
vol.Required("dateiname"): cv.string,
vol.Required("daten_base64"): cv.string,
}),
D_BILD_LOESCHEN: vol.Schema({vol.Required("dateiname"): cv.string}),
D_REIFEN_WECHSELN: vol.Schema({vol.Required("satz"): vol.In(reifen.SAETZE)}),
D_REIFEN_KM_SETZEN: vol.Schema({
vol.Required("satz"): vol.In(reifen.SAETZE),
vol.Required("km"): vol.Coerce(float),
}),
D_REIFEN_ARCHIVIEREN: vol.Schema({vol.Required("satz"): vol.In(reifen.SAETZE)}),
D_REIFEN_ARCHIV_AKTUALISIEREN: vol.Schema({
vol.Required("id"): cv.string, **_REIFEN_ARCHIV_FELDER
}),
D_REIFEN_ARCHIV_LOESCHEN: vol.Schema({vol.Required("id"): cv.string}),
D_BATTERIEVERLAUF_LOESCHEN: vol.Schema({vol.Required("datum"): cv.string}),
D_BACKUP_WIEDERHERSTELLEN: vol.Schema({
vol.Optional("profil_json"): _JSON,
vol.Optional("fahrten_json"): _JSON,
vol.Optional("tankvorgaenge_json"): _JSON,
vol.Optional("entitaeten_json"): _JSON,
}),
D_HISTORIE_IMPORTIEREN: vol.Schema({
vol.Required("start"): cv.string,
vol.Required("ende"): cv.string,
}),
D_CSV_IMPORTIEREN: vol.Schema({
vol.Required("art"): vol.In(["fahrten", "tanken", "service"]),
vol.Required("inhalt"): cv.string,
}),
D_UPDATE_PRUEFEN: vol.Schema({}),
D_UPDATE_INSTALLIEREN: vol.Schema({}),
}
Fabrik = Callable[[ServiceCall], Coroutine[Any, Any, None]]
def _als_dienst(fabrik: Fabrik) -> Callable[[ServiceCall], Coroutine[Any, Any, None]]:
"""Macht aus einer Funktion, die eine Koroutine zurückgibt, eine echte
Koroutinen-Funktion.
Home Assistant prüft mit iscoroutinefunction(), ob es einen Dienst
erwarten muss. Ein Lambda, das eine Koroutine nur zurückgibt, besteht
diese Prüfung nicht - der Aufruf würde stillschweigend nichts tun."""
async def behandler(aufruf: ServiceCall) -> None:
await fabrik(aufruf)
return behandler
def _objekt(wert: object) -> Any:
return json.loads(wert) if isinstance(wert, str) else wert
def registrieren(hass: HomeAssistant, k: Koordinator) -> None:
"""Meldet alle Dienste an. Gegenstück: abmelden()."""
async def jetzt_aktualisieren(_aufruf: ServiceCall) -> None:
"""Gegenstück zum "Ziehen zum Aktualisieren": stößt den von der
Datenquelle bereitgestellten Refresh-Knopf an, statt auf den nächsten
Takt zu warten. Ohne eingetragenen Knopf nur die direkte
Neuveröffentlichung - besser als gar nichts, auch wenn dann kein
echter neuer Abruf beim Fahrzeug passiert."""
knopf = k.zuordnung.werte.REFRESH_BUTTON
if knopf:
await hass.services.async_call(
"button", "press", {"entity_id": knopf}, blocking=True
)
await k.fahrzeugstatus_veroeffentlichen()
async def profil_schreiben(aufruf: ServiceCall) -> None:
"""Überschreibt das komplette Fahrzeugprofil.
Die Oberfläche hält Profil, Fahrzeugdaten und Einstellungen ohnehin
als einen zusammenhängenden Stand im Speicher und schickt ihn nach
jeder Änderung komplett zurück - deshalb ein einziger Dienst statt
eines eigenen je Feld.
Danach auch ein Versuch, die Fahrgestellnummer zu synchronisieren
(identitaet.py) - unter anderem der Weg, über den das Umschalten von
"automatisch" selbst wirkt: dieser Dienst schreibt auch das gerade
geänderte fin_automatisch-Flag, bevor die Ableitung es liest."""
await k.ablage.profil_schreiben(_objekt(aufruf.data["profil_json"]))
await identitaet.fin_ableiten_und_uebernehmen(k.hass, k.zuordnung, k.ablage)
await k.profil_veroeffentlichen()
# Die Reglerstellung "Fahrt beenden" ist zugleich der Sollwert für den
# Schlaf-Timeout des Dongles (103) - siehe flespi.py. Muss NACH
# profil_veroeffentlichen() stehen: erst dort liest der Koordinator
# die neue Reglerstellung mit.
#
# Als eigene Aufgabe und ausdrücklich NICHT abgewartet: an diesem
# Dienst hängen ~40 Aufrufstellen im Panel, eine je Feldänderung, und
# jede würde sonst auf bis zu drei HTTP-Runden zu flespi warten. Das
# Speichern selbst ist zu diesem Zeitpunkt längst erledigt.
k.hass.async_create_task(flespi.wunsch_uebernehmen(k))
async def dongle_lesen(_aufruf: ServiceCall) -> None:
"""Liest die Konfiguration des Dongles - auf ausdrücklichen Knopfdruck.
Nötig ist der Knopf nicht: flespi antwortet auch bei schlafendem Gerät
aus seinem eigenen Speicher (siehe flespi.py). Er ist dafür da,
nachzusehen, ob eine geschriebene Änderung inzwischen am Gerät
angekommen ist.
Ein Fehler wird als Anzeigestand hinterlegt statt geworfen: er gehört
in die Kachel, aus der der Knopf gedrückt wurde, nicht ins Protokoll."""
try:
await flespi.jetzt_lesen(k)
except flespi.FlespiFehler as fehler:
_LOGGER.warning("Geräte-Konfiguration nicht gelesen: %s", fehler)
await k.flespi_stand_setzen(
{"fehler": str(fehler), "gelesen_am": flespi.jetzt_iso()}
)
async def entitaeten_schreiben(aufruf: ServiceCall) -> None:
"""Speichert die im Setup-Menü gewählte Sensor-Zuordnung, wendet sie
sofort an, bindet die Zustandsbeobachter neu und veröffentlicht
Katalog und Status neu.
Das Neubinden ist der Grund, warum der frühere Hinweis "wirkt erst
nach einem Neustart" und der Neustart-Knopf daneben entfallen konnten:
auch Zündung, Kilometerstand und Tankfüllstand wirken jetzt sofort.
Danach auch ein erneuter Versuch, die Fahrgestellnummer aus den
(jetzt ggf. neuen) zugeordneten Sensoren abzuleiten (identitaet.py) -
eine leere FIN kann durch diese Zuordnung gerade erst ableitbar
geworden sein."""
await k.zuordnung.speichern(_objekt(aufruf.data["zuordnung_json"]))
await k.trigger_neu_binden()
await k.zuordnung_veroeffentlichen()
await k.fahrzeugstatus_veroeffentlichen()
await identitaet.fin_ableiten_und_uebernehmen(k.hass, k.zuordnung, k.ablage)
await k.profil_veroeffentlichen()
async def historie_importieren(aufruf: ServiceCall) -> None:
"""Startet den Import und kehrt sofort zurück.
Der Import kann bei einem Jahr Verlauf Minuten dauern. Die Oberfläche
wartet deshalb nicht auf den Dienstaufruf, sondern verfolgt den
Fortschritt über sensor.audi_dashboard_import_status - genau so, wie
sie es schon bei der pyscript-Fassung tat, deren Dienste ebenfalls
sofort zurückkehrten."""
k.entry.async_create_background_task(
hass,
historienimport.importieren(k, aufruf.data["start"], aufruf.data["ende"]),
"audi_dashboard_historienimport",
)
async def csv_importieren(aufruf: ServiceCall) -> None:
"""Liest eine per "Datensatz laden" hochgeladene CSV-Datei ein (siehe
csv_import.py). Läuft synchron statt im Hintergrund wie
historie_importieren - eine einzelne CSV-Datei ist klein, der Aufruf
kehrt erst zurück, wenn das Ergebnis auch wirklich in
sensor.audi_dashboard_import_status steht, kein Abfragen nötig."""
ergebnis = await csv_import.importieren(k, aufruf.data["art"], aufruf.data["inhalt"])
await k.import_status_veroeffentlichen("csv_fertig", {"art": aufruf.data["art"], **ergebnis})
async def update_pruefen(_aufruf: ServiceCall) -> None:
"""Fragt bei Gitea nach, ob eine neuere Fassung vorliegt - ohne
etwas herunterzuladen. Läuft nur auf Tastendruck, siehe
aktualisierung.py's Moduldocstring."""
token = k.entry.options.get(CONF_GITEA_TOKEN, "")
jetzt = datetime.datetime.now(datetime.timezone.utc).isoformat()
try:
ergebnis = await aktualisierung.version_pruefen(hass, token, k.version)
_LOGGER.info(
"Update-Prüfung: Gitea hat Version %s (installiert: %s, verfügbar: %s)",
ergebnis["version"], k.version, ergebnis["verfuegbar"],
)
k.letzte_update_pruefung = {
"verfuegbar": ergebnis["verfuegbar"],
"version": ergebnis["version"],
"geprueft_am": ergebnis["geprueft_am"],
"installiert": False,
"fehler": None,
}
except aktualisierung.AktualisierungsFehler as fehler:
_LOGGER.warning("Update-Prüfung fehlgeschlagen: %s", fehler)
k.letzte_update_pruefung = {
"verfuegbar": False,
"version": None,
"geprueft_am": jetzt,
"installiert": False,
"fehler": str(fehler),
}
await k.app_version_veroeffentlichen()
async def update_installieren(_aufruf: ServiceCall) -> None:
"""Laedt die neueste Fassung von Gitea und ersetzt den
Integrationsordner.
Ob danach ein Neustart noetig ist, entscheidet der Updater am
Datei-Vergleich (aktualisierung.neustart_noetig): betrifft das Update
nur frontend/, reicht ein Neuladen der Seite - die Oberflaechen-Dateien
werden direkt von der Platte ausgeliefert (StaticPathConfig mit
cache_headers=False). Nur geaenderter Python-Code braucht wirklich
einen Neustart, weil Home Assistant seine Module einmal beim Start
importiert und danach mit lebenden Objekten daran haengt.
Kein automatischer Reload aus dem eigenen, noch laufenden Aufruf
heraus - siehe aktualisierung.py."""
token = k.entry.options.get(CONF_GITEA_TOKEN, "")
jetzt = datetime.datetime.now(datetime.timezone.utc).isoformat()
try:
ergebnis = await aktualisierung.update_installieren(hass, token)
neustart = bool(ergebnis.get("neustart_noetig", True))
_LOGGER.info(
"Update installiert: Version %s - %s",
ergebnis["version"],
"Home Assistant muss neu gestartet werden."
if neustart
else "nur Oberflaechen-Dateien, ein Neuladen der Seite genuegt.",
)
if not neustart:
# Ohne Neustart bleibt Home Assistants zwischengespeichertes
# Manifest stehen - die neue Nummer muss von der Platte kommen.
await k.version_neu_lesen()
# Und die Modul-URL des Panels muss sie tragen, sonst kann das
# Panel nach dem Neuladen nicht erkennen, dass es bereits die
# neuen Dateien ausfuehrt - und die Meldung "Seite neu laden"
# bliebe stehen.
await k.panel_neu_anmelden()
k.letzte_update_pruefung = {
"verfuegbar": False,
"version": ergebnis["version"],
"geprueft_am": jetzt,
"installiert": True,
"neustart_noetig": neustart,
"fehler": None,
}
except aktualisierung.AktualisierungsFehler as fehler:
_LOGGER.warning("Update-Installation fehlgeschlagen: %s", fehler)
k.letzte_update_pruefung = {
"verfuegbar": (k.letzte_update_pruefung or {}).get("verfuegbar", False),
"version": (k.letzte_update_pruefung or {}).get("version"),
"geprueft_am": jetzt,
"installiert": False,
"neustart_noetig": True,
"fehler": str(fehler),
}
await k.app_version_veroeffentlichen()
bilder_ordner = hass.config.path(BILDER_ORDNER)
behandler: dict[str, Callable[[ServiceCall], Coroutine[Any, Any, None]]] = {
D_JETZT_AKTUALISIEREN: jetzt_aktualisieren,
D_PROFIL_SCHREIBEN: profil_schreiben,
D_ENTITAETEN_SCHREIBEN: entitaeten_schreiben,
D_HISTORIE_IMPORTIEREN: historie_importieren,
D_CSV_IMPORTIEREN: csv_importieren,
D_SCREENING_JETZT: _als_dienst(lambda _a: screening.durchfuehren(k)),
D_FAHRT_MANUELL_ANLEGEN: _als_dienst(
lambda a: fahrterkennung.manuell_anlegen(k, dict(a.data))
),
D_FAHRT_AKTUALISIEREN: _als_dienst(
lambda a: fahrterkennung.aktualisieren(k, dict(a.data))
),
D_FAHRT_JETZT_BEENDEN: _als_dienst(lambda _a: fahrterkennung.jetzt_beenden(k)),
D_ZUGANG_SETZEN: _als_dienst(
lambda a: zugaenge.setzen(k, a.data["dienst"], a.data.get("token", ""))
),
D_DONGLE_LESEN: dongle_lesen,
D_FAHRT_LOESCHEN: _als_dienst(
lambda a: fahrterkennung.loeschen(k, a.data["trip_id"])
),
D_BELEG_HOCHLADEN: _als_dienst(lambda a: belege.hochladen(k, dict(a.data))),
D_TANKVORGANG_MANUELL: _als_dienst(
lambda a: belege.tankvorgang_manuell(k, dict(a.data))
),
D_TANKVORGANG_AKTUALISIEREN: _als_dienst(
lambda a: belege.tankvorgang_aktualisieren(k, dict(a.data))
),
D_TANKVORGANG_LOESCHEN: _als_dienst(
lambda a: belege.tankvorgang_loeschen(k, a.data["tank_id"])
),
# Nach dem Schreiben sofort neu veroeffentlichen: der Bildstand
# haengt an der Datei, und ohne diese Zeile saehe die Oberflaeche
# das neue Foto erst beim naechsten Minutentakt.
D_BILD_HOCHLADEN: _als_dienst(
lambda a: _bild_hochladen(k, hass, bilder_ordner, a)
),
D_BILD_LOESCHEN: _als_dienst(
lambda a: _bild_loeschen(k, hass, bilder_ordner, a)
),
D_REIFEN_WECHSELN: _als_dienst(lambda a: reifen.wechseln(k, a.data["satz"])),
D_REIFEN_KM_SETZEN: _als_dienst(
lambda a: reifen.km_setzen(k, a.data["satz"], a.data["km"])
),
D_REIFEN_ARCHIVIEREN: _als_dienst(lambda a: reifen.archivieren(k, a.data["satz"])),
D_REIFEN_ARCHIV_AKTUALISIEREN: _als_dienst(
lambda a: reifen.archiv_aktualisieren(k, a.data["id"], dict(a.data))
),
D_REIFEN_ARCHIV_LOESCHEN: _als_dienst(
lambda a: reifen.archiv_loeschen(k, a.data["id"])
),
D_BATTERIEVERLAUF_LOESCHEN: _als_dienst(
lambda a: batterie.eintrag_loeschen(k, a.data["datum"])
),
D_BACKUP_JETZT: _als_dienst(lambda _a: sicherung.jetzt(k)),
D_BACKUP_WIEDERHERSTELLEN: _als_dienst(
lambda a: sicherung.wiederherstellen(k, dict(a.data))
),
D_UPDATE_PRUEFEN: update_pruefen,
D_UPDATE_INSTALLIEREN: update_installieren,
}
for name, funktion in behandler.items():
hass.services.async_register(DOMAIN, name, funktion, schema=SCHEMATA[name])
def abmelden(hass: HomeAssistant) -> None:
for name in SCHEMATA:
hass.services.async_remove(DOMAIN, name)
async def _bild_hochladen(k, hass, ordner, aufruf) -> None:
await bilder.hochladen(
hass, ordner, aufruf.data["dateiname"], aufruf.data["daten_base64"]
)
await k.app_version_veroeffentlichen()
async def _bild_loeschen(k, hass, ordner, aufruf) -> None:
await bilder.loeschen(hass, ordner, aufruf.data["dateiname"])
await k.app_version_veroeffentlichen()
Reference in New Issue View Git Blame Copy Permalink