from __future__ import annotations import enum import logging from typing import TYPE_CHECKING, Any from zigpy.const import APS_REPLY_TIMEOUT import zigpy.exceptions import zigpy.profiles import zigpy.types as t import zigpy.util import zigpy.zcl from zigpy.zcl.foundation import GENERAL_COMMANDS, GeneralCommand, Status as ZCLStatus from zigpy.zdo.types import Status as ZDOStatus if TYPE_CHECKING: from zigpy.device import Device LOGGER = logging.getLogger(__name__) class Status(enum.IntEnum): """The status of an Endpoint""" # No initialization is done NEW = 0 # Endpoint information (device type, clusters, etc) init done ZDO_INIT = 1 # Endpoint Inactive ENDPOINT_INACTIVE = 3 class Endpoint(zigpy.util.LocalLogMixin, zigpy.util.ListenableMixin): """An endpoint on a device on the network""" def __init__(self, device: Device, endpoint_id: int) -> None: self._device: Device = device self._endpoint_id: int = endpoint_id self._listeners: dict = {} self.status: Status = Status.NEW self.profile_id: int | None = None self.device_type: zigpy.profiles.zha.DeviceType | None = None self.in_clusters: dict = {} self.out_clusters: dict = {} self._cluster_attr: dict = {} self._member_of: dict = {} self._manufacturer: str | None = None self._model: str | None = None async def initialize(self) -> None: self.info("Discovering endpoint information") if self.profile_id is not None or self.status == Status.ENDPOINT_INACTIVE: self.info("Endpoint descriptor already queried") else: status, _, sd = await self._device.zdo.Simple_Desc_req( self._device.nwk, self._endpoint_id ) if status == ZDOStatus.NOT_ACTIVE: # These endpoints are essentially junk but this lets the device join self.status = Status.ENDPOINT_INACTIVE return elif status != ZDOStatus.SUCCESS: raise zigpy.exceptions.InvalidResponse( "Failed to retrieve service descriptor: %s", status ) self.info("Discovered endpoint information: %s", sd) self.profile_id = sd.profile self.device_type = sd.device_type if self.profile_id == zigpy.profiles.zha.PROFILE_ID: self.device_type = zigpy.profiles.zha.DeviceType(self.device_type) elif self.profile_id == zigpy.profiles.zll.PROFILE_ID: self.device_type = zigpy.profiles.zll.DeviceType(self.device_type) for cluster in sd.input_clusters: self.add_input_cluster(cluster) for cluster in sd.output_clusters: self.add_output_cluster(cluster) self.status = Status.ZDO_INIT @property def clusters(self) -> list[zigpy.zcl.Cluster]: """Return all clusters on this endpoint.""" return [*self.in_clusters.values(), *self.out_clusters.values()] def add_input_cluster( self, cluster_id: int, cluster: zigpy.zcl.Cluster | None = None ) -> zigpy.zcl.Cluster: """Adds an endpoint's input cluster (a server cluster supported by the device) """ if cluster is None: if cluster_id in self.in_clusters: return self.in_clusters[cluster_id] cluster = zigpy.zcl.Cluster.from_id(self, cluster_id, is_server=True) self.in_clusters[cluster_id] = cluster if cluster.ep_attribute is not None: self._cluster_attr[cluster.ep_attribute] = cluster if self._device.application._dblistener is not None: self._device.application._dblistener.register_cluster_events(cluster) return cluster def add_output_cluster( self, cluster_id: int, cluster: zigpy.zcl.Cluster | None = None ) -> zigpy.zcl.Cluster: """Adds an endpoint's output cluster (a client cluster supported by the device) """ if cluster is None: if cluster_id in self.out_clusters: return self.out_clusters[cluster_id] cluster = zigpy.zcl.Cluster.from_id(self, cluster_id, is_server=False) self.out_clusters[cluster_id] = cluster if self._device.application._dblistener is not None: self._device.application._dblistener.register_cluster_events(cluster) return cluster async def add_to_group(self, grp_id: int, name: str | None = None) -> ZCLStatus: try: res = await self.groups.add(grp_id, name) except AttributeError: self.debug("Cannot add 0x%04x group, no groups cluster", grp_id) return ZCLStatus.FAILURE if res[0] not in (ZCLStatus.SUCCESS, ZCLStatus.DUPLICATE_EXISTS): self.debug("Couldn't add to 0x%04x group: %s", grp_id, res[0]) return res[0] group = self.device.application.groups.add_group(grp_id, name) group.add_member(self) return res[0] async def remove_from_group(self, grp_id: int) -> ZCLStatus: try: res = await self.groups.remove(grp_id) except AttributeError: self.debug("Cannot remove 0x%04x group, no groups cluster", grp_id) return ZCLStatus.FAILURE if res[0] not in (ZCLStatus.SUCCESS, ZCLStatus.NOT_FOUND): self.debug("Couldn't remove to 0x%04x group: %s", grp_id, res[0]) return res[0] if grp_id in self.device.application.groups: self.device.application.groups[grp_id].remove_member(self) return res[0] async def group_membership_scan(self) -> None: """Sync up group membership.""" try: res = await self.groups.get_membership(groups=[]) except AttributeError: return except (TimeoutError, zigpy.exceptions.ZigbeeException): self.debug("Failed to sync-up group membership") return if isinstance(res, GENERAL_COMMANDS[GeneralCommand.Default_Response].schema): self.debug("Device does not support group commands: %s", res) return groups = set(res[1]) self.device.application.groups.update_group_membership(self, groups) async def get_model_info(self) -> tuple[str | None, str | None]: if zigpy.zcl.clusters.general.Basic.cluster_id not in self.in_clusters: return None, None # Some devices can't handle multiple attributes in the same read request for names in (["manufacturer", "model"], ["manufacturer"], ["model"]): try: success, failure = await self.basic.read_attributes( names, allow_cache=True ) except TimeoutError: # Only swallow the `TimeoutError` on the double attribute read if len(names) == 2: continue raise if "model" in success: self._model = success["model"] if "manufacturer" in success: self._manufacturer = success["manufacturer"] return self._model, self._manufacturer async def request( self, cluster: t.ClusterId, sequence: t.uint8_t, data: bytes, command_id: GeneralCommand | t.uint8_t = 0x00, timeout=APS_REPLY_TIMEOUT, expect_reply: bool = True, use_ieee: bool = False, ask_for_ack: bool | None = None, priority: int | None = None, retries: int | None = None, retry_delay: float | None = None, ): if self.profile_id == zigpy.profiles.zll.PROFILE_ID and not ( cluster == zigpy.zcl.clusters.lightlink.LightLink.cluster_id and command_id < 0x40 ): profile_id = zigpy.profiles.zha.PROFILE_ID else: profile_id = self.profile_id return await self.device.request( profile=profile_id, cluster=cluster, src_ep=self._endpoint_id, dst_ep=self._endpoint_id, sequence=sequence, data=data, timeout=timeout, expect_reply=expect_reply, use_ieee=use_ieee, ask_for_ack=ask_for_ack, priority=priority, retries=retries, retry_delay=retry_delay, ) async def reply( self, cluster: t.ClusterId, sequence: t.uint8_t, data: bytes, command_id: GeneralCommand | t.uint8_t = 0x00, timeout=APS_REPLY_TIMEOUT, expect_reply: bool = False, use_ieee: bool = False, ask_for_ack: bool | None = None, priority: int | None = None, retries: int | None = None, retry_delay: float | None = None, ) -> None: if self.profile_id == zigpy.profiles.zll.PROFILE_ID and not ( cluster == zigpy.zcl.clusters.lightlink.LightLink.cluster_id and command_id < 0x40 ): profile_id = zigpy.profiles.zha.PROFILE_ID else: profile_id = self.profile_id return await self.device.reply( profile=profile_id, cluster=cluster, src_ep=self._endpoint_id, dst_ep=self._endpoint_id, sequence=sequence, data=data, timeout=timeout, expect_reply=expect_reply, use_ieee=use_ieee, ask_for_ack=ask_for_ack, priority=priority, retries=retries, retry_delay=retry_delay, ) def log(self, lvl: int, msg: str, *args: Any, **kwargs: Any) -> None: msg = "[0x%04x:%s] " + msg args = (self._device.nwk, self._endpoint_id, *args) LOGGER.log(lvl, msg, *args, **kwargs) @property def device(self) -> Device: return self._device @property def endpoint_id(self) -> int: return self._endpoint_id @property def manufacturer(self) -> str | None: if self._manufacturer is not None: return self._manufacturer return self.device.manufacturer @manufacturer.setter def manufacturer(self, value) -> None: self.warning( "Overriding manufacturer from quirks is not supported and " "will be removed in the next zigpy version" ) self._manufacturer = value @property def manufacturer_id(self) -> int | None: """Return device's manufacturer id code.""" return self.device.manufacturer_id @property def member_of(self) -> dict: return self._member_of @property def model(self) -> str | None: if self._model is not None: return self._model return self.device.model @model.setter def model(self, value) -> None: self.warning( "Overriding model from quirks is not supported and " "will be removed in the next version" ) self._model = value @property def unique_id(self) -> tuple[t.EUI64, int]: return self.device.ieee, self.endpoint_id def __getattr__(self, name: str) -> zigpy.zcl.Cluster: try: return self._cluster_attr[name] except KeyError as exc: raise AttributeError from exc def __repr__(self) -> str: def cluster_repr(clusters): return ", ".join( [f"{c.ep_attribute}:0x{c.cluster_id:04X}" for c in clusters] ) return ( f"<{type(self).__name__}" f" id={self.endpoint_id}" f" in=[{cluster_repr(self.in_clusters.values())}]" f" out=[{cluster_repr(self.out_clusters.values())}]" f" status={self.status!r}" f">" )