"""Der Koordinator: hält den Zustand der App und verdrahtet alles miteinander. Was hier zusammenkommt, lag in der pyscript-Fassung über die Dateien verteilt, weil es dort keinen Ort dafür gab: die Dekoratoren @service, @state_trigger und @time_trigger banden ihre Registrierung direkt an die Funktion daneben. Eine Integration hat einen Lebenszyklus (Einrichten, Entladen, Neu-Laden), und genau deshalb braucht es eine Stelle, die weiß, was sie angemeldet hat und es wieder abmelden kann. DREI DINGE, DIE DIESER AUFBAU MÖGLICH MACHT UND DIE ES VORHER NICHT GAB ----------------------------------------------------------------------- 1. Zustandsbeobachter lassen sich zur Laufzeit neu binden. Früher backte @state_trigger die Entity-ID einmalig beim Modul-Laden ein - ein im Setup-Menü geänderter Zündungs-, Kilometer- oder Tanksensor wirkte erst nach einem HA-Neustart, und die Oberfläche musste einen eigenen Neustart-Knopf dafür anbieten. trigger_neu_binden() ersetzt beides. 2. Der Zwischenstand einer laufenden Fahrt überlebt einen Neustart. Er liegt in einem Store statt nur im Arbeitsspeicher; dasselbe gilt für den Tank-Tiefststand. Beides war als bekannte Schwäche vermerkt. 3. Die Nutzlast landet nicht mehr in der Recorder-Datenbank (siehe sensor.py). Der Weg von hier zur Oberfläche: `setzen()` legt Zustand und Nutzlast ab und schickt ein Signal; die Entität in sensor.py hört darauf und schreibt sich neu. Die Entitäten holen sich ihren Wert beim Anlegen selbst aus `werte` - deshalb darf schon veröffentlicht werden, bevor die Plattform überhaupt eingerichtet ist. """ from __future__ import annotations import asyncio import datetime import json import logging import os from collections.abc import Callable, Coroutine from typing import Any from homeassistant.config_entries import ConfigEntry from homeassistant.core import CALLBACK_TYPE, Event, EventStateChangedData, HomeAssistant from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.event import ( async_track_state_change_event, async_track_time_change, async_track_time_interval, ) from homeassistant.helpers.start import async_at_started from homeassistant.helpers.storage import Store from . import ( batterie, fahrterkennung, identitaet, reifen, screening, verbrauchskorrektur, sicherung, tankerkennung, verlauf, ) from .ablage import Ablage from .geokodierung import Geokodierung from .const import ( BUENDEL_INFO, BUENDEL_ORDNER, BUENDEL_URL, E_APP_VERSION, E_BATTERIEVERLAUF, E_BELEG_ERGEBNIS, E_FAHRTEN, E_FAHRZEUGSTATUS, E_IMPORT_STATUS, E_PROFIL, E_REIFEN_AKTIV, E_REIFEN_SOMMER, E_REIFEN_WINTER, E_TANKVORGAENGE, E_ZUORDNUNG, SIGNAL_AKTUALISIERT, ) from .veroeffentlichung import ( fahrzeugstatus, spannung_cache_vorladen, zustand_oder_none, ) from .zuordnung import Zuordnung _LOGGER = logging.getLogger(__name__) # Wie weit zurück nach dem letzten Zündungswechsel gesucht wird, wenn der # Parkbeginn nach einem Neustart fehlt. Sieben Tage decken jede realistische # Standzeit ab und bleiben eine Abfrage, die der recorder in einem Rutsch # beantwortet; weiter zurück reicht seine Aufbewahrung ohnehin selten. PARKBEGINN_FENSTER = datetime.timedelta(days=7) # Wie weit um den Zündungswechsel herum nach der zugehörigen Meldezeit gesucht # wird. Beide stammen aus demselben Datensatz und liegen deshalb # Millisekunden auseinander (gemessen: 2 ms) - eine Minute ist reichlich. MELDEZEIT_FENSTER = datetime.timedelta(minutes=1) # Wie oft der Live-Fahrzeugstatus neu aus den Quellsensoren gelesen wird. TAKT_STATUS = datetime.timedelta(seconds=20) # Profil/Fahrten/Tankvorgänge ändern sich normalerweise nur durch Aktionen, # die ohnehin sofort selbst veröffentlichen. Dieser Takt ist nur das # Sicherheitsnetz für Änderungen direkt im Dateisystem. TAKT_DATEN = datetime.timedelta(seconds=60) TAKT_BATTERIE = datetime.timedelta(minutes=5) STORE_VERSION = 1 class Koordinator: """Zustand, Zeittakte und Zustandsbeobachter der Integration.""" def __init__(self, hass: HomeAssistant, entry: ConfigEntry, version: str) -> None: self.hass = hass self.entry = entry self.version = version self.ablage = Ablage(hass, hass.config.path("audi_dashboard")) self.zuordnung = Zuordnung(self.ablage) # Ortsnamen werden einmal hier aufgeloest und mit der Fahrt # gespeichert, statt in jedem Geraet erneut - siehe geokodierung.py. self.geokodierung = Geokodierung(hass, self.ablage.basis) # Was die Entitäten anzeigen: Schlüssel -> (Zustand, Nutzlast). self.werte: dict[str, tuple[str, dict | list | None]] = {} self._store = Store[dict](hass, STORE_VERSION, f"{entry.entry_id}_laufzeit") self.fahrt_start_ts: datetime.datetime | None = None self.tiefststand_pct: float | None = None self.tiefststand_liter: float | None = None # Seit wann das Fahrzeug steht (Zeit des Geraets). None = unbekannt. self.geparkt_seit_ts: datetime.datetime | None = None # Ein eingeplanter zweiter Screening-Lauf, siehe screening.py # (_nachfassen_planen). None = keiner offen. self.screening_nachlauf: CALLBACK_TYPE | None = None # Ergebnis des letzten "Auf Update prüfen"/"Update installieren" - # nur auf Tastendruck gesetzt (siehe dienste.py), nie im normalen # Veröffentlichungs-Takt neu berechnet, damit keine automatischen # Gitea-Abfragen im Hintergrund laufen. app_version_veroeffentlichen() # liest diesen Wert bei jedem Takt erneut mit, damit er nicht # zwischen zwei Prüfungen verschwindet. self.letzte_update_pruefung: dict | None = None self._takte: list[CALLBACK_TYPE] = [] self._beobachter: list[CALLBACK_TYPE] = [] self._ende_aufgabe: asyncio.Task | None = None # ------------------------------------------------------------ Lebenszyklus async def starten(self) -> None: await self.ablage.ordner_sicherstellen() if await self.ablage.vorlage_anlegen( os.path.join(os.path.dirname(__file__), "vorlage", "fahrzeugprofil.json") ): _LOGGER.info( "Erste Einrichtung: Fahrzeugprofil aus der Vorlage angelegt. " "Die Fahrzeugdaten werden in der App unter Einstellungen -> " "Fahrzeug einrichten eingetragen." ) await self.zuordnung.anwenden() await self._laufzeit_laden() # Die "Zustand"-Kachel soll die zuletzt gemessene Batteriespannung # auch direkt nach einem Neustart zeigen, nicht erst "unbekannt" bis # zur nächsten Live-Messung (siehe spannung_cache_vorladen()) - vor # dem ersten alles_veroeffentlichen(), damit schon dessen erste # Veröffentlichung den vorgeladenen Wert trägt. spannung_cache_vorladen(await self.ablage.batterieverlauf_lesen()) await self.alles_veroeffentlichen() # Der Radzähler hängt an den Fahrten, nicht am Kilometerstand - was # während eines Ausfalls gefahren wurde, kommt über die dabei # erkannten Fahrten herein (siehe reifen.py). await reifen.nachtragen(self) self._takte_anmelden() await self.trigger_neu_binden() # Erst wenn Home Assistant vollständig oben ist: bis dahin haben die # Datenquellen ihre Zustände oft noch nicht gemeldet, und eine noch # nicht gemeldete Zündung ließe sich nicht von "aus" unterscheiden # (siehe fahrterkennung.nach_neustart_fortsetzen). self._takte.append( async_at_started(self.hass, self._nach_start) ) async def _nach_start(self, _hass: HomeAssistant) -> None: await fahrterkennung.nach_neustart_fortsetzen(self) await self._parkbeginn_nachholen() # Erst hier (nicht in starten()): das Geräte-/Entity-Register der # zugeordneten Sensoren ist vor async_at_started oft noch nicht # vollständig - siehe die Begründung von trigger_neu_binden() oben. await identitaet.fin_ableiten_und_uebernehmen(self.hass, self.zuordnung, self.ablage) await self.profil_veroeffentlichen() # Genau hier kommt eine frisch aktualisierte Integration an, und was # ihre neuen Felder nachtragen koennen, soll nicht bis zur naechsten # Fahrt warten. Reicht das Ortsbudget eines Laufs nicht, plant das # Screening selbst einen zweiten ein (screening._nachfassen_planen). await screening.durchfuehren(self) async def beenden(self) -> None: if self.screening_nachlauf is not None: self.screening_nachlauf() self.screening_nachlauf = None for abmelden in (*self._takte, *self._beobachter): abmelden() self._takte.clear() self._beobachter.clear() self.warte_ende_ab_abbrechen() # ------------------------------------------------------------- Zeittakte def _takte_anmelden(self) -> None: self._takte.append( async_track_time_interval(self.hass, self._takt_status, TAKT_STATUS) ) self._takte.append( async_track_time_interval(self.hass, self._takt_daten, TAKT_DATEN) ) self._takte.append( async_track_time_interval(self.hass, self._takt_batterie, TAKT_BATTERIE) ) # Ein einzelner täglicher Termin genügt für alle Sicherungs-Intervalle # - welche heute fällig ist, entscheidet sicherung.automatisch(). self._takte.append( async_track_time_change(self.hass, self._takt_sicherung, hour=4, minute=0, second=0) ) async def _takt_status(self, _jetzt: datetime.datetime) -> None: await self.fahrzeugstatus_veroeffentlichen() async def _takt_daten(self, _jetzt: datetime.datetime) -> None: await self.profil_veroeffentlichen() await self.fahrten_veroeffentlichen() await self.tankvorgaenge_veroeffentlichen() async def _takt_batterie(self, _jetzt: datetime.datetime) -> None: await batterie.pruefen(self) async def _takt_sicherung(self, _jetzt: datetime.datetime) -> None: await sicherung.automatisch(self) # -------------------------------------------------- Zustandsbeobachter async def trigger_neu_binden(self) -> None: """Meldet die Beobachter für Zündung, Kilometerstand und Tankfüllstand neu an - nach jeder Änderung der Sensor-Zuordnung. Ohne zugeordnete Entität wird bewusst gar kein Beobachter angemeldet, statt einen gegen eine nicht existierende Entity-ID zu registrieren.""" for abmelden in self._beobachter: abmelden() self._beobachter.clear() werte = self.zuordnung.werte # Fahrten löst der Trip-Status aus (oder die Zündung, wenn er fehlt). self._beobachten(verlauf.fahrtsignal(werte), self._fahrtsignal_geaendert) # Die Zündung selbst nur noch für die Anzeige "fährt/steht" - sofort # neu veröffentlichen, statt bis zum nächsten 20-Sekunden-Takt zu # warten. Entfällt, wenn sie ohnehin schon das Fahrtsignal ist. if werte.ZUENDUNG_SENSOR != verlauf.fahrtsignal(werte): self._beobachten(werte.ZUENDUNG_SENSOR, self._zuendung_angezeigt) self._beobachten(werte.KM_SENSOR, self._kilometerstand_geaendert) self._beobachten(werte.TANK_SENSOR, self._tankfuellstand_geaendert) self._beobachten(werte.TANK_LITER_SENSOR, self._tankvolumen_geaendert) def _beobachten( self, entity_id: str, rueckruf: Callable[[Event[EventStateChangedData]], Coroutine[Any, Any, None]] ) -> None: if not entity_id: return self._beobachter.append( async_track_state_change_event(self.hass, [entity_id], rueckruf) ) @staticmethod def _zustaende(ereignis: Event[EventStateChangedData]) -> tuple[str | None, str | None]: alt = ereignis.data["old_state"] neu = ereignis.data["new_state"] return (neu.state if neu else None, alt.state if alt else None) async def _zuendung_angezeigt(self, ereignis: Event[EventStateChangedData]) -> None: """Die Zündung dient nur der Anzeige - und hält den Parkbeginn fest. "Geparkt seit" hing bis 2026-09-01 am Ende der jüngsten Fahrt in der Liste. Das ist keine Aussage über das Parken, sondern über die Fahrterkennung: eine verpasste Fahrt lässt das Fahrzeug seit Tagen stehen, obwohl es gestern unterwegs war (real vorgekommen am 31.08.2026). Der Zeitpunkt, zu dem die Zündung zuletzt ausging, ist die direkte Antwort. Mit der Gerätezeit, nicht mit der Ankunft: der Datensatz kommt bei niedriger Priorität erst mit der nächsten Übertragung an, trägt aber seinen eigenen Zeitstempel (siehe verlauf.geraetezeit()).""" neu, alt = self._zustaende(ereignis) # Kein Vorzustand heisst: die Entitaet taucht gerade erst auf (Start von # Home Assistant, Neuladen der Integration). Das ist kein Wechsel der # Zuendung. Ohne diese Pruefung stempelte JEDER Neustart den Parkbeginn # neu - live beobachtet beim Ausliefern von 2026.9.1.6 - und die # Parkdauer begann wieder bei null. Der gespeicherte Wert ueberlebt den # Neustart ohnehin (_laufzeit_laden). Dieselbe Pruefung wie in # fahrterkennung.zuendung_geaendert(). if alt is None: await self.fahrzeugstatus_veroeffentlichen() return if neu == "off": zustand = ereignis.data["new_state"] seit = await verlauf.geraetezeit( self.hass, self.zuordnung.werte.MELDEZEIT_SENSOR, zustand.last_updated if zustand else None, datetime.datetime.now(datetime.UTC), ) await self.geparkt_seit_setzen(seit) _LOGGER.info("Zündung aus - geparkt seit %s", seit.isoformat()) elif neu == "on" and self.geparkt_seit_ts is not None: await self.geparkt_seit_setzen(None) await self.fahrzeugstatus_veroeffentlichen() async def _fahrtsignal_geaendert(self, ereignis: Event[EventStateChangedData]) -> None: neu, alt = self._zustaende(ereignis) # Der Zeitpunkt, zu dem Home Assistant diesen Wechsel gesehen hat - # nicht "jetzt". Die Fahrterkennung braucht ihn, um die Meldezeit des # Geräts demselben Datensatz zuzuordnen (siehe _geraetezeit dort). zustand = ereignis.data["new_state"] await fahrterkennung.zuendung_geaendert( self, neu, alt, zustand.last_updated if zustand else None ) async def _kilometerstand_geaendert(self, ereignis: Event[EventStateChangedData]) -> None: neu, alt = self._zustaende(ereignis) if neu is None or alt is None: return # Der Radzaehler haengt nicht mehr am Kilometerstand, sondern an den # Fahrten - das Screening rechnet ihn am Ende selbst nach. await screening.durchfuehren(self) async def _tankfuellstand_geaendert(self, ereignis: Event[EventStateChangedData]) -> None: neu, _alt = self._zustaende(ereignis) zustand = ereignis.data["new_state"] await tankerkennung.fuellstand_geaendert( self, neu, zustand.last_updated if zustand else None ) async def _tankvolumen_geaendert(self, ereignis: Event[EventStateChangedData]) -> None: neu, _alt = self._zustaende(ereignis) zustand = ereignis.data["new_state"] await tankerkennung.tankvolumen_geaendert( self, neu, zustand.last_updated if zustand else None ) # ------------------------------------------------- Laufende Fahrt merken async def _laufzeit_laden(self) -> None: gespeichert = await self._store.async_load() or {} roh = gespeichert.get("fahrt_start_ts") if roh: try: self.fahrt_start_ts = datetime.datetime.fromisoformat(roh) _LOGGER.info( "Laufende Fahrt seit %s aus dem letzten Lauf übernommen", roh ) except ValueError: self.fahrt_start_ts = None self.tiefststand_pct = gespeichert.get("tiefststand_pct") self.tiefststand_liter = gespeichert.get("tiefststand_liter") roh = gespeichert.get("geparkt_seit_ts") if roh: try: geladen = datetime.datetime.fromisoformat(roh) except ValueError: geladen = None # Auch beim LADEN pruefen, nicht nur beim Setzen: ein einmal # falsch gespeicherter Wert (auf der Instanz des Eigentuemers am # 02.09.2026 die Unix-Epoche, weil der Meldezeit-Sensor 0 lieferte) # ueberlebt sonst jeden Neustart - und _parkbeginn_nachholen() # springt nicht ein, weil ja etwas gespeichert ist. if geladen is not None and not verlauf.geraetezeit_plausibel( geladen, datetime.datetime.now(datetime.UTC) ): _LOGGER.warning( "Gespeicherter Parkbeginn %s ist unglaubwuerdig - verworfen", geladen.isoformat(), ) geladen = None self.geparkt_seit_ts = geladen async def _laufzeit_sichern(self) -> None: await self._store.async_save({ "fahrt_start_ts": self.fahrt_start_ts.isoformat() if self.fahrt_start_ts else None, "tiefststand_pct": self.tiefststand_pct, "tiefststand_liter": self.tiefststand_liter, "geparkt_seit_ts": ( self.geparkt_seit_ts.isoformat() if self.geparkt_seit_ts else None ), }) async def fahrt_start_setzen(self, ts: datetime.datetime | None) -> None: self.fahrt_start_ts = ts await self._laufzeit_sichern() async def _parkbeginn_nachholen(self) -> None: """Holt den Parkbeginn aus der Aufzeichnung, wenn er fehlt. Der Zuhörer (_zuendung_angezeigt) bekommt den Wechsel nur mit, wenn Home Assistant läuft. Ging die Zündung aus, während es unten war - über Nacht der Normalfall -, feuert nichts; die Registrierung der Entität beim Hochfahren wird bewusst ignoriert, sonst begänne die Parkdauer bei jedem Neustart neu. Der Zeitpunkt steht aber im recorder, und dort wird er hier einmalig nachgeschlagen. Nur wenn WIRKLICH ein Wechsel an -> aus im Fenster liegt. Steht die Zündung schon die ganze Zeit auf aus, wissen wir den Beginn nicht - dann bleibt es bei \"unbekannt\", statt den Rand des Fensters zu erfinden. Eine falsche Zahl ist schlechter als keine (Ansage des Eigentümers, 2026-09-01).""" if self.geparkt_seit_ts is not None: return sensor = self.zuordnung.werte.ZUENDUNG_SENSOR if not sensor or zustand_oder_none(self.hass, sensor) != "off": return jetzt = datetime.datetime.now(datetime.UTC) punkte = await verlauf.verlauf_lesen( self.hass, sensor, jetzt - PARKBEGINN_FENSTER, jetzt ) # verlauf_lesen() filtert unknown/unavailable schon heraus - eine # Funklücke trennt die Reihe also nicht in zwei Wechsel auf. wechsel: datetime.datetime | None = None vorher: str | None = None for ts, wert in punkte: if wert == "off" and vorher == "on": wechsel = ts vorher = wert if wechsel is None: return # Die Ankunftszeit des Datensatzes ist nicht seine Entstehungszeit. Die # Meldezeit desselben Datensatzes ist es - dieselbe Begründung wie in # verlauf.geraetezeit(), hier nur rückblickend statt live. seit = wechsel meldesensor = self.zuordnung.werte.MELDEZEIT_SENSOR if meldesensor: reihe = await verlauf.verlauf_lesen( self.hass, meldesensor, wechsel - MELDEZEIT_FENSTER, wechsel + MELDEZEIT_FENSTER ) epoche = verlauf.naechster_wert(wechsel, reihe) if epoche is not None: try: gemeldet = datetime.datetime.fromtimestamp(epoche, datetime.UTC) except (ValueError, OSError, OverflowError): gemeldet = None if verlauf.geraetezeit_plausibel(gemeldet, jetzt): seit = gemeldet await self.geparkt_seit_setzen(seit) _LOGGER.info( "Parkbeginn aus der Aufzeichnung nachgeholt: %s", seit.isoformat() ) async def geparkt_seit_setzen(self, ts: datetime.datetime | None) -> None: if ts is not None and not verlauf.geraetezeit_plausibel( ts, datetime.datetime.now(datetime.UTC) ): _LOGGER.warning("Parkbeginn %s ist unglaubwuerdig - nicht gesetzt", ts.isoformat()) ts = None self.geparkt_seit_ts = ts await self._laufzeit_sichern() async def tiefststand_setzen(self, wert: float | None) -> None: self.tiefststand_pct = wert await self._laufzeit_sichern() async def tiefststand_liter_setzen(self, wert: float | None) -> None: self.tiefststand_liter = wert await self._laufzeit_sichern() def warte_ende_ab(self, koroutine: Coroutine[Any, Any, None]) -> None: self.warte_ende_ab_abbrechen() self._ende_aufgabe = self.entry.async_create_background_task( self.hass, koroutine, "audi_dashboard_fahrtende" ) def warte_ende_ab_abbrechen(self) -> None: if self._ende_aufgabe is not None and not self._ende_aufgabe.done(): self._ende_aufgabe.cancel() self._ende_aufgabe = None # ---------------------------------------------------------- Veröffentlichen def setzen( self, schluessel: str, zustand: str | float | None, daten: dict | list | None = None ) -> None: self.werte[schluessel] = (zustand, daten) async_dispatcher_send(self.hass, f"{SIGNAL_AKTUALISIERT}_{schluessel}") async def profil_veroeffentlichen(self, profil: dict | None = None) -> None: if profil is None: profil = await self.ablage.profil_lesen() if profil is None: return self.setzen(E_PROFIL, "aktuell", profil) # Die drei Reifen-Kontrollwerte hängen am selben Profil - zusammen # veröffentlichen, damit sie nicht auseinanderlaufen können. # Die beiden km-Stände bleiben Zahlen (die Entitäten tragen die # Einheit km); None heißt "noch nie gezählt" und wird von Home # Assistant als "unknown" angezeigt - richtiger als eine erfundene 0. stand = reifen.zustaende(profil) self.setzen(E_REIFEN_SOMMER, stand["sommer"]) self.setzen(E_REIFEN_WINTER, stand["winter"]) self.setzen(E_REIFEN_AKTIV, str(stand["aktiv"])) async def fahrten_veroeffentlichen(self) -> None: self.setzen(E_FAHRTEN, "aktuell", await self.ablage.fahrten_lesen()) async def tankvorgaenge_veroeffentlichen(self) -> None: self.setzen(E_TANKVORGAENGE, "aktuell", await self.ablage.tankvorgaenge_lesen()) async def tankvorgang_nachbereiten(self) -> None: """Nach jedem Schreiben eines Tankvorgangs: veroeffentlichen und den Verbrauchsfaktor rechnen lassen. Eine Betankung ist der Moment, in dem sich ein Intervall SCHLIESST - vorher gibt es kein Paar aus zwei Tankvorgaengen und damit keinen Faktor. Bis hierher haing die Korrektur allein am Screening, das bei jeder Aenderung des Kilometerstands laeuft: gerechnet wurde also erst bei der naechsten Fahrt, nicht beim Tanken. Der Aufruf im Screening bleibt trotzdem bestehen, und zwar aus dem umgekehrten Grund: eine Fahrt im Intervall kann ihren Rohverbrauch erst NACH der Betankung bekommen (der Tankstand-Verlauf trifft verzoegert ein), und dann muss der Faktor sie noch einholen. Der 60-Sekunden-Takt ruft bewusst weiter nur tankvorgaenge_veroeffentlichen() - dort aendert sich nichts, was einen neuen Faktor ergeben koennte.""" await self.tankvorgaenge_veroeffentlichen() await verbrauchskorrektur.anwenden(self) async def batterieverlauf_veroeffentlichen(self) -> None: self.setzen(E_BATTERIEVERLAUF, "aktuell", await self.ablage.batterieverlauf_lesen()) async def zuordnung_veroeffentlichen(self) -> None: self.setzen(E_ZUORDNUNG, "aktuell", self.zuordnung.aktueller_stand()) async def fahrzeugstatus_veroeffentlichen(self) -> None: self.setzen(E_FAHRZEUGSTATUS, "aktuell", fahrzeugstatus(self.hass, self.zuordnung.werte, self.geparkt_seit_ts)) async def beleg_ergebnis_veroeffentlichen(self, daten: dict) -> None: self.setzen(E_BELEG_ERGEBNIS, "aktuell", daten) async def import_status_veroeffentlichen(self, zustand: str, daten: dict) -> None: self.setzen(E_IMPORT_STATUS, zustand, daten) def _buendel_lesen(self) -> dict | None: """Angaben zum mitgelieferten OTA-Bündel, oder None. None ist ein völlig normaler Zustand: wer nur das Panel benutzt, braucht kein App-Bündel, und eine Installation ohne bundle.json ist deshalb nicht kaputt, sondern schlicht ohne diese Möglichkeit.""" pfad = os.path.join( os.path.dirname(__file__), BUENDEL_ORDNER.replace("/", os.sep), BUENDEL_INFO ) if not os.path.exists(pfad): return None try: with open(pfad, encoding="utf-8") as datei: info = json.load(datei) except (ValueError, OSError) as fehler: _LOGGER.warning("%s ist nicht lesbar (%s)", pfad, fehler) return None if not isinstance(info, dict) or not info.get("version") or not info.get("sha256"): _LOGGER.warning("%s hat nicht die erwartete Form - Bündel wird nicht gemeldet", pfad) return None return {**info, "url": BUENDEL_URL} async def app_version_veroeffentlichen(self) -> None: buendel = await self.hass.async_add_executor_job(self._buendel_lesen) # Die App liest alle drei aus derselben Entität: welchen Stand diese # Installation ausliefert, ob ein nachladbares OTA-Bündel danebenliegt, # und ob eine neuere Fassung bei Gitea bereitliegt. Zusammen an einer # Stelle, damit die App nicht mehrere Quellen abgleichen muss, die # auseinanderlaufen können. self.setzen( E_APP_VERSION, self.version, { "app": self.version, "buendel": buendel, "integration_update": self.letzte_update_pruefung, }, ) async def alles_veroeffentlichen(self) -> None: await self.app_version_veroeffentlichen() await self.profil_veroeffentlichen() await self.fahrten_veroeffentlichen() await self.tankvorgaenge_veroeffentlichen() await self.fahrzeugstatus_veroeffentlichen() await self.batterieverlauf_veroeffentlichen() await self.zuordnung_veroeffentlichen()