# -*- coding: utf-8 -*- """ Homematic Device Manager. Kapselt Geraetelogik: Raum→Geraete-Aufloesung, Filterung nach Kategorie, Caching und Zustandsabfragen/-aenderungen. Verwendet device_types.py fuer praezise Geraetetyp-Klassifizierung, sodass z.B. "Licht im Wohnzimmer an" keine Steckdosen mitschaltet. """ from __future__ import annotations import asyncio import time from dataclasses import dataclass, field from typing import Any from trixy_core.utils.debug import pdebug, pwarn from plugins.homematic.api import HomematicApi, Device, Channel, Room from plugins.homematic.device_types import ( DeviceCategory, DeviceTypeInfo, classify_device, LIGHT_CATEGORIES, PLUG_CATEGORIES, CLIMATE_CATEGORIES, CONTACT_CATEGORIES, MOTION_CATEGORIES, ) @dataclass class ResolvedChannel: """Ein aufgeloester Channel mit Geraete- und Typ-Informationen.""" device: Device channel: Channel type_info: DeviceTypeInfo room_name: str = "" @property def category(self) -> DeviceCategory: return self.type_info.category @property def name(self) -> str: return self.channel.name @property def channel_id(self) -> str: return self.channel.ise_id @property def device_name(self) -> str: return self.device.name @property def device_type(self) -> str: return self.device.device_type class DeviceManager: """Verwaltet Homematic-Geraete mit Raum-Zuordnung und Typ-Klassifizierung. Laedt Geraete, Raeume und Programme von der CCU, klassifiziert nach Geraetetyp und bietet Filtermethoden. """ def __init__( self, api: HomematicApi, room_aliases: dict[str, str] | None = None, satellite_rooms: dict[str, str] | None = None, ) -> None: self._api = api self._room_aliases = room_aliases or {} self._satellite_rooms = satellite_rooms or {} # Caches self._devices: dict[str, Device] = {} # ise_id → Device self._device_types: dict[str, DeviceTypeInfo] = {} # device_ise_id → TypeInfo self._rooms: dict[str, Room] = {} # name_lower → Room self._room_channel_ids: dict[str, set[str]] = {} # room_name_lower → {channel_ids} self._channel_to_device: dict[str, str] = {} # channel_ise_id → device_ise_id self._channel_to_room: dict[str, str] = {} # channel_ise_id → room_name_lower self._cache_time: float = 0.0 self._cache_ttl: float = 300.0 async def refresh(self) -> None: """Laedt alle Geraete und Raeume von der CCU.""" devices, rooms = await asyncio.gather( self._api.get_device_list(), self._api.get_room_list(), ) self._devices.clear() self._device_types.clear() self._channel_to_device.clear() for device in devices: self._devices[device.ise_id] = device type_info = classify_device(device.device_type) self._device_types[device.ise_id] = type_info for channel in device.channels.values(): self._channel_to_device[channel.ise_id] = device.ise_id self._rooms.clear() self._room_channel_ids.clear() self._channel_to_room.clear() for room in rooms: key = room.name.lower() self._rooms[key] = room ch_ids = set(room.channel_ids) self._room_channel_ids[key] = ch_ids for ch_id in ch_ids: self._channel_to_room[ch_id] = key self._cache_time = time.time() pdebug(f"[DeviceManager] {len(self._devices)} Geraete, {len(self._rooms)} Raeume geladen") async def ensure_cache(self) -> None: """Stellt sicher dass der Cache aktuell ist.""" if time.time() - self._cache_time > self._cache_ttl: await self.refresh() def normalize_room(self, room_name: str) -> str: """Normalisiert einen Raumnamen (Aliases, Lowercase). Prueft: Alias-Tabelle → direkt in Raeume vorhanden → unveraendert. """ lower = room_name.lower().strip() # 1. Expliziter Alias if lower in self._room_aliases: return self._room_aliases[lower] # 2. Schon ein gueltiger Raumname if lower in self._rooms: return lower return lower def resolve_room_for_satellite( self, satellite_id: str, room_id: str, alias: str = "" ) -> str: """Loest den Homematic-Raum fuer einen Satellite auf. Reihenfolge: 1. Explizites Mapping in satellite_rooms (Satellite-ID oder Alias) 2. room_id direkt als Raumname pruefen 3. room_id/alias ueber room_aliases aufloesen 4. Leer (kein Raum gefunden) Args: satellite_id: Satellite-ID (z.B. MAC-Adresse) room_id: Room-ID des Satellites (z.B. "livingroom") alias: Satellite-Alias (z.B. "Wohnzimmer") Returns: Normalisierter Homematic-Raumname oder "" """ # 1. Explizites Satellite-Mapping (per ID oder Alias) for key in (satellite_id, alias.lower(), room_id.lower()): if not key: continue mapped = self._satellite_rooms.get(key) if mapped: normalized = mapped.lower() if normalized in self._rooms: return normalized # 2. room_id direkt als Raumname if room_id: lower = room_id.lower() if lower in self._rooms: return lower # 3. room_id oder alias ueber Alias-Tabelle for candidate in (room_id, alias): if not candidate: continue resolved = self._room_aliases.get(candidate.lower()) if resolved and resolved in self._rooms: return resolved # 4. Alias direkt pruefen if alias: lower = alias.lower() if lower in self._rooms: return lower return "" def get_room_names(self) -> list[str]: """Gibt alle Raumnamen zurueck.""" return [r.name for r in self._rooms.values()] # ======================================================================== # Geraete-Filter # ======================================================================== def get_channels_in_room( self, room_name: str, categories: frozenset[DeviceCategory] | None = None, ) -> list[ResolvedChannel]: """Gibt alle Channels in einem Raum zurueck, optional nach Kategorie gefiltert. Args: room_name: Raum (wird normalisiert) categories: Nur diese Kategorien (None = alle) Returns: Liste von ResolvedChannel """ room_key = self.normalize_room(room_name) channel_ids = self._room_channel_ids.get(room_key, set()) if not channel_ids: return [] results: list[ResolvedChannel] = [] for device in self._devices.values(): type_info = self._device_types[device.ise_id] if categories and type_info.category not in categories: continue for channel in device.channels.values(): if channel.ise_id in channel_ids: results.append(ResolvedChannel( device=device, channel=channel, type_info=type_info, room_name=room_key, )) return results def get_lights_in_room(self, room_name: str) -> list[ResolvedChannel]: """Gibt alle Licht-Channels in einem Raum zurueck (keine Steckdosen!).""" return self.get_channels_in_room(room_name, LIGHT_CATEGORIES) def get_thermostats_in_room(self, room_name: str) -> list[ResolvedChannel]: """Gibt alle Thermostat-Channels in einem Raum zurueck.""" return self.get_channels_in_room(room_name, CLIMATE_CATEGORIES) def get_contacts_in_room(self, room_name: str) -> list[ResolvedChannel]: """Gibt alle Fenster-/Tuer-Kontakte in einem Raum zurueck.""" return self.get_channels_in_room(room_name, CONTACT_CATEGORIES) def get_motion_sensors_in_room(self, room_name: str) -> list[ResolvedChannel]: """Gibt alle Bewegungsmelder in einem Raum zurueck.""" return self.get_channels_in_room(room_name, MOTION_CATEGORIES) def get_all_channels( self, categories: frozenset[DeviceCategory] | None = None, ) -> list[ResolvedChannel]: """Gibt alle Channels zurueck, optional nach Kategorie gefiltert.""" results: list[ResolvedChannel] = [] for device in self._devices.values(): type_info = self._device_types[device.ise_id] if categories and type_info.category not in categories: continue for channel in device.channels.values(): room = self._channel_to_room.get(channel.ise_id, "") results.append(ResolvedChannel( device=device, channel=channel, type_info=type_info, room_name=room, )) return results def room_exists(self, room_name: str) -> bool: """Prueft ob ein Raum existiert.""" return self.normalize_room(room_name) in self._rooms # ======================================================================== # Aktionen # ======================================================================== async def switch_lights(self, room_name: str, on: bool) -> int: """Schaltet alle Lichter in einem Raum ein/aus. Beruecksichtigt Geraetetyp: Nur LIGHT_SWITCH, DIMMER, LED_RGBW, etc. Steckdosen (PSM, PS) werden NICHT geschaltet. Returns: Anzahl geschalteter Channels """ lights = self.get_lights_in_room(room_name) if not lights: return 0 switched = 0 for resolved in lights: # Nur Channels mit sichtbarem Schalt-/Dimm-Kanal if resolved.channel.visible != "true": continue try: dps = await self._api.get_channel_state(resolved.channel_id) if "STATE" in dps: await self._api.set_state( dps["STATE"].ise_id, "true" if on else "false", ) switched += 1 elif "LEVEL" in dps: await self._api.set_state( dps["LEVEL"].ise_id, "1.0" if on else "0.0", ) switched += 1 except Exception as e: pdebug(f"[DeviceManager] Licht-Fehler {resolved.name}: {e}") return switched async def get_light_states(self, room_name: str) -> dict[str, bool]: """Gibt Licht-Zustaende in einem Raum zurueck. Returns: Dict von {channel_name: is_on} """ lights = self.get_lights_in_room(room_name) states: dict[str, bool] = {} for resolved in lights: if resolved.channel.visible != "true": continue try: dps = await self._api.get_channel_state(resolved.channel_id) if "STATE" in dps: states[resolved.name] = dps["STATE"].value_bool elif "LEVEL" in dps: states[resolved.name] = dps["LEVEL"].value_float > 0 except Exception: pass return states async def set_led_color(self, room_name: str, color_value: int) -> int: """Setzt die Farbe aller LED-RGBW-Controller in einem Raum. Args: color_value: HM-LC-RGBW-WM: 0-200 (Hue + 200=Weiss) HmIP-RGBW: HSV-basiert Returns: Anzahl geaenderter Channels """ led_channels = self.get_channels_in_room( room_name, frozenset({DeviceCategory.LED_RGBW, DeviceCategory.LED_CONTROLLER}), ) changed = 0 for resolved in led_channels: try: dps = await self._api.get_channel_state(resolved.channel_id) if "COLOR" in dps: await self._api.set_state(dps["COLOR"].ise_id, str(color_value)) changed += 1 elif "HUE" in dps: # HmIP-RGBW: HUE Datenpunkt await self._api.set_state(dps["HUE"].ise_id, str(color_value)) changed += 1 except Exception as e: pdebug(f"[DeviceManager] LED-Farbe Fehler {resolved.name}: {e}") return changed async def set_led_level(self, room_name: str, level: float) -> int: """Setzt die Helligkeit aller LED-Controller in einem Raum. Args: level: 0.0 (aus) bis 1.0 (voll) Returns: Anzahl geaenderter Channels """ led_channels = self.get_channels_in_room( room_name, frozenset({DeviceCategory.LED_RGBW, DeviceCategory.LED_CONTROLLER, DeviceCategory.DIMMER}), ) changed = 0 for resolved in led_channels: # Nur Dimmer/LED-Channels, nicht Farb- oder Programm-Channels if resolved.channel.visible != "true": continue try: dps = await self._api.get_channel_state(resolved.channel_id) if "LEVEL" in dps: await self._api.set_state( dps["LEVEL"].ise_id, str(max(0.0, min(1.0, level))), ) changed += 1 except Exception as e: pdebug(f"[DeviceManager] LED-Level Fehler {resolved.name}: {e}") return changed async def get_temperatures( self, room_name: str = "" ) -> list[dict[str, Any]]: """Gibt Thermostat-Temperaturen zurueck. Args: room_name: Raum (leer = alle Raeume) Returns: Liste von {name, actual, target, channel_id, room} """ if room_name: channels = self.get_thermostats_in_room(room_name) else: channels = self.get_all_channels(CLIMATE_CATEGORIES) results: list[dict[str, Any]] = [] for resolved in channels: try: dps = await self._api.get_channel_state(resolved.channel_id) actual = dps.get("ACTUAL_TEMPERATURE") target = dps.get("SET_TEMPERATURE") if actual and target: results.append({ "name": resolved.name, "actual": actual.value_float, "target": target.value_float, "channel_id": resolved.channel_id, "room": resolved.room_name, }) except Exception: pass return results async def set_temperature(self, room_name: str, temp: float) -> int: """Setzt die Soll-Temperatur aller Thermostate in einem Raum. Returns: Anzahl geaenderter Thermostate """ thermostat_cats = frozenset({DeviceCategory.THERMOSTAT, DeviceCategory.WALL_THERMOSTAT}) channels = self.get_channels_in_room(room_name, thermostat_cats) changed = 0 for resolved in channels: try: success = await self._api.set_temperature(resolved.channel_id, temp) if success: changed += 1 except Exception as e: pdebug(f"[DeviceManager] Temperatur-Fehler {resolved.name}: {e}") return changed async def get_window_states( self, room_name: str = "" ) -> dict[str, bool]: """Gibt Fenster-/Tuer-Zustaende zurueck. Returns: Dict von {channel_name: is_open} """ if room_name: channels = self.get_contacts_in_room(room_name) else: channels = self.get_all_channels(CONTACT_CATEGORIES) states: dict[str, bool] = {} for resolved in channels: if resolved.channel.visible != "true": continue try: dps = await self._api.get_channel_state(resolved.channel_id) state = dps.get("STATE") if state: states[resolved.name] = state.value_bool except Exception: pass return states async def get_last_motion(self, room_name: str) -> dict[str, Any] | None: """Gibt den letzten Bewegungszeitpunkt in einem Raum zurueck. Returns: {name, timestamp, brightness} oder None """ channels = self.get_motion_sensors_in_room(room_name) for resolved in channels: try: dps = await self._api.get_channel_state(resolved.channel_id) motion = dps.get("MOTION") if motion: result: dict[str, Any] = { "name": resolved.name, "timestamp": int(motion.timestamp), } brightness = dps.get("BRIGHTNESS") or dps.get("ILLUMINATION") if brightness: result["brightness"] = brightness.value_float return result except Exception: pass return None