From ddcaf1555d668f8093d26a0a0f3c406962565b26 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Mon, 17 Aug 2026 19:06:15 +0000 Subject: [PATCH 1/5] Make `Device` work with the new `ZigbeeDevice` base --- zha/zigbee/device.py | 1013 +++++++++++++++++++++++------------------- 1 file changed, 551 insertions(+), 462 deletions(-) diff --git a/zha/zigbee/device.py b/zha/zigbee/device.py index 90dfd4547..b8f280131 100644 --- a/zha/zigbee/device.py +++ b/zha/zigbee/device.py @@ -17,7 +17,7 @@ import time from typing import TYPE_CHECKING, Any, Final -from zigpy.device import Device as ZigpyDevice +from zigpy.device import BaseDevice as ZigpyBaseDevice, Device as ZigpyDevice import zigpy.exceptions from zigpy.profiles import PROFILES from zigpy.types import uint1_t, uint8_t, uint16_t @@ -147,7 +147,7 @@ def _cluster_entry(cluster_id: int, cluster: Cluster) -> dict[str, Any]: def get_device_automation_triggers( - device: zigpy.device.Device, + device: zigpy.device.BaseDevice, ) -> dict[tuple[str, str], dict[str, str]]: """Get the supported device automation triggers for a zigpy device.""" return { @@ -329,38 +329,24 @@ class ExtendedDeviceInfo(DeviceInfo): endpoint_names: list[EndpointNameInfo] -class Device(LogMixin, EventBase): - """ZHA Zigbee device object.""" - - # Authoring surface for hand-written quirks; `None` marks the unquirked fallback. - _device_match: DeviceMatch | None = None - _zigpy_device_cls: ReplacingZigpyDeviceFactory | None = None - _zigpy_device_transforms: tuple[ - Callable[[zigpy.device.Device], zigpy.device.Device], ... - ] = () +class BaseDevice(LogMixin, EventBase): + """Base class for all ZHA device types.""" # Cached properties that depend on the zigpy device and must be invalidated # when the underlying device is swapped (e.g. after a re-interview). - _ZIGPY_CACHED_PROPERTIES: Final = ( + _ZIGPY_CACHED_PROPERTIES: tuple[str, ...] = ( "name", "manufacturer", "model", "device_alerts", - "manufacturer_code", - "is_mains_powered", - "device_type", - "is_router", - "is_coordinator", - "is_end_device", "skip_configuration", "device_automation_commands", "device_automation_triggers", - "zigbee_signature", ) def __init__( self, - zigpy_device: zigpy.device.Device, + zigpy_device: zigpy.device.BaseDevice, _gateway: Gateway, ) -> None: """Initialize the gateway.""" @@ -379,7 +365,6 @@ def __init__( self._initialized: bool = False self.semaphore: asyncio.Semaphore = asyncio.Semaphore(3) self._on_remove_callbacks: list[Callable[[], None]] = [] - self._endpoints: dict[int, Endpoint] = {} self._available: bool = False self._checkins_missed_count: int = 0 @@ -387,7 +372,7 @@ def __init__( self._init_from_zigpy_device(zigpy_device) - def _init_from_zigpy_device(self, zigpy_device: zigpy.device.Device) -> None: + def _init_from_zigpy_device(self, zigpy_device: zigpy.device.BaseDevice) -> None: """(Re-)initialize device state from a zigpy device. Sets up the zigpy device reference, quirk metadata, cluster handlers, @@ -399,11 +384,10 @@ def _init_from_zigpy_device(self, zigpy_device: zigpy.device.Device) -> None: # the old handlers/entities but the lists themselves still hold stale # references. self._on_remove_callbacks.clear() - self._endpoints.clear() self._pending_entities.clear() self._discovered_entities.clear() - self._zigpy_device: ZigpyDevice = zigpy_device + self._zigpy_device: ZigpyBaseDevice = zigpy_device # Invalidate cached properties that depend on the zigpy device before # they are read below (e.g. is_mains_powered, is_coordinator). @@ -445,12 +429,6 @@ def _init_from_zigpy_device(self, zigpy_device: zigpy.device.Device) -> None: self.status: DeviceStatus = DeviceStatus.CREATED - for ep_id, endpoint in zigpy_device.endpoints.items(): - if ep_id != 0: - ep = Endpoint.new(endpoint, self) - self._endpoints[ep_id] = ep - self._on_remove_callbacks.append(ep.on_remove) - def __repr__(self) -> str: """Return a string representation of the device.""" return ( @@ -461,7 +439,7 @@ def __repr__(self) -> str: ) @property - def device(self) -> zigpy.device.Device: + def device(self) -> zigpy.device.BaseDevice: """Return underlying Zigpy device.""" return self._zigpy_device @@ -515,17 +493,8 @@ def manufacturer(self) -> str: return self._resolve_manufacturer() def _resolve_manufacturer(self) -> str: - """Resolve the manufacturer name (declarative quirks override this).""" - if self.is_active_coordinator: - manufacturer = ( - self.gateway.application_controller.state.node_info.manufacturer - ) - return manufacturer if manufacturer is not None else "" - - if self._zigpy_device.manufacturer is None: - return UNKNOWN_MANUFACTURER - - return self._zigpy_device.manufacturer + """Resolve the manufacturer name (subclasses and declarative quirks override).""" + raise NotImplementedError @cached_property def model(self) -> str: @@ -533,30 +502,18 @@ def model(self) -> str: return self._resolve_model() def _resolve_model(self) -> str: - """Resolve the model name (declarative quirks override this).""" - if self.is_active_coordinator: - model = self.gateway.application_controller.state.node_info.model - if model is None: - return f"Generic Zigbee Coordinator ({self.gateway.radio_type.pretty_name})" - return model - - if self._zigpy_device.model is None: - return UNKNOWN_MODEL - - return self._zigpy_device.model + """Resolve the model name (subclasses and declarative quirks override).""" + raise NotImplementedError @cached_property def device_alerts(self) -> Iterable[Any]: """Return device alerts for this device (declarative quirks override this).""" return [] - @cached_property + @property def manufacturer_code(self) -> int | None: """Return the manufacturer code for the device.""" - if self._zigpy_device.node_desc is None: - return None - - return self._zigpy_device.node_desc.manufacturer_code + raise NotImplementedError @property def nwk(self) -> NWK: @@ -578,21 +535,15 @@ def last_seen(self) -> float | None: """Return last_seen for device.""" return self._zigpy_device.last_seen - @cached_property + @property def is_mains_powered(self) -> bool | None: """Return true if device is mains powered.""" - if self._zigpy_device.node_desc is None: - return None - - return self._zigpy_device.node_desc.is_mains_powered + raise NotImplementedError - @cached_property + @property def device_type(self) -> str: """Return the logical device type for the device.""" - if self._zigpy_device.node_desc is None: - return UNKNOWN - - return self._zigpy_device.node_desc.logical_type.name + raise NotImplementedError @property def power_source(self) -> str: @@ -601,51 +552,15 @@ def power_source(self) -> str: POWER_MAINS_POWERED if self.is_mains_powered else POWER_BATTERY_OR_UNKNOWN ) - @cached_property - def is_router(self) -> bool | None: - """Return true if this is a routing capable device.""" - if self._zigpy_device.node_desc is None: - return None - - return self._zigpy_device.node_desc.is_router - - @cached_property - def is_coordinator(self) -> bool | None: - """Return true if this device represents a coordinator.""" - if self._zigpy_device.node_desc is None: - return None - - return self._zigpy_device.node_desc.is_coordinator - @property def is_active_coordinator(self) -> bool: """Return true if this device is the active coordinator.""" - if not self.is_coordinator: - return False - - return self.ieee == self.gateway.state.node_info.ieee - - @cached_property - def is_end_device(self) -> bool | None: - """Return true if this device is an end device.""" - if self._zigpy_device.node_desc is None: - return None - - return self._zigpy_device.node_desc.is_end_device - - @property - def is_groupable(self) -> bool: - """Return true if this device has a group cluster.""" - return self.is_active_coordinator or ( - self.available and bool(self.async_get_groupable_endpoints()) - ) + return False @cached_property def skip_configuration(self) -> bool: """Return true if the device should not issue configuration related commands.""" - if self._quirk_skip_configuration(): - return True - return self._zigpy_device.skip_configuration or bool(self.is_active_coordinator) + return self._quirk_skip_configuration() @property def gateway(self): @@ -691,45 +606,10 @@ def on_network(self, new_on_network: bool) -> None: if not new_on_network: self.debug("Device is not on the network, marking unavailable") - def _first_in_cluster(self, cluster_id: int) -> zigpy.zcl.Cluster | None: - """Return the first in_cluster with the given cluster_id across endpoints.""" - for ep_id, ep in self._zigpy_device.endpoints.items(): - if ep_id == 0: - continue - cluster = ep.in_clusters.get(cluster_id) - if cluster is not None: - return cluster - return None - @property - def basic_cluster(self) -> zigpy.zcl.Cluster | None: - """Return the first Basic cluster across endpoints, if present.""" - return self._first_in_cluster(Basic.cluster_id) - - @property - def identify_cluster(self) -> zigpy.zcl.Cluster | None: - """Return the first Identify cluster across endpoints, if present.""" - return self._first_in_cluster(Identify.cluster_id) - - @property - def endpoints(self) -> dict[int, Endpoint]: - """Return the endpoints for this device.""" - return self._endpoints - - @cached_property - def zigbee_signature(self) -> dict[str, Any]: - """Get zigbee signature for this device.""" - return { - ATTR_NODE_DESCRIPTOR: self._zigpy_device.node_desc, - ATTR_ENDPOINTS: { - signature[0]: signature[1] - for signature in [ - endpoint.zigbee_signature for endpoint in self._endpoints.values() - ] - }, - ATTR_MANUFACTURER: self.manufacturer, - ATTR_MODEL: self.model, - } + def signature(self) -> dict[str, Any]: + """Return the device signature reported in the device info.""" + raise NotImplementedError @property def firmware_version(self) -> str | None: @@ -778,22 +658,6 @@ def get_entity( ) return matches[0] - @classmethod - def new( - cls, - zigpy_dev: zigpy.device.Device, - gateway: Gateway, - ) -> Device: - """Create new device, dispatching to the factory matched during resolution.""" - if zigpy_dev.ieee == gateway.state.node_info.ieee: - return CoordinatorDevice(zigpy_dev, gateway) - - entry = getattr(zigpy_dev, QUIRK_REGISTRY_ENTRY_ATTR, None) - if entry is not None and entry.zha_device_factory is not None: - return entry.zha_device_factory(zigpy_dev, gateway) - - return cls(zigpy_dev, gateway) - def async_update_firmware_version(self, firmware_version: str) -> None: """Update device firmware version.""" if firmware_version == self._firmware_version: @@ -811,54 +675,7 @@ def async_update_firmware_version(self, firmware_version: str) -> None: ) async def _check_available(self, *_: Any) -> None: - # don't flip the availability state of the coordinator - if self.is_active_coordinator: - return - if self.last_seen is None: - self.debug("last_seen is None, marking the device unavailable") - self.update_available(False) - return - - difference = time.time() - self.last_seen - if difference < self.consider_unavailable_time: - self.debug( - "Device seen - marking the device available and resetting counter" - ) - self.update_available(True) - self._checkins_missed_count = 0 - return - - if self._gateway.config.allow_polling: - if ( - self._checkins_missed_count >= _CHECKIN_GRACE_PERIODS - or self.manufacturer == "LUMI" - or not self._endpoints - ): - self.debug( - ( - "last_seen is %s seconds ago and ping attempts have been exhausted," - " marking the device unavailable" - ), - difference, - ) - self.update_available(False) - return - - self._checkins_missed_count += 1 - self.debug( - "Attempting to checkin with device - missed checkins: %s", - self._checkins_missed_count, - ) - basic = self.basic_cluster - if basic is None: - self.debug("does not have a mandatory basic cluster") - self.update_available(False) - return - res = await safe_read( - basic, [ATTR_MANUFACTURER], allow_cache=False, only_cache=False - ) - if res.get(ATTR_MANUFACTURER) is not None: - self._checkins_missed_count = 0 + raise NotImplementedError def update_available(self, available: bool) -> None: """Update device availability and signal entities.""" @@ -934,111 +751,13 @@ def device_info(self) -> DeviceInfo: last_seen=update_time, available=self.available, device_type=self.device_type, - signature=self.zigbee_signature, + signature=self.signature, ) @property def extended_device_info(self) -> ExtendedDeviceInfo: """Get extended device information.""" - topology = self.gateway.application_controller.topology - names: list[EndpointNameInfo] = [] - for endpoint in (ep for epid, ep in self.device.endpoints.items() if epid): - profile = PROFILES.get(endpoint.profile_id) - if profile and endpoint.device_type is not None: - # DeviceType provides undefined enums - names.append( - EndpointNameInfo(name=profile.DeviceType(endpoint.device_type).name) - ) - else: - names.append( - EndpointNameInfo( - name=( - f"unknown {endpoint.device_type} device_type " - f"of 0x{(endpoint.profile_id or 0xFFFF):04x} profile id" - ) - ) - ) - - return ExtendedDeviceInfo( - **self.device_info.__dict__, - active_coordinator=self.is_active_coordinator, - entities={ - platform_entity.unique_id: platform_entity.state - for platform_entity in self.platform_entities.values() - }, - neighbors=[ - NeighborInfo( - device_type=neighbor.device_type.name, - rx_on_when_idle=neighbor.rx_on_when_idle.name, - relationship=neighbor.relationship.name, - extended_pan_id=neighbor.extended_pan_id, - ieee=neighbor.ieee, - nwk=neighbor.nwk, - permit_joining=neighbor.permit_joining.name, - depth=neighbor.depth, - lqi=neighbor.lqi, - ) - for neighbor in topology.neighbors[self.ieee] - ], - routes=[ - RouteInfo( - dest_nwk=route.DstNWK, - route_status=route.RouteStatus.name, - memory_constrained=route.MemoryConstrained, - many_to_one=route.ManyToOne, - route_record_required=route.RouteRecordRequired, - next_hop=route.NextHop, - ) - for route in topology.routes[self.ieee] - ], - endpoint_names=names, - ) - - async def async_configure(self) -> None: - """Configure the device.""" - self.debug("started configuration") - - if hasattr(self._zigpy_device, "apply_custom_configuration"): - self.debug("applying quirks custom device configuration") - await self._zigpy_device.apply_custom_configuration() - - self._discover_new_entities() - - # Configure binding and reporting from entity-level cluster configs - aggregated = aggregate_cluster_configs(self._discovered_entities) - if aggregated and not self.skip_configuration: - await configure_cluster_configs(self, aggregated) - - self.emit_reconfigure_done() - - self.debug("completed configuration") - - identify_cluster = self.identify_cluster - if ( - self.gateway.config.config.device_options.enable_identify_on_join - and identify_cluster is not None - and not self.skip_configuration - ): - self._gateway.async_create_task( - identify_cluster.trigger_effect( - effect_id=Identify.EffectIdentifier.Okay, - effect_variant=Identify.EffectVariant.Default, - ), - name=f"({self.nwk},{self.model}) trigger_effect identify", - eager_start=True, - ) - - async def async_rebuild_from_zigpy_device( - self, zigpy_device: zigpy.device.Device - ) -> None: - """Tear down and rebuild this device from a new zigpy device. - - Called by the gateway after a successful re-interview swaps the - underlying zigpy device. Emits entity removal events so listeners - (e.g. HA) can clean up stale entities. - """ - await self.async_teardown(emit_entity_events=True) - self._init_from_zigpy_device(zigpy_device) + raise NotImplementedError def emit_reconfigure_done(self) -> None: """Emit `DeviceConfiguredEvent`. @@ -1052,25 +771,12 @@ def emit_reconfigure_done(self) -> None: ) def discover_entities(self) -> Iterator[BaseEntity]: - """Yield the default (ZCL) entities for this device. + """Yield the entities for this device. Declarative quirks add their exposed entities by overriding this in zhaquirks' `QuirkV2Device`; hand-written quirks override it directly. """ - # TODO: purge old coordinator entities - if self.is_coordinator: - return - - for ep_id, endpoint in self.endpoints.items(): - if ep_id == 0: - continue - - _LOGGER.debug( - "Discovering entities for endpoint: %s-%s", - str(endpoint.device.ieee), - endpoint.id, - ) - yield from discovery.discover_entities_for_endpoint(endpoint) + raise NotImplementedError def _discover_new_entities(self) -> None: self._discovered_entities.clear() @@ -1204,110 +910,542 @@ async def recompute_entities(self) -> None: entities = list(self._platform_entities.values()) - # Remove all entities that are no longer supported - for entity in entities[:]: - entity.recompute_capabilities() + # Remove all entities that are no longer supported + for entity in entities[:]: + entity.recompute_capabilities() + + if not entity.is_supported() or not entity.is_supported_in_list(entities): + self.debug("Removing unsupported entity %s", entity) + await self._remove_entity(entity, remove=True) + entities.remove(entity) + + # Discover new entities + self._discover_new_entities() + await self._add_pending_entities() + + async def async_initialize(self, from_cache: bool = False) -> None: + """Initialize cluster handlers.""" + self.debug("started initialization") + + # We discover prospective entities before initialization + self._discover_new_entities() + + # Read initial attributes from entity-level cluster configs + aggregated = aggregate_cluster_configs(self._discovered_entities) + if aggregated and not self.skip_configuration: + await initialize_cluster_configs(aggregated, from_cache) + + # And add them after. Emit events only on re-initialization, not the first. + await self._add_pending_entities(emit_event=self._initialized) + self._initialized = True + + # Sync the device's firmware version with the first platform entity + for (platform, _unique_id), entity in self.platform_entities.items(): + if platform != Platform.UPDATE: + continue + + assert isinstance(entity, BaseFirmwareUpdateEntity) + self._firmware_version = entity.installed_version + + def entity_update_listener(event: EntityStateChangedEvent) -> None: + """Listen to firmware update entity changes.""" + entity = self.get_platform_entity(event.platform, event.unique_id) + assert isinstance(entity, BaseFirmwareUpdateEntity) + self.async_update_firmware_version(entity.installed_version) + + self._on_remove_callbacks.append( + entity.on_event(STATE_CHANGED, entity_update_listener) + ) + + break + + self.debug("power source: %s", self.power_source) + self.status = DeviceStatus.INITIALIZED + self.debug("completed initialization") + + async def async_teardown(self, *, emit_entity_events: bool) -> None: + """Tear down handlers, entities, and endpoints. + + Args: + emit_entity_events: When True, emit ``DeviceEntityRemovedEvent`` + for each removed entity so that listeners (e.g. HA) can clean + up. Shutdown paths pass False to avoid unnecessary traffic. + + """ + for callback in self._on_remove_callbacks: + try: + callback() + except Exception: + _LOGGER.warning( + "Failed to execute on_remove callback %s for device %s", + callback, + self, + exc_info=True, + ) + + for platform_entity in list(self._platform_entities.values()): + try: + await self._remove_entity( + platform_entity, emit_event=emit_entity_events + ) + except Exception: + _LOGGER.warning( + "Failed to remove platform entity %s for device %s", + platform_entity, + self, + exc_info=True, + ) + + for entity in self._pending_entities: + try: + await entity.on_remove() + except Exception: + _LOGGER.warning( + "Failed to remove pending entity %s for device %s", + entity, + self, + exc_info=True, + ) + + # Ensure stale pending entities aren't reprocessed if the device is + # re-initialized after removal (e.g. re-interview). + self._pending_entities.clear() + + async def on_remove(self) -> None: + """Cancel tasks this device owns (shutdown path).""" + await self.async_teardown(emit_entity_events=False) + + def log(self, level: int, msg: str, *args: Any, **kwargs: Any) -> None: + """Log a message.""" + msg = f"[%s](%s): {msg}" + args = (self.nwk, self.model) + args + _LOGGER.log(level, msg, *args, **kwargs) + + def _compute_primary_entity(self, entities: Sequence[PlatformEntity]) -> None: + """Compute the primary entity from a given set of entities.""" + + # First, check if any entity is explicitly primary + explicitly_primary = [entity for entity in entities if entity.primary] + + if len(explicitly_primary) == 1: + self.debug( + "Device has a single explicitly primary entity," + " not performing weight matching" + ) + return + + # It should not be possible for there to be more than one + assert not explicitly_primary + + # For weight matching, only consider entities with a non-zero primary weight + # which are not explicitly marked as not primary + candidates = [ + e + for e in entities + if e.enabled and e._attr_primary is not False and e.primary_weight > 0 + ] + candidates.sort(reverse=True, key=lambda e: e.primary_weight) + + if not candidates: + return + + winner = candidates[0] + others = candidates[1:] + + # We have a clear winner + if not others or winner.primary_weight > others[0].primary_weight: + winner.primary = True + + for entity in others: + entity.primary = False + + return + + self.debug( + "Primary entity tie between %s and %s, no primary entity", winner, others[0] + ) + + for entity in candidates: + entity.primary = False + + +class ZigbeeDevice(BaseDevice): + """ZHA Zigbee device object.""" + + # Authoring surface for hand-written quirks; `None` marks the unquirked fallback. + _device_match: DeviceMatch | None = None + _zigpy_device_cls: ReplacingZigpyDeviceFactory | None = None + _zigpy_device_transforms: tuple[ + Callable[[zigpy.device.Device], zigpy.device.Device], ... + ] = () + + _ZIGPY_CACHED_PROPERTIES = ( + *BaseDevice._ZIGPY_CACHED_PROPERTIES, + "manufacturer_code", + "is_mains_powered", + "device_type", + "is_router", + "is_coordinator", + "is_end_device", + "zigbee_signature", + ) + + _zigpy_device: ZigpyDevice + + def __init__( + self, + zigpy_device: ZigpyDevice, + _gateway: Gateway, + ) -> None: + """Initialize the Zigbee device.""" + self._endpoints: dict[int, Endpoint] = {} + super().__init__(zigpy_device, _gateway) + + def _init_from_zigpy_device(self, zigpy_device: ZigpyDevice) -> None: + self._endpoints.clear() + super()._init_from_zigpy_device(zigpy_device) + + for ep_id, endpoint in zigpy_device.endpoints.items(): + if ep_id != 0: + ep = Endpoint.new(endpoint, self) + self._endpoints[ep_id] = ep + self._on_remove_callbacks.append(ep.on_remove) + + @property + def device(self) -> ZigpyDevice: + """Return underlying Zigpy device.""" + return self._zigpy_device + + def _resolve_manufacturer(self) -> str: + """Resolve the manufacturer name (declarative quirks override this).""" + if self.is_active_coordinator: + manufacturer = ( + self.gateway.application_controller.state.node_info.manufacturer + ) + return manufacturer if manufacturer is not None else "" + + if self._zigpy_device.manufacturer is None: + return UNKNOWN_MANUFACTURER + + return self._zigpy_device.manufacturer + + def _resolve_model(self) -> str: + """Resolve the model name (declarative quirks override this).""" + if self.is_active_coordinator: + model = self.gateway.application_controller.state.node_info.model + if model is None: + return f"Generic Zigbee Coordinator ({self.gateway.radio_type.pretty_name})" + return model + + if self._zigpy_device.model is None: + return UNKNOWN_MODEL + + return self._zigpy_device.model + + @cached_property + def manufacturer_code(self) -> int | None: + """Return the manufacturer code for the device.""" + if self._zigpy_device.node_desc is None: + return None + + return self._zigpy_device.node_desc.manufacturer_code + + @cached_property + def is_mains_powered(self) -> bool | None: + """Return true if device is mains powered.""" + if self._zigpy_device.node_desc is None: + return None + + return self._zigpy_device.node_desc.is_mains_powered + + @cached_property + def device_type(self) -> str: + """Return the logical device type for the device.""" + if self._zigpy_device.node_desc is None: + return UNKNOWN + + return self._zigpy_device.node_desc.logical_type.name + + @cached_property + def is_router(self) -> bool | None: + """Return true if this is a routing capable device.""" + if self._zigpy_device.node_desc is None: + return None + + return self._zigpy_device.node_desc.is_router + + @cached_property + def is_coordinator(self) -> bool | None: + """Return true if this device represents a coordinator.""" + if self._zigpy_device.node_desc is None: + return None + + return self._zigpy_device.node_desc.is_coordinator + + @property + def is_active_coordinator(self) -> bool: + """Return true if this device is the active coordinator.""" + if not self.is_coordinator: + return False + + return self.ieee == self.gateway.state.node_info.ieee + + @cached_property + def is_end_device(self) -> bool | None: + """Return true if this device is an end device.""" + if self._zigpy_device.node_desc is None: + return None + + return self._zigpy_device.node_desc.is_end_device + + @property + def is_groupable(self) -> bool: + """Return true if this device has a group cluster.""" + return self.is_active_coordinator or ( + self.available and bool(self.async_get_groupable_endpoints()) + ) + + @cached_property + def skip_configuration(self) -> bool: + """Return true if the device should not issue configuration related commands.""" + if self._quirk_skip_configuration(): + return True + return self._zigpy_device.skip_configuration or bool(self.is_active_coordinator) + + def _first_in_cluster(self, cluster_id: int) -> zigpy.zcl.Cluster | None: + """Return the first in_cluster with the given cluster_id across endpoints.""" + for ep_id, ep in self._zigpy_device.endpoints.items(): + if ep_id == 0: + continue + cluster = ep.in_clusters.get(cluster_id) + if cluster is not None: + return cluster + return None + + @property + def basic_cluster(self) -> zigpy.zcl.Cluster | None: + """Return the first Basic cluster across endpoints, if present.""" + return self._first_in_cluster(Basic.cluster_id) + + @property + def identify_cluster(self) -> zigpy.zcl.Cluster | None: + """Return the first Identify cluster across endpoints, if present.""" + return self._first_in_cluster(Identify.cluster_id) + + @property + def endpoints(self) -> dict[int, Endpoint]: + """Return the endpoints for this device.""" + return self._endpoints + + @cached_property + def zigbee_signature(self) -> dict[str, Any]: + """Get zigbee signature for this device.""" + return { + ATTR_NODE_DESCRIPTOR: self._zigpy_device.node_desc, + ATTR_ENDPOINTS: { + signature[0]: signature[1] + for signature in [ + endpoint.zigbee_signature for endpoint in self._endpoints.values() + ] + }, + ATTR_MANUFACTURER: self.manufacturer, + ATTR_MODEL: self.model, + } + + @property + def signature(self) -> dict[str, Any]: + """Return the device signature reported in the device info.""" + return self.zigbee_signature + + @classmethod + def new( + cls, + zigpy_dev: zigpy.device.Device, + gateway: Gateway, + ) -> Device: + """Create new device, dispatching to the factory matched during resolution.""" + if zigpy_dev.ieee == gateway.state.node_info.ieee: + return CoordinatorDevice(zigpy_dev, gateway) + + entry = getattr(zigpy_dev, QUIRK_REGISTRY_ENTRY_ATTR, None) + if entry is not None and entry.zha_device_factory is not None: + return entry.zha_device_factory(zigpy_dev, gateway) + + return cls(zigpy_dev, gateway) + + async def _check_available(self, *_: Any) -> None: + # don't flip the availability state of the coordinator + if self.is_active_coordinator: + return + if self.last_seen is None: + self.debug("last_seen is None, marking the device unavailable") + self.update_available(False) + return + + difference = time.time() - self.last_seen + if difference < self.consider_unavailable_time: + self.debug( + "Device seen - marking the device available and resetting counter" + ) + self.update_available(True) + self._checkins_missed_count = 0 + return + + if self._gateway.config.allow_polling: + if ( + self._checkins_missed_count >= _CHECKIN_GRACE_PERIODS + or self.manufacturer == "LUMI" + or not self._endpoints + ): + self.debug( + ( + "last_seen is %s seconds ago and ping attempts have been exhausted," + " marking the device unavailable" + ), + difference, + ) + self.update_available(False) + return + + self._checkins_missed_count += 1 + self.debug( + "Attempting to checkin with device - missed checkins: %s", + self._checkins_missed_count, + ) + basic = self.basic_cluster + if basic is None: + self.debug("does not have a mandatory basic cluster") + self.update_available(False) + return + res = await safe_read( + basic, [ATTR_MANUFACTURER], allow_cache=False, only_cache=False + ) + if res.get(ATTR_MANUFACTURER) is not None: + self._checkins_missed_count = 0 + + @property + def extended_device_info(self) -> ExtendedDeviceInfo: + """Get extended device information.""" + topology = self.gateway.application_controller.topology + names: list[EndpointNameInfo] = [] + for endpoint in (ep for epid, ep in self.device.endpoints.items() if epid): + profile = PROFILES.get(endpoint.profile_id) + if profile and endpoint.device_type is not None: + # DeviceType provides undefined enums + names.append( + EndpointNameInfo(name=profile.DeviceType(endpoint.device_type).name) + ) + else: + names.append( + EndpointNameInfo( + name=( + f"unknown {endpoint.device_type} device_type " + f"of 0x{(endpoint.profile_id or 0xFFFF):04x} profile id" + ) + ) + ) - if not entity.is_supported() or not entity.is_supported_in_list(entities): - self.debug("Removing unsupported entity %s", entity) - await self._remove_entity(entity, remove=True) - entities.remove(entity) + return ExtendedDeviceInfo( + **self.device_info.__dict__, + active_coordinator=self.is_active_coordinator, + entities={ + platform_entity.unique_id: platform_entity.state + for platform_entity in self.platform_entities.values() + }, + neighbors=[ + NeighborInfo( + device_type=neighbor.device_type.name, + rx_on_when_idle=neighbor.rx_on_when_idle.name, + relationship=neighbor.relationship.name, + extended_pan_id=neighbor.extended_pan_id, + ieee=neighbor.ieee, + nwk=neighbor.nwk, + permit_joining=neighbor.permit_joining.name, + depth=neighbor.depth, + lqi=neighbor.lqi, + ) + for neighbor in topology.neighbors[self.ieee] + ], + routes=[ + RouteInfo( + dest_nwk=route.DstNWK, + route_status=route.RouteStatus.name, + memory_constrained=route.MemoryConstrained, + many_to_one=route.ManyToOne, + route_record_required=route.RouteRecordRequired, + next_hop=route.NextHop, + ) + for route in topology.routes[self.ieee] + ], + endpoint_names=names, + ) - # Discover new entities - self._discover_new_entities() - await self._add_pending_entities() + async def async_configure(self) -> None: + """Configure the device.""" + self.debug("started configuration") - async def async_initialize(self, from_cache: bool = False) -> None: - """Initialize cluster handlers.""" - self.debug("started initialization") + if hasattr(self._zigpy_device, "apply_custom_configuration"): + self.debug("applying quirks custom device configuration") + await self._zigpy_device.apply_custom_configuration() - # We discover prospective entities before initialization self._discover_new_entities() - # Read initial attributes from entity-level cluster configs + # Configure binding and reporting from entity-level cluster configs aggregated = aggregate_cluster_configs(self._discovered_entities) if aggregated and not self.skip_configuration: - await initialize_cluster_configs(aggregated, from_cache) - - # And add them after. Emit events only on re-initialization, not the first. - await self._add_pending_entities(emit_event=self._initialized) - self._initialized = True - - # Sync the device's firmware version with the first platform entity - for (platform, _unique_id), entity in self.platform_entities.items(): - if platform != Platform.UPDATE: - continue + await configure_cluster_configs(self, aggregated) - assert isinstance(entity, BaseFirmwareUpdateEntity) - self._firmware_version = entity.installed_version + self.emit_reconfigure_done() - def entity_update_listener(event: EntityStateChangedEvent) -> None: - """Listen to firmware update entity changes.""" - entity = self.get_platform_entity(event.platform, event.unique_id) - assert isinstance(entity, BaseFirmwareUpdateEntity) - self.async_update_firmware_version(entity.installed_version) + self.debug("completed configuration") - self._on_remove_callbacks.append( - entity.on_event(STATE_CHANGED, entity_update_listener) + identify_cluster = self.identify_cluster + if ( + self.gateway.config.config.device_options.enable_identify_on_join + and identify_cluster is not None + and not self.skip_configuration + ): + self._gateway.async_create_task( + identify_cluster.trigger_effect( + effect_id=Identify.EffectIdentifier.Okay, + effect_variant=Identify.EffectVariant.Default, + ), + name=f"({self.nwk},{self.model}) trigger_effect identify", + eager_start=True, ) - break - - self.debug("power source: %s", self.power_source) - self.status = DeviceStatus.INITIALIZED - self.debug("completed initialization") - - async def async_teardown(self, *, emit_entity_events: bool) -> None: - """Tear down handlers, entities, and endpoints. - - Args: - emit_entity_events: When True, emit ``DeviceEntityRemovedEvent`` - for each removed entity so that listeners (e.g. HA) can clean - up. Shutdown paths pass False to avoid unnecessary traffic. + async def async_rebuild_from_zigpy_device( + self, zigpy_device: zigpy.device.Device + ) -> None: + """Tear down and rebuild this device from a new zigpy device. + Called by the gateway after a successful re-interview swaps the + underlying zigpy device. Emits entity removal events so listeners + (e.g. HA) can clean up stale entities. """ - for callback in self._on_remove_callbacks: - try: - callback() - except Exception: - _LOGGER.warning( - "Failed to execute on_remove callback %s for device %s", - callback, - self, - exc_info=True, - ) + await self.async_teardown(emit_entity_events=True) + self._init_from_zigpy_device(zigpy_device) - for platform_entity in list(self._platform_entities.values()): - try: - await self._remove_entity( - platform_entity, emit_event=emit_entity_events - ) - except Exception: - _LOGGER.warning( - "Failed to remove platform entity %s for device %s", - platform_entity, - self, - exc_info=True, - ) + def discover_entities(self) -> Iterator[BaseEntity]: + """Yield the default (ZCL) entities for this device. - for entity in self._pending_entities: - try: - await entity.on_remove() - except Exception: - _LOGGER.warning( - "Failed to remove pending entity %s for device %s", - entity, - self, - exc_info=True, - ) + Declarative quirks add their exposed entities by overriding this in + zhaquirks' `QuirkV2Device`; hand-written quirks override it directly. + """ + # TODO: purge old coordinator entities + if self.is_coordinator: + return - # Ensure stale pending entities aren't reprocessed if the device is - # re-initialized after removal (e.g. re-interview). - self._pending_entities.clear() + for ep_id, endpoint in self.endpoints.items(): + if ep_id == 0: + continue - async def on_remove(self) -> None: - """Cancel tasks this device owns (shutdown path).""" - await self.async_teardown(emit_entity_events=False) + _LOGGER.debug( + "Discovering entities for endpoint: %s-%s", + str(endpoint.device.ieee), + endpoint.id, + ) + yield from discovery.discover_entities_for_endpoint(endpoint) def async_get_clusters(self) -> dict[int, dict[str, dict[int, Cluster]]]: """Get all clusters for this device.""" @@ -1603,59 +1741,6 @@ async def _async_group_binding_operation( fmt = f"{log_msg[1]} completed: %s" zdo.debug(fmt, *(log_msg[2] + (outcome,))) - def log(self, level: int, msg: str, *args: Any, **kwargs: Any) -> None: - """Log a message.""" - msg = f"[%s](%s): {msg}" - args = (self.nwk, self.model) + args - _LOGGER.log(level, msg, *args, **kwargs) - - def _compute_primary_entity(self, entities: Sequence[PlatformEntity]) -> None: - """Compute the primary entity from a given set of entities.""" - - # First, check if any entity is explicitly primary - explicitly_primary = [entity for entity in entities if entity.primary] - - if len(explicitly_primary) == 1: - self.debug( - "Device has a single explicitly primary entity," - " not performing weight matching" - ) - return - - # It should not be possible for there to be more than one - assert not explicitly_primary - - # For weight matching, only consider entities with a non-zero primary weight - # which are not explicitly marked as not primary - candidates = [ - e - for e in entities - if e.enabled and e._attr_primary is not False and e.primary_weight > 0 - ] - candidates.sort(reverse=True, key=lambda e: e.primary_weight) - - if not candidates: - return - - winner = candidates[0] - others = candidates[1:] - - # We have a clear winner - if not others or winner.primary_weight > others[0].primary_weight: - winner.primary = True - - for entity in others: - entity.primary = False - - return - - self.debug( - "Primary entity tie between %s and %s, no primary entity", winner, others[0] - ) - - for entity in candidates: - entity.primary = False - def get_diagnostics_json(self): """Get ZHA device information.""" @@ -1818,7 +1903,7 @@ def get_diagnostics_json(self): return info -class CoordinatorDevice(Device): +class CoordinatorDevice(ZigbeeDevice): """ZHA wrapper for the active coordinator device.""" def discover_entities(self) -> Iterator[BaseEntity]: @@ -1845,3 +1930,7 @@ def discover_entities(self) -> Iterator[BaseEntity]: sensor.DeviceCounterSensor.__name__, f"counter groups[{counter_groups}] counter group[{counter_group}] counter[{counter}]", ) + + +# Backwards-compatible alias +Device = ZigbeeDevice From 2c1e39314f6d4259cead7f6a44cc3186de151f31 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Mon, 17 Aug 2026 19:11:59 +0000 Subject: [PATCH 2/5] Migrate `PlatformEntity` to `ZclPlatformEntity` --- zha/application/discovery.py | 8 +- zha/application/platforms/__init__.py | 137 +++++++++++------- .../platforms/alarm_control_panel/__init__.py | 4 +- .../platforms/binary_sensor/__init__.py | 4 +- zha/application/platforms/button/__init__.py | 4 +- zha/application/platforms/climate/__init__.py | 4 +- zha/application/platforms/cover/__init__.py | 4 +- zha/application/platforms/device_tracker.py | 4 +- zha/application/platforms/fan/__init__.py | 6 +- zha/application/platforms/light/__init__.py | 4 +- zha/application/platforms/lock/__init__.py | 4 +- zha/application/platforms/number/__init__.py | 4 +- zha/application/platforms/select.py | 6 +- zha/application/platforms/sensor/__init__.py | 14 +- zha/application/platforms/siren.py | 4 +- zha/application/platforms/switch.py | 8 +- zha/application/platforms/update.py | 4 +- zha/application/platforms/virtual.py | 4 +- zha/zigbee/cluster_config.py | 4 +- 19 files changed, 136 insertions(+), 95 deletions(-) diff --git a/zha/application/discovery.py b/zha/application/discovery.py index 25906dfa8..1121b13db 100644 --- a/zha/application/discovery.py +++ b/zha/application/discovery.py @@ -19,8 +19,8 @@ BaseEntity, ClusterConfig, ClusterMatch, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, alarm_control_panel, binary_sensor, button, @@ -159,7 +159,7 @@ def _is_renamed_cluster(cluster: Cluster) -> bool: return cluster.ep_attribute != standard.ep_attribute -def discover_entities_for_endpoint(endpoint: Endpoint) -> Iterator[PlatformEntity]: # noqa: C901 +def discover_entities_for_endpoint(endpoint: Endpoint) -> Iterator[ZclPlatformEntity]: # noqa: C901 """Discover entities for an endpoint using the new registry-based discovery.""" device = endpoint.device @@ -178,7 +178,7 @@ def discover_entities_for_endpoint(endpoint: Endpoint) -> Iterator[PlatformEntit PlatformFeatureGroup | None, defaultdict[ int, # Weight - list[tuple[ClusterMatch, type[PlatformEntity]]], + list[tuple[ClusterMatch, type[ZclPlatformEntity]]], ], ] = defaultdict(lambda: defaultdict(list)) @@ -290,7 +290,7 @@ def discover_entities_for_endpoint(endpoint: Endpoint) -> Iterator[PlatformEntit if platform_override is not None and feature is not None: override_by_priority: defaultdict[ int, - list[tuple[ClusterMatch, type[PlatformEntity]]], + list[tuple[ClusterMatch, type[ZclPlatformEntity]]], ] = defaultdict(list) for priority, priority_matches in matches_by_priority.items(): diff --git a/zha/application/platforms/__init__.py b/zha/application/platforms/__init__.py index e1c789f77..c266ecda9 100644 --- a/zha/application/platforms/__init__.py +++ b/zha/application/platforms/__init__.py @@ -40,7 +40,9 @@ DEFAULT_UPDATE_GROUP_FROM_CHILD_DELAY: float = 0.5 -ENTITY_REGISTRY: dict[ClusterId | int, list[type[PlatformEntity]]] = defaultdict(list) +ENTITY_REGISTRY: dict[ClusterId | int, list[type[ZclPlatformEntity]]] = defaultdict( + list +) GROUP_ENTITY_REGISTRY: list[type[GroupEntity]] = [] @@ -142,7 +144,7 @@ def __post_init__(self) -> None: ) -def register_entity[T: type[PlatformEntity]]( +def register_entity[T: type[ZclPlatformEntity]]( cluster_id: ClusterId | int, ) -> Callable[[T], T]: """Register an entity class for discovery.""" @@ -218,7 +220,7 @@ class PlatformEntityIdentifiers(BaseIdentifiers): """Identifiers for the platform entity.""" device_ieee: EUI64 - endpoint_id: int + endpoint_id: int | None = None @dataclasses.dataclass(frozen=True, kw_only=True) @@ -508,7 +510,7 @@ def log(self, level: int, msg: str, *args: Any, **kwargs: Any) -> None: class PlatformEntity(BaseEntity): - """Class that represents an entity for a device platform.""" + """Class that represents a device-bound entity for a device platform.""" # suffix to add to the unique_id of the entity. Used for multi # entities using the same cluster handler/cluster id for the entity. @@ -516,20 +518,11 @@ class PlatformEntity(BaseEntity): _migrate_platform_unique_ids: tuple[tuple[UniqueIdMigration, str]] | None = None - # Direct cluster matching for discovery - _cluster_match: ClusterMatch | None = None - - # Per-cluster configuration (keyed by cluster ID) - _server_cluster_config: Mapping[int, ClusterConfig] = MappingProxyType({}) - - _client_cluster_config: Mapping[int, ClusterConfig] = MappingProxyType({}) - def __init__( self, - endpoint: Endpoint, device: Device, *, - cluster: zigpy.zcl.Cluster, + unique_id: str, from_quirk: bool = False, fallback_name: str | None = None, translation_key: str | None = None, @@ -538,15 +531,13 @@ def __init__( entity_type: EntityType | None = None, primary: bool | None = None, initially_disabled: bool = False, - legacy_discovery_unique_id: str | None = None, **kwargs: Any, ): """Initialize the platform entity. Quirk entities are constructed with `from_quirk=True` and the generic config keywords (`fallback_name`, `translation_key`, `entity_type`, etc.); - the platform subclasses add their own keywords. Default-discovery - entities pass none of these. + the platform subclasses add their own keywords. """ if from_quirk: self._apply_quirk_entity_config( @@ -559,24 +550,12 @@ def __init__( initially_disabled=initially_disabled, ) - if legacy_discovery_unique_id is None: - if from_quirk: - legacy_discovery_unique_id = f"{device.ieee}-{endpoint.id}" - else: - legacy_discovery_unique_id = ( - f"{device.ieee}-{endpoint.id}-{cluster.cluster_id}" - ) - if self._unique_id_suffix is not None: - unique_id = f"{legacy_discovery_unique_id}-{self._unique_id_suffix}" - else: - unique_id = legacy_discovery_unique_id + unique_id = f"{unique_id}-{self._unique_id_suffix}" super().__init__(unique_id=unique_id, **kwargs) self._device: Device = device - self._endpoint = endpoint - self._cluster: zigpy.zcl.Cluster = cluster def _apply_quirk_entity_config( self, @@ -625,7 +604,6 @@ def identifiers(self) -> PlatformEntityIdentifiers: unique_id=self.unique_id, platform=self.PLATFORM, device_ieee=self.device.ieee, - endpoint_id=self.endpoint.id, ) @property @@ -633,6 +611,76 @@ def device(self) -> Device: """Return the device.""" return self._device + @property + def should_poll(self) -> bool: + """Return True if we need to poll for state changes.""" + return False + + @property + def available(self) -> bool: + """Return true if the device this entity belongs to is available.""" + return self.device.available + + async def async_update(self) -> None: + """Retrieve latest state. + + Default no-op: subclasses that need polling override this to read their + own attributes directly from the relevant cluster(s). + """ + + @property + def state(self) -> BaseEntityState: + """Return the state of this entity.""" + return dataclasses.replace( + super().state, + device_ieee=self._device.ieee, + available=self.available, + ) + + +class ZclPlatformEntity(PlatformEntity): + """Platform entity backed by ZCL cluster(s) on an endpoint.""" + + # Direct cluster matching for discovery + _cluster_match: ClusterMatch | None = None + + # Per-cluster configuration (keyed by cluster ID) + _server_cluster_config: Mapping[int, ClusterConfig] = MappingProxyType({}) + + _client_cluster_config: Mapping[int, ClusterConfig] = MappingProxyType({}) + + def __init__( + self, + endpoint: Endpoint, + device: Device, + *, + cluster: zigpy.zcl.Cluster, + from_quirk: bool = False, + legacy_discovery_unique_id: str | None = None, + **kwargs: Any, + ): + """Initialize the ZCL platform entity. + + Default-discovery entities pass no config keywords. + """ + if legacy_discovery_unique_id is None: + if from_quirk: + legacy_discovery_unique_id = f"{device.ieee}-{endpoint.id}" + else: + legacy_discovery_unique_id = ( + f"{device.ieee}-{endpoint.id}-{cluster.cluster_id}" + ) + + self._endpoint = endpoint + self._cluster: zigpy.zcl.Cluster = cluster + + super().__init__( + device, + unique_id=legacy_discovery_unique_id, + from_quirk=from_quirk, + **kwargs, + ) + @property def endpoint(self) -> Endpoint: """Return the endpoint.""" @@ -678,31 +726,22 @@ def targets_cluster( return in_client return in_server or in_client - @property - def should_poll(self) -> bool: - """Return True if we need to poll for state changes.""" - return False - - @property - def available(self) -> bool: - """Return true if the device this entity belongs to is available.""" - return self.device.available - - async def async_update(self) -> None: - """Retrieve latest state. - - Default no-op: subclasses that need polling override this to read their - own attributes directly from the relevant cluster(s). - """ + @cached_property + def identifiers(self) -> PlatformEntityIdentifiers: + """Return a dict with the information necessary to identify this entity.""" + return PlatformEntityIdentifiers( + unique_id=self.unique_id, + platform=self.PLATFORM, + device_ieee=self.device.ieee, + endpoint_id=self.endpoint.id, + ) @property def state(self) -> BaseEntityState: """Return the state of this entity.""" return dataclasses.replace( super().state, - device_ieee=self._device.ieee, endpoint_id=self._endpoint.id, - available=self.available, ) diff --git a/zha/application/platforms/alarm_control_panel/__init__.py b/zha/application/platforms/alarm_control_panel/__init__.py index b1e5e5e4f..442ea66e8 100644 --- a/zha/application/platforms/alarm_control_panel/__init__.py +++ b/zha/application/platforms/alarm_control_panel/__init__.py @@ -21,7 +21,7 @@ BaseEntityState, ClusterConfig, ClusterMatch, - PlatformEntity, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.alarm_control_panel.const import ( @@ -55,7 +55,7 @@ class AlarmControlPanelState(BaseEntityState): supported_features: int -class BaseAlarmControlPanel(PlatformEntity, ABC): +class BaseAlarmControlPanel(ZclPlatformEntity, ABC): """Abstract base class for ZHA alarm control panel entities.""" PLATFORM = Platform.ALARM_CONTROL_PANEL diff --git a/zha/application/platforms/binary_sensor/__init__.py b/zha/application/platforms/binary_sensor/__init__.py index 07e1e8de4..7ab03d6b3 100644 --- a/zha/application/platforms/binary_sensor/__init__.py +++ b/zha/application/platforms/binary_sensor/__init__.py @@ -29,8 +29,8 @@ ClusterConfig, ClusterMatch, EntityCategory, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.binary_sensor.const import ( @@ -61,7 +61,7 @@ class BinarySensorState(BaseEntityState): attribute_name: str -class BaseBinarySensor(PlatformEntity, ABC): +class BaseBinarySensor(ZclPlatformEntity, ABC): """Abstract base class for ZHA binary sensors.""" PLATFORM: Platform = Platform.BINARY_SENSOR diff --git a/zha/application/platforms/button/__init__.py b/zha/application/platforms/button/__init__.py index 03eb39aee..3f0bc8b73 100644 --- a/zha/application/platforms/button/__init__.py +++ b/zha/application/platforms/button/__init__.py @@ -17,7 +17,7 @@ BaseEntityState, ClusterMatch, EntityCategory, - PlatformEntity, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.button.const import DEFAULT_DURATION, ButtonDeviceClass @@ -55,7 +55,7 @@ class WriteAttributeButtonState(ButtonState): attribute_value: Any -class BaseButton(PlatformEntity, ABC): +class BaseButton(ZclPlatformEntity, ABC): """Base representation of a ZHA button.""" PLATFORM = Platform.BUTTON diff --git a/zha/application/platforms/climate/__init__.py b/zha/application/platforms/climate/__init__.py index 31acebd1b..03405adcb 100644 --- a/zha/application/platforms/climate/__init__.py +++ b/zha/application/platforms/climate/__init__.py @@ -32,8 +32,8 @@ BaseEntityState, ClusterConfig, ClusterMatch, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.climate.const import ( @@ -94,7 +94,7 @@ class ThermostatState(ClimateState): unoccupied_heating_setpoint: int | None = None -class BaseThermostat(PlatformEntity, ABC): +class BaseThermostat(ZclPlatformEntity, ABC): """Abstract base class for climate entities.""" PLATFORM = Platform.CLIMATE diff --git a/zha/application/platforms/cover/__init__.py b/zha/application/platforms/cover/__init__.py index 7256db53c..d71630a47 100644 --- a/zha/application/platforms/cover/__init__.py +++ b/zha/application/platforms/cover/__init__.py @@ -35,8 +35,8 @@ BaseEntityState, ClusterConfig, ClusterMatch, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.cover.const import ( @@ -77,7 +77,7 @@ class CoverEntityState(BaseEntityState): supported_features: CoverEntityFeature -class BaseCover(PlatformEntity, ABC): +class BaseCover(ZclPlatformEntity, ABC): """Abstract base class for ZHA covers.""" PLATFORM = Platform.COVER diff --git a/zha/application/platforms/device_tracker.py b/zha/application/platforms/device_tracker.py index f283f621e..9972c8873 100644 --- a/zha/application/platforms/device_tracker.py +++ b/zha/application/platforms/device_tracker.py @@ -25,7 +25,7 @@ BaseEntityState, ClusterConfig, ClusterMatch, - PlatformEntity, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.sensor import Battery @@ -58,7 +58,7 @@ class DeviceTrackerState(BaseEntityState): battery_level: float | None -class BaseDeviceTracker(PlatformEntity, ABC): +class BaseDeviceTracker(ZclPlatformEntity, ABC): """Abstract base class for ZHA device tracker entities.""" PLATFORM = Platform.DEVICE_TRACKER diff --git a/zha/application/platforms/fan/__init__.py b/zha/application/platforms/fan/__init__.py index ed37f19dd..93feb91bd 100644 --- a/zha/application/platforms/fan/__init__.py +++ b/zha/application/platforms/fan/__init__.py @@ -26,8 +26,8 @@ ClusterConfig, ClusterMatch, GroupEntity, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, register_group_entity, ) @@ -242,7 +242,7 @@ def percentage_to_speed(self, percentage: int) -> str: @register_entity(hvac.Fan.cluster_id) -class Fan(BaseFan, PlatformEntity): +class Fan(BaseFan, ZclPlatformEntity): """Representation of a ZHA fan.""" _cluster_match = ClusterMatch( @@ -385,7 +385,7 @@ def update(self, _: Any = None) -> None: @register_entity(IKEA_AIR_PURIFIER_CLUSTER) -class IkeaFan(BaseFan, PlatformEntity): +class IkeaFan(BaseFan, ZclPlatformEntity): """Representation of an Ikea fan.""" _attr_supported_features: FanEntityFeature = ( diff --git a/zha/application/platforms/light/__init__.py b/zha/application/platforms/light/__init__.py index ea1282609..cb702b486 100644 --- a/zha/application/platforms/light/__init__.py +++ b/zha/application/platforms/light/__init__.py @@ -34,8 +34,8 @@ ClusterConfig, ClusterMatch, GroupEntity, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, register_group_entity, ) @@ -873,7 +873,7 @@ async def on_remove(self) -> None: @register_entity(OnOff.cluster_id) -class Light(BaseSharedLight, PlatformEntity): +class Light(BaseSharedLight, ZclPlatformEntity): """Representation of a ZHA or ZLL light.""" _attr_translation_key: str = "light" diff --git a/zha/application/platforms/lock/__init__.py b/zha/application/platforms/lock/__init__.py index be4e8b3fc..4e0810ba0 100644 --- a/zha/application/platforms/lock/__init__.py +++ b/zha/application/platforms/lock/__init__.py @@ -26,7 +26,7 @@ BaseEntityState, ClusterConfig, ClusterMatch, - PlatformEntity, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.lock.const import ( @@ -47,7 +47,7 @@ class LockState(BaseEntityState): is_locked: bool -class BaseLock(PlatformEntity, ABC): +class BaseLock(ZclPlatformEntity, ABC): """Abstract base class for ZHA lock entities.""" PLATFORM = Platform.LOCK diff --git a/zha/application/platforms/number/__init__.py b/zha/application/platforms/number/__init__.py index 3d0a5ab51..2e80082c4 100644 --- a/zha/application/platforms/number/__init__.py +++ b/zha/application/platforms/number/__init__.py @@ -28,8 +28,8 @@ ClusterConfig, ClusterMatch, EntityCategory, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.const import ( @@ -63,7 +63,7 @@ class NumberState(BaseEntityState): native_unit_of_measurement: str | None -class BaseNumber(PlatformEntity, ABC): +class BaseNumber(ZclPlatformEntity, ABC): """Representation of a ZHA Number entity.""" PLATFORM = Platform.NUMBER diff --git a/zha/application/platforms/select.py b/zha/application/platforms/select.py index dcdc0c288..1612c5c79 100644 --- a/zha/application/platforms/select.py +++ b/zha/application/platforms/select.py @@ -36,7 +36,7 @@ ClusterConfig, ClusterMatch, EntityCategory, - PlatformEntity, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.const import ( @@ -112,7 +112,7 @@ async def async_select_option(self, option: str) -> None: """Change the selected option.""" -class SirenDefaultSelectEntity(BaseSelectEntity, PlatformEntity): +class SirenDefaultSelectEntity(BaseSelectEntity, ZclPlatformEntity): """Select entity whose state lives on the AdvancedSiren on the same cluster.""" _attr_entity_category = EntityCategory.CONFIG @@ -246,7 +246,7 @@ class DefaultStrobeSelectEntity(SirenDefaultSelectEntity): ) -class ZCLEnumSelectEntity(BaseSelectEntity, PlatformEntity): +class ZCLEnumSelectEntity(BaseSelectEntity, ZclPlatformEntity): """Representation of a ZHA ZCL enum select entity.""" _attribute_name: str diff --git a/zha/application/platforms/sensor/__init__.py b/zha/application/platforms/sensor/__init__.py index af816e216..a018786f8 100644 --- a/zha/application/platforms/sensor/__init__.py +++ b/zha/application/platforms/sensor/__init__.py @@ -65,8 +65,8 @@ ClusterConfig, ClusterMatch, EntityCategory, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.climate.const import HVACAction @@ -190,7 +190,7 @@ class DeviceCounterSensorIdentifiers(BaseIdentifiers): device_ieee: str -class BaseSensor(PlatformEntity, ABC): +class BaseSensor(ZclPlatformEntity, ABC): """Abstract base class for ZHA sensor entities.""" PLATFORM = Platform.SENSOR @@ -795,7 +795,9 @@ class Battery(Sensor): def _is_supported(self) -> bool: # XXX: We intentionally ignore the presence of this attribute - return PlatformEntity._is_supported(self) and not self.device.is_mains_powered + return ( + ZclPlatformEntity._is_supported(self) and not self.device.is_mains_powered + ) @staticmethod def formatter(value: int) -> float | None: # pylint: disable=arguments-differ @@ -3088,7 +3090,7 @@ class ThermostatHVACAction(Sensor): } def _is_supported(self) -> bool: - return PlatformEntity._is_supported(self) + return ZclPlatformEntity._is_supported(self) @property def _pi_heating_demand(self) -> int | None: @@ -3207,7 +3209,7 @@ def _rm_rs_action(self) -> HVACAction: class RSSISensor(Sensor): """RSSI sensor for a device.""" - # TODO: migrate this away from `PlatformEntity` + # TODO: migrate this away from `ZclPlatformEntity` _unique_id_suffix: str = "rssi" _attr_state_class: SensorStateClass = SensorStateClass.MEASUREMENT _attr_device_class: SensorDeviceClass | None = SensorDeviceClass.SIGNAL_STRENGTH @@ -3282,7 +3284,7 @@ def update(self): class LQISensor(RSSISensor): """LQI sensor for a device.""" - # TODO: migrate this away from `PlatformEntity` + # TODO: migrate this away from `ZclPlatformEntity` _unique_id_suffix: str = "lqi" _attr_device_class = None _attr_native_unit_of_measurement = None diff --git a/zha/application/platforms/siren.py b/zha/application/platforms/siren.py index 2e4579db6..9229d4424 100644 --- a/zha/application/platforms/siren.py +++ b/zha/application/platforms/siren.py @@ -27,8 +27,8 @@ BaseEntityState, ClusterConfig, ClusterMatch, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, ) from zha.quirks import SIREN_BASIC @@ -65,7 +65,7 @@ class SirenState(BaseEntityState): supported_features: SirenEntityFeature -class BaseSiren(PlatformEntity, ABC): +class BaseSiren(ZclPlatformEntity, ABC): """Abstract base class for ZHA siren entities.""" PLATFORM = Platform.SIREN diff --git a/zha/application/platforms/switch.py b/zha/application/platforms/switch.py index aa9eb4fb1..f19cc947f 100644 --- a/zha/application/platforms/switch.py +++ b/zha/application/platforms/switch.py @@ -32,8 +32,8 @@ ClusterMatch, EntityCategory, GroupEntity, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, register_group_entity, ) @@ -102,7 +102,7 @@ async def async_turn_off(self) -> None: @register_entity(OnOff.cluster_id) -class Switch(PlatformEntity, BaseSwitch): # type: ignore[misc] +class Switch(ZclPlatformEntity, BaseSwitch): # type: ignore[misc] """ZHA switch.""" _attr_translation_key = "switch" @@ -243,7 +243,7 @@ async def async_update(self) -> None: @register_entity(BinaryOutput.cluster_id) -class BinaryOutputSwitch(PlatformEntity, BaseSwitch): # type: ignore[misc] +class BinaryOutputSwitch(ZclPlatformEntity, BaseSwitch): # type: ignore[misc] """BinaryOutputCluster switch.""" _attr_primary_weight = 10 @@ -390,7 +390,7 @@ def update(self, _: Any | None = None) -> None: self.maybe_emit_state_changed_event() -class ConfigurableAttributeSwitch(PlatformEntity): +class ConfigurableAttributeSwitch(ZclPlatformEntity): """Representation of a ZHA switch configuration entity.""" PLATFORM = Platform.SWITCH diff --git a/zha/application/platforms/update.py b/zha/application/platforms/update.py index 6b8ef0ebe..86f28904f 100644 --- a/zha/application/platforms/update.py +++ b/zha/application/platforms/update.py @@ -27,8 +27,8 @@ ClusterConfig, ClusterMatch, EntityCategory, - PlatformEntity, PlatformFeatureGroup, + ZclPlatformEntity, register_entity, ) from zha.exceptions import ZHAException @@ -80,7 +80,7 @@ class UpdateState(BaseEntityState): supported_features: UpdateEntityFeature -class BaseFirmwareUpdateEntity(PlatformEntity, ABC): +class BaseFirmwareUpdateEntity(ZclPlatformEntity, ABC): """Abstract base class for ZHA firmware update entities.""" PLATFORM = Platform.UPDATE diff --git a/zha/application/platforms/virtual.py b/zha/application/platforms/virtual.py index 4585ca960..763332af3 100644 --- a/zha/application/platforms/virtual.py +++ b/zha/application/platforms/virtual.py @@ -35,7 +35,7 @@ AttrConfig, ClusterConfig, ClusterMatch, - PlatformEntity, + ZclPlatformEntity, register_entity, ) from zha.application.platforms.const import ( @@ -67,7 +67,7 @@ from zha.zigbee.endpoint import Endpoint -class VirtualEntity(PlatformEntity): +class VirtualEntity(ZclPlatformEntity): """Cluster-level background driver that isn't registered as a HA entity. Virtual entities participate in discovery and cluster-config aggregation diff --git a/zha/zigbee/cluster_config.py b/zha/zigbee/cluster_config.py index f0d487d01..fc63dc896 100644 --- a/zha/zigbee/cluster_config.py +++ b/zha/zigbee/cluster_config.py @@ -16,7 +16,7 @@ ZHA_CLUSTER_BIND_EVENT, ZHA_CLUSTER_CONFIGURE_REPORTING_EVENT, ) -from zha.application.platforms import AttrConfig, PlatformEntity +from zha.application.platforms import AttrConfig, ZclPlatformEntity if TYPE_CHECKING: from collections.abc import Iterable @@ -76,7 +76,7 @@ def aggregate_cluster_configs( result: dict[tuple[int, int, bool], AggregatedClusterConfig] = {} for entity in entities: - if not isinstance(entity, PlatformEntity): + if not isinstance(entity, ZclPlatformEntity): continue if not entity._server_cluster_config and not entity._client_cluster_config: From 9156bee99573ed18fbf4e1ae3f5f011a48f4dca9 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Mon, 17 Aug 2026 19:39:52 +0000 Subject: [PATCH 3/5] Base ZGP device --- tests/test_green_power_device.py | 116 +++++++++++++++++++++ zha/application/gateway.py | 30 +++++- zha/application/helpers.py | 2 +- zha/quirks.py | 6 +- zha/zigbee/device.py | 170 +++++++++++++++++++++++++++---- zha/zigbee/group.py | 26 ++--- 6 files changed, 309 insertions(+), 41 deletions(-) create mode 100644 tests/test_green_power_device.py diff --git a/tests/test_green_power_device.py b/tests/test_green_power_device.py new file mode 100644 index 000000000..0a20d4e08 --- /dev/null +++ b/tests/test_green_power_device.py @@ -0,0 +1,116 @@ +"""Tests for ZHA Green Power devices.""" + +from __future__ import annotations + +from zigpy.device import GreenPowerDevice as ZigpyGreenPowerDevice +from zigpy.zgp.types import ApplicationID, DeviceID, GPDCommandID, SrcID + +from zha.application.const import UNKNOWN_MANUFACTURER, UNKNOWN_MODEL +from zha.application.gateway import Gateway +from zha.quirks import DEVICE_REGISTRY +from zha.zigbee.device import GreenPowerDevice + + +def make_zigpy_gpd(zha_gateway: Gateway) -> ZigpyGreenPowerDevice: + """Create a commissioned zigpy GPD on the test application.""" + zigpy_gpd = ZigpyGreenPowerDevice( + zha_gateway.application_controller, + application_id=ApplicationID.SrcID, + src_id=SrcID(0x12345678), + ) + zigpy_gpd.device_id = DeviceID.OnOffSwitch + zigpy_gpd.commands = [ + GPDCommandID.Toggle, + GPDCommandID.Press1of1, + GPDCommandID.Release1of1, + ] + + zha_gateway.application_controller.devices[zigpy_gpd.ieee] = zigpy_gpd + return zigpy_gpd + + +async def test_green_power_device_creation(zha_gateway: Gateway) -> None: + """Test that a zigpy GPD is wrapped in the ZHA Green Power device class.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + zha_device = zha_gateway.get_or_create_device(zigpy_gpd) + + assert isinstance(zha_device, GreenPowerDevice) + assert zha_device.device is zigpy_gpd + assert zha_gateway.devices[zigpy_gpd.ieee] is zha_device + + +async def test_green_power_device_properties(zha_gateway: Gateway) -> None: + """Test the Green Power device property surface.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + zha_device = zha_gateway.get_or_create_device(zigpy_gpd) + + assert zha_device.available is True + assert zha_device.is_mains_powered is False + assert zha_device.is_active_coordinator is False + assert zha_device.device_type == "GreenPower" + assert zha_device.manufacturer == UNKNOWN_MANUFACTURER + assert zha_device.model == UNKNOWN_MODEL + assert zha_device.manufacturer_code is None + assert zha_device.signature["device_id"] == DeviceID.OnOffSwitch + assert zha_device.signature["src_id"] == 0x12345678 + assert zha_device.signature["commands"] == [ + GPDCommandID.Toggle, + GPDCommandID.Press1of1, + GPDCommandID.Release1of1, + ] + + # A GPD has no heartbeat: the availability check never flips it + await zha_device._check_available() + assert zha_device.available is True + + +async def test_green_power_device_identifiers(zha_gateway: Gateway) -> None: + """Test manufacturer and model resolution from commissioning identifiers.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + zigpy_gpd.gpd_manufacturer_id = 0x1234 + zigpy_gpd.gpd_model_id = 0x0007 + + zha_device = zha_gateway.get_or_create_device(zigpy_gpd) + + assert zha_device.manufacturer == "0x1234" + assert zha_device.model == "0x0007" + assert zha_device.manufacturer_code == 0x1234 + assert zha_device.name == "0x1234 0x0007" + + +async def test_green_power_device_info(zha_gateway: Gateway) -> None: + """Test device info, extended device info, and diagnostics.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + zha_device = zha_gateway.get_or_create_device(zigpy_gpd) + + device_info = zha_device.device_info + assert device_info.ieee == zigpy_gpd.ieee + assert device_info.device_type == "GreenPower" + assert device_info.signature["device_id"] == 0x02 + + extended_info = zha_device.extended_device_info + assert extended_info.active_coordinator is False + assert extended_info.entities == {} + assert extended_info.neighbors == [] + assert extended_info.routes == [] + assert extended_info.endpoint_names == [] + + diagnostics = zha_device.get_diagnostics_json() + assert diagnostics["ieee"] == str(zigpy_gpd.ieee) + assert diagnostics["device_type"] == "GreenPower" + assert diagnostics["signature"]["device_id"] == 0x02 + + +async def test_green_power_device_quirk_passthrough(zha_gateway: Gateway) -> None: + """Test that the quirk registry passes GP devices through untouched.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + assert DEVICE_REGISTRY.resolve(zigpy_gpd) is zigpy_gpd + + +async def test_green_power_device_initialize(zha_gateway: Gateway) -> None: + """Test that a Green Power device initializes without entities.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + zha_device = zha_gateway.get_or_create_device(zigpy_gpd) + + await zha_device.async_initialize(from_cache=True) + assert zha_device.platform_entities == {} diff --git a/zha/application/gateway.py b/zha/application/gateway.py index d53a1f520..78b3d6429 100644 --- a/zha/application/gateway.py +++ b/zha/application/gateway.py @@ -54,7 +54,13 @@ ) from zha.event import EventBase from zha.quirks import DEVICE_REGISTRY, QUIRK_REGISTRY_ENTRY_ATTR -from zha.zigbee.device import Device, DeviceInfo, DeviceStatus, ExtendedDeviceInfo +from zha.zigbee.device import ( + BaseDevice, + Device, + DeviceInfo, + DeviceStatus, + ExtendedDeviceInfo, +) from zha.zigbee.group import Group, GroupInfo, GroupMemberReference BLOCK_LOG_TIMEOUT: Final[int] = 60 @@ -176,7 +182,7 @@ def __init__(self, config: ZHAData) -> None: """Initialize the gateway.""" super().__init__() self.config: ZHAData = config - self._devices: dict[EUI64, Device] = {} + self._devices: dict[EUI64, BaseDevice] = {} self._groups: dict[int, Group] = {} self.application_controller: ControllerApplication = None self.coordinator_zha_device: Device | None = None @@ -488,6 +494,9 @@ async def _async_device_reinterviewed( ) return + # Only Zigbee devices can be re-interviewed + assert isinstance(zha_device, Device) + old_entry = getattr(zha_device.device, QUIRK_REGISTRY_ENTRY_ATTR, None) old_factory = ( old_entry.zha_device_factory @@ -521,6 +530,7 @@ async def _async_device_reinterviewed( await zha_device.async_teardown(emit_entity_events=True) zha_device = Device.new(new_zigpy_device, self) + assert isinstance(zha_device, Device) self._devices[new_zigpy_device.ieee] = zha_device zha_device.available = True @@ -701,7 +711,7 @@ def state(self) -> State: return self.application_controller.state @property - def devices(self) -> dict[EUI64, Device]: + def devices(self) -> dict[EUI64, BaseDevice]: """Return devices.""" return self._devices @@ -710,7 +720,13 @@ def groups(self) -> dict[int, Group]: """Return groups.""" return self._groups - def get_or_create_device(self, zigpy_device: zigpy.device.Device) -> Device: + def get_zigbee_device(self, ieee: EUI64) -> Device: + """Look up a Zigbee device by IEEE address.""" + device = self._devices[ieee] + assert isinstance(device, Device) + return device + + def get_or_create_device(self, zigpy_device: zigpy.device.BaseDevice) -> BaseDevice: """Get or create a ZHA device.""" if (zha_device := self._devices.get(zigpy_device.ieee)) is None: zha_device = Device.new(zigpy_device, self) @@ -847,8 +863,10 @@ async def async_create_zigpy_group( name, group_id, ) + member_device = self.devices[member.ieee] + assert isinstance(member_device, Device) tasks.append( - self.devices[member.ieee].async_add_endpoint_to_group( + member_device.async_add_endpoint_to_group( member.endpoint_id, group_id ) ) @@ -866,6 +884,8 @@ async def async_remove_device(self, ieee: EUI64) -> None: for group_id, group in self.groups.items(): for member_ieee_endpoint_id in list(group.zigpy_group.members.keys()): if member_ieee_endpoint_id[0] == ieee: + # Only Zigbee devices can be group members + assert isinstance(device, Device) await device.async_remove_from_group(group_id) await self.application_controller.remove(ieee) diff --git a/zha/application/helpers.py b/zha/application/helpers.py index df65e5b55..18df88be0 100644 --- a/zha/application/helpers.py +++ b/zha/application/helpers.py @@ -548,7 +548,7 @@ async def check_device_availability(self): *( dev._check_available() for dev in self._gateway.devices.values() - if not dev.is_coordinator + if not dev.is_active_coordinator ), ) _LOGGER.debug("Device availability checker interval finished") diff --git a/zha/quirks.py b/zha/quirks.py index f297c89b8..5a42d5dfe 100644 --- a/zha/quirks.py +++ b/zha/quirks.py @@ -227,13 +227,17 @@ def match_entry( return None - def resolve(self, zigpy_device: zigpy.device.Device) -> zigpy.device.Device: + def resolve(self, zigpy_device: zigpy.device.BaseDevice) -> zigpy.device.BaseDevice: """Apply the quirk transforms registered for `zigpy_device` and return the result.""" # Resolution is idempotent: an already-quirked device is returned as-is if hasattr(zigpy_device, QUIRK_REGISTRY_ENTRY_ATTR): return zigpy_device + # Green Power devices will resolve through their own registry + if not isinstance(zigpy_device, zigpy.device.ZigbeeDevice): + return zigpy_device + entry = self.match_entry(zigpy_device) if entry is None: return zigpy_device diff --git a/zha/zigbee/device.py b/zha/zigbee/device.py index b8f280131..9ca587a71 100644 --- a/zha/zigbee/device.py +++ b/zha/zigbee/device.py @@ -17,7 +17,11 @@ import time from typing import TYPE_CHECKING, Any, Final -from zigpy.device import BaseDevice as ZigpyBaseDevice, Device as ZigpyDevice +from zigpy.device import ( + BaseDevice as ZigpyBaseDevice, + Device as ZigpyDevice, + GreenPowerDevice as ZigpyGreenPowerDevice, +) import zigpy.exceptions from zigpy.profiles import PROFILES from zigpy.types import uint1_t, uint8_t, uint16_t @@ -80,6 +84,7 @@ BaseEntityState, EntityStateChangedEvent, PlatformEntity, + ZclPlatformEntity, sensor, ) from zha.application.platforms.update import BaseFirmwareUpdateEntity @@ -438,6 +443,25 @@ def __repr__(self) -> str: f"exposes_features: {self.exposes_features}" ) + @classmethod + def new( + cls, + zigpy_dev: zigpy.device.BaseDevice, + gateway: Gateway, + ) -> BaseDevice: + """Create new device, dispatching to the factory matched during resolution.""" + if zigpy_dev.ieee == gateway.state.node_info.ieee: + return CoordinatorDevice(zigpy_dev, gateway) + + entry = getattr(zigpy_dev, QUIRK_REGISTRY_ENTRY_ATTR, None) + if entry is not None and entry.zha_device_factory is not None: + return entry.zha_device_factory(zigpy_dev, gateway) + + if isinstance(zigpy_dev, ZigpyGreenPowerDevice): + return GreenPowerDevice(zigpy_dev, gateway) + + return ZigbeeDevice(zigpy_dev, gateway) + @property def device(self) -> zigpy.device.BaseDevice: """Return underlying Zigpy device.""" @@ -645,10 +669,13 @@ def get_entity( for entity in self._platform_entities.values(): if platform != entity.PLATFORM: continue - if endpoint_id is not None and entity.endpoint.id != endpoint_id: - continue - if cluster_id is not None and entity.cluster.cluster_id != cluster_id: - continue + if endpoint_id is not None or cluster_id is not None: + if not isinstance(entity, ZclPlatformEntity): + continue + if endpoint_id is not None and entity.endpoint.id != endpoint_id: + continue + if cluster_id is not None and entity.cluster.cluster_id != cluster_id: + continue matches.append(entity) if not matches or (not pick_first and len(matches) != 1): raise LookupError( @@ -1257,22 +1284,6 @@ def signature(self) -> dict[str, Any]: """Return the device signature reported in the device info.""" return self.zigbee_signature - @classmethod - def new( - cls, - zigpy_dev: zigpy.device.Device, - gateway: Gateway, - ) -> Device: - """Create new device, dispatching to the factory matched during resolution.""" - if zigpy_dev.ieee == gateway.state.node_info.ieee: - return CoordinatorDevice(zigpy_dev, gateway) - - entry = getattr(zigpy_dev, QUIRK_REGISTRY_ENTRY_ATTR, None) - if entry is not None and entry.zha_device_factory is not None: - return entry.zha_device_factory(zigpy_dev, gateway) - - return cls(zigpy_dev, gateway) - async def _check_available(self, *_: Any) -> None: # don't flip the availability state of the coordinator if self.is_active_coordinator: @@ -1932,5 +1943,122 @@ def discover_entities(self) -> Iterator[BaseEntity]: ) +class GreenPowerDevice(BaseDevice): + """ZHA Green Power device object.""" + + _zigpy_device: ZigpyGreenPowerDevice + + def _init_from_zigpy_device(self, zigpy_device: ZigpyGreenPowerDevice) -> None: + super()._init_from_zigpy_device(zigpy_device) + + # A GPD has no heartbeat to age against: it is available until removed + self._available = True + + @property + def device(self) -> ZigpyGreenPowerDevice: + """Return underlying Zigpy device.""" + return self._zigpy_device + + def _resolve_manufacturer(self) -> str: + """Resolve the manufacturer name (declarative quirks override this).""" + if self._zigpy_device.gpd_manufacturer_id is not None: + return f"0x{self._zigpy_device.gpd_manufacturer_id:04X}" + + return UNKNOWN_MANUFACTURER + + def _resolve_model(self) -> str: + """Resolve the model name (declarative quirks override this).""" + if self._zigpy_device.gpd_model_id is not None: + return f"0x{self._zigpy_device.gpd_model_id:04X}" + + return UNKNOWN_MODEL + + @property + def manufacturer_code(self) -> int | None: + """Return the manufacturer code for the device.""" + return self._zigpy_device.gpd_manufacturer_id + + @property + def is_mains_powered(self) -> bool | None: + """Return true if device is mains powered.""" + return False + + @property + def device_type(self) -> str: + """Return the logical device type for the device.""" + return "GreenPower" + + @property + def signature(self) -> dict[str, Any]: + """Return the device signature reported in the device info.""" + return self._zigpy_device.get_signature() + + async def _check_available(self, *_: Any) -> None: + """Do nothing: a GPD has no heartbeat and nothing to ping.""" + + def discover_entities(self) -> Iterator[BaseEntity]: + """Yield the entities for this device. + + GP quirks contribute event entities by overriding this; an unquirked GPD + exposes nothing yet. + """ + yield from () + + @property + def extended_device_info(self) -> ExtendedDeviceInfo: + """Get extended device information.""" + return ExtendedDeviceInfo( + **self.device_info.__dict__, + active_coordinator=False, + entities={ + platform_entity.unique_id: platform_entity.state + for platform_entity in self.platform_entities.values() + }, + neighbors=[], + routes=[], + endpoint_names=[], + ) + + def get_diagnostics_json(self) -> dict[str, Any]: + """Get ZHA device information.""" + info: dict[str, Any] = {} + info["version"] = DIAGNOSTICS_JSON_VERSION + info["ieee"] = str(self.ieee) + info["nwk"] = str(self.nwk) + info["friendly_manufacturer"] = self.manufacturer + info["friendly_model"] = self.model + info["name"] = self.name + info["quirk_applied"] = self.quirk_applied + info["quirk_class"] = self.quirk_class + info["exposes_features"] = self.exposes_features + info["manufacturer_code"] = self.manufacturer_code + info["power_source"] = self.power_source + info["lqi"] = self.lqi + info["rssi"] = self.rssi + info["last_seen"] = self.last_seen + info["available"] = self.available + info["device_type"] = self.device_type + info["signature"] = self.signature + + info["zha_lib_entities"] = defaultdict(list) + + for (platform, _unique_id), platform_entity in sorted( + self.platform_entities.items() + ): + if platform is Platform.VIRTUAL: + continue + + state_dict = dataclasses.asdict(platform_entity.state) + state_dict["migrate_unique_ids"] = list(state_dict["migrate_unique_ids"]) + state_dict["device_ieee"] = str(state_dict["device_ieee"]) + state_dict["extra_state_attribute_names"] = sorted( + state_dict["extra_state_attribute_names"] + ) + + info["zha_lib_entities"][platform].append(state_dict) + + return info + + # Backwards-compatible alias Device = ZigbeeDevice diff --git a/zha/zigbee/group.py b/zha/zigbee/group.py index ab09301f5..d7017e3ce 100644 --- a/zha/zigbee/group.py +++ b/zha/zigbee/group.py @@ -191,7 +191,7 @@ def gateway(self) -> Gateway: def members(self) -> list[GroupMember]: """Return the ZHA devices that are members of this group.""" return [ - GroupMember(self, self._gateway.devices[member_ieee], endpoint_id) + GroupMember(self, self._gateway.get_zigbee_device(member_ieee), endpoint_id) for (member_ieee, endpoint_id) in self._zigpy_group.members if member_ieee in self._gateway.devices ] @@ -288,40 +288,40 @@ def update_entity_subscriptions(self) -> None: async def async_add_members(self, members: list[GroupMemberReference]) -> None: """Add members to this group.""" - devices: dict[EUI64, Device] = self._gateway.devices if len(members) > 1: tasks = [] for member in members: tasks.append( - devices[member.ieee].async_add_endpoint_to_group( - member.endpoint_id, self.group_id - ) + self._gateway.get_zigbee_device( + member.ieee + ).async_add_endpoint_to_group(member.endpoint_id, self.group_id) ) await asyncio.gather(*tasks) else: member = members[0] - await devices[member.ieee].async_add_endpoint_to_group( - member.endpoint_id, self.group_id - ) + await self._gateway.get_zigbee_device( + member.ieee + ).async_add_endpoint_to_group(member.endpoint_id, self.group_id) self.update_entity_subscriptions() async def async_remove_members(self, members: list[GroupMemberReference]) -> None: """Remove members from this group.""" - devices: dict[EUI64, Device] = self._gateway.devices if len(members) > 1: tasks = [] for member in members: tasks.append( - devices[member.ieee].async_remove_endpoint_from_group( + self._gateway.get_zigbee_device( + member.ieee + ).async_remove_endpoint_from_group( member.endpoint_id, self.group_id ) ) await asyncio.gather(*tasks) else: member = members[0] - await devices[member.ieee].async_remove_endpoint_from_group( - member.endpoint_id, self.group_id - ) + await self._gateway.get_zigbee_device( + member.ieee + ).async_remove_endpoint_from_group(member.endpoint_id, self.group_id) self.update_entity_subscriptions() def get_platform_entities(self, platform: str) -> list[PlatformEntity]: From 613d19b8b5d351bee00538a78b9cd144138ac393 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Mon, 17 Aug 2026 20:38:41 +0000 Subject: [PATCH 4/5] GP quirk registry --- tests/test_green_power_device.py | 97 +++++++++++++++++++- zha/quirks.py | 151 ++++++++++++++++++++++++++----- 2 files changed, 225 insertions(+), 23 deletions(-) diff --git a/tests/test_green_power_device.py b/tests/test_green_power_device.py index 0a20d4e08..9a1cae02b 100644 --- a/tests/test_green_power_device.py +++ b/tests/test_green_power_device.py @@ -3,11 +3,17 @@ from __future__ import annotations from zigpy.device import GreenPowerDevice as ZigpyGreenPowerDevice +from zigpy.types import EUI64 from zigpy.zgp.types import ApplicationID, DeviceID, GPDCommandID, SrcID from zha.application.const import UNKNOWN_MANUFACTURER, UNKNOWN_MODEL from zha.application.gateway import Gateway -from zha.quirks import DEVICE_REGISTRY +from zha.quirks import ( + DEVICE_REGISTRY, + QUIRK_REGISTRY_ENTRY_ATTR, + GreenPowerDeviceMatch, + GreenPowerQuirkRegistryEntry, +) from zha.zigbee.device import GreenPowerDevice @@ -114,3 +120,92 @@ async def test_green_power_device_initialize(zha_gateway: Gateway) -> None: await zha_device.async_initialize(from_cache=True) assert zha_device.platform_entities == {} + + +async def test_green_power_device_match(zha_gateway: Gateway) -> None: + """Test the Green Power quirk matching criteria.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + zigpy_gpd.gpd_manufacturer_id = 0x1234 + + assert GreenPowerDeviceMatch().matches(zigpy_gpd) + assert GreenPowerDeviceMatch(device_id=DeviceID.OnOffSwitch).matches(zigpy_gpd) + assert not GreenPowerDeviceMatch(device_id=DeviceID.GenericSwitch).matches( + zigpy_gpd + ) + assert GreenPowerDeviceMatch(manufacturer_id=0x1234).matches(zigpy_gpd) + assert not GreenPowerDeviceMatch(manufacturer_id=0x5678).matches(zigpy_gpd) + assert not GreenPowerDeviceMatch(model_id=0x0001).matches(zigpy_gpd) + + assert GreenPowerDeviceMatch(src_id_ranges=((0x12000000, 0x12FFFFFF),)).matches( + zigpy_gpd + ) + assert not GreenPowerDeviceMatch(src_id_ranges=((0x00, 0xFF),)).matches(zigpy_gpd) + + # The synthetic IEEE of a SrcID-addressed GPD carries no vendor prefix + assert not GreenPowerDeviceMatch(ieee_prefixes=(bytes([0x04, 0xCD]),)).matches( + zigpy_gpd + ) + + assert GreenPowerDeviceMatch( + filters=(lambda device: GPDCommandID.Toggle in device.commands,) + ).matches(zigpy_gpd) + assert not GreenPowerDeviceMatch( + filters=(lambda device: GPDCommandID.Off in device.commands,) + ).matches(zigpy_gpd) + + +async def test_green_power_device_match_ieee(zha_gateway: Gateway) -> None: + """Test matching an IEEE-addressed GPD by address prefix.""" + zigpy_gpd = ZigpyGreenPowerDevice( + zha_gateway.application_controller, + application_id=ApplicationID.IEEE, + ieee=EUI64.convert("04:cd:15:00:11:22:33:44"), + endpoint=1, + ) + + assert GreenPowerDeviceMatch(ieee_prefixes=(bytes([0x04, 0xCD, 0x15]),)).matches( + zigpy_gpd + ) + assert not GreenPowerDeviceMatch( + ieee_prefixes=(bytes([0x04, 0xCD, 0x16]),) + ).matches(zigpy_gpd) + + # Either identity criterion is sufficient when both are declared + assert GreenPowerDeviceMatch( + src_id_ranges=((0x00, 0xFF),), + ieee_prefixes=(bytes([0x04, 0xCD, 0x15]),), + ).matches(zigpy_gpd) + + +async def test_green_power_quirk_resolution(zha_gateway: Gateway) -> None: + """Test that a registered Green Power quirk resolves and builds the ZHA device.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + + class QuirkedGreenPowerDevice(GreenPowerDevice): + """Quirk-supplied device class.""" + + generic_entry = GreenPowerQuirkRegistryEntry( + device_match=GreenPowerDeviceMatch(device_id=DeviceID.OnOffSwitch), + ) + entry = GreenPowerQuirkRegistryEntry( + device_match=GreenPowerDeviceMatch( + device_id=DeviceID.OnOffSwitch, + src_id_ranges=((0x12000000, 0x12FFFFFF),), + ), + zha_device_factory=QuirkedGreenPowerDevice, + ) + + with DEVICE_REGISTRY.preserve_state(): + DEVICE_REGISTRY.register(generic_entry) + DEVICE_REGISTRY.register(entry) + + # The most recently registered matching entry wins + assert DEVICE_REGISTRY.match_green_power_entry(zigpy_gpd) is entry + + resolved = DEVICE_REGISTRY.resolve(zigpy_gpd) + assert resolved is zigpy_gpd + assert getattr(resolved, QUIRK_REGISTRY_ENTRY_ATTR) is entry + + zha_device = zha_gateway.get_or_create_device(resolved) + assert isinstance(zha_device, QuirkedGreenPowerDevice) + assert zha_device.quirk_applied diff --git a/zha/quirks.py b/zha/quirks.py index 5a42d5dfe..17f4e13d1 100644 --- a/zha/quirks.py +++ b/zha/quirks.py @@ -9,7 +9,7 @@ import inspect import logging from pathlib import Path -from typing import TYPE_CHECKING, NamedTuple +from typing import TYPE_CHECKING, NamedTuple, TypeVar, overload from zigpy.application import ControllerApplication import zigpy.device @@ -19,12 +19,15 @@ from zigpy.zcl.clusters.general import Ota if TYPE_CHECKING: - from zha.zigbee.device import Device + from zha.zigbee.device import Device, GreenPowerDevice _LOGGER = logging.getLogger(__name__) QUIRK_REGISTRY_ENTRY_ATTR = "_quirk_registry_entry" FilterType = Callable[[zigpy.device.Device], bool] +GreenPowerFilterType = Callable[[zigpy.device.GreenPowerDevice], bool] + +_EntryT = TypeVar("_EntryT", "QuirkRegistryEntry", "GreenPowerQuirkRegistryEntry") DEVICE_REGISTRY: DeviceRegistry @@ -154,8 +157,71 @@ class QuirkRegistryEntry: source: QuirkSource | None = field(default=None, compare=False) +@dataclass(frozen=True) +class GreenPowerDeviceMatch: + """Criteria matching a Green Power quirk to a GPD's commissioning signature. + + Every specified field must match. `src_id_ranges` and `ieee_prefixes` + together form the identity criterion: when either is present, the device's + SrcID must fall within one of the ranges or its IEEE address must begin + with one of the prefixes. + """ + + device_id: int | None = None + manufacturer_id: int | None = None + model_id: int | None = None + src_id_ranges: tuple[tuple[int, int], ...] = () # inclusive bounds + ieee_prefixes: tuple[bytes, ...] = () # most-significant-byte first + filters: tuple[GreenPowerFilterType, ...] = () + + def matches(self, device: zigpy.device.GreenPowerDevice) -> bool: + """Return True if `device` satisfies all criteria.""" + if self.device_id is not None and device.device_id != self.device_id: + return False + + if ( + self.manufacturer_id is not None + and device.gpd_manufacturer_id != self.manufacturer_id + ): + return False + + if self.model_id is not None and device.gpd_model_id != self.model_id: + return False + + if self.src_id_ranges or self.ieee_prefixes: + in_range = device.src_id is not None and any( + lower <= device.src_id <= upper for lower, upper in self.src_id_ranges + ) + # An EUI64 serializes least-significant-byte first + ieee = bytes(reversed(device.ieee.serialize())) + has_prefix = any(ieee.startswith(prefix) for prefix in self.ieee_prefixes) + + if not in_range and not has_prefix: + return False + + return all(matcher(device) for matcher in self.filters) + + +@dataclass(frozen=True) +class GreenPowerQuirkRegistryEntry: + """A registered Green Power quirk: how to match, mutate, and build a device.""" + + device_match: GreenPowerDeviceMatch + zigpy_transforms: tuple[ + Callable[[zigpy.device.GreenPowerDevice], zigpy.device.GreenPowerDevice], ... + ] = () + zha_device_factory: Callable[..., GreenPowerDevice] | None = None + # Excluded from equality so identical quirks registered at different sites still + # deduplicate. + source: QuirkSource | None = field(default=None, compare=False) + + class DeviceRegistry: - """Registry of quirk entries, keyed by (manufacturer, model).""" + """Registry of quirk entries for all device types. + + Zigbee entries are indexed by (manufacturer, model); Green Power entries + are kept in a flat list matched in registration order. + """ def __init__(self) -> None: """Initialize the registry.""" @@ -167,9 +233,26 @@ def __init__(self) -> None: # Matched against every device by their filters alone, used mostly for legacy v1 # quirks without model/manufacturer filters. self._wildcard_registry: list[QuirkRegistryEntry] = [] + # Green Power entries, matched in registration order. + self._gp_registry: list[GreenPowerQuirkRegistryEntry] = [] + + @overload + def register(self, entry: QuirkRegistryEntry) -> QuirkRegistryEntry: ... - def register(self, entry: QuirkRegistryEntry) -> QuirkRegistryEntry: + @overload + def register( + self, entry: GreenPowerQuirkRegistryEntry + ) -> GreenPowerQuirkRegistryEntry: ... + + def register( + self, entry: QuirkRegistryEntry | GreenPowerQuirkRegistryEntry + ) -> QuirkRegistryEntry | GreenPowerQuirkRegistryEntry: """Add a quirk entry to the registry, ignoring exact duplicates.""" + if isinstance(entry, GreenPowerQuirkRegistryEntry): + if entry not in self._gp_registry: + self._gp_registry.insert(0, entry) + return entry + if not entry.device_match.applies_to: if entry not in self._wildcard_registry: self._wildcard_registry.insert(0, entry) @@ -227,6 +310,16 @@ def match_entry( return None + def match_green_power_entry( + self, zigpy_device: zigpy.device.GreenPowerDevice + ) -> GreenPowerQuirkRegistryEntry | None: + """Return the first registered entry matching the Green Power device.""" + for entry in self._gp_registry: + if entry.device_match.matches(zigpy_device): + return entry + + return None + def resolve(self, zigpy_device: zigpy.device.BaseDevice) -> zigpy.device.BaseDevice: """Apply the quirk transforms registered for `zigpy_device` and return the result.""" @@ -234,21 +327,18 @@ def resolve(self, zigpy_device: zigpy.device.BaseDevice) -> zigpy.device.BaseDev if hasattr(zigpy_device, QUIRK_REGISTRY_ENTRY_ATTR): return zigpy_device - # Green Power devices will resolve through their own registry - if not isinstance(zigpy_device, zigpy.device.ZigbeeDevice): + entry: QuirkRegistryEntry | GreenPowerQuirkRegistryEntry | None + if isinstance(zigpy_device, zigpy.device.GreenPowerDevice): + entry = self.match_green_power_entry(zigpy_device) + elif isinstance(zigpy_device, zigpy.device.ZigbeeDevice): + entry = self.match_entry(zigpy_device) + else: return zigpy_device - entry = self.match_entry(zigpy_device) if entry is None: return zigpy_device - _LOGGER.debug( - "Resolved %s/%s (%s) to quirk %s", - zigpy_device.manufacturer, - zigpy_device.model, - zigpy_device.ieee, - entry, - ) + _LOGGER.debug("Resolved %s to quirk %s", zigpy_device, entry) # A failing quirk must not prevent the device from loading: log and fall # back to the bare device rather than letting the exception propagate. @@ -264,7 +354,9 @@ def resolve(self, zigpy_device: zigpy.device.BaseDevice) -> zigpy.device.BaseDev return resolved_device - def __iter__(self) -> Iterator[QuirkRegistryEntry]: + def __iter__( + self, + ) -> Iterator[QuirkRegistryEntry | GreenPowerQuirkRegistryEntry]: """Yield every registered entry once (deduplicated across model keys).""" seen: set[int] = set() for entries in (*self._registry.values(), self._wildcard_registry): @@ -273,8 +365,14 @@ def __iter__(self) -> Iterator[QuirkRegistryEntry]: seen.add(id(entry)) yield entry - def remove(self, entry: QuirkRegistryEntry) -> None: + yield from self._gp_registry + + def remove(self, entry: QuirkRegistryEntry | GreenPowerQuirkRegistryEntry) -> None: """Remove a quirk entry from the registry.""" + if isinstance(entry, GreenPowerQuirkRegistryEntry): + self._gp_registry.remove(entry) + return + if not entry.device_match.applies_to: self._wildcard_registry.remove(entry) return @@ -282,29 +380,38 @@ def remove(self, entry: QuirkRegistryEntry) -> None: for manufacturer, model in entry.device_match.applies_to: self._registry[ModelInfo(manufacturer, model)].remove(entry) + @staticmethod + def _purge_custom_entries(entries: list[_EntryT], custom_quirks_root: Path) -> None: + """Remove entries defined within `custom_quirks_root` from `entries`.""" + for entry in list(entries): + if entry.source is None or entry.source.file is None: + continue + if Path(entry.source.file).is_relative_to(custom_quirks_root): + _LOGGER.debug("Removing stale custom quirk: %s", entry) + entries.remove(entry) + def purge_custom_quirks(self, custom_quirks_root: Path) -> None: """Remove quirks loaded from the custom quirks directory.""" # Prefer the explicit registry to the wildcard registry for entries in (*self._registry.values(), self._wildcard_registry): - for entry in list(entries): - if entry.source is None or entry.source.file is None: - continue - if Path(entry.source.file).is_relative_to(custom_quirks_root): - _LOGGER.debug("Removing stale custom quirk: %s", entry) - entries.remove(entry) + self._purge_custom_entries(entries, custom_quirks_root) + + self._purge_custom_entries(self._gp_registry, custom_quirks_root) @contextlib.contextmanager def preserve_state(self) -> Iterator[None]: """Snapshot the registry and restore it on exit.""" saved = {key: list(entries) for key, entries in self._registry.items()} saved_wildcard = list(self._wildcard_registry) + saved_gp = list(self._gp_registry) try: yield finally: self._registry.clear() self._registry.update(saved) self._wildcard_registry[:] = saved_wildcard + self._gp_registry[:] = saved_gp DEVICE_REGISTRY = DeviceRegistry() From 3300e9fe8872fac060af4d5e49646014ded95440 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:07:25 +0000 Subject: [PATCH 5/5] Adapt event platform --- tests/test_platform_event.py | 6 +++--- zha/application/platforms/event/__init__.py | 10 +++------- 2 files changed, 6 insertions(+), 10 deletions(-) diff --git a/tests/test_platform_event.py b/tests/test_platform_event.py index 388f6a956..d00b3bb02 100644 --- a/tests/test_platform_event.py +++ b/tests/test_platform_event.py @@ -16,7 +16,7 @@ ) from zha.application import Platform from zha.application.gateway import Gateway -from zha.application.platforms import EntityStateChangedEvent +from zha.application.platforms import EntityStateChangedEvent, ZclPlatformEntity from zha.application.platforms.event import ( BaseEvent, EntityEventTriggeredEvent, @@ -31,8 +31,8 @@ from zha.zigbee.device import Device -class FakeEvent(BaseEvent): - """Event entity with a fixed set of event types.""" +class FakeEvent(BaseEvent, ZclPlatformEntity): + """ZCL-backed event entity with a fixed set of event types.""" _unique_id_suffix = "fake" _attr_device_class = EventDeviceClass.BUTTON diff --git a/zha/application/platforms/event/__init__.py b/zha/application/platforms/event/__init__.py index ffab05b4c..265ed8a9f 100644 --- a/zha/application/platforms/event/__init__.py +++ b/zha/application/platforms/event/__init__.py @@ -3,7 +3,7 @@ from __future__ import annotations import dataclasses -from typing import TYPE_CHECKING, Any, Final +from typing import Any, Final from zigpy.types.named import EUI64 @@ -11,10 +11,6 @@ from zha.application.platforms import BaseEntityState, PlatformEntity from zha.application.platforms.event.const import DoorbellEventType, EventDeviceClass -if TYPE_CHECKING: - from zha.zigbee.device import Device - from zha.zigbee.endpoint import Endpoint - @dataclasses.dataclass(frozen=True, kw_only=True) class EventState(BaseEntityState): @@ -53,9 +49,9 @@ class BaseEvent(PlatformEntity): _attr_device_class: EventDeviceClass | None = None _attr_event_types: list[str] - def __init__(self, endpoint: Endpoint, device: Device, **kwargs: Any) -> None: + def __init__(self, *args: Any, **kwargs: Any) -> None: """Initialize the event entity.""" - super().__init__(endpoint=endpoint, device=device, **kwargs) + super().__init__(*args, **kwargs) # Doorbells are expected to ring: the `doorbell.rang` trigger matches on it if (