| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475 |
- # -*- coding: utf-8 -*-
- """
- Zone-Management für Satellites.
- Zonen sind hierarchische Gruppierungen von Räumen/Bereichen
- für koordinierte Aktionen über mehrere Räume hinweg.
- """
- from __future__ import annotations
- from dataclasses import dataclass, field
- from typing import Any, Iterator, TYPE_CHECKING
- if TYPE_CHECKING:
- from trixy_core.satellite.satellite import Satellite
- from trixy_core.satellite.groups.group import SatelliteGroup
- @dataclass
- class ZoneConfig:
- """
- Konfiguration einer Zone.
- Attributes:
- priority: Priorität der Zone (höher = wichtiger)
- fallback_zone: Fallback-Zone wenn keine Satellites verfügbar
- audio_sync: Audio zwischen Räumen synchronisieren
- propagate_commands: Befehle an alle Räume senden
- """
- priority: int = 0
- fallback_zone: str | None = None
- audio_sync: bool = False
- propagate_commands: bool = True
- class Zone:
- """
- Hierarchische Zone aus mehreren Räumen.
- Ermöglicht die Verwaltung von Bereichen wie "Erdgeschoss",
- "Obergeschoss", "Außenbereich" etc.
- Example:
- zone = Zone(
- zone_id="erdgeschoss",
- name="Erdgeschoss",
- rooms=["wohnzimmer", "küche", "flur"]
- )
- # Alle Satellites in der Zone
- satellites = zone.get_all_satellites()
- # Prüfen ob Raum in Zone
- if "wohnzimmer" in zone:
- print("Wohnzimmer ist im Erdgeschoss")
- """
- def __init__(
- self,
- zone_id: str,
- name: str = "",
- rooms: list[str] | None = None,
- config: ZoneConfig | None = None,
- parent_zone: "Zone | None" = None,
- ) -> None:
- """
- Initialisiert eine Zone.
- Args:
- zone_id: Eindeutige Zone-ID
- name: Anzeigename
- rooms: Liste der Raum-IDs
- config: Zone-Konfiguration
- parent_zone: Optionale übergeordnete Zone
- """
- self._id = zone_id
- self._name = name or zone_id
- self._rooms: set[str] = set(rooms or [])
- self._config = config or ZoneConfig()
- self._parent = parent_zone
- self._children: dict[str, "Zone"] = {}
- self._metadata: dict[str, Any] = {}
- # Gruppen-Referenzen (werden vom ZoneManager gesetzt)
- self._room_groups: dict[str, "SatelliteGroup"] = {}
- @property
- def id(self) -> str:
- """Zone-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 config(self) -> ZoneConfig:
- """Zone-Konfiguration."""
- return self._config
- @property
- def rooms(self) -> set[str]:
- """Räume in dieser Zone (nur direkte)."""
- return self._rooms.copy()
- @property
- def all_rooms(self) -> set[str]:
- """Alle Räume inklusive Kinder-Zonen."""
- result = self._rooms.copy()
- for child in self._children.values():
- result.update(child.all_rooms)
- return result
- @property
- def parent(self) -> "Zone | None":
- """Übergeordnete Zone."""
- return self._parent
- @property
- def children(self) -> list["Zone"]:
- """Untergeordnete Zonen."""
- return list(self._children.values())
- @property
- def is_root(self) -> bool:
- """Prüft, ob dies eine Root-Zone ist."""
- return self._parent is None
- @property
- def depth(self) -> int:
- """Tiefe in der Hierarchie (Root = 0)."""
- if self._parent is None:
- return 0
- return self._parent.depth + 1
- @property
- def metadata(self) -> dict[str, Any]:
- """Zusätzliche Metadaten."""
- return self._metadata
- def add_room(self, room_id: str) -> None:
- """Fügt einen Raum zur Zone hinzu."""
- self._rooms.add(room_id)
- def remove_room(self, room_id: str) -> bool:
- """
- Entfernt einen Raum aus der Zone.
- Returns:
- True wenn entfernt
- """
- if room_id in self._rooms:
- self._rooms.discard(room_id)
- self._room_groups.pop(room_id, None)
- return True
- return False
- def add_child(self, zone: "Zone") -> None:
- """Fügt eine Kinder-Zone hinzu."""
- zone._parent = self
- self._children[zone.id] = zone
- def remove_child(self, zone_id: str) -> "Zone | None":
- """Entfernt eine Kinder-Zone."""
- zone = self._children.pop(zone_id, None)
- if zone:
- zone._parent = None
- return zone
- def contains_room(self, room_id: str, recursive: bool = True) -> bool:
- """
- Prüft, ob ein Raum in der Zone ist.
- Args:
- room_id: Raum-ID
- recursive: Auch in Kinder-Zonen suchen
- Returns:
- True wenn enthalten
- """
- if room_id in self._rooms:
- return True
- if recursive:
- for child in self._children.values():
- if child.contains_room(room_id, recursive=True):
- return True
- return False
- def get_all_satellites(self) -> list["Satellite"]:
- """
- Gibt alle Satellites in der Zone zurück.
- Beinhaltet alle Räume und Kinder-Zonen.
- Returns:
- Liste aller Satellites
- """
- satellites: list["Satellite"] = []
- for group in self._room_groups.values():
- satellites.extend(group.get_all())
- for child in self._children.values():
- satellites.extend(child.get_all_satellites())
- return satellites
- def get_connected_satellites(self) -> list["Satellite"]:
- """Gibt alle verbundenen Satellites zurück."""
- return [s for s in self.get_all_satellites() if s.is_connected]
- def get_path(self) -> list["Zone"]:
- """
- Gibt den Pfad von der Root-Zone zu dieser Zone.
- Returns:
- Liste der Zonen von Root bis hier
- """
- path = [self]
- current = self._parent
- while current is not None:
- path.insert(0, current)
- current = current._parent
- return path
- def __contains__(self, item: str) -> bool:
- """Prüft, ob ein Raum in der Zone ist."""
- return self.contains_room(item, recursive=True)
- def __iter__(self) -> Iterator[str]:
- """Iterator über alle Räume."""
- return iter(self.all_rooms)
- def __len__(self) -> int:
- """Anzahl Räume (inklusive Kinder)."""
- return len(self.all_rooms)
- def to_dict(self) -> dict[str, Any]:
- """Konvertiert zu Dictionary."""
- return {
- "id": self._id,
- "name": self._name,
- "rooms": list(self._rooms),
- "all_rooms": list(self.all_rooms),
- "parent": self._parent.id if self._parent else None,
- "children": [c.id for c in self._children.values()],
- "depth": self.depth,
- "config": {
- "priority": self._config.priority,
- "fallback_zone": self._config.fallback_zone,
- "audio_sync": self._config.audio_sync,
- "propagate_commands": self._config.propagate_commands,
- },
- "satellite_count": len(self.get_all_satellites()),
- }
- def __repr__(self) -> str:
- return f"Zone(id={self._id!r}, name={self._name!r}, rooms={len(self._rooms)})"
- class ZoneManager:
- """
- Verwaltet alle Zonen.
- Bietet hierarchische Organisation von Zonen und
- automatische Zuordnung von Räumen.
- Example:
- manager = ZoneManager()
- # Root-Zone erstellen
- gesamt = manager.create("gesamt", "Gesamtes Haus")
- # Kind-Zonen
- eg = manager.create("eg", "Erdgeschoss", parent="gesamt")
- eg.add_room("wohnzimmer")
- eg.add_room("küche")
- og = manager.create("og", "Obergeschoss", parent="gesamt")
- og.add_room("schlafzimmer")
- og.add_room("bad")
- # Zone für Raum finden
- zone = manager.get_zone_for_room("wohnzimmer")
- """
- def __init__(self) -> None:
- """Initialisiert den ZoneManager."""
- self._zones: dict[str, Zone] = {}
- self._room_mapping: dict[str, str] = {} # room_id -> zone_id
- @property
- def count(self) -> int:
- """Anzahl Zonen."""
- return len(self._zones)
- def create(
- self,
- zone_id: str,
- name: str = "",
- rooms: list[str] | None = None,
- config: ZoneConfig | None = None,
- parent: str | None = None,
- ) -> Zone:
- """
- Erstellt eine neue Zone.
- Args:
- zone_id: Eindeutige Zone-ID
- name: Anzeigename
- rooms: Liste der Raum-IDs
- config: Zone-Konfiguration
- parent: ID der übergeordneten Zone
- Returns:
- Die erstellte Zone
- Raises:
- ValueError: Wenn ID bereits existiert oder Parent nicht gefunden
- """
- if zone_id in self._zones:
- raise ValueError(f"Zone existiert bereits: {zone_id}")
- parent_zone = None
- if parent:
- parent_zone = self._zones.get(parent)
- if parent_zone is None:
- raise ValueError(f"Parent-Zone nicht gefunden: {parent}")
- zone = Zone(
- zone_id=zone_id,
- name=name,
- rooms=rooms,
- config=config,
- parent_zone=parent_zone,
- )
- self._zones[zone_id] = zone
- if parent_zone is not None:
- parent_zone.add_child(zone)
- # Room-Mapping aktualisieren
- for room_id in zone.rooms:
- self._room_mapping[room_id] = zone_id
- return zone
- def get(self, zone_id: str) -> Zone | None:
- """Gibt eine Zone nach ID zurück."""
- return self._zones.get(zone_id)
- def delete(self, zone_id: str, recursive: bool = False) -> Zone | None:
- """
- Löscht eine Zone.
- Args:
- zone_id: ID der zu löschenden Zone
- recursive: Auch alle Kinder-Zonen löschen
- Returns:
- Die gelöschte Zone oder None
- """
- zone = self._zones.get(zone_id)
- if zone is None:
- return None
- if recursive:
- for child in list(zone.children):
- self.delete(child.id, recursive=True)
- # Aus Parent entfernen
- if zone.parent is not None:
- zone.parent.remove_child(zone_id)
- # Room-Mapping bereinigen
- for room_id in zone.rooms:
- if self._room_mapping.get(room_id) == zone_id:
- del self._room_mapping[room_id]
- return self._zones.pop(zone_id, None)
- def get_zone_for_room(self, room_id: str) -> Zone | None:
- """
- Gibt die Zone für einen Raum zurück.
- Args:
- room_id: Raum-ID
- Returns:
- Die Zone oder None
- """
- zone_id = self._room_mapping.get(room_id)
- if zone_id:
- return self._zones.get(zone_id)
- return None
- def get_root_zones(self) -> list[Zone]:
- """Gibt alle Root-Zonen zurück (ohne Parent)."""
- return [z for z in self._zones.values() if z.is_root]
- def get_all(self) -> list[Zone]:
- """Gibt alle Zonen zurück."""
- return list(self._zones.values())
- def move_room(self, room_id: str, target_zone_id: str) -> bool:
- """
- Verschiebt einen Raum in eine andere Zone.
- Args:
- room_id: Raum-ID
- target_zone_id: Ziel-Zone-ID
- Returns:
- True wenn erfolgreich
- """
- target = self._zones.get(target_zone_id)
- if target is None:
- return False
- # Aus alter Zone entfernen
- old_zone_id = self._room_mapping.get(room_id)
- if old_zone_id:
- old_zone = self._zones.get(old_zone_id)
- if old_zone:
- old_zone.remove_room(room_id)
- # In neue Zone einfügen
- target.add_room(room_id)
- self._room_mapping[room_id] = target_zone_id
- return True
- def get_hierarchy(self) -> dict[str, Any]:
- """
- Gibt die Zone-Hierarchie als verschachteltes Dictionary zurück.
- Returns:
- Hierarchie-Dictionary
- """
- def build_tree(zone: Zone) -> dict[str, Any]:
- return {
- "id": zone.id,
- "name": zone.name,
- "rooms": list(zone.rooms),
- "children": [build_tree(c) for c in zone.children],
- }
- return {
- "root_zones": [build_tree(z) for z in self.get_root_zones()]
- }
- def __iter__(self) -> Iterator[Zone]:
- """Iterator über alle Zonen."""
- return iter(self._zones.values())
- def __len__(self) -> int:
- """Anzahl Zonen."""
- return len(self._zones)
- def __contains__(self, zone_id: str) -> bool:
- """Prüft, ob eine Zone existiert."""
- return zone_id in self._zones
- def __getitem__(self, zone_id: str) -> Zone:
- """Zugriff per ID."""
- if zone_id not in self._zones:
- raise KeyError(f"Zone nicht gefunden: {zone_id}")
- return self._zones[zone_id]
|