# -*- coding: utf-8 -*- """ Homematic Geraeteueberwachung. Prueft periodisch Batterie-Status, Verbindung und Geraetezustaende. Gibt Warnungen per TTS aus und persistiert den Status in status.json. """ from __future__ import annotations import asyncio import json import time from pathlib import Path from typing import Any from trixy_core.utils.debug import pdebug, pinfo, pwarn from plugins.homematic.api import HomematicApi # Batterie-Schwellwerte (Volt) fuer HM-CC-RT-DN (2x AA, 3.0V max) BATTERY_LOW_VOLTAGE = 2.2 BATTERY_CRITICAL_VOLTAGE = 1.8 class DeviceStatus: """Gespeicherter Geraete-Status fuer Aenderungserkennung.""" def __init__(self, status_file: Path) -> None: self._file = status_file self._devices: dict[str, dict[str, Any]] = {} self._load() def _load(self) -> None: """Laedt den gespeicherten Status.""" if self._file.exists(): try: with open(self._file, "r", encoding="utf-8") as f: self._devices = json.load(f) except Exception: self._devices = {} def _save(self) -> None: """Speichert den aktuellen Status.""" try: with open(self._file, "w", encoding="utf-8") as f: json.dump(self._devices, f, indent=2, ensure_ascii=False) except Exception as e: pwarn(f"[Homematic Monitor] Status speichern fehlgeschlagen: {e}") def get(self, device_id: str) -> dict[str, Any]: """Gibt den gespeicherten Status eines Geraets zurueck.""" return self._devices.get(device_id, {}) def update(self, device_id: str, status: dict[str, Any]) -> None: """Aktualisiert den Status eines Geraets.""" self._devices[device_id] = status self._save() def update_batch(self, updates: dict[str, dict[str, Any]]) -> None: """Aktualisiert mehrere Geraete und speichert einmal.""" self._devices.update(updates) self._save() class HomematicMonitor: """Ueberwacht Homematic-Geraete im Hintergrund. Prueft periodisch: - Batterie-Status (LOWBAT, BATTERY_STATE) - Verbindungsstatus (UNREACH) - Geraetezustaende (STATE, LEVEL, etc.) Gibt Warnungen per Event aus, die vom Plugin als TTS gesprochen werden. """ def __init__( self, api: HomematicApi, plugin_path: Path, event_callback: Any = None, ) -> None: self._api = api self._status = DeviceStatus(plugin_path / "status.json") self._event_callback = event_callback self._task: asyncio.Task | None = None # Konfiguration (wird vom Plugin gesetzt) self.interval: int = 120 self.warn_battery_low: bool = True self.warn_connection_changed: bool = True self.warn_status_changed: bool = False @property def is_enabled(self) -> bool: """Prueft ob mindestens eine Ueberwachung aktiv ist.""" return ( self.interval > 0 and (self.warn_battery_low or self.warn_connection_changed or self.warn_status_changed) ) def start(self) -> None: """Startet den Monitor-Task.""" if self._task and not self._task.done(): return if not self.is_enabled: return self._task = asyncio.create_task(self._monitor_loop()) pinfo(f"[Homematic Monitor] Gestartet (Intervall: {self.interval}s)") def stop(self) -> None: """Stoppt den Monitor-Task.""" if self._task and not self._task.done(): self._task.cancel() self._task = None pinfo("[Homematic Monitor] Gestoppt") def restart(self) -> None: """Neustart mit aktueller Konfiguration.""" self.stop() if self.is_enabled: self.start() async def _monitor_loop(self) -> None: """Periodische Pruefschleife.""" # Erster Check nach kurzem Delay (Server-Start abwarten) await asyncio.sleep(30) while True: try: await self._check_devices() except asyncio.CancelledError: break except Exception as e: pdebug(f"[Homematic Monitor] Pruefung fehlgeschlagen: {e}") try: await asyncio.sleep(self.interval) except asyncio.CancelledError: break async def _check_devices(self) -> None: """Prueft alle Geraete und gibt bei Aenderungen Warnungen aus.""" try: devices = await self._api.get_state_list() except Exception as e: pdebug(f"[Homematic Monitor] API-Fehler: {e}") return new_battery_low: list[str] = [] new_connection_lost: list[str] = [] new_connection_restored: list[str] = [] status_changes: list[str] = [] status_updates: dict[str, dict[str, Any]] = {} for dev in devices: dev_id = dev.ise_id old = self._status.get(dev_id) current: dict[str, Any] = {"name": dev.name} # Batterie und Connection aus allen Channels sammeln unreach = False lowbat = False battery_voltage = 0.0 for ch in dev.channels.values(): ur = ch.datapoints.get("UNREACH") if ur: unreach = ur.value_bool lb = ch.datapoints.get("LOWBAT") if lb and lb.value_bool: lowbat = True bs = ch.datapoints.get("BATTERY_STATE") if bs and bs.value: try: battery_voltage = float(bs.value) except (ValueError, TypeError): pass battery_low = lowbat or (battery_voltage > 0 and battery_voltage <= BATTERY_LOW_VOLTAGE) current["unreach"] = unreach current["battery_low"] = battery_low current["battery_voltage"] = battery_voltage # Batterie-Warnung (nur wenn sich Status geaendert hat) if self.warn_battery_low and battery_low: was_low = old.get("battery_low", False) if not was_low: new_battery_low.append(dev.name) # Verbindungs-Warnung if self.warn_connection_changed: was_unreach = old.get("unreach", False) if unreach and not was_unreach: new_connection_lost.append(dev.name) elif not unreach and was_unreach: new_connection_restored.append(dev.name) # Status-Aenderungen (Fenster, Schalter, etc.) if self.warn_status_changed: for ch in dev.channels.values(): if ch.visible != "true": continue state = ch.datapoints.get("STATE") if state: ch_key = f"state_{ch.ise_id}" old_state = old.get(ch_key) if old_state is not None and old_state != state.value: if state.value_bool: status_changes.append(f"{ch.name} geoeffnet") else: status_changes.append(f"{ch.name} geschlossen") current[ch_key] = state.value status_updates[dev_id] = current # Status speichern self._status.update_batch(status_updates) # Warnungen zusammenbauen warnings: list[str] = [] if new_battery_low: if len(new_battery_low) == 1: warnings.append(f"Die Batterie von {new_battery_low[0]} ist fast leer") else: aufzaehlung = ", ".join(new_battery_low[:-1]) + " und " + new_battery_low[-1] warnings.append(f"Die Batterien von {aufzaehlung} sind fast leer") if new_connection_lost: if len(new_connection_lost) == 1: warnings.append(f"{new_connection_lost[0]} hat keine Verbindung") else: aufzaehlung = ", ".join(new_connection_lost[:-1]) + " und " + new_connection_lost[-1] warnings.append(f"{aufzaehlung} haben keine Verbindung") if new_connection_restored: if len(new_connection_restored) == 1: warnings.append(f"{new_connection_restored[0]} ist wieder erreichbar") else: aufzaehlung = ", ".join(new_connection_restored[:-1]) + " und " + new_connection_restored[-1] warnings.append(f"{aufzaehlung} sind wieder erreichbar") if status_changes: warnings.append(", ".join(status_changes)) # Warnungen ausgeben if warnings: text = " und ".join(warnings) + "." pinfo(f"[Homematic Monitor] {text}") if self._event_callback: await self._event_callback("homematic_warning", { "text": text, "battery_low": new_battery_low, "connection_lost": new_connection_lost, "connection_restored": new_connection_restored, "status_changes": status_changes, })