"""Die Dienste, die beide Oberflächen aufrufen. Aus `pyscript.audi_dashboard_` ist `audi_dashboard.` 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()