Source code for pyisy.nodes.group

"""Representation of groups (scenes) from an ISY."""

from __future__ import annotations

from typing import TYPE_CHECKING

from ..constants import (
    FAMILY_GENERIC,
    INSTEON_STATELESS_NODEDEFID,
    ISY_VALUE_UNKNOWN,
    PROTO_GROUP,
)
from ..helpers import now
from .node import Node
from .nodebase import NodeBase

if TYPE_CHECKING:
    from . import Nodes


[docs] class Group(NodeBase): """ Interact with ISY groups (scenes). | nodes: The node manager object. | address: The node ID. | name: The node name. | members: List of the members in this group. | controllers: List of the controllers in this group. | spoken: The string of the Notes Spoken field. :ivar has_children: Boolean value indicating that group has no children. :ivar members: List of the members of this group. :ivar controllers: List of the controllers of this group. :ivar name: The name of this group. :ivar status: Watched property indicating the status of the group. :ivar group_all_on: Watched property indicating if all devices in group are on. """
[docs] def __init__( self, nodes: Nodes, address: str, name: str, members: list[str] | None = None, controllers: list[str] | None = None, family_id: str = FAMILY_GENERIC, pnode: str | None = None, flag: int = 0, ) -> None: """Initialize a Group class.""" self._all_on: bool = False self._controllers: list[str] = controllers or [] self._members: list[str] = members or [] super().__init__(nodes, address, name, 0, family_id=family_id, pnode=pnode, flag=flag) # listen for changes in children self._members_handlers = [ self._nodes[m].status_events.subscribe(self.update_callback) for m in self.members ] # get and update the status self._update()
[docs] def __del__(self) -> None: """Cleanup event handlers before deleting.""" for handler in self._members_handlers: handler.unsubscribe()
@property def controllers(self) -> list[str]: """Get the controller nodes of the scene/group.""" return self._controllers @property def group_all_on(self) -> bool: """Return the current node state.""" return self._all_on @group_all_on.setter def group_all_on(self, value: bool) -> bool: """Set the current node state and notify listeners.""" if self._all_on != value: self._all_on = value self._last_changed = now() # Re-publish the current status. Let users pick up the all on change. self.status_events.notify(self._status) return self._all_on @property def members(self) -> list[str]: """Get the members of the scene/group.""" return self._members @property def protocol(self) -> str: """Return the protocol for this entity.""" return PROTO_GROUP
[docs] async def update(self, event=None, wait_time: float = 0.0, xmldoc=None): """Update the group with values from the controller.""" return self._update(event, wait_time, xmldoc)
def _update(self, event=None, wait_time: float = 0.0, xmldoc=None): """Update the group with values from the controller.""" self._last_update = now() address_to_node: dict[str, Node] = {address: self._nodes[address] for address in self.members} valid_nodes = [ address for address, node_obj in address_to_node.items() if ( node_obj.status is not None and node_obj.status != ISY_VALUE_UNKNOWN and node_obj.node_def_id not in INSTEON_STATELESS_NODEDEFID ) ] on_nodes = [node for node in valid_nodes if int(address_to_node[node].status) > 0] if on_nodes: self.group_all_on = len(on_nodes) == len(valid_nodes) self.status = 255 return self.status = 0 self.group_all_on = False
[docs] def update_callback(self, event=None): """Handle synchronous callbacks for subscriber events.""" self._update(event)