# -*- coding: utf-8 -*- """ Satellite-Gruppierung. Ermöglicht die Organisation von Satellites in logischen Gruppen für koordinierte Aktionen und Verwaltung. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from enum import IntEnum, auto from typing import Any, Callable, Iterator, TYPE_CHECKING if TYPE_CHECKING: from trixy_core.satellite.satellite import Satellite class GroupType(IntEnum): """Typ einer Satellite-Gruppe.""" ROOM = auto() # Raum-basierte Gruppe ZONE = auto() # Zone/Bereich CUSTOM = auto() # Benutzerdefiniert DYNAMIC = auto() # Dynamisch basierend auf Kriterien @dataclass class GroupConfig: """ Konfiguration einer Gruppe. Attributes: auto_add: Satellites automatisch hinzufügen bei Match auto_remove: Satellites automatisch entfernen wenn nicht mehr passend max_members: Maximale Anzahl Mitglieder (0 = unbegrenzt) priority: Priorität der Gruppe (höher = wichtiger) """ auto_add: bool = True auto_remove: bool = True max_members: int = 0 priority: int = 0 class SatelliteGroup: """ Logische Gruppe von Satellites. Ermöglicht koordinierte Aktionen auf mehreren Satellites. Example: # Gruppe erstellen group = SatelliteGroup( group_id="wohnbereich", name="Wohnbereich", group_type=GroupType.ZONE ) # Satellites hinzufügen group.add(satellite1) group.add(satellite2) # Aktion auf allen Satellites for satellite in group: await satellite.speak("Willkommen") # Filtern connected = group.filter(lambda s: s.is_connected) """ def __init__( self, group_id: str, name: str = "", group_type: GroupType = GroupType.CUSTOM, config: GroupConfig | None = None, criteria: Callable[["Satellite"], bool] | None = None, ) -> None: """ Initialisiert eine Satellite-Gruppe. Args: group_id: Eindeutige Gruppen-ID name: Anzeigename group_type: Typ der Gruppe config: Gruppenkonfiguration criteria: Optionale Funktion zur automatischen Zuordnung """ self._id = group_id self._name = name or group_id self._type = group_type self._config = config or GroupConfig() self._criteria = criteria self._members: dict[str, "Satellite"] = {} self._created_at = datetime.now() self._metadata: dict[str, Any] = {} @property def id(self) -> str: """Gruppen-ID.""" return self._id @property def name(self) -> str: """Anzeigename.""" return self._name @name.setter def name(self, value: str) -> None: self._name = value @property def group_type(self) -> GroupType: """Gruppentyp.""" return self._type @property def config(self) -> GroupConfig: """Gruppenkonfiguration.""" return self._config @property def size(self) -> int: """Anzahl Mitglieder.""" return len(self._members) @property def is_empty(self) -> bool: """Prüft, ob die Gruppe leer ist.""" return len(self._members) == 0 @property def metadata(self) -> dict[str, Any]: """Zusätzliche Metadaten.""" return self._metadata def add(self, satellite: "Satellite") -> bool: """ Fügt einen Satellite zur Gruppe hinzu. Args: satellite: Der hinzuzufügende Satellite Returns: True wenn hinzugefügt, False wenn bereits vorhanden oder Limit erreicht """ if satellite.id in self._members: return False if self._config.max_members > 0 and self.size >= self._config.max_members: return False self._members[satellite.id] = satellite return True def remove(self, satellite_id: str) -> "Satellite | None": """ Entfernt einen Satellite aus der Gruppe. Args: satellite_id: ID des zu entfernenden Satellites Returns: Der entfernte Satellite oder None """ return self._members.pop(satellite_id, None) def contains(self, satellite_id: str) -> bool: """Prüft, ob ein Satellite in der Gruppe ist.""" return satellite_id in self._members def get(self, satellite_id: str) -> "Satellite | None": """Gibt einen Satellite nach ID zurück.""" return self._members.get(satellite_id) def get_all(self) -> list["Satellite"]: """Gibt alle Satellites der Gruppe zurück.""" return list(self._members.values()) def get_connected(self) -> list["Satellite"]: """Gibt alle verbundenen Satellites zurück.""" return [s for s in self._members.values() if s.is_connected] def matches(self, satellite: "Satellite") -> bool: """ Prüft, ob ein Satellite zu dieser Gruppe passt. Verwendet das Kriterium wenn definiert. Args: satellite: Der zu prüfende Satellite Returns: True wenn passend """ if self._criteria is None: return False return self._criteria(satellite) def filter( self, predicate: Callable[["Satellite"], bool] ) -> list["Satellite"]: """ Filtert Satellites nach einem Prädikat. Args: predicate: Filter-Funktion Returns: Liste der passenden Satellites """ return [s for s in self._members.values() if predicate(s)] def clear(self) -> int: """ Entfernt alle Satellites aus der Gruppe. Returns: Anzahl entfernter Satellites """ count = len(self._members) self._members.clear() return count def __iter__(self) -> Iterator["Satellite"]: """Iterator über alle Satellites.""" return iter(self._members.values()) def __len__(self) -> int: """Anzahl Satellites.""" return len(self._members) def __contains__(self, item: str | "Satellite") -> bool: """Prüft Mitgliedschaft.""" if isinstance(item, str): return item in self._members return item.id in self._members def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "id": self._id, "name": self._name, "type": self._type.name, "size": self.size, "members": list(self._members.keys()), "config": { "auto_add": self._config.auto_add, "auto_remove": self._config.auto_remove, "max_members": self._config.max_members, "priority": self._config.priority, }, "created_at": self._created_at.isoformat(), "metadata": self._metadata, } def __repr__(self) -> str: return f"SatelliteGroup(id={self._id!r}, name={self._name!r}, size={self.size})" class GroupManager: """ Verwaltet alle Satellite-Gruppen. Bietet zentrale Verwaltung von Gruppen und automatische Zuordnung von Satellites basierend auf Kriterien. Example: manager = GroupManager() # Raum-Gruppe erstellen manager.create_room_group("wohnzimmer", "Wohnzimmer") # Zone erstellen manager.create_zone_group( "erdgeschoss", "Erdgeschoss", criteria=lambda s: s.room_id in ["wohnzimmer", "küche", "flur"] ) # Satellite automatisch zuordnen manager.auto_assign(satellite) # Alle Gruppen eines Satellites groups = manager.get_groups_for_satellite(satellite.id) """ def __init__(self) -> None: """Initialisiert den GroupManager.""" self._groups: dict[str, SatelliteGroup] = {} self._room_groups: dict[str, str] = {} # room_id -> group_id @property def count(self) -> int: """Anzahl Gruppen.""" return len(self._groups) def create( self, group_id: str, name: str = "", group_type: GroupType = GroupType.CUSTOM, config: GroupConfig | None = None, criteria: Callable[["Satellite"], bool] | None = None, ) -> SatelliteGroup: """ Erstellt eine neue Gruppe. Args: group_id: Eindeutige Gruppen-ID name: Anzeigename group_type: Typ der Gruppe config: Konfiguration criteria: Optionales Kriterium für Auto-Zuordnung Returns: Die erstellte Gruppe Raises: ValueError: Wenn ID bereits existiert """ if group_id in self._groups: raise ValueError(f"Gruppe existiert bereits: {group_id}") group = SatelliteGroup( group_id=group_id, name=name, group_type=group_type, config=config, criteria=criteria, ) self._groups[group_id] = group return group def create_room_group(self, room_id: str, name: str = "") -> SatelliteGroup: """ Erstellt eine Raum-basierte Gruppe. Args: room_id: Raum-ID name: Anzeigename (Standard: room_id) Returns: Die erstellte Gruppe """ group = self.create( group_id=f"room:{room_id}", name=name or room_id, group_type=GroupType.ROOM, criteria=lambda s: s.room_id == room_id, ) self._room_groups[room_id] = group.id return group def create_zone_group( self, zone_id: str, name: str = "", criteria: Callable[["Satellite"], bool] | None = None, rooms: list[str] | None = None, ) -> SatelliteGroup: """ Erstellt eine Zone-Gruppe. Args: zone_id: Zone-ID name: Anzeigename criteria: Optionales Kriterium rooms: Liste von Raum-IDs für automatisches Kriterium Returns: Die erstellte Gruppe """ if criteria is None and rooms: criteria = lambda s: s.room_id in rooms return self.create( group_id=f"zone:{zone_id}", name=name or zone_id, group_type=GroupType.ZONE, criteria=criteria, ) def get(self, group_id: str) -> SatelliteGroup | None: """Gibt eine Gruppe nach ID zurück.""" return self._groups.get(group_id) def get_by_room(self, room_id: str) -> SatelliteGroup | None: """Gibt die Gruppe für einen Raum zurück.""" group_id = self._room_groups.get(room_id) if group_id: return self._groups.get(group_id) return None def get_all(self) -> list[SatelliteGroup]: """Gibt alle Gruppen zurück.""" return list(self._groups.values()) def delete(self, group_id: str) -> SatelliteGroup | None: """ Löscht eine Gruppe. Args: group_id: ID der zu löschenden Gruppe Returns: Die gelöschte Gruppe oder None """ group = self._groups.pop(group_id, None) # Room-Mapping bereinigen if group and group.group_type == GroupType.ROOM: room_id = group_id.replace("room:", "") self._room_groups.pop(room_id, None) return group def add_to_group(self, group_id: str, satellite: "Satellite") -> bool: """ Fügt einen Satellite zu einer Gruppe hinzu. Args: group_id: Gruppen-ID satellite: Der Satellite Returns: True wenn hinzugefügt """ group = self._groups.get(group_id) if group is None: return False return group.add(satellite) def remove_from_group(self, group_id: str, satellite_id: str) -> bool: """ Entfernt einen Satellite aus einer Gruppe. Args: group_id: Gruppen-ID satellite_id: Satellite-ID Returns: True wenn entfernt """ group = self._groups.get(group_id) if group is None: return False return group.remove(satellite_id) is not None def auto_assign(self, satellite: "Satellite") -> list[str]: """ Ordnet einen Satellite automatisch passenden Gruppen zu. Args: satellite: Der zuzuordnende Satellite Returns: Liste der Gruppen-IDs zu denen hinzugefügt wurde """ assigned = [] for group in self._groups.values(): if group.config.auto_add and group.matches(satellite): if group.add(satellite): assigned.append(group.id) return assigned def auto_remove(self, satellite: "Satellite") -> list[str]: """ Entfernt einen Satellite aus nicht mehr passenden Gruppen. Args: satellite: Der zu prüfende Satellite Returns: Liste der Gruppen-IDs aus denen entfernt wurde """ removed = [] for group in self._groups.values(): if satellite.id in group: if group.config.auto_remove and not group.matches(satellite): group.remove(satellite.id) removed.append(group.id) return removed def get_groups_for_satellite(self, satellite_id: str) -> list[SatelliteGroup]: """ Gibt alle Gruppen zurück, in denen ein Satellite ist. Args: satellite_id: Satellite-ID Returns: Liste der Gruppen """ return [g for g in self._groups.values() if satellite_id in g] def find_groups( self, group_type: GroupType | None = None, min_size: int = 0, has_connected: bool = False, ) -> list[SatelliteGroup]: """ Sucht Gruppen nach Kriterien. Args: group_type: Filter nach Typ min_size: Minimale Gruppengröße has_connected: Nur Gruppen mit verbundenen Satellites Returns: Liste passender Gruppen """ result = [] for group in self._groups.values(): if group_type is not None and group.group_type != group_type: continue if group.size < min_size: continue if has_connected and not any(s.is_connected for s in group): continue result.append(group) return result def __iter__(self) -> Iterator[SatelliteGroup]: """Iterator über alle Gruppen.""" return iter(self._groups.values()) def __len__(self) -> int: """Anzahl Gruppen.""" return len(self._groups) def __contains__(self, group_id: str) -> bool: """Prüft, ob eine Gruppe existiert.""" return group_id in self._groups def __getitem__(self, group_id: str) -> SatelliteGroup: """Zugriff per ID.""" if group_id not in self._groups: raise KeyError(f"Gruppe nicht gefunden: {group_id}") return self._groups[group_id]