# -*- coding: utf-8 -*- """ Trixy Fritzbox-Plugin. Ueberwacht die Fritzbox auf: - Hohen Upload/Download (konfigurierbare Schwellwerte) - Neue/unbekannte Geraete im Netzwerk - Offline-gegangene Geraete (optional) Benachrichtigung per TTS auf konfigurierten Satellites. Sprachsteuerung fuer Status-Abfragen. """ from __future__ import annotations import asyncio from pathlib import Path from trixy_core.plugins.trixy_plugin import TrixyPlugin from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn from trixy_core.nlp.decorators import intent, admin_only from trixy_core.nlp.handler import IntentResult, IntentReceivedData from plugins.fritzbox.monitor import FritzboxMonitor, TrafficSnapshot, NetworkDevice class FritzboxPlugin(TrixyPlugin): """Fritzbox-Ueberwachung mit Sprachsteuerung.""" NAME = "fritzbox" VERSION = "1.0.0" DESCRIPTION = "Fritzbox-Netzwerkueberwachung mit Warnungen" AUTHOR = "Trixy" def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self._monitor: FritzboxMonitor | None = None self._check_task: asyncio.Task | None = None self._notification_pcm: bytes = b"" self._known_macs: set[str] = set() # Cooldown: verhindert Spam bei anhaltendem hohem Traffic self._last_upload_alert: float = 0.0 self._last_download_alert: float = 0.0 self._alert_cooldown_seconds: float = 300.0 # 5 Min zwischen gleichen Warnungen async def on_load(self) -> None: """Initialisiert Monitor und startet periodische Pruefung.""" fritz_cfg = self.config.get("fritzbox", {}) host = fritz_cfg.get("host", "192.168.178.1") port = fritz_cfg.get("port", 49000) username = fritz_cfg.get("username", "") password = fritz_cfg.get("password", "") use_tls = fritz_cfg.get("use_tls", False) if not password: pwarn("[Fritzbox] Kein Passwort konfiguriert — Plugin inaktiv") return self._monitor = FritzboxMonitor(host, port, username, password, use_tls) if not self._monitor.is_available: perror("[Fritzbox] fritzconnection nicht installiert") return # Verbindung herstellen if not self._monitor.connect(): return # Benachrichtigungs-Sound laden self._load_notification_sound() # Bekannte Geraete aus Config laden known = self.config.get("known_devices", []) self._known_macs = {mac.upper() for mac in known} # Initial alle aktiven Geraete als bekannt erfassen if not self._known_macs: await self._init_known_devices() # Periodische Pruefung starten interval = self.get_config_value("check_interval_seconds", 30) self._check_task = asyncio.create_task(self._check_loop(interval)) pinfo(f"[Fritzbox] Plugin geladen — Monitoring alle {interval}s, " f"{len(self._known_macs)} bekannte Geraete") async def on_unload(self) -> None: """Stoppt Monitoring und trennt Verbindung.""" if self._check_task and not self._check_task.done(): self._check_task.cancel() try: await self._check_task except asyncio.CancelledError: pass if self._monitor: self._monitor.disconnect() def _load_notification_sound(self) -> None: """Laedt den Benachrichtigungs-Sound.""" try: asset_service = self.application.services.get_service("AssetService") if asset_service: sound_name = self.get_config_value("notification_sound", "notification.wav") audio_path = asset_service.get_audio(sound_name) if audio_path and audio_path.exists(): raw = audio_path.read_bytes() self._notification_pcm = raw[44:] # WAV-Header entfernen except Exception: pass async def _init_known_devices(self) -> None: """Erfasst alle aktuell aktiven Geraete als bekannt.""" if not self._monitor: return devices = await asyncio.get_event_loop().run_in_executor( None, self._monitor.get_active_devices, ) for dev in devices: self._known_macs.add(dev.mac.upper()) # In Config speichern self.set_config_value("known_devices", sorted(self._known_macs)) self.save_config() pdebug(f"[Fritzbox] {len(self._known_macs)} Geraete als bekannt erfasst") # === Periodische Pruefung === async def _check_loop(self, interval: float) -> None: """Hauptschleife fuer periodische Pruefungen.""" while True: try: await asyncio.sleep(interval) await self._run_checks() except asyncio.CancelledError: break except Exception as e: perror(f"[Fritzbox] Check-Fehler: {e}") await asyncio.sleep(interval) async def _run_checks(self) -> None: """Fuehrt alle aktivierten Pruefungen durch.""" if not self._monitor or not self._monitor.is_connected: return loop = asyncio.get_event_loop() alerts = self.config.get("alerts", {}) # Traffic pruefen traffic = await loop.run_in_executor(None, self._monitor.get_traffic) if traffic: await self._check_traffic(traffic, alerts) # Geraete pruefen if alerts.get("new_device", {}).get("enabled") or alerts.get("device_offline", {}).get("enabled"): devices = await loop.run_in_executor(None, self._monitor.get_active_devices) await self._check_devices(devices, alerts) async def _check_traffic(self, traffic: TrafficSnapshot, alerts: dict) -> None: """Prueft Traffic gegen Schwellwerte.""" import time now = time.monotonic() # Upload up_cfg = alerts.get("high_upload", {}) if up_cfg.get("enabled") and traffic.upload_mbit >= up_cfg.get("threshold_mbit", 5.0): if now - self._last_upload_alert >= self._alert_cooldown_seconds: self._last_upload_alert = now msg = up_cfg.get("message", "Hoher Upload: {rate} Mbit/s") msg = msg.replace("{rate}", f"{traffic.upload_mbit:.1f}") pinfo(f"[Fritzbox] WARNUNG: {msg}") await self._notify(msg) # Download down_cfg = alerts.get("high_download", {}) if down_cfg.get("enabled") and traffic.download_mbit >= down_cfg.get("threshold_mbit", 50.0): if now - self._last_download_alert >= self._alert_cooldown_seconds: self._last_download_alert = now msg = down_cfg.get("message", "Hoher Download: {rate} Mbit/s") msg = msg.replace("{rate}", f"{traffic.download_mbit:.1f}") pinfo(f"[Fritzbox] WARNUNG: {msg}") await self._notify(msg) async def _check_devices(self, devices: list[NetworkDevice], alerts: dict) -> None: """Prueft auf neue/offline Geraete.""" active_macs = {d.mac.upper() for d in devices} # Neue Geraete new_cfg = alerts.get("new_device", {}) if new_cfg.get("enabled"): for dev in devices: mac = dev.mac.upper() if mac not in self._known_macs: self._known_macs.add(mac) msg = new_cfg.get("message", "Neues Geraet: {name} ({ip})") msg = msg.replace("{name}", dev.name or "Unbekannt") msg = msg.replace("{ip}", dev.ip) msg = msg.replace("{mac}", dev.mac) pinfo(f"[Fritzbox] {msg}") await self._notify(msg) # Bekannte Geraete aktualisieren self.set_config_value("known_devices", sorted(self._known_macs)) self.save_config() # Offline-Geraete offline_cfg = alerts.get("device_offline", {}) if offline_cfg.get("enabled"): watch = {m.upper() for m in offline_cfg.get("watch_devices", [])} for mac in watch: if mac not in active_macs: # Geraetename finden name = mac for dev in devices: if dev.mac.upper() == mac: name = dev.name or mac break msg = offline_cfg.get("message", "Geraet offline: {name}") msg = msg.replace("{name}", name) await self._notify(msg) # === Benachrichtigung an Satellites === async def _notify(self, text: str) -> None: """Sendet Warnung per TTS an konfigurierte Satellites.""" satellites = getattr(self.application, "satellites", None) if not satellites: return notify_cfg = self.config.get("notify_satellites", {}) mode = notify_cfg.get("mode", "all") target_aliases = [a.lower() for a in notify_cfg.get("aliases", [])] target_rooms = [r.lower() for r in notify_cfg.get("rooms", [])] for sat in satellites.get_connected(): if not sat.is_connected: continue if mode == "all": pass # Alle Satellites elif mode == "alias": alias = getattr(sat, "alias", "").lower() if alias not in target_aliases: continue elif mode == "room": room = getattr(sat, "room_id", "").lower() if room not in target_rooms: continue else: continue try: if self._notification_pcm: await sat.say(self._notification_pcm) await sat.speak(text) except Exception as e: pdebug(f"[Fritzbox] Benachrichtigung an {sat_id} fehlgeschlagen: {e}") # === Sprachsteuerung === @intent("fritzbox_status", description="Fritzbox-Netzwerkstatus abfragen") @admin_only async def handle_status(self, data: IntentReceivedData) -> IntentResult: """Gibt aktuellen Traffic und Geraete-Anzahl aus.""" if not self._monitor or not self._monitor.is_connected: return IntentResult.failure("not_connected", "Die Fritzbox ist nicht verbunden.") loop = asyncio.get_event_loop() traffic = await loop.run_in_executor(None, self._monitor.get_traffic) devices = await loop.run_in_executor(None, self._monitor.get_active_devices) parts = [] if traffic: parts.append( f"Download {traffic.download_mbit:.1f} Mbit, " f"Upload {traffic.upload_mbit:.1f} Mbit pro Sekunde" ) parts.append(f"{len(devices)} Geraete im Netzwerk") return IntentResult.success_with_response(". ".join(parts)) @intent("fritzbox_devices", description="Verbundene Geraete auflisten") @admin_only async def handle_devices(self, data: IntentReceivedData) -> IntentResult: """Listet verbundene Geraete auf.""" if not self._monitor or not self._monitor.is_connected: return IntentResult.failure("not_connected", "Die Fritzbox ist nicht verbunden.") loop = asyncio.get_event_loop() devices = await loop.run_in_executor(None, self._monitor.get_active_devices) if not devices: return IntentResult.success_with_response("Keine aktiven Geraete gefunden.") # Max 10 per Sprache ausgeben names = [d.name or d.ip for d in devices[:10]] rest = len(devices) - 10 if len(devices) > 10 else 0 text = f"{len(devices)} Geraete online: {', '.join(names)}" if rest > 0: text += f" und {rest} weitere" return IntentResult.success_with_response(text) @intent("fritzbox_traffic", description="Aktuelle Internetgeschwindigkeit") @admin_only async def handle_traffic(self, data: IntentReceivedData) -> IntentResult: """Gibt aktuelle Up-/Download-Raten aus.""" if not self._monitor or not self._monitor.is_connected: return IntentResult.failure("not_connected", "Die Fritzbox ist nicht verbunden.") loop = asyncio.get_event_loop() traffic = await loop.run_in_executor(None, self._monitor.get_traffic) if not traffic: return IntentResult.failure("no_data", "Konnte keine Traffic-Daten abrufen.") return IntentResult.success_with_response( f"Download: {traffic.download_mbit:.1f} Mbit pro Sekunde. " f"Upload: {traffic.upload_mbit:.1f} Mbit pro Sekunde." )