audi-app/custom_components/audi_dashboard/dienste.py

"""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,
    belege,
    bilder,
    fahrterkennung,
    historienimport,
    reifen,
    screening,
    sicherung,
)
from .const import (
    BILDER_ORDNER,
    CONF_GITEA_TOKEN,
    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,
    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,
}

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,
    }),
    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."""
        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",
        )

    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)
            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:
        """Lädt die neueste Fassung von Gitea und ersetzt den
        Integrationsordner. Home Assistant muss danach von Hand neu
        gestartet werden - 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)
            k.letzte_update_pruefung = {
                "verfuegbar": False,
                "version": ergebnis["version"],
                "geprueft_am": jetzt,
                "installiert": True,
                "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,
                "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_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))
        ),
        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)