Trixy7/plugins/fritzbox/monitor.py

# -*- coding: utf-8 -*-
"""
Fritzbox-Monitor — Abfrage von Traffic und Geraeten via TR-064.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime

from trixy_core.utils.debug import pdebug, perror

try:
    from fritzconnection import FritzConnection
    from fritzconnection.lib.fritzstatus import FritzStatus
    from fritzconnection.lib.fritzhosts import FritzHosts
    _HAS_FRITZ = True
except ImportError:
    _HAS_FRITZ = False


@dataclass
class TrafficSnapshot:
    """Momentaufnahme der Transferraten."""
    timestamp: datetime
    upload_bytes_sec: float = 0.0
    download_bytes_sec: float = 0.0

    @property
    def upload_mbit(self) -> float:
        return self.upload_bytes_sec * 8 / 1_000_000

    @property
    def download_mbit(self) -> float:
        return self.download_bytes_sec * 8 / 1_000_000


@dataclass
class NetworkDevice:
    """Ein Geraet im Netzwerk."""
    name: str
    ip: str
    mac: str
    active: bool
    interface: str = ""


class FritzboxMonitor:
    """Verbindet sich mit der Fritzbox und liest Traffic/Geraete."""

    def __init__(self, host: str, port: int, username: str, password: str,
                 use_tls: bool = False) -> None:
        self._host = host
        self._port = port
        self._username = username
        self._password = password
        self._use_tls = use_tls
        self._connection: FritzConnection | None = None
        self._status: FritzStatus | None = None
        self._hosts: FritzHosts | None = None
        self._connected = False

    @property
    def is_available(self) -> bool:
        return _HAS_FRITZ

    @property
    def is_connected(self) -> bool:
        return self._connected

    def connect(self) -> bool:
        """Verbindet sich mit der Fritzbox."""
        if not _HAS_FRITZ:
            perror("[Fritzbox] fritzconnection nicht installiert (pip install fritzconnection)")
            return False

        try:
            self._connection = FritzConnection(
                address=self._host,
                port=self._port,
                user=self._username,
                password=self._password,
                use_tls=self._use_tls,
                timeout=10,
            )
            self._status = FritzStatus(fc=self._connection)
            self._hosts = FritzHosts(fc=self._connection)
            self._connected = True
            pdebug(f"[Fritzbox] Verbunden mit {self._host}:{self._port}")
            return True
        except Exception as e:
            perror(f"[Fritzbox] Verbindung fehlgeschlagen: {e}")
            self._connected = False
            return False

    def get_traffic(self) -> TrafficSnapshot | None:
        """Liest aktuelle Transferraten."""
        if not self._status:
            return None

        try:
            # transmission_rate liefert (upload_bytes/s, download_bytes/s)
            up, down = self._status.transmission_rate
            return TrafficSnapshot(
                timestamp=datetime.now(),
                upload_bytes_sec=float(up),
                download_bytes_sec=float(down),
            )
        except Exception as e:
            pdebug(f"[Fritzbox] Traffic-Abfrage fehlgeschlagen: {e}")
            return None

    def get_devices(self) -> list[NetworkDevice]:
        """Liest alle bekannten Geraete."""
        if not self._hosts:
            return []

        try:
            count = self._hosts.host_numbers
            devices = []
            for i in range(1, count + 1):
                try:
                    host = self._hosts.get_generic_host_entry(i)
                    devices.append(NetworkDevice(
                        name=host.get("NewHostName", ""),
                        ip=host.get("NewIPAddress", ""),
                        mac=host.get("NewMACAddress", ""),
                        active=host.get("NewActive", False),
                        interface=host.get("NewInterfaceType", ""),
                    ))
                except Exception:
                    continue
            return devices
        except Exception as e:
            pdebug(f"[Fritzbox] Geraete-Abfrage fehlgeschlagen: {e}")
            return []

    def get_active_devices(self) -> list[NetworkDevice]:
        """Gibt nur aktive (verbundene) Geraete zurueck."""
        return [d for d in self.get_devices() if d.active]

    def disconnect(self) -> None:
        """Trennt die Verbindung."""
        self._connection = None
        self._status = None
        self._hosts = None
        self._connected = False