from __future__ import annotations import asyncio from asyncio import timeout as asyncio_timeout from collections.abc import Callable, Coroutine import contextlib from dataclasses import dataclass from datetime import UTC, datetime import enum import itertools import logging import math import time import typing from typing import Any, TypeVar import warnings from zigpy import zdo from zigpy.const import ( APS_REPLY_TIMEOUT, APS_REPLY_TIMEOUT_EXTENDED, SIG_ENDPOINTS, SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE, SIG_MANUFACTURER, SIG_MODEL, SIG_NODE_DESC, ) import zigpy.datastructures import zigpy.endpoint import zigpy.exceptions from zigpy.exceptions import DeliveryError import zigpy.listeners from zigpy.ota.manager import update_firmware from zigpy.profiles import zha, zll import zigpy.types as t import zigpy.util from zigpy.zcl import Cluster, ClusterType, OtaQueryCacheClearedEvent, foundation from zigpy.zcl.clusters.general import Ota, PollControl, QueryNextImageCommand import zigpy.zdo.types as zdo_t if typing.TYPE_CHECKING: _R = TypeVar("_R") from zigpy.application import ControllerApplication from zigpy.ota.providers import OtaImageWithMetadata LOGGER = logging.getLogger(__name__) PACKET_DEBOUNCE_WINDOW = 10 MAX_DEVICE_CONCURRENCY = 2 DEFAULT_FAST_POLL_TIMEOUT = 30 DEFAULT_REQUEST_RETRIES = 2 DEFAULT_REQUEST_RETRY_DELAY = 0.1 AFTER_OTA_ATTR_READ_DELAY = 10 @dataclass(frozen=True, slots=True) class ResponseKey: """Key for request/response matching.""" endpoint_id: int cluster_id: int direction: foundation.Direction | None tsn: int class Status(enum.IntEnum): """The status of a Device. Maintained for backwards compatibility.""" # No initialization done NEW = 0 # ZDO endpoint discovery done ZDO_INIT = 1 # Endpoints initialized ENDPOINTS_INIT = 2 class Device(zigpy.util.LocalLogMixin, zigpy.util.ListenableMixin): """A device on the network""" manufacturer_id_override = None def __init__(self, application: ControllerApplication, ieee: t.EUI64, nwk: t.NWK): self._application: ControllerApplication = application self._ieee: t.EUI64 = ieee self.nwk: t.NWK = t.NWK(nwk) self.zdo: zdo.ZDO = zdo.ZDO(self) self.endpoints: dict[int, zdo.ZDO | zigpy.endpoint.Endpoint] = {0: self.zdo} # Persist the original signature for the device, before quirks are applied self._original_signature: dict[str, Any] | None = None self.lqi: int | None = None self.rssi: int | None = None self.ota_in_progress: bool = False self._last_seen: datetime | None = None self._initialize_task: asyncio.Task | None = None self._reinterview_in_progress: bool = False self._group_scan_task: asyncio.Task | None = None self._fast_polling_reset_task: asyncio.Task | None = None self._listeners = {} self._manufacturer: str | None = None self._model: str | None = None self.node_desc: zdo_t.NodeDescriptor | None = None self._requests: dict[ResponseKey, asyncio.Future] = {} self._relays: t.Relays | None = None self._skip_configuration: bool = False self._send_sequence: int = 0 self._fast_polling = False self._on_remove_callbacks: list[typing.Callable[[], None]] = [] self._tasks: set[asyncio.Future[Any]] = set() self._packet_debouncer = zigpy.datastructures.Debouncer() self._concurrent_requests_semaphore = zigpy.datastructures.RequestLimiter( max_concurrency=MAX_DEVICE_CONCURRENCY, capacities={ t.PacketPriority.HIGH: 0.5, # t.PacketPriority.NORMAL is shared with LOW t.PacketPriority.LOW: 0.5, }, ) # Retained for backwards compatibility, will be removed in a future release self.status = Status.NEW self._on_remove_callbacks.append( self._application.register_callback_listener( src=self, filters=[PollControl.ClientCommandDefs.checkin.schema()], callback=self.poll_control_checkin_callback, ) ) def create_task( self, target: Coroutine[Any, Any, _R], name: str | None = None ) -> asyncio.Task[_R]: """Create a task and store a reference to it until the task completes. target: target to call. """ task = asyncio.get_running_loop().create_task(target, name=name) self._tasks.add(task) task.add_done_callback(self._tasks.remove) return task def on_remove(self) -> None: """Call on remove callbacks.""" for callback in self._on_remove_callbacks: callback() self._on_remove_callbacks.clear() for task in self._tasks: task.cancel() self._tasks.clear() @contextlib.asynccontextmanager async def _limit_concurrency(self, *, priority: int | None = None): """Async context manager to limit device request concurrency.""" # Defer to the current app-level priority if not specified if priority is None: priority = self._application._packet_priority_var.get() start_time = time.monotonic() manager: contextlib.AbstractAsyncContextManager if priority >= t.PacketPriority.CRITICAL: LOGGER.debug( "Critical priority request received (%s), skipping queue with %d requests", priority, self._concurrent_requests_semaphore.waiting_requests, ) manager = contextlib.nullcontext() was_locked = False else: manager = self._concurrent_requests_semaphore(priority=priority) was_locked = self._concurrent_requests_semaphore.locked(priority=priority) if was_locked: LOGGER.debug( "Device concurrency (%s) reached, delaying request (%s enqueued)", self._concurrent_requests_semaphore.active_requests, self._concurrent_requests_semaphore.waiting_requests, ) async with manager: if was_locked: LOGGER.debug( "Previously delayed device request is now running, delayed by %0.2fs", time.monotonic() - start_time, ) yield def get_sequence(self) -> int: self._send_sequence = (self._send_sequence + 1) % 256 return self._send_sequence @property def name(self) -> str: return f"0x{self.nwk:04X}" def update_last_seen(self) -> None: """Update the `last_seen` attribute to the current time and emit an event.""" warnings.warn( "Calling `update_last_seen` directly is deprecated", DeprecationWarning ) self.last_seen = datetime.now(UTC) @property def last_seen(self) -> float | None: return self._last_seen.timestamp() if self._last_seen is not None else None @last_seen.setter def last_seen(self, value: datetime | float | None): if isinstance(value, int | float): value = datetime.fromtimestamp(value, UTC) self._last_seen = value self.listener_event("device_last_seen_updated", self._last_seen) @property def non_zdo_endpoints(self) -> list[zigpy.endpoint.Endpoint]: return [ ep for epid, ep in self.endpoints.items() if not (isinstance(ep, zdo.ZDO)) ] @property def has_non_zdo_endpoints(self) -> bool: return bool(self.non_zdo_endpoints) @property def all_endpoints_init(self) -> bool: return self.has_non_zdo_endpoints and all( ep.status != zigpy.endpoint.Status.NEW for ep in self.non_zdo_endpoints ) @property def is_initialized(self) -> bool: return self.node_desc is not None and self.all_endpoints_init def schedule_group_membership_scan(self) -> asyncio.Task: """Rescan device group's membership.""" if self._group_scan_task and not self._group_scan_task.done(): self.debug("Cancelling old group rescan") self._group_scan_task.cancel() self._group_scan_task = self.create_task( self.group_membership_scan(), name="group_membership_scan" ) return self._group_scan_task async def group_membership_scan(self) -> None: """Sync up group membership.""" for ep in self.non_zdo_endpoints: await ep.group_membership_scan() @property def initializing(self) -> bool: """Return True if device is being initialized.""" return self._initialize_task is not None and not self._initialize_task.done() @property def reinterviewing(self) -> bool: """Return True if device is being re-interviewed.""" return self._reinterview_in_progress def cancel_initialization(self) -> None: """Cancel initialization call.""" if self.initializing: self.debug("Canceling old initialize call") self._initialize_task.cancel() # type:ignore[union-attr] def schedule_initialize(self) -> asyncio.Task | None: # Already-initialized devices don't need to be re-initialized if self.is_initialized: self.debug("Skipping initialization, device is fully initialized") self._application.device_initialized(self) return None if self.reinterviewing: self.debug("Skipping initialization, re-interview in progress") return None self.debug("Scheduling initialization") self.cancel_initialization() self._initialize_task = self.create_task(self.initialize(), name="initialize") return self._initialize_task async def reinterview(self) -> None: """Re-interview this device using a shadow device for safety. Creates a fresh Device, discovers its endpoints/clusters/model info, and on success swaps the old device out. On failure, the old device and its DB state are preserved. """ if self.reinterviewing: self.debug("Re-interview already in progress, skipping") return if self.initializing: self.debug("Initialization in progress, skipping re-interview") return if self.ota_in_progress: self.debug("OTA in progress, skipping re-interview") return self._reinterview_in_progress = True try: shadow = Device(self._application, self._ieee, self.nwk) shadow._reinterview_in_progress = True # prevent auto-initialization # Temporarily register the shadow in app.devices so it receives # ZDO responses routed by packet_received(). self._application.devices[self._ieee] = shadow try: async with self._application.request_priority( t.PacketPriority.CRITICAL ): await shadow._discover() except Exception: # Discovery failed — restore old device, clean up shadow self._application.devices[self._ieee] = self shadow.on_remove() raise # Discovery succeeded — swap the old device for the new one. # If this somehow fails, the state may be partially swapped; attempting # to undo would likely make things worse. await self._application._device_reinterviewed(self, shadow) except Exception: # noqa: BLE001 self.warning( "Re-interview failed, keeping existing device", exc_info=True, ) self._application.listener_event("device_reinterview_failure", self) finally: self._reinterview_in_progress = False async def get_node_descriptor(self) -> zdo_t.NodeDescriptor: self.info("Requesting 'Node Descriptor'") status, _, node_desc = await self.zdo.Node_Desc_req(self.nwk) if status != zdo_t.Status.SUCCESS: raise zigpy.exceptions.InvalidResponse( f"Requesting Node Descriptor failed: {status}" ) self.node_desc = node_desc self.info("Got Node Descriptor: %s", node_desc) return node_desc async def initialize(self) -> None: try: # Perform initialization with critical priority async with self._application.request_priority(t.PacketPriority.CRITICAL): await self._initialize() except (TimeoutError, zigpy.exceptions.ZigbeeException): self.application.listener_event("device_init_failure", self) except Exception: # noqa: BLE001 LOGGER.warning( "Device %r failed to initialize due to unexpected error", self, exc_info=True, ) self.application.listener_event("device_init_failure", self) def find_cluster( self, cluster_id: int, cluster_type: ClusterType = ClusterType.Server ) -> Cluster: """Find the first cluster by its ID and type on any endpoint.""" for ep in self.non_zdo_endpoints: if cluster_type == ClusterType.Server and cluster_id in ep.in_clusters: return ep.in_clusters[cluster_id] elif cluster_type == ClusterType.Client and cluster_id in ep.out_clusters: return ep.out_clusters[cluster_id] raise ValueError( f"Cluster {cluster_id:#06x} not found in any endpoint of device {self}" ) async def poll_control_checkin_callback( self, zcl_hdr: foundation.ZCLHeader, command: foundation.CommandSchema, ) -> None: """Handle Poll Control check-in callback.""" poll_control = self.find_cluster(cluster_id=PollControl.cluster_id) async with self._application.request_priority(t.PacketPriority.CRITICAL): # Initiate fast polling mode if we are initializing or waiting for requests # to be sent if ( self.initializing or self.reinterviewing or self._concurrent_requests_semaphore.active_requests > 0 or self._fast_polling ): # Initiate fast polling mode if we are initializing or waiting for # requests to be sent await poll_control.checkin_response( start_fast_polling=True, fast_poll_timeout=int(DEFAULT_FAST_POLL_TIMEOUT * 4), tsn=zcl_hdr.tsn, expect_reply=False, disable_default_response=True, ) else: await poll_control.checkin_response( start_fast_polling=False, fast_poll_timeout=0, tsn=zcl_hdr.tsn, expect_reply=False, disable_default_response=True, ) async def begin_fast_polling( self, timeout: float = DEFAULT_FAST_POLL_TIMEOUT, *, reset_after: bool = True ) -> None: """Ask the device to enter fast polling mode.""" try: poll_control = self.find_cluster(cluster_id=PollControl.cluster_id) except ValueError: LOGGER.debug("Device does not support fast polling") # The device doesn't have the cluster, there's nothing more we can do return # Cancel any fast poll reset tasks if self._fast_polling_reset_task is not None: self._fast_polling_reset_task.cancel() self._fast_polling_reset_task = None # The units are quarter seconds, we round up to the nearest one adjusted_timeout = math.ceil(timeout * 4) / 4 LOGGER.debug("Beginning fast polling for %0.2fs", adjusted_timeout) # We must first bind to the cluster, otherwise the device will not send a check- # in command await poll_control.bind() await poll_control.write_attributes( {PollControl.AttributeDefs.fast_poll_timeout.id: int(4 * adjusted_timeout)} ) self._fast_polling = True if reset_after: async def reset_fast_polling() -> None: await asyncio.sleep(adjusted_timeout) self._fast_polling = False self._fast_polling_reset_task = self.create_task( reset_fast_polling(), name="reset_fast_polling" ) @contextlib.asynccontextmanager async def fast_poll_mode( self, initial_timeout: float = DEFAULT_FAST_POLL_TIMEOUT ) -> None: """Ask the device to enter fast polling mode.""" await self.begin_fast_polling(timeout=initial_timeout, reset_after=False) try: yield finally: LOGGER.debug("Stopping fast polling on next device check-in") self._fast_polling = False async def _discover(self) -> None: """Discover all basic information about a device: its node descriptor, all endpoints and clusters, and the model and manufacturer attributes from any Basic cluster exposing those attributes. Does not signal completion. """ # Some devices are improperly initialized and are missing a node descriptor if self.node_desc is None: await self.get_node_descriptor() # Devices should have endpoints other than ZDO if self.has_non_zdo_endpoints: self.info("Already have endpoints: %s", self.endpoints) else: self.info("Discovering endpoints") status, _, endpoints = await self.zdo.Active_EP_req(self.nwk) if status != zdo_t.Status.SUCCESS: raise zigpy.exceptions.InvalidResponse( f"Endpoint request failed: {status}" ) self.info("Discovered endpoints: %s", endpoints) for endpoint_id in endpoints: if endpoint_id != 0: self.add_endpoint(endpoint_id) self.status = Status.ZDO_INIT # Initialize all of the discovered endpoints initiated_fast_polling = self._fast_polling if self.all_endpoints_init: self.info( "All endpoints are already initialized: %s", self.non_zdo_endpoints ) if not initiated_fast_polling: # Begin fast polling if we are re-initializing await self.begin_fast_polling() else: self.info("Initializing endpoints %s", self.non_zdo_endpoints) for ep in self.non_zdo_endpoints: await ep.initialize() if not initiated_fast_polling: # Ask the device to enter fast polling mode as soon as we are # aware of a PollControl cluster try: await self.begin_fast_polling() except (TimeoutError, DeliveryError): pass else: initiated_fast_polling = True # Query model info if self.model is not None and self.manufacturer is not None: self.info("Already have model and manufacturer info") else: for ep in self.non_zdo_endpoints: if self.model is None or self.manufacturer is None: model, manufacturer = await ep.get_model_info() self.info( "Read model %r and manufacturer %r from %s", model, manufacturer, ep, ) if model is not None: self.model = model if manufacturer is not None: self.manufacturer = manufacturer try: ota = self.find_cluster( cluster_id=Ota.cluster_id, cluster_type=ClusterType.Client ) except ValueError: self.debug("Device does not support OTA cluster") else: await ota.read_attributes([Ota.AttributeDefs.current_file_version.name]) # Notify the device that OTA images may be available so it queries us try: await ota.image_notify( payload_type=Ota.ImageNotifyCommand.PayloadType.QueryJitter, query_jitter=100, ) except Exception: # noqa: BLE001 self.debug( "OTA image_notify failed during initialization", exc_info=True ) self.status = Status.ENDPOINTS_INIT self.info("Discovered basic device information for %s", self) async def _initialize(self) -> None: """Discover device information and signal to the application.""" await self._discover() # Signal to the application that the device is ready self._application.device_initialized(self) def add_endpoint(self, endpoint_id) -> zigpy.endpoint.Endpoint: ep = zigpy.endpoint.Endpoint(self, endpoint_id) self.endpoints[endpoint_id] = ep return ep async def add_to_group(self, grp_id: int, name: str | None = None) -> None: for ep in self.non_zdo_endpoints: await ep.add_to_group(grp_id, name) async def remove_from_group(self, grp_id: int) -> None: for ep in self.non_zdo_endpoints: await ep.remove_from_group(grp_id) async def request( self, profile, cluster, src_ep, dst_ep, sequence, data, expect_reply=True, timeout=APS_REPLY_TIMEOUT, use_ieee=False, ask_for_ack: bool | None = None, priority: int | None = None, retries: int | None = None, retry_delay: float | None = None, **kwargs, ): if retries is None: retries = DEFAULT_REQUEST_RETRIES if retry_delay is None: retry_delay = DEFAULT_REQUEST_RETRY_DELAY extended_timeout = False if self.node_desc is None or self.node_desc.is_end_device: self.debug("Extending timeout for 0x%02x request", sequence) timeout = APS_REPLY_TIMEOUT_EXTENDED extended_timeout = True if expect_reply: if dst_ep == zdo.ZDO_ENDPOINT: rsp_key = ResponseKey( endpoint_id=dst_ep, # e.g. Node_Desc_req = 0x0002 corresponds to Node_Desc_rsp = 0x8002 cluster_id=cluster ^ 0x8000, direction=None, tsn=sequence, ) else: zcl_hdr, _ = foundation.ZCLHeader.deserialize(data) rsp_key = ResponseKey( endpoint_id=dst_ep, cluster_id=cluster, direction=zcl_hdr.frame_control.direction.flip(), tsn=sequence, ) if rsp_key in self._requests: self.debug( "Duplicate request key %s, pending requests %s", rsp_key, self._requests, ) raise zigpy.exceptions.ControllerException( f"Duplicate request key: {rsp_key}" ) else: rsp_key = None max_attempts = retries + 1 # Use a lambda so we don't leave the coroutine unawaited in case of an exception send_request = lambda attempt: self._application.request( # noqa: E731 device=self, profile=profile, cluster=cluster, src_ep=src_ep, dst_ep=dst_ep, sequence=sequence, data=data, expect_reply=expect_reply, use_ieee=use_ieee, extended_timeout=extended_timeout, ask_for_ack=ask_for_ack, priority=priority, force_route_discovery=(attempt > 0), **kwargs, ) for attempt in range(max_attempts): try: async with self._limit_concurrency(priority=priority): if not expect_reply: await send_request(attempt=attempt) return None assert rsp_key is not None future: asyncio.Future[list[Any] | foundation.CommandSchema] = ( asyncio.Future() ) self._requests[rsp_key] = future try: await send_request(attempt=attempt) async with asyncio_timeout(timeout): return await future finally: if not future.done(): future.cancel() self._requests.pop(rsp_key, None) except zigpy.exceptions.ParsingError: raise except Exception: LOGGER.debug( "Failed to send request, attempt %d of %d", attempt + 1, max_attempts, exc_info=True, ) if attempt >= max_attempts - 1: raise await asyncio.sleep(retry_delay) continue def handle_message( self, profile: int, cluster: int, src_ep: int, dst_ep: int, message: bytes, ): """Deprecated compatibility function. Use `packet_received` instead.""" warnings.warn( "`handle_message` is deprecated, use `packet_received`", DeprecationWarning ) self.packet_received( t.ZigbeePacket( profile_id=profile, cluster_id=cluster, src_ep=src_ep, dst_ep=dst_ep, data=t.SerializableBytes(message), dst=t.AddrModeAddress( addr_mode=t.AddrMode.NWK, address=self.nwk, ), ) ) def _find_zcl_cluster_strict( self, hdr: foundation.ZCLHeader, packet: t.ZigbeePacket ) -> Cluster: """Find the ZCL cluster for a given header and packet, strict.""" assert packet.src_ep is not None ep = self.endpoints[packet.src_ep] if hdr.frame_control.direction == foundation.Direction.Client_to_Server: return ep.out_clusters[packet.cluster_id] else: return ep.in_clusters[packet.cluster_id] def _find_zcl_cluster( self, hdr: foundation.ZCLHeader, packet: t.ZigbeePacket ) -> Cluster: """Find the ZCL cluster for a given header and packet.""" try: return self._find_zcl_cluster_strict(hdr, packet) except KeyError: # If the cluster is not found, try to find it with flipped direction. This # will be removed in 2025.9.0. cluster = self._find_zcl_cluster_strict( hdr.replace( frame_control=hdr.frame_control.replace( direction=hdr.frame_control.direction.flip() ) ), packet, ) LOGGER.debug( ( "Cluster 0x%04x on %r has incorrect direction (got %r for %r cluster)." " Please report this here: https://github.com/zigpy/zigpy/issues/1640" ), packet.cluster_id, self, hdr.frame_control.direction, cluster.cluster_type, ) return cluster def custom_profile_packet_received(self, packet: t.ZigbeePacket) -> None: """Handle packets with a custom profile ID.""" self.debug( "Received packet with custom profile 0x%04x, ignoring", packet.profile_id, ) def _should_filter_packet(self, packet: t.ZigbeePacket) -> bool: """Check if packet should be filtered as duplicate.""" return self._packet_debouncer.filter( # Be conservative with deduplication obj=packet.replace(timestamp=None, tsn=None, lqi=None, rssi=None), expire_in=PACKET_DEBOUNCE_WINDOW, ) def _parse_packet_header( self, packet: t.ZigbeePacket ) -> tuple[zdo_t.ZDOHeader | foundation.ZCLHeader, ResponseKey] | tuple[None, None]: """Parse packet header and create response key.""" data = packet.data.serialize() if packet.src_ep == zdo.ZDO_ENDPOINT: hdr, _ = zdo_t.ZDOHeader.deserialize(packet.cluster_id, data) rsp_key = ResponseKey( endpoint_id=packet.src_ep, cluster_id=packet.cluster_id, direction=None, tsn=hdr.tsn, ) return hdr, rsp_key elif packet.profile_id in (zha.PROFILE_ID, zll.PROFILE_ID): hdr, _ = foundation.ZCLHeader.deserialize(data) rsp_key = ResponseKey( endpoint_id=packet.src_ep, cluster_id=packet.cluster_id, direction=hdr.frame_control.direction, tsn=hdr.tsn, ) return hdr, rsp_key else: return None, None def _match_packet_endpoint_cluster( self, packet: t.ZigbeePacket, hdr: zdo_t.ZDOHeader | foundation.ZCLHeader ) -> ( tuple[zigpy.endpoint.Endpoint, Cluster] | tuple[zigpy.zdo.ZDO, None] | tuple[None, None] ): """Validate packet routing and find target endpoint and cluster.""" if packet.src_ep not in self.endpoints: self.debug( "Ignoring message on unknown endpoint %s (expected one of %s)", packet.src_ep, self.endpoints, ) return None, None endpoint = self.endpoints[packet.src_ep] if packet.src_ep == zdo.ZDO_ENDPOINT: assert isinstance(endpoint, zigpy.zdo.ZDO) return endpoint, None else: assert isinstance(endpoint, zigpy.endpoint.Endpoint) try: zcl_cluster = self._find_zcl_cluster(hdr, packet) except KeyError: self.debug( "Ignoring message on unknown cluster: 0x%04x", packet.cluster_id, ) return None, None else: return endpoint, zcl_cluster def _parse_packet_command( self, packet: t.ZigbeePacket, endpoint: typing.Any, zcl_cluster: Cluster | None ) -> typing.Any: """Deserialize packet data.""" data = packet.data.serialize() if packet.src_ep == zdo.ZDO_ENDPOINT: _, cmd = endpoint.deserialize(packet.cluster_id, data) else: assert zcl_cluster is not None _, cmd = zcl_cluster.deserialize(data) return cmd def _maybe_match_response( self, rsp_key: ResponseKey, cmd: typing.Any | None, error: Exception | None ) -> bool: """Handle response matching for pending requests, returns True if packet was matched.""" future = self._requests.get(rsp_key) if future is None: return False # Attribute reports often collide with command responses, they should never be # matched up if isinstance( cmd, foundation.GENERAL_COMMANDS[ foundation.GeneralCommand.Report_Attributes ].schema, ): return False try: if error is not None: future.set_exception(error) else: future.set_result(cmd) except asyncio.InvalidStateError: self.debug( "Invalid state on future for %s -- probably duplicate response", rsp_key, ) return True def packet_received(self, packet: t.ZigbeePacket) -> None: """Process received packet through the device's packet handling pipeline.""" self.last_seen = packet.timestamp if packet.lqi is not None: self.lqi = packet.lqi if packet.rssi is not None: self.rssi = packet.rssi # Filter duplicate packets if self._should_filter_packet(packet): self.debug("Filtering duplicate packet") return # Parse packet header and create response key hdr, rsp_key = self._parse_packet_header(packet) if hdr is None: self.custom_profile_packet_received(packet) return # Validate packet routing and find target endpoint/cluster endpoint, zcl_cluster = self._match_packet_endpoint_cluster(packet, hdr) if endpoint is None: return # Deserialize packet data try: cmd = self._parse_packet_command(packet, endpoint, zcl_cluster) except Exception as exc: # noqa: BLE001 cmd = None error = zigpy.exceptions.ParsingError() error.__cause__ = exc self.debug("Failed to parse packet %r", packet, exc_info=error) else: error = None # Handle response matching for pending requests if self._maybe_match_response(rsp_key, cmd, error): return # Skip further processing if there was a parsing error if error is not None: return # Pass the request off to a listener, if one is registered for listener in itertools.chain( self._application._req_listeners[zigpy.listeners.ANY_DEVICE], self._application._req_listeners[self], ): # Resolve only until the first future listener if listener.resolve(hdr, cmd) and isinstance( listener, zigpy.listeners.FutureListener ): break # Finally, pass it off to the cluster message handler. This will be removed. if zcl_cluster is not None: zcl_cluster.handle_message(hdr, cmd) else: assert isinstance(endpoint, zdo.ZDO) endpoint.handle_message(packet.profile_id, packet.cluster_id, hdr, cmd) async def reply( self, profile, cluster, src_ep, dst_ep, sequence, data, timeout=APS_REPLY_TIMEOUT, expect_reply: bool = False, use_ieee: bool = False, ask_for_ack: bool | None = None, priority: int | None = None, **kwargs, ): return await self.request( profile=profile, cluster=cluster, src_ep=src_ep, dst_ep=dst_ep, sequence=sequence, data=data, expect_reply=expect_reply, timeout=timeout, use_ieee=use_ieee, ask_for_ack=ask_for_ack, priority=priority, **kwargs, ) async def update_firmware( self, image: OtaImageWithMetadata, progress_callback: Callable[[int, int, float], None] | None = None, force: bool = False, ) -> foundation.Status | None: """Update device firmware.""" if self.ota_in_progress: self.debug("OTA already in progress") return None self.ota_in_progress = True try: result = await update_firmware( device=self, image=image, progress_callback=progress_callback, force=force, ) except Exception as exc: # noqa: BLE001 self.debug("OTA failed!", exc_info=exc) raise finally: self.ota_in_progress = False if result != foundation.Status.SUCCESS: return result # Clear the current file version when the update succeeds ota = self.find_cluster( cluster_id=Ota.cluster_id, cluster_type=ClusterType.Client ) ota.update_attribute(Ota.AttributeDefs.current_file_version.id, None) ota.last_query_cmd = None ota.emit( OtaQueryCacheClearedEvent.event_type, OtaQueryCacheClearedEvent( device_ieee=str(self.ieee), endpoint_id=ota.endpoint.endpoint_id, ), ) await asyncio.sleep(AFTER_OTA_ATTR_READ_DELAY) await ota.read_attributes( [Ota.AttributeDefs.current_file_version.name], retries=10, retry_delay=AFTER_OTA_ATTR_READ_DELAY, ) # Prompt device to send QueryNextImage with updated version for query cache try: await ota.image_notify( payload_type=Ota.ImageNotifyCommand.PayloadType.QueryJitter, query_jitter=100, ) except Exception: # noqa: BLE001 self.debug("Post-OTA image_notify failed", exc_info=True) # Re-interview the device after successful OTA to pick up # any changes in clusters/endpoints/model and re-apply quirks. # reinterview() handles its own errors internally. await self.reinterview() return result def get_last_ota_query_cmd(self) -> QueryNextImageCommand | None: """Return the last cached QueryNextImageCommand, preferring client clusters.""" for ep in self.non_zdo_endpoints: # Prefer client (out) clusters since runtime routing places # query_next_image there when both cluster types exist. for clusters in (ep.out_clusters, ep.in_clusters): cluster = clusters.get(Ota.cluster_id) if isinstance(cluster, Ota) and cluster.last_query_cmd is not None: return cluster.last_query_cmd return None def radio_details(self, lqi=None, rssi=None) -> None: if lqi is not None: self.lqi = lqi if rssi is not None: self.rssi = rssi def log(self, lvl, msg, *args, **kwargs) -> None: msg = "[0x%04x] " + msg args = (self.nwk, *args) LOGGER.log(lvl, msg, *args, **kwargs) @property def application(self) -> ControllerApplication: return self._application @property def ieee(self) -> t.EUI64: return self._ieee @property def manufacturer(self) -> str | None: return self._manufacturer @manufacturer.setter def manufacturer(self, value) -> None: if isinstance(value, str): self._manufacturer = value @property def manufacturer_id(self) -> int | None: """Return manufacturer id.""" if self.manufacturer_id_override: return self.manufacturer_id_override elif self.node_desc is not None: return self.node_desc.manufacturer_code else: return None @property def model(self) -> str | None: return self._model @model.setter def model(self, value) -> None: if isinstance(value, str): self._model = value @property def original_signature(self) -> dict[str, Any] | None: return self._original_signature @original_signature.setter def original_signature(self, value: dict[str, Any] | None) -> None: if self._original_signature is not None: return self._original_signature = value @property def skip_configuration(self) -> bool: return self._skip_configuration @skip_configuration.setter def skip_configuration(self, should_skip_configuration) -> None: if isinstance(should_skip_configuration, bool): self._skip_configuration = should_skip_configuration else: self._skip_configuration = False @property def relays(self) -> t.Relays | None: """Relay list.""" return self._relays @relays.setter def relays(self, relays: t.Relays | None) -> None: if relays is None: pass elif not isinstance(relays, t.Relays): relays = t.Relays(relays) self._relays = relays self.listener_event("device_relays_updated", relays) def __getitem__(self, key): return self.endpoints[key] def get_signature(self) -> dict[str, typing.Any]: # return the device signature by providing essential device information # - Model Identifier ( Attribute 0x0005 of Basic Cluster 0x0000 ) # - Manufacturer Name ( Attribute 0x0004 of Basic Cluster 0x0000 ) # - Endpoint list # - Profile Id, Device Id, Cluster Out, Cluster In signature: dict[str, typing.Any] = {} if self._manufacturer is not None: signature[SIG_MANUFACTURER] = self.manufacturer if self._model is not None: signature[SIG_MODEL] = self._model if self.node_desc is not None: signature[SIG_NODE_DESC] = self.node_desc.as_dict() for endpoint_id, endpoint in self.endpoints.items(): if endpoint_id == zdo.ZDO_ENDPOINT: # ZDO continue signature.setdefault(SIG_ENDPOINTS, {}) in_clusters = list(endpoint.in_clusters) out_clusters = list(endpoint.out_clusters) signature[SIG_ENDPOINTS][endpoint_id] = { SIG_EP_PROFILE: endpoint.profile_id, SIG_EP_TYPE: endpoint.device_type, SIG_EP_INPUT: in_clusters, SIG_EP_OUTPUT: out_clusters, } return signature def __repr__(self) -> str: return ( f"<" f"{type(self).__name__}" f" model={self.model!r}" f" manuf={self.manufacturer!r}" f" nwk={t.NWK(self.nwk)}" f" ieee={self.ieee}" f" is_initialized={self.is_initialized}" f">" ) async def broadcast( app, profile, cluster, src_ep, dst_ep, grpid, radius, sequence, data, broadcast_address=t.BroadcastAddress.RX_ON_WHEN_IDLE, ): return await app.broadcast( profile, cluster, src_ep, dst_ep, grpid, radius, sequence, data, broadcast_address=broadcast_address, )