Trixy7/plugins/homematic/api.py

# -*- coding: utf-8 -*-
"""
Homematic XML-API Wrapper.

Vollstaendiger async Wrapper fuer die Homematic XML-API.
Unterstuetzt alle Endpoints des XML-API Addons (v2.x).

Verwendung:
    api = HomematicApi("http://homematic-raspi", session_id="xxx")
    devices = await api.get_device_list()
    await api.set_state(ise_id="1445", new_value="true")
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from typing import Any
from xml.etree import ElementTree as ET

import aiohttp

from trixy_core.utils.debug import pdebug, perror, pinfo, pwarn


# ============================================================================
# Datenklassen
# ============================================================================

@dataclass
class Datapoint:
    """Ein einzelner Datenpunkt eines Channels."""
    ise_id: str
    name: str
    type: str
    value: str
    valuetype: str = ""
    valueunit: str = ""
    timestamp: str = "0"
    operations: str = ""

    @property
    def value_bool(self) -> bool:
        """Wert als Boolean."""
        return self.value.lower() in ("true", "1")

    @property
    def value_float(self) -> float:
        """Wert als Float."""
        try:
            return float(self.value)
        except (ValueError, TypeError):
            return 0.0

    @property
    def value_int(self) -> int:
        """Wert als Integer."""
        try:
            return int(float(self.value))
        except (ValueError, TypeError):
            return 0

    @property
    def is_writable(self) -> bool:
        """Prueft ob der Datenpunkt beschreibbar ist (Bit 1 in operations)."""
        try:
            return bool(int(self.operations) & 2)
        except (ValueError, TypeError):
            return False


@dataclass
class Channel:
    """Ein Channel eines Geraets."""
    ise_id: str
    name: str
    index: str = ""
    direction: str = ""
    parent_device: str = ""
    address: str = ""
    channel_type: str = ""
    visible: str = ""
    operate: str = ""
    datapoints: dict[str, Datapoint] = field(default_factory=dict)


@dataclass
class Device:
    """Ein Homematic-Geraet."""
    ise_id: str
    name: str
    address: str = ""
    interface: str = ""
    device_type: str = ""
    ready_config: str = ""
    channels: dict[str, Channel] = field(default_factory=dict)
    unreach: str = ""
    config_pending: str = ""


@dataclass
class Room:
    """Ein Raum mit zugeordneten Channels."""
    ise_id: str
    name: str
    channel_ids: list[str] = field(default_factory=list)


@dataclass
class Program:
    """Ein CCU-Programm."""
    ise_id: str
    name: str
    active: str = ""
    timestamp: str = "0"
    description: str = ""
    visible: str = ""
    operate: str = ""


@dataclass
class SystemVariable:
    """Eine Systemvariable."""
    ise_id: str
    name: str
    variable: str = ""
    value: str = ""
    value_list: str = ""
    value_text: str = ""
    var_type: str = ""  # type in XML
    subtype: str = ""
    unit: str = ""
    min_value: str = ""
    max_value: str = ""
    timestamp: str = "0"
    info: str = ""
    value_name_0: str = ""
    value_name_1: str = ""
    logged: str = ""

    @property
    def value_bool(self) -> bool:
        return self.value.lower() in ("true", "1")

    @property
    def value_float(self) -> float:
        try:
            return float(self.variable)
        except (ValueError, TypeError):
            return 0.0


@dataclass
class Favorite:
    """Ein Favorit / Favoritenliste."""
    ise_id: str
    name: str
    channels: list[dict[str, str]] = field(default_factory=list)


@dataclass
class RssiEntry:
    """RSSI-Wert eines Geraets."""
    device: str
    address: str
    rx: str = ""
    tx: str = ""


@dataclass
class ProtocolEntry:
    """Ein Protokolleintrag."""
    timestamp: str = ""
    name: str = ""
    value: str = ""
    info: str = ""


@dataclass
class Notification:
    """Eine Systembenachrichtigung."""
    ise_id: str = ""
    name: str = ""
    type: str = ""
    timestamp: str = ""
    message: str = ""


# ============================================================================
# API Wrapper
# ============================================================================

class HomematicApi:
    """Async Wrapper fuer die Homematic XML-API.

    Unterstuetzt alle Endpoints des XML-API Addons:
    - Geraete, Channels, Datenpunkte (devicelist, statelist, state, statechange)
    - Raeume (roomlist)
    - Programme (programlist, runprogram)
    - Systemvariablen (sysvarlist, sysvar)
    - Favoriten (favoritelist)
    - RSSI, Protokoll, Benachrichtigungen
    - Mastervalues, Token-Verwaltung
    - Version und Updates
    """

    def __init__(
        self,
        base_url: str,
        session_id: str = "",
        username: str = "",
        password: str = "",
        timeout: int = 30,
    ) -> None:
        self._base_url = base_url.rstrip("/")
        self._api_base = f"{self._base_url}/addons/xmlapi"
        self._session_id = session_id
        self._username = username
        self._password = password
        self._timeout = aiohttp.ClientTimeout(total=timeout)
        self._session: aiohttp.ClientSession | None = None

    # ========================================================================
    # Session-Management
    # ========================================================================

    async def _ensure_session(self) -> aiohttp.ClientSession:
        """Stellt sicher, dass eine aiohttp-Session existiert."""
        if self._session is None or self._session.closed:
            self._session = aiohttp.ClientSession(timeout=self._timeout)
        return self._session

    async def close(self) -> None:
        """Schliesst die HTTP-Session."""
        if self._session and not self._session.closed:
            await self._session.close()
            self._session = None

    async def login(self, username: str = "", password: str = "") -> bool:
        """Prueft ob die CCU-Zugangsdaten gueltig sind (JSON-API).

        Hinweis: Die XML-API nutzt persistente Tokens, die im WebUI unter
        Einstellungen → Sicherheit registriert werden muessen.
        Ein JSON-API Login kann keine XML-API Tokens erstellen.

        Diese Methode prueft nur ob die Zugangsdaten korrekt sind.

        Args:
            username: Benutzername
            password: Passwort

        Returns:
            True wenn Login erfolgreich
        """
        user = username or self._username
        pw = password or self._password
        if not user or not pw:
            raise ValueError("Username und Passwort erforderlich fuer Login")

        session = await self._ensure_session()
        url = f"{self._base_url}/api/homematic.cgi"

        # Login pruefen
        try:
            async with session.post(url, json={
                "method": "Session.login",
                "params": {"username": user, "password": pw},
            }) as resp:
                data = await resp.json(content_type=None)
                sid = data.get("result", "")
                if not sid or sid == "null":
                    return False

                # Session direkt wieder aufraeumen
                try:
                    await session.post(url, json={
                        "method": "Session.logout",
                        "params": {"_session_id_": sid},
                    })
                except Exception:
                    pass

                return True
        except aiohttp.ClientError as e:
            raise RuntimeError(f"Login-Verbindungsfehler: {e}")

    async def logout(self) -> bool:
        """Meldet die aktuelle Session ab."""
        if not self._session_id:
            return True
        session = await self._ensure_session()
        url = f"{self._base_url}/api/homematic.cgi"
        payload = {
            "method": "Session.logout",
            "params": {"_session_id_": self._session_id},
        }
        try:
            async with session.post(url, json=payload) as resp:
                data = await resp.json(content_type=None)
                self._session_id = ""
                return data.get("result", False)
        except aiohttp.ClientError:
            return False

    @property
    def session_id(self) -> str:
        return self._session_id

    @session_id.setter
    def session_id(self, value: str) -> None:
        self._session_id = value

    # ========================================================================
    # HTTP-Anfragen
    # ========================================================================

    async def _request(self, endpoint: str, **params: str) -> ET.Element:
        """Fuehrt eine XML-API-Anfrage aus und gibt das XML-Root zurueck.

        Args:
            endpoint: CGI-Endpoint (z.B. "devicelist.cgi")
            **params: Zusaetzliche Query-Parameter

        Returns:
            XML-Root-Element
        """
        session = await self._ensure_session()
        url = f"{self._api_base}/{endpoint}"
        query = {"sid": self._session_id}
        query.update(params)

        try:
            async with session.get(url, params=query) as resp:
                if resp.status != 200:
                    raise RuntimeError(
                        f"HTTP {resp.status} von {endpoint}"
                    )
                # XML-API liefert ISO-8859-1
                raw = await resp.read()
                text = raw.decode("iso-8859-1")
                return ET.fromstring(text)
        except ET.ParseError as e:
            raise RuntimeError(f"XML-Parse-Fehler bei {endpoint}: {e}")
        except aiohttp.ClientError as e:
            raise RuntimeError(f"Verbindungsfehler bei {endpoint}: {e}")

    async def _request_raw(self, endpoint: str, **params: str) -> str:
        """Fuehrt eine Anfrage aus und gibt den Rohtext zurueck."""
        session = await self._ensure_session()
        url = f"{self._api_base}/{endpoint}"
        query = {"sid": self._session_id}
        query.update(params)

        try:
            async with session.get(url, params=query) as resp:
                raw = await resp.read()
                return raw.decode("iso-8859-1")
        except aiohttp.ClientError as e:
            raise RuntimeError(f"Verbindungsfehler bei {endpoint}: {e}")

    # ========================================================================
    # Geraete
    # ========================================================================

    async def get_device_list(self) -> list[Device]:
        """Gibt alle Geraete mit Channels zurueck (devicelist.cgi)."""
        root = await self._request("devicelist.cgi")
        devices: list[Device] = []

        for dev_elem in root.findall("device"):
            device = Device(
                ise_id=dev_elem.get("ise_id", ""),
                name=dev_elem.get("name", ""),
                address=dev_elem.get("address", ""),
                interface=dev_elem.get("interface", ""),
                device_type=dev_elem.get("device_type", ""),
                ready_config=dev_elem.get("ready_config", ""),
            )
            for ch_elem in dev_elem.findall("channel"):
                channel = Channel(
                    ise_id=ch_elem.get("ise_id", ""),
                    name=ch_elem.get("name", ""),
                    index=ch_elem.get("index", ""),
                    direction=ch_elem.get("direction", ""),
                    parent_device=ch_elem.get("parent_device", ""),
                    address=ch_elem.get("address", ""),
                    channel_type=ch_elem.get("type", ""),
                    visible=ch_elem.get("visible", ""),
                    operate=ch_elem.get("operate", ""),
                )
                device.channels[channel.ise_id] = channel
            devices.append(device)

        return devices

    async def get_device_type_list(self) -> list[dict[str, str]]:
        """Gibt alle Geraetetypen zurueck (devicetypelist.cgi)."""
        root = await self._request("devicetypelist.cgi")
        types: list[dict[str, str]] = []
        for dt in root.findall("deviceType"):
            types.append({
                "name": dt.get("name", ""),
                "description": dt.get("description", ""),
                "thumbnail": dt.get("thumbnailPath", ""),
                "image": dt.get("imagePath", ""),
            })
        return types

    # ========================================================================
    # State / Zustaende
    # ========================================================================

    async def get_state(
        self,
        device_id: str = "",
        channel_id: str = "",
        datapoint_id: str = "",
    ) -> list[Device]:
        """Gibt Zustaende fuer Geraete/Channels/Datenpunkte zurueck (state.cgi).

        Mindestens ein Parameter muss angegeben werden.
        """
        params: dict[str, str] = {}
        if device_id:
            params["device_id"] = device_id
        if channel_id:
            params["channel_id"] = channel_id
        if datapoint_id:
            params["datapoint_id"] = datapoint_id

        root = await self._request("state.cgi", **params)
        return self._parse_state_devices(root)

    async def get_state_list(self) -> list[Device]:
        """Gibt alle Geraetezustaende zurueck (statelist.cgi)."""
        root = await self._request("statelist.cgi")
        return self._parse_state_devices(root)

    async def set_state(self, ise_id: str, new_value: str) -> bool:
        """Setzt einen Datenpunkt-Wert (statechange.cgi).

        Args:
            ise_id: ISE-ID des Datenpunkts
            new_value: Neuer Wert

        Returns:
            True bei Erfolg
        """
        root = await self._request(
            "statechange.cgi",
            ise_id=ise_id,
            new_value=new_value,
        )
        changed = root.find("changed")
        if changed is not None:
            return changed.get("new_value", "") == new_value
        return False

    async def set_states(self, changes: dict[str, str]) -> dict[str, bool]:
        """Setzt mehrere Datenpunkt-Werte gleichzeitig.

        Args:
            changes: Dict von {ise_id: new_value}

        Returns:
            Dict von {ise_id: success}
        """
        results: dict[str, bool] = {}
        tasks = [self.set_state(ise_id, val) for ise_id, val in changes.items()]
        outcomes = await asyncio.gather(*tasks, return_exceptions=True)
        for (ise_id, _), result in zip(changes.items(), outcomes):
            results[ise_id] = result is True
        return results

    def _parse_state_devices(self, root: ET.Element) -> list[Device]:
        """Parst Geraete aus state/statelist XML."""
        devices: list[Device] = []
        for dev_elem in root.findall("device"):
            device = Device(
                ise_id=dev_elem.get("ise_id", ""),
                name=dev_elem.get("name", ""),
                unreach=dev_elem.get("unreach", ""),
                config_pending=dev_elem.get("config_pending", ""),
            )
            for ch_elem in dev_elem.findall("channel"):
                channel = Channel(
                    ise_id=ch_elem.get("ise_id", ""),
                    name=ch_elem.get("name", ""),
                    index=ch_elem.get("index", ""),
                    visible=ch_elem.get("visible", ""),
                    operate=ch_elem.get("operate", ""),
                )
                for dp_elem in ch_elem.findall("datapoint"):
                    dp = Datapoint(
                        ise_id=dp_elem.get("ise_id", ""),
                        name=dp_elem.get("name", ""),
                        type=dp_elem.get("type", ""),
                        value=dp_elem.get("value", ""),
                        valuetype=dp_elem.get("valuetype", ""),
                        valueunit=dp_elem.get("valueunit", ""),
                        timestamp=dp_elem.get("timestamp", "0"),
                        operations=dp_elem.get("operations", ""),
                    )
                    channel.datapoints[dp.type] = dp
                device.channels[channel.ise_id] = channel
            devices.append(device)
        return devices

    # ========================================================================
    # Raeume
    # ========================================================================

    async def get_room_list(self) -> list[Room]:
        """Gibt alle Raeume mit zugeordneten Channels zurueck (roomlist.cgi)."""
        root = await self._request("roomlist.cgi")
        rooms: list[Room] = []
        for room_elem in root.findall("room"):
            room = Room(
                ise_id=room_elem.get("ise_id", ""),
                name=room_elem.get("name", ""),
                channel_ids=[
                    ch.get("ise_id", "")
                    for ch in room_elem.findall("channel")
                ],
            )
            rooms.append(room)
        return rooms

    # ========================================================================
    # Programme
    # ========================================================================

    async def get_program_list(self) -> list[Program]:
        """Gibt alle CCU-Programme zurueck (programlist.cgi)."""
        root = await self._request("programlist.cgi")
        programs: list[Program] = []
        for prog_elem in root.findall("program"):
            programs.append(Program(
                ise_id=prog_elem.get("id", ""),
                name=prog_elem.get("name", ""),
                active=prog_elem.get("active", ""),
                timestamp=prog_elem.get("timestamp", "0"),
                description=prog_elem.get("description", ""),
                visible=prog_elem.get("visible", ""),
                operate=prog_elem.get("operate", ""),
            ))
        return programs

    async def run_program(self, program_id: str) -> bool:
        """Fuehrt ein CCU-Programm aus (runprogram.cgi).

        Args:
            program_id: Programm-ID (aus programlist.cgi)

        Returns:
            True bei Erfolg
        """
        root = await self._request("runprogram.cgi", program_id=program_id)
        started = root.find("started")
        return started is not None

    async def get_program_actions(self, ise_id: str) -> ET.Element:
        """Gibt Programm-Aktionen zurueck (programactions.cgi)."""
        return await self._request("programactions.cgi", ise_id=ise_id)

    # ========================================================================
    # Systemvariablen
    # ========================================================================

    async def get_sysvar_list(self) -> list[SystemVariable]:
        """Gibt alle Systemvariablen zurueck (sysvarlist.cgi)."""
        root = await self._request("sysvarlist.cgi")
        return self._parse_sysvars(root)

    async def get_sysvar(self, ise_id: str) -> SystemVariable | None:
        """Gibt eine einzelne Systemvariable zurueck (sysvar.cgi)."""
        root = await self._request("sysvar.cgi", ise_id=ise_id)
        sysvars = self._parse_sysvars(root)
        return sysvars[0] if sysvars else None

    async def set_sysvar(self, ise_id: str, value: str) -> bool:
        """Setzt den Wert einer Systemvariable (statechange.cgi).

        Args:
            ise_id: ISE-ID der Variable
            value: Neuer Wert

        Returns:
            True bei Erfolg
        """
        return await self.set_state(ise_id, value)

    def _parse_sysvars(self, root: ET.Element) -> list[SystemVariable]:
        """Parst Systemvariablen aus XML."""
        sysvars: list[SystemVariable] = []
        for sv_elem in root.findall("systemVariable"):
            sysvars.append(SystemVariable(
                ise_id=sv_elem.get("ise_id", ""),
                name=sv_elem.get("name", ""),
                variable=sv_elem.get("variable", ""),
                value=sv_elem.get("value", ""),
                value_list=sv_elem.get("value_list", ""),
                value_text=sv_elem.get("value_text", ""),
                var_type=sv_elem.get("type", ""),
                subtype=sv_elem.get("subtype", ""),
                unit=sv_elem.get("unit", ""),
                min_value=sv_elem.get("min", ""),
                max_value=sv_elem.get("max", ""),
                timestamp=sv_elem.get("timestamp", "0"),
                info=sv_elem.get("info", ""),
                value_name_0=sv_elem.get("value_name_0", ""),
                value_name_1=sv_elem.get("value_name_1", ""),
                logged=sv_elem.get("logged", ""),
            ))
        return sysvars

    # ========================================================================
    # Favoriten
    # ========================================================================

    async def get_favorite_list(self) -> list[Favorite]:
        """Gibt alle Favoriten zurueck (favoritelist.cgi)."""
        root = await self._request("favoritelist.cgi")
        favorites: list[Favorite] = []
        for fav_elem in root.findall("favorite"):
            channels: list[dict[str, str]] = []
            for ch in fav_elem.findall("channel"):
                channels.append({
                    "ise_id": ch.get("ise_id", ""),
                    "name": ch.get("name", ""),
                    "type": ch.get("type", ""),
                })
            favorites.append(Favorite(
                ise_id=fav_elem.get("ise_id", ""),
                name=fav_elem.get("name", ""),
                channels=channels,
            ))
        return favorites

    # ========================================================================
    # Mastervalues
    # ========================================================================

    async def get_master_value(
        self, device_id: str = "", channel_id: str = ""
    ) -> ET.Element:
        """Gibt Mastervalues zurueck (mastervalue.cgi)."""
        params: dict[str, str] = {}
        if device_id:
            params["device_id"] = device_id
        if channel_id:
            params["channel_id"] = channel_id
        return await self._request("mastervalue.cgi", **params)

    async def set_master_value(
        self, ise_id: str, name: str, value: str
    ) -> bool:
        """Setzt einen Mastervalue (mastervaluechange.cgi).

        Args:
            ise_id: ISE-ID des Channels/Geraets
            name: Parameter-Name
            value: Neuer Wert
        """
        root = await self._request(
            "mastervaluechange.cgi",
            ise_id=ise_id,
            name=name,
            value=value,
        )
        changed = root.find("changed")
        return changed is not None

    # ========================================================================
    # RSSI / Signalstaerke
    # ========================================================================

    async def get_rssi_list(self) -> list[RssiEntry]:
        """Gibt RSSI-Werte aller Geraete zurueck (rssilist.cgi)."""
        root = await self._request("rssilist.cgi")
        entries: list[RssiEntry] = []
        for entry in root.findall(".//rssi"):
            entries.append(RssiEntry(
                device=entry.get("device", ""),
                address=entry.get("address", ""),
                rx=entry.get("rx", ""),
                tx=entry.get("tx", ""),
            ))
        return entries

    # ========================================================================
    # Protokoll / Log
    # ========================================================================

    async def get_protocol(
        self,
        start: str = "",
        show: str = "",
    ) -> list[ProtocolEntry]:
        """Gibt Protokolleintraege zurueck (protocol.cgi).

        Args:
            start: Start-Index
            show: Anzahl Eintraege
        """
        params: dict[str, str] = {}
        if start:
            params["start"] = start
        if show:
            params["show"] = show
        root = await self._request("protocol.cgi", **params)
        entries: list[ProtocolEntry] = []
        for row in root.findall(".//row"):
            entries.append(ProtocolEntry(
                timestamp=row.get("timestamp", ""),
                name=row.get("name", ""),
                value=row.get("value", ""),
                info=row.get("info", ""),
            ))
        return entries

    # ========================================================================
    # Benachrichtigungen
    # ========================================================================

    async def get_notifications(self) -> list[Notification]:
        """Gibt Systembenachrichtigungen zurueck (systemNotification.cgi)."""
        root = await self._request("systemNotification.cgi")
        notifications: list[Notification] = []
        for notif in root.findall("notification"):
            notifications.append(Notification(
                ise_id=notif.get("ise_id", ""),
                name=notif.get("name", ""),
                type=notif.get("type", ""),
                timestamp=notif.get("timestamp", ""),
                message=notif.get("message", notif.text or ""),
            ))
        return notifications

    async def clear_notifications(self) -> bool:
        """Loescht alle Systembenachrichtigungen (systemNotificationClear.cgi)."""
        try:
            await self._request("systemNotificationClear.cgi")
            return True
        except Exception:
            return False

    # ========================================================================
    # Script-Fehler
    # ========================================================================

    async def get_script_errors(self) -> str:
        """Gibt Script-Fehler zurueck (scripterrors.cgi)."""
        return await self._request_raw("scripterrors.cgi")

    # ========================================================================
    # ReGaHss-Script ausfuehren
    # ========================================================================

    async def exec_script(self, script: str) -> str:
        """Fuehrt ein ReGaHss-Script auf der CCU aus (exec.cgi).

        Args:
            script: HM-Script Code

        Returns:
            Script-Ausgabe als String
        """
        session = await self._ensure_session()
        url = f"{self._api_base}/exec.cgi"
        params = {"sid": self._session_id}
        data = aiohttp.FormData()
        data.add_field("script", script)
        try:
            async with session.post(url, params=params, data=data) as resp:
                raw = await resp.read()
                return raw.decode("iso-8859-1")
        except aiohttp.ClientError as e:
            raise RuntimeError(f"Script-Ausfuehrung fehlgeschlagen: {e}")

    # ========================================================================
    # Token-Verwaltung
    # ========================================================================

    async def get_token_list(self) -> list[dict[str, str]]:
        """Gibt registrierte Tokens zurueck (tokenlist.cgi).

        Returns:
            Liste von {token: str, description: str}
        """
        root = await self._request("tokenlist.cgi")
        tokens: list[dict[str, str]] = []
        for token_elem in root.findall("token"):
            tokens.append({
                "token": token_elem.text or "",
                "description": token_elem.get("desc", ""),
            })
        return tokens

    async def register_token(self, description: str = "") -> str:
        """Registriert einen neuen Token (tokenregister.cgi).

        Args:
            description: Beschreibung des Tokens

        Returns:
            Token-Wert als String
        """
        params: dict[str, str] = {}
        if description:
            params["desc"] = description
        root = await self._request("tokenregister.cgi", **params)
        # Response: <tokenregister><token desc="...">TOKEN_VALUE</token></tokenregister>
        token = root.find("token")
        return token.text if token is not None and token.text else ""

    async def revoke_token(self, token: str) -> bool:
        """Widerruft einen Token (tokenrevoke.cgi).

        Args:
            token: Der Token-Wert der widerrufen werden soll
        """
        try:
            await self._request("tokenrevoke.cgi", sid=token)
            return True
        except Exception:
            return False

    # ========================================================================
    # System
    # ========================================================================

    async def get_version(self) -> str:
        """Gibt die XML-API-Version zurueck (version.cgi)."""
        root = await self._request("version.cgi")
        return root.text or ""

    async def check_update(self) -> dict[str, str]:
        """Prueft auf Updates (update.cgi)."""
        root = await self._request("update.cgi")
        return {
            "current": root.get("current_version", root.findtext("current", "")),
            "latest": root.get("latest_version", root.findtext("latest", "")),
            "update_available": root.get("update_available", ""),
        }

    # ========================================================================
    # Komfort-Methoden
    # ========================================================================

    async def find_device_by_name(self, name: str) -> Device | None:
        """Sucht ein Geraet nach Name (case-insensitive)."""
        devices = await self.get_device_list()
        name_lower = name.lower()
        for device in devices:
            if device.name.lower() == name_lower:
                return device
        return None

    async def find_channel_by_name(self, name: str) -> tuple[Device, Channel] | None:
        """Sucht einen Channel nach Name (case-insensitive)."""
        devices = await self.get_device_list()
        name_lower = name.lower()
        for device in devices:
            for channel in device.channels.values():
                if channel.name.lower() == name_lower:
                    return device, channel
        return None

    async def find_room_by_name(self, name: str) -> Room | None:
        """Sucht einen Raum nach Name (case-insensitive)."""
        rooms = await self.get_room_list()
        name_lower = name.lower()
        for room in rooms:
            if room.name.lower() == name_lower:
                return room
        return None

    async def find_program_by_name(self, name: str) -> Program | None:
        """Sucht ein Programm nach Name (case-insensitive)."""
        programs = await self.get_program_list()
        name_lower = name.lower()
        for prog in programs:
            if prog.name.lower() == name_lower:
                return prog
        return None

    async def find_sysvar_by_name(self, name: str) -> SystemVariable | None:
        """Sucht eine Systemvariable nach Name (case-insensitive)."""
        sysvars = await self.get_sysvar_list()
        name_lower = name.lower()
        for sv in sysvars:
            if sv.name.lower() == name_lower:
                return sv
        return None

    async def get_channel_state(self, channel_id: str) -> dict[str, Datapoint]:
        """Gibt alle Datenpunkte eines Channels zurueck.

        Returns:
            Dict von {datapoint_type: Datapoint}
        """
        devices = await self.get_state(channel_id=channel_id)
        for device in devices:
            for channel in device.channels.values():
                if channel.ise_id == channel_id:
                    return channel.datapoints
        return {}

    async def get_datapoint_value(
        self, channel_id: str, dp_type: str
    ) -> str | None:
        """Gibt den Wert eines bestimmten Datenpunkts zurueck.

        Args:
            channel_id: Channel ISE-ID
            dp_type: Datenpunkt-Typ (z.B. "STATE", "LEVEL", "ACTUAL_TEMPERATURE")
        """
        dps = await self.get_channel_state(channel_id)
        dp = dps.get(dp_type)
        return dp.value if dp else None

    async def get_rooms_with_devices(self) -> list[dict[str, Any]]:
        """Gibt Raeume mit aufgeloesten Geraete-Informationen zurueck."""
        rooms = await self.get_room_list()
        devices = await self.get_device_list()

        # Channel-ID → Device/Channel Mapping
        channel_map: dict[str, tuple[Device, Channel]] = {}
        for device in devices:
            for channel in device.channels.values():
                channel_map[channel.ise_id] = (device, channel)

        result: list[dict[str, Any]] = []
        for room in rooms:
            room_devices: list[dict[str, str]] = []
            for ch_id in room.channel_ids:
                if ch_id in channel_map:
                    dev, ch = channel_map[ch_id]
                    room_devices.append({
                        "device_name": dev.name,
                        "device_type": dev.device_type,
                        "channel_name": ch.name,
                        "channel_id": ch.ise_id,
                    })
            result.append({
                "name": room.name,
                "ise_id": room.ise_id,
                "devices": room_devices,
            })
        return result

    async def get_temperature(self, channel_id: str) -> float | None:
        """Gibt die aktuelle Temperatur eines Thermostats zurueck."""
        val = await self.get_datapoint_value(channel_id, "ACTUAL_TEMPERATURE")
        if val is not None:
            try:
                return float(val)
            except ValueError:
                pass
        return None

    async def set_temperature(self, channel_id: str, temperature: float) -> bool:
        """Setzt die Soll-Temperatur eines Thermostats.

        Args:
            channel_id: Channel ISE-ID des Thermostat-Kanals
            temperature: Zieltemperatur in Grad Celsius
        """
        dps = await self.get_channel_state(channel_id)
        dp = dps.get("SET_TEMPERATURE")
        if dp is None:
            pwarn(f"[Homematic] Kein SET_TEMPERATURE Datenpunkt fuer Channel {channel_id}")
            return False
        return await self.set_state(dp.ise_id, str(temperature))

    async def switch(self, channel_id: str, on: bool) -> bool:
        """Schaltet einen Schaltaktor ein/aus.

        Args:
            channel_id: Channel ISE-ID
            on: True = ein, False = aus
        """
        dps = await self.get_channel_state(channel_id)
        dp = dps.get("STATE")
        if dp is None:
            pwarn(f"[Homematic] Kein STATE Datenpunkt fuer Channel {channel_id}")
            return False
        return await self.set_state(dp.ise_id, "true" if on else "false")

    async def set_dimmer(self, channel_id: str, level: float) -> bool:
        """Setzt den Dimmer-Level (0.0 - 1.0).

        Args:
            channel_id: Channel ISE-ID des Dimmer-Kanals
            level: Helligkeit (0.0 = aus, 1.0 = voll)
        """
        dps = await self.get_channel_state(channel_id)
        dp = dps.get("LEVEL")
        if dp is None:
            pwarn(f"[Homematic] Kein LEVEL Datenpunkt fuer Channel {channel_id}")
            return False
        return await self.set_state(dp.ise_id, str(max(0.0, min(1.0, level))))

    async def is_window_open(self, channel_id: str) -> bool | None:
        """Prueft ob ein Fenster-/Tuerkontakt geoeffnet ist."""
        val = await self.get_datapoint_value(channel_id, "STATE")
        if val is not None:
            return val.lower() in ("true", "1")
        return None

    async def get_all_temperatures(self) -> dict[str, dict[str, float]]:
        """Gibt alle Thermostat-Temperaturen zurueck.

        Sucht Channels mit ACTUAL_TEMPERATURE und SET_TEMPERATURE Datenpunkten
        (nur Channels die beide haben sind echte Thermostate).

        Returns:
            Dict von {channel_name: {actual: float, target: float, channel_id: str}}
        """
        devices = await self.get_state_list()
        temps: dict[str, dict[str, float]] = {}

        for device in devices:
            for channel in device.channels.values():
                actual = channel.datapoints.get("ACTUAL_TEMPERATURE")
                target = channel.datapoints.get("SET_TEMPERATURE")
                # Nur echte Thermostate haben beide Datenpunkte
                if actual is not None and target is not None:
                    temps[channel.name] = {
                        "actual": actual.value_float,
                        "target": target.value_float,
                        "channel_id": channel.ise_id,
                    }
        return temps

    async def get_all_window_states(self) -> dict[str, bool]:
        """Gibt alle Fenster-/Tuerkontakt-Zustaende zurueck.

        Nutzt devicelist fuer Typ-Filterung und statelist fuer Zustaende.
        Filtert auf: HM-Sec-SCo, HM-Sec-SC, HM-Sec-RHS, HM-SCI-3-FM, HmIP-SWDO

        Returns:
            Dict von {channel_name: is_open}
        """
        contact_types = ("Sec-SC", "Sec-RHS", "SCI-", "SWDO", "SWDM")

        # Geraete mit Typen laden + State-Daten
        device_list, state_list = await asyncio.gather(
            self.get_device_list(),
            self.get_state_list(),
        )

        # ISE-IDs der Kontakt-Geraete sammeln
        contact_device_ids: set[str] = set()
        for device in device_list:
            if any(t in device.device_type for t in contact_types):
                contact_device_ids.add(device.ise_id)

        # States filtern
        windows: dict[str, bool] = {}
        for device in state_list:
            if device.ise_id not in contact_device_ids:
                continue
            for channel in device.channels.values():
                state_dp = channel.datapoints.get("STATE")
                if state_dp and channel.visible == "true":
                    windows[channel.name] = state_dp.value_bool
        return windows