pyscript-Backend zur echten HA-Integration umgebaut (HACS-fähig)
Das Backend liegt jetzt als custom_components/audi_dashboard/ vor - eine normale Home-Assistant-Integration mit Config-Flow, einer sensor-Plattform und 18 Diensten. Damit ist die App über HACS installierbar; bis das Repo auf GitHub gespiegelt ist (HACS spricht ausschließlich mit GitHub), installiert homeassistant/installationspaket/install.ps1 denselben Ordner ohne HACS. Fünf Installationsschritte entfallen ersatzlos: der pyscript:-Block, der panel_custom:-Block, das Kopieren der Oberfläche nach www/, das langlebige Zugriffstoken (der Verlauf wird direkt über die recorder-API gelesen) und "pip install pypdf" (steht in manifest.json). Das Fahrzeugprofil legt die Integration beim ersten Start aus ihrer Vorlage an. Drei alte Schwächen sind dabei mit erledigt: - Die Nutzlast landet nicht mehr in der Recorder-Datenbank (_unrecorded_attributes - das kann nur eine echte Entität). - Eine laufende Fahrt überlebt einen Neustart (Store statt Arbeitsspeicher); fiel sie während eines Ausfalls ins Ende, schließt nach_neustart_fortsetzen() sie beim letzten aufgezeichneten Zeitpunkt. - Sensor-Zuordnungen wirken sofort - die Zustandsbeobachter werden neu gebunden, der Neustart-Hinweis und der Neustart-Dienst sind weg. Namensvertrag geändert, beide Oberflächen mitgezogen: pyscript.audi_dashboard_x -> sensor.audi_dashboard_x, pyscript.audi_dashboard_y -> audi_dashboard.y. Eine Companion-App vom alten Stand findet nach dem Umstieg nichts mehr und muss neu gebaut werden; das Panel liegt in der Integration und kann nicht driften. Der selbstgebaute Updater entfällt - HACS ist die Update-Mechanik, die Home Assistant kennt. Die Versionierung schrumpft auf eine Quelle: manifest.json. Geprüft am laufenden Testcontainer (Container byteweise identisch mit dem Repo): alle 18 Dienste, Panel, Config-Entry neu laden, Historienimport, echter Shell-Beleg in-process, Neuinstallation im Wegwerf-Container blank mit automatisch nachinstalliertem pypdf. Companion-App: tsc sauber, 112/112 Tests, beide Rauchtests gegen das laufende Backend grün. Belegparser 8/8. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
@@ -0,0 +1,127 @@
|
||||
"""Lesezugriff auf den aufgezeichneten Zustandsverlauf (recorder).
|
||||
|
||||
Über Home Assistants eigene recorder-API
|
||||
(`homeassistant.components.recorder.history.get_significant_states`), nicht
|
||||
über direkten SQL-Zugriff auf home-assistant_v2.db: das Datenbankschema des
|
||||
recorders ist HA-intern und ändert sich zwischen Versionen, die Funktion
|
||||
dagegen ist die von HA selbst benutzte und stabile Schnittstelle.
|
||||
|
||||
`significant_changes_only=False` ist wichtig: bei numerischen Sensoren
|
||||
(Kilometerstand, Tankfüllstand) liefert der Standardmodus nur "auffällige"
|
||||
Änderungen und verschluckt genau die kleinen Schritte, aus denen sich
|
||||
Fahrstrecke und Tankvorgänge zusammensetzen.
|
||||
|
||||
WAS HIER GEGENÜBER DER PYSCRIPT-FASSUNG WEGFÄLLT: das Fahrtabschluss-
|
||||
Screening las den Verlauf früher über die HTTP-REST-API (/api/history/period)
|
||||
und brauchte dafür ein langlebiges Zugriffstoken in
|
||||
audi_dashboard/ha_token.txt - ein eigener Installationsschritt, der bei jeder
|
||||
Neuinstallation vergessen werden konnte und dessen Fehlen sich nur als
|
||||
Warnung im Protokoll zeigte (Fahrten blieben dann stumm ohne Strecke). Als
|
||||
echte Integration liest die App den Verlauf direkt; Token und
|
||||
Installationsschritt entfallen ersatzlos.
|
||||
|
||||
Die recorder-Abfrage läuft über den Executor: ein Datenbankzugriff hat im
|
||||
Event-Loop nichts verloren.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import logging
|
||||
|
||||
from homeassistant.components.recorder import get_instance, history
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
Verlaufspunkt = tuple[datetime.datetime, str]
|
||||
|
||||
|
||||
def _rohverlauf(
|
||||
hass: HomeAssistant, entity_id: str, start: datetime.datetime, ende: datetime.datetime
|
||||
) -> list[Verlaufspunkt]:
|
||||
roh = history.get_significant_states(
|
||||
hass, start, ende, [entity_id], None, True, False
|
||||
)
|
||||
ergebnis: list[Verlaufspunkt] = []
|
||||
for zustand in roh.get(entity_id) or []:
|
||||
# "unknown"/"unavailable" bedeuten "keine Meldung", nicht "Wert 0" -
|
||||
# würden sie durchgereicht, ergäbe ein Ausfall der Datenquelle eine
|
||||
# Fahrt mit absurder Kilometerdifferenz.
|
||||
if zustand.state in ("unknown", "unavailable", None, ""):
|
||||
continue
|
||||
ergebnis.append((zustand.last_updated, zustand.state))
|
||||
ergebnis.sort(key=lambda p: p[0])
|
||||
return ergebnis
|
||||
|
||||
|
||||
async def verlauf_lesen(
|
||||
hass: HomeAssistant,
|
||||
entity_id: str | None,
|
||||
start: datetime.datetime,
|
||||
ende: datetime.datetime,
|
||||
) -> list[Verlaufspunkt]:
|
||||
"""Zustandsverlauf einer Entität als aufsteigende Liste von
|
||||
(Zeitpunkt, Rohwert). Leere Liste, wenn die Entität nicht zugeordnet ist,
|
||||
im Zeitraum nichts vorliegt oder der recorder nicht erreichbar ist."""
|
||||
if not entity_id:
|
||||
return []
|
||||
try:
|
||||
return await get_instance(hass).async_add_executor_job(
|
||||
_rohverlauf, hass, entity_id, start, ende
|
||||
)
|
||||
except Exception: # noqa: BLE001 - der Verlauf ist Beiwerk, nie der Kern
|
||||
# Ohne Verlauf bleibt eine Fahrt ohne Strecke bzw. der Import leer -
|
||||
# beides ist verkraftbar. Den aufrufenden Ablauf deswegen abzubrechen
|
||||
# wäre es nicht.
|
||||
_LOGGER.warning("Verlauf von %s nicht lesbar", entity_id, exc_info=True)
|
||||
return []
|
||||
|
||||
|
||||
def zahl(wert: object) -> float | None:
|
||||
try:
|
||||
return float(wert) # type: ignore[arg-type]
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def wert_bei(
|
||||
verlauf: list[Verlaufspunkt], zeitpunkt: datetime.datetime
|
||||
) -> float | None:
|
||||
"""Der zuletzt vor `zeitpunkt` gemeldete Zahlenwert, sonst der erste
|
||||
danach, sonst None.
|
||||
|
||||
"Zuletzt davor" ist die richtige Wahl für einen Zählerstand: der
|
||||
Kilometerstand bei Fahrtbeginn ist der, der zuletzt gemeldet wurde, nicht
|
||||
der nächste (der schon Strecke enthält)."""
|
||||
davor = None
|
||||
for ts, wert in verlauf:
|
||||
gezahlt = zahl(wert)
|
||||
if gezahlt is None:
|
||||
continue
|
||||
if ts <= zeitpunkt:
|
||||
davor = gezahlt
|
||||
else:
|
||||
return davor if davor is not None else gezahlt
|
||||
return davor
|
||||
|
||||
|
||||
def naechster_wert(
|
||||
zielzeit: datetime.datetime, verlauf: list[Verlaufspunkt]
|
||||
) -> float | None:
|
||||
"""Der Wert, dessen Zeitstempel am nächsten an `zielzeit` liegt.
|
||||
|
||||
Anders als wert_bei() ist hier bewusst egal, ob der Wert davor oder danach
|
||||
liegt: der Kilometerstand kommt laut Datenquelle nicht sicher mit
|
||||
Fahrtende, sondern teils erst mit Beginn der nächsten Fahrt."""
|
||||
bester_wert = None
|
||||
beste_diff = None
|
||||
for ts, wert in verlauf:
|
||||
gezahlt = zahl(wert)
|
||||
if gezahlt is None:
|
||||
continue
|
||||
diff = abs((ts - zielzeit).total_seconds())
|
||||
if beste_diff is None or diff < beste_diff:
|
||||
bester_wert = gezahlt
|
||||
beste_diff = diff
|
||||
return bester_wert
|
||||
Reference in New Issue
Block a user