tobias
6b35de458d
- Neues Profilfeld fahrzeug.fin_automatisch (Standard: an). - identitaet.py: bei "aus" läuft die Ableitung (samt Mehrfach-Sensor-Abgleich) gar nicht erst; bei "an" überschreibt der abgeleitete Wert auch einen vorher von Hand eingetragenen - der Schalter selbst entscheidet jetzt, statt aus einem leeren Feld zu raten. - dienste.py: profil_schreiben löst die Ableitung jetzt auch aus, damit das Umschalten des Schalters selbst sofort wirkt. - Panel: Schalter "automatisch" zwischen Label und Textfeld in der Fahrgestellnummer-Zeile, Feld wird bei "an" deaktiviert. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
401 lines
16 KiB
Python
401 lines
16 KiB
Python
"""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,
|
|
historienimport,
|
|
identitaet,
|
|
reifen,
|
|
screening,
|
|
sicherung,
|
|
)
|
|
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_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_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()
|
|
|
|
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:
|
|
"""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)
|
|
_LOGGER.info(
|
|
"Update installiert: Version %s - Home Assistant muss neu gestartet werden.",
|
|
ergebnis["version"],
|
|
)
|
|
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_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_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_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)
|