"""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 json 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 belege, bilder, fahrterkennung, historienimport, reifen, screening, sicherung from .const import ( BILDER_ORDNER, D_BACKUP_JETZT, D_BACKUP_WIEDERHERSTELLEN, D_BELEG_HOCHLADEN, D_BILD_HOCHLADEN, D_BILD_LOESCHEN, D_ENTITAETEN_SCHREIBEN, D_FAHRT_AKTUALISIEREN, D_FAHRT_LOESCHEN, D_FAHRT_MANUELL_ANLEGEN, D_HISTORIE_IMPORTIEREN, D_JETZT_AKTUALISIEREN, D_PROFIL_SCHREIBEN, D_REIFEN_KM_SETZEN, D_REIFEN_WECHSELN, D_SCREENING_JETZT, D_TANKVORGANG_AKTUALISIEREN, D_TANKVORGANG_LOESCHEN, D_TANKVORGANG_MANUELL, DOMAIN, ) from .koordinator import Koordinator # 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, } 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_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_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, }), } 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.""" await k.ablage.profil_schreiben(_objekt(aufruf.data["profil_json"])) await k.profil_veroeffentlichen() 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.""" await k.zuordnung.speichern(_objekt(aufruf.data["zuordnung_json"])) await k.trigger_neu_binden() await k.zuordnung_veroeffentlichen() await k.fahrzeugstatus_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", ) 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_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_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"]) ), D_BILD_HOCHLADEN: _als_dienst( lambda a: bilder.hochladen( hass, bilder_ordner, a.data["dateiname"], a.data["daten_base64"] ) ), D_BILD_LOESCHEN: _als_dienst( lambda a: bilder.loeschen(hass, bilder_ordner, a.data["dateiname"]) ), 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_BACKUP_JETZT: _als_dienst(lambda _a: sicherung.jetzt(k)), D_BACKUP_WIEDERHERSTELLEN: _als_dienst( lambda a: sicherung.wiederherstellen(k, dict(a.data)) ), } 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)