diff --git a/pyproject.toml b/pyproject.toml index 614c2b925..d5c240579 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -102,6 +102,7 @@ non_interactive = true # `python -m tools.regenerate_mypy_ignores`. They are sorted easiest-to-fix # first; clean up a module's errors and re-run the tool to shrink the list. # Autogenerated mypy overrides: start +# Total ignored errors: 182 across 20 modules [[tool.mypy.overrides]] module = "zha.application.platforms.virtual" disable_error_code = [ @@ -109,29 +110,34 @@ disable_error_code = [ ] [[tool.mypy.overrides]] -module = "zha.quirks" +module = "zha.zigbee.group" disable_error_code = [ - "no-any-return", # 1 + "return-value", # 1 ] [[tool.mypy.overrides]] -module = "zha.application.platforms.number" +module = "zha.application.gateway" disable_error_code = [ - "no-any-return", # 2 + "assignment", # 2 ] [[tool.mypy.overrides]] -module = "zha.application.discovery" +module = "zha.application.platforms" disable_error_code = [ - "no-any-return", # 3 + "arg-type", # 1 + "assignment", # 1 ] [[tool.mypy.overrides]] -module = "zha.application.platforms" +module = "zha.application.platforms.number" disable_error_code = [ - "arg-type", # 1 - "assignment", # 1 - "no-any-return", # 1 + "no-any-return", # 2 +] + +[[tool.mypy.overrides]] +module = "zha.application.discovery" +disable_error_code = [ + "no-any-return", # 3 ] [[tool.mypy.overrides]] @@ -177,29 +183,19 @@ disable_error_code = [ "no-untyped-def", # 4 ] -[[tool.mypy.overrides]] -module = "zha.zigbee.group" -disable_error_code = [ - "no-untyped-def", # 3 - "return-value", # 2 -] - [[tool.mypy.overrides]] module = "zha.application.platforms.fan" disable_error_code = [ "arg-type", # 2 - "index", # 1 "no-any-return", # 3 "no-untyped-def", # 1 ] [[tool.mypy.overrides]] -module = "zha.application.gateway" +module = "zha.application.helpers" disable_error_code = [ - "arg-type", # 3 - "assignment", # 2 - "no-untyped-call", # 4 - "return-value", # 1 + "arg-type", # 2 + "no-untyped-def", # 5 ] [[tool.mypy.overrides]] @@ -218,52 +214,41 @@ disable_error_code = [ ] [[tool.mypy.overrides]] -module = "zha.application.helpers" +module = "zha.zigbee.device" disable_error_code = [ - "arg-type", # 2 - "no-untyped-call", # 4 - "no-untyped-def", # 10 + "arg-type", # 1 + "no-untyped-call", # 2 + "union-attr", # 13 ] [[tool.mypy.overrides]] module = "zha.application.platforms.light" disable_error_code = [ "arg-type", # 3 - "no-any-return", # 4 + "no-any-return", # 3 "no-untyped-call", # 11 "no-untyped-def", # 8 "type-var", # 2 ] -[[tool.mypy.overrides]] -module = "zha.application.platforms.climate" -disable_error_code = [ - "call-overload", # 4 - "no-any-return", # 23 - "no-untyped-def", # 8 -] - [[tool.mypy.overrides]] module = "zha.application.platforms.sensor" disable_error_code = [ - "arg-type", # 12 + "arg-type", # 11 "assignment", # 2 - "attr-defined", # 5 + "attr-defined", # 2 "no-any-return", # 9 "no-untyped-call", # 4 "no-untyped-def", # 3 - "operator", # 4 + "operator", # 2 ] [[tool.mypy.overrides]] -module = "zha.zigbee.device" +module = "zha.application.platforms.climate" disable_error_code = [ - "arg-type", # 12 - "no-any-return", # 4 - "no-untyped-call", # 3 - "no-untyped-def", # 10 - "return-value", # 1 - "union-attr", # 18 + "call-overload", # 4 + "no-any-return", # 23 + "no-untyped-def", # 8 ] # Autogenerated mypy overrides: end diff --git a/tests/test_device.py b/tests/test_device.py index e4b1a6289..670c71cba 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -315,6 +315,80 @@ async def test_check_available_no_basic_cluster( assert "does not have a mandatory basic cluster" in caplog.text +async def test_check_available_skips_stale_coordinator( + zha_gateway: Gateway, +) -> None: + """Check that a former coordinator is not marked unavailable.""" + old_coord_dev = zigpy_device( + zha_gateway, + with_basic_cluster=False, + ieee="aa:bb:cc:dd:ee:ff:00:12", + nwk=0x0000, + ) + old_coord_dev.node_desc = old_coord_dev.node_desc.replace( + logical_type=zdo_t.LogicalType.Coordinator + ) + zha_device = await join_zigpy_device(zha_gateway, old_coord_dev) + + assert not zha_device.is_active_coordinator + assert zha_device.available is True + + old_coord_dev.last_seen = time.time() - zha_device.consider_unavailable_time - 2 + await _send_time_changed( + zha_gateway, zha_gateway._device_availability_checker.__polling_interval + 1 + ) + + assert zha_device.available is True + + +async def test_check_available_skips_active_coordinator( + zha_gateway: Gateway, +) -> None: + """Check that the active coordinator is never marked unavailable.""" + coord_dev = zigpy_device( + zha_gateway, + with_basic_cluster=False, + ieee="aa:bb:cc:dd:ee:ff:00:11", + nwk=0x0000, + ) + coord_dev.node_desc = coord_dev.node_desc.replace( + logical_type=zdo_t.LogicalType.Coordinator + ) + zha_device = await join_zigpy_device(zha_gateway, coord_dev) + zha_gateway.application_controller.state.node_info.ieee = coord_dev.ieee + + assert zha_device.is_active_coordinator + assert zha_device.available is True + + coord_dev.last_seen = time.time() - zha_device.consider_unavailable_time - 2 + await zha_device._check_available() + + # `available` is always True for the active coordinator, so check the raw flag + assert zha_device._available is True + assert zha_device.available is True + + +async def test_discover_entities_skips_zdo_endpoint( + zha_gateway: Gateway, +) -> None: + """Check that endpoint 0 is never passed to endpoint entity discovery.""" + zha_device = await join_zigpy_device( + zha_gateway, zigpy_device(zha_gateway, with_basic_cluster=True) + ) + # `Device.endpoints` only holds non-ZDO endpoints, so ep 0 is injected + zha_device._endpoints[0] = mock.MagicMock() + + with patch( + "zha.zigbee.device.discovery.discover_entities_for_endpoint", + return_value=iter(()), + ) as discover: + list(zha_device.discover_entities()) + + discovered = {c.args[0] for c in discover.mock_calls} + assert zha_device._endpoints[0] not in discovered + assert discovered == {ep for i, ep in zha_device._endpoints.items() if i != 0} + + async def test_device_is_active_coordinator( zha_gateway: Gateway, ) -> None: diff --git a/tests/test_green_power_device.py b/tests/test_green_power_device.py new file mode 100644 index 000000000..dde2a32f4 --- /dev/null +++ b/tests/test_green_power_device.py @@ -0,0 +1,921 @@ +"""Tests for ZHA Green Power devices.""" + +from __future__ import annotations + +from collections.abc import Iterator +from datetime import UTC, datetime +import json +import logging +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock + +import pytest +import zigpy.device +from zigpy.device import ( + GreenPowerCommandReceived, + GreenPowerDevice as ZigpyGreenPowerDevice, +) +from zigpy.profiles import zha +from zigpy.types import EUI64 +from zigpy.zcl.clusters.general import Basic, OnOff +from zigpy.zgp.types import ApplicationID, DeviceID, GPDCommandID, SrcID + +from tests.common import ( + SIG_EP_INPUT, + SIG_EP_OUTPUT, + SIG_EP_PROFILE, + SIG_EP_TYPE, + ZhaJsonEncoder, + create_mock_zigpy_device, + join_zigpy_device, +) +from zha.application import Platform +from zha.application.const import ( + UNKNOWN_MANUFACTURER, + UNKNOWN_MODEL, + ZHA_DEVICE_CONFIGURED_EVENT, + ZHA_DEVICE_ENTITY_ADDED_EVENT, + ZHA_DEVICE_ENTITY_REMOVED_EVENT, + ZHA_GW_MSG_DEVICE_FULL_INIT, + ZHA_GW_MSG_DEVICE_JOINED, + ZHA_GW_MSG_DEVICE_LEFT, + ZHA_GW_MSG_DEVICE_REMOVED, + ZHA_GW_MSG_RAW_INIT, +) +from zha.application.gateway import ( + DeviceFullInitEvent, + DeviceJoinedEvent, + DeviceLeftEvent, + DevicePairingStatus, + DeviceRemovedEvent, + Gateway, +) +from zha.application.helpers import async_is_bindable_target, get_matched_clusters +from zha.application.platforms import PlatformEntity +from zha.application.platforms.event import ( + BaseEvent, + EntityEventTriggeredEvent, + TriggeredEvent, +) +from zha.exceptions import ZHAException +from zha.quirks import ( + DEVICE_REGISTRY, + QUIRK_REGISTRY_ENTRY_ATTR, + DeviceMatch, + DeviceRegistry, + GreenPowerDeviceMatch, + GreenPowerQuirkRegistryEntry, + ModelInfo, + QuirkRegistryEntry, + QuirkSource, +) +from zha.zigbee.device import ( + BaseDevice, + Device, + DeviceEntityAddedEvent, + DeviceEntityRemovedEvent, + 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.skip_configuration is False + assert zha_device.is_groupable is False + assert zha_device.async_get_std_clusters() == {} + 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 == [] + + zigpy_gpd.last_seen = datetime(2026, 1, 1, tzinfo=UTC) + diagnostics = zha_device.get_diagnostics_json() + assert diagnostics["ieee"] == str(zigpy_gpd.ieee) + assert diagnostics["last_seen"] == "2026-01-01T00:00:00+00:00" + assert diagnostics["active_coordinator"] is False + 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 == {} + + +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"), + ) + + 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(entry) + DEVICE_REGISTRY.register(generic_entry) + + # The more specific entry wins regardless of registration order + 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 + + +async def test_green_power_quirk_specificity(zha_gateway: Gateway) -> None: + """Test that Green Power quirks are ranked by specificity, then recency.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + zigpy_gpd.gpd_manufacturer_id = 0x1234 + + device_id_entry = GreenPowerQuirkRegistryEntry( + device_match=GreenPowerDeviceMatch(device_id=DeviceID.OnOffSwitch), + ) + two_field_entry = GreenPowerQuirkRegistryEntry( + device_match=GreenPowerDeviceMatch( + device_id=DeviceID.OnOffSwitch, manufacturer_id=0x1234 + ), + ) + src_id_entry = GreenPowerQuirkRegistryEntry( + device_match=GreenPowerDeviceMatch(src_id_ranges=((0x12000000, 0x12FFFFFF),)), + ) + later_device_id_entry = GreenPowerQuirkRegistryEntry( + device_match=GreenPowerDeviceMatch( + device_id=DeviceID.OnOffSwitch, filters=(lambda device: True,) + ), + ) + + # Equally specific entries fall back to the most recently registered one + with DEVICE_REGISTRY.preserve_state(): + DEVICE_REGISTRY.register(device_id_entry) + DEVICE_REGISTRY.register(later_device_id_entry) + assert DEVICE_REGISTRY.match_green_power_entry(zigpy_gpd) is ( + later_device_id_entry + ) + + with DEVICE_REGISTRY.preserve_state(): + DEVICE_REGISTRY.register(two_field_entry) + DEVICE_REGISTRY.register(device_id_entry) + assert DEVICE_REGISTRY.match_green_power_entry(zigpy_gpd) is two_field_entry + + # An identity criterion outranks any number of signature fields + with DEVICE_REGISTRY.preserve_state(): + DEVICE_REGISTRY.register(src_id_entry) + DEVICE_REGISTRY.register(two_field_entry) + assert DEVICE_REGISTRY.match_green_power_entry(zigpy_gpd) is src_id_entry + + # Specificity counts field presence, not value + assert ( + GreenPowerDeviceMatch(device_id=0).specificity + > GreenPowerDeviceMatch().specificity + ) + + +async def test_green_power_device_not_zigbee(zha_gateway: Gateway) -> None: + """Test that Zigbee-only helpers reject or skip a Green Power device.""" + zha_device = zha_gateway.get_or_create_device(make_zigpy_gpd(zha_gateway)) + + remote_zha_device = await join_zigpy_device( + zha_gateway, + create_mock_zigpy_device( + zha_gateway, + { + 1: { + SIG_EP_INPUT: [Basic.cluster_id], + SIG_EP_OUTPUT: [OnOff.cluster_id], + SIG_EP_TYPE: zha.DeviceType.REMOTE_CONTROL, + SIG_EP_PROFILE: zha.PROFILE_ID, + } + }, + ), + ) + + assert not async_is_bindable_target(remote_zha_device, zha_device) + assert not async_is_bindable_target(zha_device, remote_zha_device) + assert not async_is_bindable_target(zha_device, zha_gateway.coordinator_zha_device) + assert await get_matched_clusters(remote_zha_device, zha_device) == [] + assert await get_matched_clusters(zha_device, remote_zha_device) == [] + + with pytest.raises(ZHAException, match="is not a Zigbee device"): + zha_gateway.get_zigbee_device(zha_device.ieee) + + +async def test_green_power_device_raw_initialized(zha_gateway: Gateway) -> None: + """Test the raw init event for a GPD, which has no model or manufacturer.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + + events = [] + zha_gateway.on_event(ZHA_GW_MSG_RAW_INIT, events.append) + zha_gateway.raw_device_initialized(zigpy_gpd) + + assert len(events) == 1 + assert events[0].device_info.ieee == zigpy_gpd.ieee + assert events[0].device_info.model == UNKNOWN_MODEL + assert events[0].device_info.manufacturer == UNKNOWN_MANUFACTURER + + +async def test_green_power_device_recommissioned(zha_gateway: Gateway) -> None: + """Test that a re-commissioned GPD rebuilds the existing ZHA device.""" + zha_device = zha_gateway.get_or_create_device(make_zigpy_gpd(zha_gateway)) + + events = [] + zha_device.on_event(ZHA_DEVICE_CONFIGURED_EVENT, events.append) + + new_zigpy_gpd = make_zigpy_gpd(zha_gateway) + zha_gateway.application_controller.listener_event( + "device_reinterviewed", new_zigpy_gpd + ) + await zha_gateway.async_block_till_done() + + assert zha_gateway.devices[new_zigpy_gpd.ieee] is zha_device + assert zha_device.device is new_zigpy_gpd + assert len(events) == 1 + + +# Mimics the zhaquirks `GreenPowerQuirkBuilder` until a released quirk replaces it +class ToggleEvent(BaseEvent): + """Event entity firing when the GPD sends a toggle command.""" + + _attr_event_types = ["toggle"] + + def on_add(self) -> None: + """Subscribe to the commands sent by the device.""" + super().on_add() + self._on_remove_callbacks.append( + self.device.device.on_event( + GreenPowerCommandReceived.event_type, self._handle_gp_command + ) + ) + + def _handle_gp_command(self, event: GreenPowerCommandReceived) -> None: + if event.command_id == GPDCommandID.Toggle: + self._trigger_event("toggle") + + +class QuirkGreenPowerDevice(GreenPowerDevice): + """Quirk device exposing a single event entity.""" + + def discover_entities(self) -> Iterator[PlatformEntity]: + """Yield the quirk's entities.""" + yield from super().discover_entities() + yield ToggleEvent( + self, + unique_id=str(self.ieee), + from_quirk=True, + unique_id_suffix="toggle", + fallback_name="Toggle", + ) + + +def send_toggle(zigpy_gpd: ZigpyGreenPowerDevice) -> None: + """Emit a received toggle command from the zigpy GPD.""" + zigpy_gpd.emit( + GreenPowerCommandReceived.event_type, + GreenPowerCommandReceived( + device_ieee=str(zigpy_gpd.ieee), + endpoint_id=None, + command_id=GPDCommandID.Toggle, + command=None, + ), + ) + + +async def test_green_power_quirk_entity_lifecycle(zha_gateway: Gateway) -> None: + """Test a quirk entity on a GPD through initialization and recommissioning.""" + entry = GreenPowerQuirkRegistryEntry( + device_match=GreenPowerDeviceMatch(device_id=DeviceID.OnOffSwitch), + zha_device_factory=QuirkGreenPowerDevice, + ) + + with DEVICE_REGISTRY.preserve_state(): + DEVICE_REGISTRY.register(entry) + + zigpy_gpd = DEVICE_REGISTRY.resolve(make_zigpy_gpd(zha_gateway)) + zha_device = zha_gateway.get_or_create_device(zigpy_gpd) + assert isinstance(zha_device, QuirkGreenPowerDevice) + + await zha_device.async_initialize(from_cache=True) + + unique_id = f"{zigpy_gpd.ieee}-toggle" + entity = zha_device.get_platform_entity(Platform.EVENT, unique_id) + assert list(zha_device.platform_entities) == [(Platform.EVENT, unique_id)] + assert entity.state.available is True + assert entity.state.event_types == ["toggle"] + assert zha_device.extended_device_info.entities == {unique_id: entity.state} + + triggered = [] + entity.on_event(EntityEventTriggeredEvent.event, triggered.append) + send_toggle(zigpy_gpd) + assert [event.triggered for event in triggered] == [ + TriggeredEvent(event_type="toggle", event_attributes={}) + ] + + entity_events = [] + zha_device.on_event(ZHA_DEVICE_ENTITY_REMOVED_EVENT, entity_events.append) + zha_device.on_event(ZHA_DEVICE_ENTITY_ADDED_EVENT, entity_events.append) + + # Recommissioning replaces the zigpy device and recreates the entity + new_zigpy_gpd = DEVICE_REGISTRY.resolve(make_zigpy_gpd(zha_gateway)) + zha_gateway.application_controller.listener_event( + "device_reinterviewed", new_zigpy_gpd + ) + await zha_gateway.async_block_till_done() + + assert zha_gateway.devices[new_zigpy_gpd.ieee] is zha_device + assert zha_device.device is new_zigpy_gpd + assert entity_events == [ + DeviceEntityRemovedEvent(platform=Platform.EVENT, unique_id=unique_id), + DeviceEntityAddedEvent(platform=Platform.EVENT, unique_id=unique_id), + ] + + new_entity = zha_device.get_platform_entity(Platform.EVENT, unique_id) + assert new_entity is not entity + assert list(zha_device.platform_entities) == [(Platform.EVENT, unique_id)] + + # The old entity is unsubscribed from the old zigpy device + new_triggered = [] + new_entity.on_event(EntityEventTriggeredEvent.event, new_triggered.append) + send_toggle(zigpy_gpd) + assert len(triggered) == 1 + assert new_triggered == [] + + send_toggle(new_zigpy_gpd) + assert len(triggered) == 1 + assert [event.triggered for event in new_triggered] == [ + TriggeredEvent(event_type="toggle", event_attributes={}) + ] + + +async def test_green_power_device_diagnostics_entities(zha_gateway: Gateway) -> None: + """Test that diagnostics serialize entities and skip virtual ones.""" + entry = GreenPowerQuirkRegistryEntry( + device_match=GreenPowerDeviceMatch(device_id=DeviceID.OnOffSwitch), + zha_device_factory=QuirkGreenPowerDevice, + ) + + with DEVICE_REGISTRY.preserve_state(): + DEVICE_REGISTRY.register(entry) + + zigpy_gpd = DEVICE_REGISTRY.resolve(make_zigpy_gpd(zha_gateway)) + zigpy_gpd.last_seen = datetime(2026, 1, 1, tzinfo=UTC) + zha_device = zha_gateway.get_or_create_device(zigpy_gpd) + await zha_device.async_initialize(from_cache=True) + + unique_id = f"{zigpy_gpd.ieee}-toggle" + event_entity = zha_device.get_platform_entity(Platform.EVENT, unique_id) + + # Virtual entities have no state surface and are left out of diagnostics + zha_device.platform_entities[(Platform.VIRTUAL, "virtual")] = MagicMock( + PLATFORM=Platform.VIRTUAL, unique_id="virtual", on_remove=AsyncMock() + ) + + diagnostics = json.loads( + json.dumps(zha_device.get_diagnostics_json(), cls=ZhaJsonEncoder) + ) + + assert list(diagnostics["zha_lib_entities"]) == [Platform.EVENT] + (state,) = diagnostics["zha_lib_entities"][Platform.EVENT] + assert state["unique_id"] == event_entity.state.unique_id + assert state["event_types"] == ["toggle"] + assert state["device_ieee"] == str(zigpy_gpd.ieee) + assert isinstance(state["migrate_unique_ids"], list) + assert state["extra_state_attribute_names"] == sorted( + state["extra_state_attribute_names"] + ) + + +async def test_green_power_base_device_defaults(zha_gateway: Gateway) -> None: + """Test the `BaseDevice` defaults a GPD inherits.""" + zha_device = zha_gateway.get_or_create_device(make_zigpy_gpd(zha_gateway)) + + assert zha_device.quirk_metadata is None + assert zha_device.is_coordinator is False + assert zha_device.device_alerts == [] + assert zha_device.device_automation_triggers == { + ("device_offline", "device_offline"): {"device_event_type": "device_offline"} + } + assert zha_device.device_automation_commands == {} + assert zha_device._quirk_exposes_features() == set() + assert zha_device._quirk_skip_configuration() is False + assert zha_device._quirk_device_automation_triggers() == {} + assert zha_device.power_source == "Battery or Unknown" + assert zha_device.gateway is zha_gateway + + +async def test_base_device_abstract_resolvers(zha_gateway: Gateway) -> None: + """Test the `BaseDevice` hooks that subclasses must implement.""" + zha_device = zha_gateway.get_or_create_device(make_zigpy_gpd(zha_gateway)) + + for method in ( + BaseDevice._resolve_manufacturer, + BaseDevice._resolve_model, + ): + with pytest.raises(NotImplementedError): + method(zha_device) + + for prop in ( + BaseDevice.manufacturer_code, + BaseDevice.is_mains_powered, + BaseDevice.device_type, + ): + with pytest.raises(NotImplementedError): + prop.fget(zha_device) + + +async def test_green_power_device_join(zha_gateway: Gateway) -> None: + """Test the join sequence of a GPD through the gateway.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + + joined: list[DeviceJoinedEvent] = [] + raw: list = [] + full: list[DeviceFullInitEvent] = [] + zha_gateway.on_event(ZHA_GW_MSG_DEVICE_JOINED, joined.append) + zha_gateway.on_event(ZHA_GW_MSG_RAW_INIT, raw.append) + zha_gateway.on_event(ZHA_GW_MSG_DEVICE_FULL_INIT, full.append) + + zha_gateway.device_joined(zigpy_gpd) + assert len(joined) == 1 + assert joined[0].device_info.ieee == zigpy_gpd.ieee + assert joined[0].device_info.nwk == zigpy_gpd.nwk + assert joined[0].device_info.pairing_status is DevicePairingStatus.PAIRED + + zha_gateway.raw_device_initialized(zigpy_gpd) + assert len(raw) == 1 + + zha_gateway.device_initialized(zigpy_gpd) + await zha_gateway.async_block_till_done() + + zha_device = zha_gateway.devices[zigpy_gpd.ieee] + assert isinstance(zha_device, GreenPowerDevice) + assert zha_device.available is True + assert zha_device.on_network is True + + assert [e.device_info.pairing_status for e in full] == [ + DevicePairingStatus.CONFIGURED, + DevicePairingStatus.INITIALIZED, + ] + assert full[0].new_join is True + assert full[0].device_info.ieee == zigpy_gpd.ieee + assert full[0].device_info.device_type == "GreenPower" + + # Initializing it again is a rejoin of a known device + zha_gateway.device_initialized(zigpy_gpd) + await zha_gateway.async_block_till_done() + + assert zha_gateway.devices[zigpy_gpd.ieee] is zha_device + assert [e.device_info.pairing_status for e in full[2:]] == [ + DevicePairingStatus.CONFIGURED, + DevicePairingStatus.INITIALIZED, + ] + assert full[2].new_join is False + + +async def test_green_power_device_left(zha_gateway: Gateway) -> None: + """Test that a GPD leaving the network is flagged off-network and unavailable.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + zha_device = zha_gateway.get_or_create_device(zigpy_gpd) + assert zha_device.on_network is True + + left: list[DeviceLeftEvent] = [] + zha_gateway.on_event(ZHA_GW_MSG_DEVICE_LEFT, left.append) + + zha_gateway.device_left(zigpy_gpd) + await zha_gateway.async_block_till_done() + + assert zha_device.on_network is False + assert zha_device.available is False + assert left == [DeviceLeftEvent(ieee=zigpy_gpd.ieee, nwk=zigpy_gpd.nwk)] + + +async def test_green_power_device_removed(zha_gateway: Gateway) -> None: + """Test removal of a GPD, from zigpy and via the gateway API.""" + zigpy_gpd = make_zigpy_gpd(zha_gateway) + zha_device = zha_gateway.get_or_create_device(zigpy_gpd) + + removed: list[DeviceRemovedEvent] = [] + zha_gateway.on_event(ZHA_GW_MSG_DEVICE_REMOVED, removed.append) + + # No groups: the gateway just asks zigpy to remove the device + zha_gateway.application_controller.remove = AsyncMock() + await zha_gateway.async_remove_device(zigpy_gpd.ieee) + zha_gateway.application_controller.remove.assert_called_once_with(zigpy_gpd.ieee) + + # zigpy then announces the removal + zha_gateway.device_removed(zigpy_gpd) + await zha_gateway.async_block_till_done() + + assert zigpy_gpd.ieee not in zha_gateway.devices + assert zha_gateway.get_device(zigpy_gpd.ieee) is None + assert len(removed) == 1 + assert removed[0].device_info.ieee == zha_device.ieee + assert removed[0].device_info.device_type == "GreenPower" + + # Removing an unknown device is a no-op + zha_gateway.device_removed(zigpy_gpd) + assert len(removed) == 1 + + +async def test_green_power_recommissioned_factory_change(zha_gateway: Gateway) -> None: + """Test that a quirk resolving on re-commissioning replaces the ZHA device.""" + zha_device = zha_gateway.get_or_create_device(make_zigpy_gpd(zha_gateway)) + assert type(zha_device) is GreenPowerDevice + + entry = GreenPowerQuirkRegistryEntry( + device_match=GreenPowerDeviceMatch(device_id=DeviceID.OnOffSwitch), + zha_device_factory=QuirkGreenPowerDevice, + ) + + with DEVICE_REGISTRY.preserve_state(): + DEVICE_REGISTRY.register(entry) + + new_zigpy_gpd = DEVICE_REGISTRY.resolve(make_zigpy_gpd(zha_gateway)) + assert getattr(new_zigpy_gpd, QUIRK_REGISTRY_ENTRY_ATTR) is entry + + full: list[DeviceFullInitEvent] = [] + zha_gateway.on_event(ZHA_GW_MSG_DEVICE_FULL_INIT, full.append) + + zha_gateway.application_controller.listener_event( + "device_reinterviewed", new_zigpy_gpd + ) + await zha_gateway.async_block_till_done() + + new_zha_device = zha_gateway.devices[new_zigpy_gpd.ieee] + assert new_zha_device is not zha_device + assert isinstance(new_zha_device, QuirkGreenPowerDevice) + assert new_zha_device.device is new_zigpy_gpd + assert new_zha_device.available is True + assert new_zha_device.on_network is True + assert len(new_zha_device.platform_entities) == 1 + assert [e.device_info.pairing_status for e in full] == [ + DevicePairingStatus.CONFIGURED + ] + + # The same entry on both sides is rebuilt in place + newer_zigpy_gpd = DEVICE_REGISTRY.resolve(make_zigpy_gpd(zha_gateway)) + zha_gateway.application_controller.listener_event( + "device_reinterviewed", newer_zigpy_gpd + ) + await zha_gateway.async_block_till_done() + assert zha_gateway.devices[newer_zigpy_gpd.ieee] is new_zha_device + assert new_zha_device.device is newer_zigpy_gpd + + +async def test_bindable_target_source_without_clusters(zha_gateway: Gateway) -> None: + """Test bindability with sources that do and do not have clusters.""" + gpd = zha_gateway.get_or_create_device(make_zigpy_gpd(zha_gateway)) + remote = await join_zigpy_device( + zha_gateway, + create_mock_zigpy_device( + zha_gateway, + { + 1: { + SIG_EP_INPUT: [Basic.cluster_id], + SIG_EP_OUTPUT: [OnOff.cluster_id], + SIG_EP_TYPE: zha.DeviceType.REMOTE_CONTROL, + SIG_EP_PROFILE: zha.PROFILE_ID, + } + }, + ), + ) + + # A source with clusters can bind to the coordinator + assert async_is_bindable_target(remote, zha_gateway.coordinator_zha_device) + + # A source without clusters never can, not even to the coordinator + assert not async_is_bindable_target(gpd, zha_gateway.coordinator_zha_device) + assert not async_is_bindable_target(gpd, remote) + + +# Registry tests, using private registries so the global one is untouched +def make_gp_entry( + match: GreenPowerDeviceMatch | None = None, file: str | None = None +) -> GreenPowerQuirkRegistryEntry: + """Create a GP registry entry, optionally defined in `file`.""" + return GreenPowerQuirkRegistryEntry( + device_match=match or GreenPowerDeviceMatch(device_id=DeviceID.OnOffSwitch), + source=None + if file is None + else QuirkSource(module="custom", file=file, line=1, label="Custom"), + ) + + +def test_registry_gp_entries_iterate_remove_and_dedupe() -> None: + """Test that GP entries are listed, deduplicated and removable.""" + registry = DeviceRegistry() + generic = make_gp_entry() + specific = make_gp_entry(GreenPowerDeviceMatch(device_id=1, model_id=2)) + zigbee = QuirkRegistryEntry( + device_match=DeviceMatch(applies_to=(ModelInfo("m", "x"), ModelInfo("m", None))) + ) + wildcard = QuirkRegistryEntry(device_match=DeviceMatch(applies_to=())) + + for entry in (generic, zigbee, wildcard, specific): + assert registry.register(entry) is entry + # Exact duplicates are ignored + registry.register(entry) + + # Zigbee entries are listed once, then GP entries by specificity + assert list(registry) == [zigbee, wildcard, specific, generic] + + registry.remove(generic) + assert list(registry) == [zigbee, wildcard, specific] + registry.remove(wildcard) + registry.remove(zigbee) + assert list(registry) == [specific] + + with pytest.raises(ValueError): + registry.remove(generic) + + +def test_registry_purge_custom_quirks(tmp_path: Path) -> None: + """Test that quirks defined under the custom quirks root are purged.""" + registry = DeviceRegistry() + root = tmp_path / "custom_zha_quirks" + + def source(file: str | None) -> QuirkSource: + return QuirkSource(module="m", file=file, line=1, label="Q") + + custom_gp = make_gp_entry(file=str(root / "gp.py")) + other_gp = make_gp_entry( + GreenPowerDeviceMatch(device_id=1), file=str(tmp_path / "builtin.py") + ) + sourceless_gp = make_gp_entry(GreenPowerDeviceMatch(device_id=3)) + custom_zigbee = QuirkRegistryEntry( + device_match=DeviceMatch(applies_to=(ModelInfo("a", "b"),)), + source=source(str(root / "zigbee.py")), + ) + custom_wildcard = QuirkRegistryEntry( + device_match=DeviceMatch(applies_to=()), source=source(str(root / "w.py")) + ) + builtin_zigbee = QuirkRegistryEntry( + device_match=DeviceMatch(applies_to=(ModelInfo("c", "d"),)), + source=source(str(tmp_path / "builtin.py")), + ) + no_file_zigbee = QuirkRegistryEntry( + device_match=DeviceMatch(applies_to=(ModelInfo("e", "f"),)), + source=source(None), + ) + + entries = [ + custom_gp, + other_gp, + sourceless_gp, + custom_zigbee, + custom_wildcard, + builtin_zigbee, + no_file_zigbee, + ] + for entry in entries: + registry.register(entry) + + registry.purge_custom_quirks(root) + + assert set(map(id, registry)) == { + id(other_gp), + id(sourceless_gp), + id(builtin_zigbee), + id(no_file_zigbee), + } + + +def test_registry_register_validation() -> None: + """Test that unusable entries and classes are rejected.""" + registry = DeviceRegistry() + + with pytest.raises(ValueError, match="manufacturer and/or model"): + registry.register( + QuirkRegistryEntry( + device_match=DeviceMatch(applies_to=(ModelInfo(None, None),)) + ) + ) + + class NoMatch(Device): + _device_match = None + + with pytest.raises(ValueError, match="does not define `_device_match`"): + registry.register_device(NoMatch) + + +def test_registry_register_device_transforms() -> None: + """Test that `register_device` collects the zigpy device transforms.""" + registry = DeviceRegistry() + + def extra_transform(device: zigpy.device.ZigbeeDevice) -> zigpy.device.ZigbeeDevice: + return device + + class Replacement: + pass + + class QuirkedDevice(Device): + _device_match = DeviceMatch(applies_to=(ModelInfo("mfg", "mdl"),)) + _zigpy_device_cls = Replacement # type: ignore[assignment] + _zigpy_device_transforms = (extra_transform,) + + assert registry.register_device(QuirkedDevice) is QuirkedDevice + + (entry,) = list(registry) + assert entry.zha_device_factory is QuirkedDevice + assert entry.zigpy_transforms[1:] == (extra_transform,) + assert entry.zigpy_transforms[0].device_cls is Replacement + assert entry.source is not None + assert entry.source.label.endswith("QuirkedDevice") + + +def test_device_match_applies_to_mismatch() -> None: + """Test that a device matching none of the `applies_to` pairs is rejected.""" + zigpy_device = MagicMock(spec=zigpy.device.ZigbeeDevice) + zigpy_device.manufacturer = "mfg" + zigpy_device.model = "mdl" + + assert DeviceMatch(applies_to=(ModelInfo("mfg", "mdl"),)).matches(zigpy_device) + assert DeviceMatch(applies_to=(ModelInfo(None, "mdl"),)).matches(zigpy_device) + assert not DeviceMatch(applies_to=(ModelInfo("mfg", "other"),)).matches( + zigpy_device + ) + assert not DeviceMatch(applies_to=(ModelInfo("other", None),)).matches(zigpy_device) + + +def test_registry_resolve_edge_cases(caplog: pytest.LogCaptureFixture) -> None: + """Test resolving devices that are not matched or whose transform fails.""" + registry = DeviceRegistry() + + # Neither a Zigbee nor a Green Power device + other_device = MagicMock(spec=zigpy.device.BaseDevice) + assert registry.resolve(other_device) is other_device + + # A Zigbee device with no matching entry + zigbee_device = MagicMock(spec=zigpy.device.ZigbeeDevice) + zigbee_device.manufacturer = "mfg" + zigbee_device.model = "mdl" + assert registry.resolve(zigbee_device) is zigbee_device + + # A failing transform falls back to the bare device, untagged + def broken(device: zigpy.device.ZigbeeDevice) -> zigpy.device.ZigbeeDevice: + raise RuntimeError("boom") + + registry.register( + QuirkRegistryEntry( + device_match=DeviceMatch(applies_to=(ModelInfo("mfg", "mdl"),)), + zigpy_transforms=(broken,), + ) + ) + + with caplog.at_level(logging.ERROR): + assert registry.resolve(zigbee_device) is zigbee_device + + assert "Failed to load quirk" in caplog.text + assert not hasattr(zigbee_device, QUIRK_REGISTRY_ENTRY_ATTR) diff --git a/tools/regenerate_mypy_ignores.py b/tools/regenerate_mypy_ignores.py index 9c8b645c6..716dd91be 100644 --- a/tools/regenerate_mypy_ignores.py +++ b/tools/regenerate_mypy_ignores.py @@ -77,8 +77,11 @@ def generate_overrides(errors_by_module: dict[str, dict[str, int]]) -> str: errors_by_module.items(), key=lambda x: (sum(x[1].values()), x[0]) ) + total = sum(sum(codes.values()) for codes in errors_by_module.values()) + header = f"# Total ignored errors: {total} across {len(errors_by_module)} modules" + sections = [format_override(module, codes) for module, codes in sorted_modules] - return "\n\n".join(sections) + return header + "\n" + "\n\n".join(sections) def main() -> None: diff --git a/zha/application/gateway.py b/zha/application/gateway.py index 23c932c0c..187cef567 100644 --- a/zha/application/gateway.py +++ b/zha/application/gateway.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, Callable from contextlib import asynccontextmanager, suppress from dataclasses import dataclass from datetime import timedelta @@ -56,8 +56,20 @@ ) from zha.event import EventBase from zha.exceptions import ZHAException -from zha.quirks import DEVICE_REGISTRY, QUIRK_REGISTRY_ENTRY_ATTR -from zha.zigbee.device import Device, DeviceInfo, DeviceStatus, ExtendedDeviceInfo +from zha.quirks import ( + DEVICE_REGISTRY, + QUIRK_REGISTRY_ENTRY_ATTR, + GreenPowerQuirkRegistryEntry, + QuirkRegistryEntry, +) +from zha.zigbee.device import ( + BaseDevice, + Device, + DeviceInfo, + DeviceStatus, + ExtendedDeviceInfo, + GreenPowerDevice, +) from zha.zigbee.group import Group, GroupInfo, GroupMemberReference BLOCK_LOG_TIMEOUT: Final[int] = 60 @@ -107,7 +119,7 @@ class ExtendedDeviceInfoWithPairingStatus(ExtendedDeviceInfo): class DeviceJoinedDeviceInfo: """Information about a device.""" - ieee: str + ieee: EUI64 nwk: int pairing_status: DevicePairingStatus @@ -186,6 +198,20 @@ class DeviceRemovedEvent: event: Final[str] = ZHA_GW_MSG_DEVICE_REMOVED +def _device_factory(zigpy_device: zigpy.device.BaseDevice) -> Callable[..., BaseDevice]: + """Return the ZHA device class `Device.new` creates for a non-coordinator.""" + entry: QuirkRegistryEntry | GreenPowerQuirkRegistryEntry | None = getattr( + zigpy_device, QUIRK_REGISTRY_ENTRY_ATTR, None + ) + if entry is not None and entry.zha_device_factory: + return entry.zha_device_factory + + if isinstance(zigpy_device, zigpy.device.GreenPowerDevice): + return GreenPowerDevice + + return Device + + class Gateway(AsyncUtilMixin, EventBase): """Gateway that handles events that happen on the ZHA Zigbee network.""" @@ -193,7 +219,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 @@ -212,7 +238,9 @@ def radio_type(self) -> RadioType: """Get the current radio type.""" return RadioType[self.config.config.coordinator_configuration.radio_type] - def get_application_controller_data(self) -> tuple[ControllerApplication, dict]: + def get_application_controller_data( + self, + ) -> tuple[type[ControllerApplication], dict[str, Any]]: """Get an uninitialized instance of a zigpy `ControllerApplication`.""" app_config = self.config.zigpy_config app_config[CONF_DEVICE] = { @@ -274,9 +302,9 @@ async def _async_initialize(self) -> None: await self.application_controller.startup(auto_form=True) - self.coordinator_zha_device = self.get_or_create_device( - self._find_coordinator_device() - ) + coordinator = self.get_or_create_device(self._find_coordinator_device()) + assert isinstance(coordinator, Device) + self.coordinator_zha_device = coordinator await self.load_devices() self.load_groups() @@ -296,7 +324,7 @@ async def async_initialize(self) -> None: @asynccontextmanager async def request_priority( - self, priority: t.PacketPriority | None + self, priority: t.PacketPriority ) -> AsyncGenerator[None, None]: """Context manager to request a priority for radio access.""" async with self.application_controller.request_priority(priority): @@ -307,7 +335,7 @@ def connection_lost(self, exc: Exception) -> None: _LOGGER.debug("Connection to the radio was lost: %r", exc) self.emit(ZHA_GW_MSG_CONNECTION_LOST, ConnectionLostEvent(exception=exc)) - def _find_coordinator_device(self) -> zigpy.device.ZigbeeDevice: + def _find_coordinator_device(self) -> zigpy.device.BaseDevice: zigpy_coordinator = self.application_controller.get_device(nwk=0x0000) if last_backup := self.application_controller.backups.most_recent_backup(): @@ -412,7 +440,7 @@ async def fetch_updated_state() -> None: fetch_updated_state(), "zha.gateway-fetch_updated_state" ) - def device_joined(self, device: zigpy.device.ZigbeeDevice) -> None: + def device_joined(self, device: zigpy.device.BaseDevice) -> None: """Handle device joined. At this point, no information about the device is known other than its @@ -430,9 +458,15 @@ def device_joined(self, device: zigpy.device.ZigbeeDevice) -> None: ), ) - def raw_device_initialized(self, device: zigpy.device.ZigbeeDevice) -> None: # pylint: disable=unused-argument + def raw_device_initialized(self, device: zigpy.device.BaseDevice) -> None: # pylint: disable=unused-argument """Handle a device initialization without quirks loaded.""" + # A GPD has no model or manufacturer strings + if isinstance(device, zigpy.device.ZigbeeDevice): + model, manufacturer = device.model, device.manufacturer + else: + model, manufacturer = None, None + self.emit( ZHA_GW_MSG_RAW_INIT, RawDeviceInitializedEvent( @@ -440,16 +474,14 @@ def raw_device_initialized(self, device: zigpy.device.ZigbeeDevice) -> None: # ieee=device.ieee, nwk=device.nwk, pairing_status=DevicePairingStatus.INTERVIEW_COMPLETE, - model=device.model if device.model else UNKNOWN_MODEL, - manufacturer=device.manufacturer - if device.manufacturer - else UNKNOWN_MANUFACTURER, + model=model if model else UNKNOWN_MODEL, + manufacturer=manufacturer if manufacturer else UNKNOWN_MANUFACTURER, signature=device.get_signature(), ) ), ) - def device_initialized(self, device: zigpy.device.ZigbeeDevice) -> None: + def device_initialized(self, device: zigpy.device.BaseDevice) -> None: """Handle device joined and basic information discovered.""" if device.ieee in self._device_init_tasks: _LOGGER.warning( @@ -471,7 +503,7 @@ def _remove_init_task(task: asyncio.Task) -> None: init_task.add_done_callback(_remove_init_task) - def device_reinterviewed(self, device: zigpy.device.ZigbeeDevice) -> None: + def device_reinterviewed(self, device: zigpy.device.BaseDevice) -> None: """Handle zigpy device_reinterviewed event (e.g. after OTA or reconfigure).""" if device.ieee in self._device_init_tasks: _LOGGER.debug( @@ -494,7 +526,7 @@ def _remove_init_task(task: asyncio.Task) -> None: init_task.add_done_callback(_remove_init_task) async def _async_device_reinterviewed( - self, new_zigpy_device: zigpy.device.ZigbeeDevice + self, new_zigpy_device: zigpy.device.BaseDevice ) -> None: """Rebuild a ZHA device after zigpy swapped the underlying device.""" zha_device = self._devices.get(new_zigpy_device.ieee) @@ -505,19 +537,9 @@ async def _async_device_reinterviewed( ) return - old_entry = getattr(zha_device.device, QUIRK_REGISTRY_ENTRY_ATTR, None) - old_factory = ( - old_entry.zha_device_factory - if old_entry is not None and old_entry.zha_device_factory - else Device - ) - - new_entry = getattr(new_zigpy_device, QUIRK_REGISTRY_ENTRY_ATTR, None) - new_factory = ( - new_entry.zha_device_factory - if new_entry is not None and new_entry.zha_device_factory - else Device - ) + # A re-commissioned GPD is announced like a re-interview + old_factory = _device_factory(zha_device.device) + new_factory = _device_factory(new_zigpy_device) # Only a `Device` swap requires a new object if new_factory is old_factory: @@ -682,7 +704,7 @@ def _emit_group_gateway_message( # pylint: disable=unused-argument ), ) - def device_removed(self, device: zigpy.device.ZigbeeDevice) -> None: + def device_removed(self, device: zigpy.device.BaseDevice) -> None: """Handle device being removed from the network.""" _LOGGER.info("Removing device %s - %s", device.ieee, f"0x{device.nwk:04x}") zha_device = self._devices.pop(device.ieee, None) @@ -699,7 +721,7 @@ def device_removed(self, device: zigpy.device.ZigbeeDevice) -> None: DeviceRemovedEvent(device_info=device_info), ) - def get_device(self, ieee: EUI64) -> Device | None: + def get_device(self, ieee: EUI64) -> BaseDevice | None: """Return Device for given ieee.""" return self._devices.get(ieee) @@ -718,7 +740,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 @@ -727,7 +749,14 @@ def groups(self) -> dict[int, Group]: """Return groups.""" return self._groups - def get_or_create_device(self, zigpy_device: zigpy.device.ZigbeeDevice) -> Device: + def get_zigbee_device(self, ieee: EUI64) -> Device: + """Look up a Zigbee device by IEEE address.""" + device = self._devices[ieee] + if not isinstance(device, Device): + raise ZHAException(f"Device {ieee} is not a Zigbee 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) @@ -755,7 +784,7 @@ def async_update_device( if device.status is DeviceStatus.INITIALIZED: device.update_available(available) - async def async_device_initialized(self, device: zigpy.device.ZigbeeDevice) -> None: + async def async_device_initialized(self, device: zigpy.device.BaseDevice) -> None: """Handle device joined and basic information discovered (async).""" zha_device = self.get_or_create_device(device) _LOGGER.debug( @@ -791,7 +820,7 @@ async def async_device_initialized(self, device: zigpy.device.ZigbeeDevice) -> N DeviceFullInitEvent(device_info=device_info), ) - async def _async_device_joined(self, zha_device: Device) -> None: + async def _async_device_joined(self, zha_device: BaseDevice) -> None: zha_device.available = True zha_device.on_network = True @@ -810,7 +839,7 @@ async def _async_device_joined(self, zha_device: Device) -> None: ), ) - async def _async_device_rejoined(self, zha_device: Device) -> None: + async def _async_device_rejoined(self, zha_device: BaseDevice) -> None: _LOGGER.debug( "skipping discovery for previously discovered device - %s:%s", zha_device.nwk, @@ -835,8 +864,12 @@ async def _async_device_rejoined(self, zha_device: Device) -> None: def _is_groupable_member(self, member: GroupMemberReference) -> bool: """Return whether a group member refers to an endpoint that can join a group.""" - if (device := self.devices.get(member.ieee)) is None: + device = self.devices.get(member.ieee) + + # Only Zigbee devices can be group members + if not isinstance(device, Device): return False + # the endpoints with a `Groups` cluster to join, which is also what Home # Assistant offers in its "Add group" page (not a coroutine, despite the # name) @@ -913,8 +946,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 ) ) @@ -939,6 +974,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 c754481e5..e5e577e61 100644 --- a/zha/application/helpers.py +++ b/zha/application/helpers.py @@ -42,7 +42,7 @@ if TYPE_CHECKING: from zha.application.gateway import Gateway - from zha.zigbee.device import Device + from zha.zigbee.device import BaseDevice _T = TypeVar("_T") _R = TypeVar("_R") @@ -125,7 +125,7 @@ async def write_attributes_safe( async def get_matched_clusters( - source_zha_device: Device, target_zha_device: Device + source_zha_device: BaseDevice, target_zha_device: BaseDevice ) -> list[BindingPair]: """Get matched input/output cluster pairs for 2 devices.""" source_clusters = source_zha_device.async_get_std_clusters() @@ -209,7 +209,7 @@ def convert_zcl_value(value: Any, field_type: Any) -> Any: def convert_to_zcl_values( - fields: dict[str, Any], schema: CommandSchema + fields: dict[str, Any], schema: type[CommandSchema] ) -> dict[str, Any]: """Convert user input to ZCL values.""" converted_fields: dict[str, Any] = {} @@ -230,12 +230,19 @@ def convert_to_zcl_values( return converted_fields -def async_is_bindable_target(source_zha_device: Device, target_zha_device: Device): +def async_is_bindable_target( + source_zha_device: BaseDevice, target_zha_device: BaseDevice +) -> bool: """Determine if target is bindable to source.""" + source_clusters = source_zha_device.async_get_std_clusters() + + # A source without clusters, such as a Green Power device, has nothing to bind + if not source_clusters: + return False + if target_zha_device.nwk == 0x0000: return True - source_clusters = source_zha_device.async_get_std_clusters() target_clusters = target_zha_device.async_get_std_clusters() for endpoint_id in source_clusters: @@ -448,7 +455,7 @@ def __init__(self, gateway: Gateway): self._update_listeners: list[Callable] = [] self._gateway: Gateway = gateway - def start(self): + def start(self) -> None: """Start the global updater.""" self._updater_task_handle = self._gateway.async_create_background_task( self.update_listeners(), @@ -461,7 +468,7 @@ def start(self): getattr(self, "__polling_interval"), ) - def stop(self): + def stop(self) -> None: """Stop the global updater.""" _LOGGER.debug("stopping global updater") if self._updater_task_handle: @@ -513,7 +520,7 @@ def __init__(self, gateway: Gateway): self._gateway: Gateway = gateway self._device_availability_task_handle: asyncio.Task | None = None - def start(self): + def start(self) -> None: """Start the device availability checker.""" self._device_availability_task_handle = ( self._gateway.async_create_background_task( @@ -528,7 +535,7 @@ def start(self): getattr(self, "__polling_interval"), ) - def stop(self): + def stop(self) -> None: """Stop the device availability checker.""" _LOGGER.debug("stopping device availability checker") if self._device_availability_task_handle: diff --git a/zha/application/platforms/__init__.py b/zha/application/platforms/__init__.py index ed65e1991..b36bfd1d5 100644 --- a/zha/application/platforms/__init__.py +++ b/zha/application/platforms/__init__.py @@ -5,7 +5,7 @@ from abc import abstractmethod import asyncio from collections import defaultdict -from collections.abc import Callable, Mapping +from collections.abc import Callable, Collection, Mapping from contextlib import suppress import dataclasses from dataclasses import dataclass, field @@ -31,7 +31,7 @@ from zha.mixins import LogMixin if TYPE_CHECKING: - from zha.zigbee.device import Device + from zha.zigbee.device import BaseDevice, Device from zha.zigbee.endpoint import Endpoint from zha.zigbee.group import Group @@ -327,7 +327,7 @@ def _is_supported(self) -> bool: """Return if the entity is supported for the device, internal.""" return True - def is_supported_in_list(self, entities: list[BaseEntity]) -> bool: + def is_supported_in_list(self, entities: Collection[BaseEntity]) -> bool: """Return if the entity is supported given all other entities.""" return True @@ -529,7 +529,7 @@ class PlatformEntity(BaseEntity): def __init__( self, - device: Device, + device: BaseDevice, *, unique_id: str, from_quirk: bool = False, @@ -564,7 +564,7 @@ def __init__( super().__init__(unique_id=unique_id, **kwargs) - self._device: Device = device + self._device: BaseDevice = device def _apply_quirk_entity_config( self, @@ -607,7 +607,7 @@ def _apply_quirk_entity_config( self._attr_primary = primary @property - def device(self) -> Device: + def device(self) -> BaseDevice: """Return the device.""" return self._device @@ -666,6 +666,8 @@ class ZclPlatformEntity(PlatformEntity): _client_cluster_config: Mapping[int, ClusterConfig] = MappingProxyType({}) + _device: Device + def __init__( self, endpoint: Endpoint, @@ -698,6 +700,11 @@ def __init__( **kwargs, ) + @property + def device(self) -> Device: + """Return the device.""" + return self._device + @property def endpoint(self) -> Endpoint: """Return the endpoint.""" diff --git a/zha/application/platforms/button/__init__.py b/zha/application/platforms/button/__init__.py index 86fd73be9..947d3b00c 100644 --- a/zha/application/platforms/button/__init__.py +++ b/zha/application/platforms/button/__init__.py @@ -3,6 +3,7 @@ from __future__ import annotations from abc import ABC, abstractmethod +from collections.abc import Collection from dataclasses import dataclass import functools import logging @@ -141,7 +142,7 @@ class IdentifyButton(Button): server_clusters=frozenset({Identify.cluster_id}), ) - def is_supported_in_list(self, entities: list[BaseEntity]) -> bool: + def is_supported_in_list(self, entities: Collection[BaseEntity]) -> bool: """Check if this button is supported given the list of entities.""" cls = type(self) return not any(type(entity) is cls for entity in entities if entity is not self) diff --git a/zha/application/platforms/sensor/__init__.py b/zha/application/platforms/sensor/__init__.py index c42b9307a..adc9eed48 100644 --- a/zha/application/platforms/sensor/__init__.py +++ b/zha/application/platforms/sensor/__init__.py @@ -3188,6 +3188,13 @@ class RSSISensor(BaseSensor): _attr_entity_registry_enabled_default = False _attr_translation_key: str = "rssi" + _device: Device + + @property + def device(self) -> Device: + """Return the device.""" + return self._device + @staticmethod def legacy_basic_cluster(device: Device) -> Cluster | None: """Return the Basic cluster the legacy RSSI/LQI unique IDs were derived from.""" diff --git a/zha/quirks.py b/zha/quirks.py index 90b996bd6..2f2ffa028 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.ZigbeeDevice], bool] +GreenPowerFilterType = Callable[[zigpy.device.GreenPowerDevice], bool] + +_EntryT = TypeVar("_EntryT", "QuirkRegistryEntry", "GreenPowerQuirkRegistryEntry") DEVICE_REGISTRY: DeviceRegistry @@ -56,7 +59,8 @@ def _read_current_firmware_version( ) except ValueError: return None - return ota.get(Ota.AttributeDefs.current_file_version.id) + version: int | None = ota.get(Ota.AttributeDefs.current_file_version.id) + return version @dataclass(frozen=True) @@ -157,8 +161,85 @@ 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) + + @property + def specificity(self) -> int: + """Rank by identity criterion first, then by the number of fields specified.""" + has_identity = bool(self.src_id_ranges) or bool(self.ieee_prefixes) + num_fields = sum( + [ + self.device_id is not None, + self.manufacturer_id is not None, + self.model_id is not None, + ] + ) + + return 10 * has_identity + num_fields + + +@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 and ranked by specificity. The most recently + registered entry wins among equally specific matches. + """ def __init__(self) -> None: """Initialize the registry.""" @@ -170,9 +251,30 @@ 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, most specific first, then most recently registered. + self._gp_registry: list[GreenPowerQuirkRegistryEntry] = [] + + @overload + def register(self, entry: QuirkRegistryEntry) -> QuirkRegistryEntry: ... + + @overload + def register( + self, entry: GreenPowerQuirkRegistryEntry + ) -> GreenPowerQuirkRegistryEntry: ... - def register(self, entry: QuirkRegistryEntry) -> QuirkRegistryEntry: + 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) + # The sort is stable, so the new entry stays ahead of equal ones + self._gp_registry.sort( + key=lambda entry: entry.device_match.specificity, reverse=True + ) + return entry + if not entry.device_match.applies_to: if entry not in self._wildcard_registry: self._wildcard_registry.insert(0, entry) @@ -216,7 +318,7 @@ def register_device(self, cls: type[Device]) -> type[Device]: def match_entry( self, zigpy_device: zigpy.device.ZigbeeDevice ) -> QuirkRegistryEntry | None: - """Return the first registered entry matching `zigpy_device`.""" + """Return the most recently registered entry matching `zigpy_device`.""" for key in ( ModelInfo(zigpy_device.manufacturer, zigpy_device.model), ModelInfo(zigpy_device.manufacturer, None), @@ -232,32 +334,53 @@ def match_entry( return None - def resolve( - self, zigpy_device: zigpy.device.ZigbeeDevice - ) -> zigpy.device.ZigbeeDevice: + def match_green_power_entry( + self, zigpy_device: zigpy.device.GreenPowerDevice + ) -> GreenPowerQuirkRegistryEntry | None: + """Return the most specific 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.""" # Resolution is idempotent: an already-quirked device is returned as-is if hasattr(zigpy_device, QUIRK_REGISTRY_ENTRY_ATTR): return zigpy_device - entry = self.match_entry(zigpy_device) - if entry is None: - return zigpy_device + if isinstance(zigpy_device, zigpy.device.GreenPowerDevice): + gp_entry = self.match_green_power_entry(zigpy_device) + if gp_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, - ) + return self._apply_entry(zigpy_device, gp_entry, gp_entry.zigpy_transforms) + + if isinstance(zigpy_device, zigpy.device.ZigbeeDevice): + entry = self.match_entry(zigpy_device) + if entry is None: + return zigpy_device + + return self._apply_entry(zigpy_device, entry, entry.zigpy_transforms) + + return zigpy_device + + @staticmethod + def _apply_entry[DeviceT: zigpy.device.BaseDevice]( + zigpy_device: DeviceT, + entry: QuirkRegistryEntry | GreenPowerQuirkRegistryEntry, + transforms: tuple[Callable[[DeviceT], DeviceT], ...], + ) -> DeviceT: + """Apply a matched entry's transforms and tag the result with the 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. resolved_device = zigpy_device try: - for transform in entry.zigpy_transforms: + for transform in transforms: resolved_device = transform(resolved_device) except Exception: _LOGGER.exception("Failed to load quirk for %s", zigpy_device) @@ -267,7 +390,9 @@ def resolve( 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): @@ -276,8 +401,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 @@ -285,29 +416,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() diff --git a/zha/zigbee/device.py b/zha/zigbee/device.py index 40b50cef0..8b5bfb727 100644 --- a/zha/zigbee/device.py +++ b/zha/zigbee/device.py @@ -6,7 +6,7 @@ import asyncio from collections import defaultdict -from collections.abc import Callable, Iterable, Iterator, Sequence +from collections.abc import Callable, Collection, Iterable, Iterator import contextlib import copy import dataclasses @@ -17,7 +17,11 @@ import time from typing import TYPE_CHECKING, Any, Final -from zigpy.device import ZigbeeDevice +from zigpy.device import ( + BaseDevice as ZigpyBaseDevice, + GreenPowerDevice as ZigpyGreenPowerDevice, + ZigbeeDevice as ZigpyDevice, +) import zigpy.exceptions from zigpy.profiles import PROFILES from zigpy.types import uint1_t, uint8_t, uint16_t @@ -28,17 +32,11 @@ from zigpy.zcl.clusters.general import Basic, Groups, Identify, Ota from zigpy.zcl.foundation import ( Status as ZclStatus, - WriteAttributesResponse, + WriteAttributesStatusRecord, + ZCLAttributeDef, ZCLCommandDef, ) import zigpy.zdo.types as zdo_types -from zigpy.zdo.types import ( - DeviceType, - PermitJoins, - Relationship, - RouteStatus, - RxOnWhenIdle, -) from zha.application import Platform, discovery from zha.application.const import ( @@ -91,6 +89,8 @@ from zha.quirks import ( QUIRK_REGISTRY_ENTRY_ATTR, DeviceMatch, + GreenPowerQuirkRegistryEntry, + QuirkRegistryEntry, ReplacingZigpyDeviceFactory, ) from zha.zigbee.cluster_config import ( @@ -148,7 +148,7 @@ def _cluster_entry(cluster_id: int, cluster: Cluster) -> dict[str, Any]: def get_device_automation_triggers( - device: zigpy.device.ZigbeeDevice, + device: zigpy.device.BaseDevice, ) -> dict[tuple[str, str], dict[str, str]]: """Get the supported device automation triggers for a zigpy device.""" return { @@ -277,8 +277,8 @@ class DeviceInfo: exposes_features: set[str] manufacturer_code: int | None power_source: str - lqi: int - rssi: int + lqi: int | None + rssi: int | None last_seen: str available: bool device_type: str @@ -289,13 +289,13 @@ class DeviceInfo: class NeighborInfo: """Describes a neighbor.""" - device_type: DeviceType - rx_on_when_idle: RxOnWhenIdle - relationship: Relationship + device_type: str + rx_on_when_idle: str + relationship: str extended_pan_id: ExtendedPanId ieee: EUI64 nwk: NWK - permit_joining: PermitJoins + permit_joining: str depth: uint8_t lqi: uint8_t @@ -305,7 +305,7 @@ class RouteInfo: """Describes a route.""" dest_nwk: NWK - route_status: RouteStatus + route_status: str memory_constrained: uint1_t many_to_one: uint1_t route_record_required: uint1_t @@ -330,38 +330,25 @@ 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.ZigbeeDevice], zigpy.device.ZigbeeDevice], ... - ] = () +class BaseDevice[ZigpyDeviceT: ZigpyBaseDevice](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", + "is_coordinator", ) def __init__( self, - zigpy_device: zigpy.device.ZigbeeDevice, + zigpy_device: ZigpyDeviceT, _gateway: Gateway, ) -> None: """Initialize the gateway.""" @@ -381,7 +368,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 @@ -389,7 +375,7 @@ def __init__( self._init_from_zigpy_device(zigpy_device) - def _init_from_zigpy_device(self, zigpy_device: zigpy.device.ZigbeeDevice) -> None: + def _init_from_zigpy_device(self, zigpy_device: ZigpyDeviceT) -> None: """(Re-)initialize device state from a zigpy device. Sets up the zigpy device reference, quirk metadata, cluster handlers, @@ -401,11 +387,10 @@ def _init_from_zigpy_device(self, zigpy_device: zigpy.device.ZigbeeDevice) -> No # 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: ZigbeeDevice = zigpy_device + self._zigpy_device: ZigpyDeviceT = zigpy_device # Invalidate cached properties that depend on the zigpy device before # they are read below (e.g. is_mains_powered, is_coordinator). @@ -447,12 +432,6 @@ def _init_from_zigpy_device(self, zigpy_device: zigpy.device.ZigbeeDevice) -> No 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 ( @@ -462,8 +441,33 @@ 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 ( + isinstance(zigpy_dev, ZigpyDevice) + and zigpy_dev.ieee == gateway.state.node_info.ieee + ): + return CoordinatorDevice(zigpy_dev, gateway) + + entry: QuirkRegistryEntry | GreenPowerQuirkRegistryEntry | None = 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) + + assert isinstance(zigpy_dev, ZigpyDevice) + return ZigbeeDevice(zigpy_dev, gateway) + @property - def device(self) -> zigpy.device.ZigbeeDevice: + def device(self) -> ZigpyDeviceT: """Return underlying Zigpy device.""" return self._zigpy_device @@ -517,17 +521,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: @@ -535,30 +530,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: @@ -566,12 +549,12 @@ def nwk(self) -> NWK: return self._zigpy_device.nwk @property - def lqi(self): + def lqi(self) -> int | None: """Return lqi for device.""" return self._zigpy_device.lqi @property - def rssi(self): + def rssi(self) -> int | None: """Return rssi for device.""" return self._zigpy_device.rssi @@ -580,21 +563,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: @@ -603,54 +580,32 @@ 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 + return False @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 + return False @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 + + def async_get_std_clusters(self) -> dict[int, dict[str, dict[int, Cluster]]]: + """Get ZHA and ZLL clusters for this device.""" + return {} @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): + def gateway(self) -> Gateway: """Return the gateway for this device.""" return self._gateway @@ -671,7 +626,7 @@ def device_automation_triggers(self) -> dict[tuple[str, str], dict[str, str]]: return triggers @property - def available(self): + def available(self) -> bool: """Return True if device is available.""" return self.is_active_coordinator or (self._available and self.on_network) @@ -681,7 +636,7 @@ def available(self, new_availability: bool) -> None: self._available = new_availability @property - def on_network(self): + def on_network(self) -> bool: """Return True if device is currently on the network.""" return self.is_active_coordinator or self._on_network @@ -693,45 +648,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: @@ -788,23 +708,7 @@ def get_entity( ) return matches[0] - @classmethod - def new( - cls, - zigpy_dev: zigpy.device.ZigbeeDevice, - 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: + def async_update_firmware_version(self, firmware_version: str | None) -> None: """Update device firmware version.""" if firmware_version == self._firmware_version: return @@ -821,54 +725,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.""" @@ -944,103 +801,19 @@ 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, - ) + raise NotImplementedError 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() + raise NotImplementedError - 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.ZigbeeDevice - ) -> None: + async def async_rebuild_from_zigpy_device(self, zigpy_device: ZigpyDeviceT) -> None: """Tear down and rebuild this device from a new zigpy device. Called by the gateway after a successful re-interview swaps the @@ -1062,35 +835,12 @@ def emit_reconfigure_done(self) -> None: ) def discover_entities(self) -> Iterator[PlatformEntity]: - """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 - - # TODO: detach these entities from the `Basic` cluster once we finish - # implementing unique ID migrations in the Core integration - if (basic_cluster := sensor.RSSISensor.legacy_basic_cluster(self)) is not None: - unique_id_base = ( - f"{self.ieee}-{basic_cluster.endpoint.endpoint_id}-{Basic.cluster_id}" - ) - - yield sensor.RSSISensor(self, unique_id=unique_id_base) - yield sensor.LQISensor(self, unique_id=unique_id_base) - - 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() @@ -1245,93 +995,494 @@ async def async_initialize(self, from_cache: bool = False) -> None: """Initialize cluster handlers.""" self.debug("started initialization") - # We discover prospective entities before 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( + Platform(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: Collection[PlatformEntity]) -> None: + """Compute the primary entity from a given set of entities.""" + self._primary_entity = None + + # First, check if any entity is explicitly primary + explicitly_primary = [entity for entity in entities if entity._attr_primary] + + if len(explicitly_primary) == 1: + self.debug( + "Device has a single explicitly primary entity," + " not performing weight matching" + ) + self._primary_entity = explicitly_primary[0] + 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. Entities disabled at runtime + # (via the entity registry in HA) deliberately stay candidates: the primary + # entity describes the main feature of the device, which does not change when + # its entity is disabled. + candidates = [ + e for e in entities if 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: + self._primary_entity = winner + return + + self.debug( + "Primary entity tie between %s and %s, no primary entity", winner, others[0] + ) + + +class ZigbeeDevice(BaseDevice[ZigpyDevice]): + """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.ZigbeeDevice], zigpy.device.ZigbeeDevice], ... + ] = () + + _ZIGPY_CACHED_PROPERTIES = ( + *BaseDevice._ZIGPY_CACHED_PROPERTIES, + "manufacturer_code", + "is_mains_powered", + "device_type", + "is_router", + "is_end_device", + "zigbee_signature", + ) + + 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 endpoint in zigpy_device.non_zdo_endpoints: + ep = Endpoint.new(endpoint, self) + self._endpoints[endpoint.endpoint_id] = ep + self._on_remove_callbacks.append(ep.on_remove) + + 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 in self._zigpy_device.non_zdo_endpoints: + cluster: zigpy.zcl.Cluster | None = 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 + + 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" + ) + ) + ) + + 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() - # 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. + def discover_entities(self) -> Iterator[PlatformEntity]: + """Yield the default (ZCL) entities for this device. + Declarative quirks add their exposed entities by overriding this in + zhaquirks' `QuirkV2Device`; hand-written quirks override it directly. """ - 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, - ) + # TODO: purge old coordinator entities + if self.is_coordinator: + return - 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, - ) + # TODO: detach these entities from the `Basic` cluster once we finish + # implementing unique ID migrations in the Core integration + if (basic_cluster := sensor.RSSISensor.legacy_basic_cluster(self)) is not None: + unique_id_base = ( + f"{self.ieee}-{basic_cluster.endpoint.endpoint_id}-{Basic.cluster_id}" + ) - 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, - ) + yield sensor.RSSISensor(self, unique_id=unique_id_base) + yield sensor.LQISensor(self, unique_id=unique_id_base) - # 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.""" @@ -1344,7 +1495,7 @@ def async_get_clusters(self) -> dict[int, dict[str, dict[int, Cluster]]]: if ep_id != 0 } - def async_get_groupable_endpoints(self): + def async_get_groupable_endpoints(self) -> list[int]: """Get device endpoints that have a group 'in' cluster.""" return [ ep_id @@ -1352,7 +1503,7 @@ def async_get_groupable_endpoints(self): if Groups.cluster_id in clusters[CLUSTER_TYPE_IN] ] - def async_get_std_clusters(self): + def async_get_std_clusters(self) -> dict[int, dict[str, dict[int, Cluster]]]: """Get ZHA and ZLL clusters for this device.""" return { @@ -1372,14 +1523,14 @@ def async_get_cluster( return clusters[endpoint_id][cluster_type][cluster_id] def async_get_cluster_attributes( - self, endpoint_id, cluster_id, cluster_type=CLUSTER_TYPE_IN - ): + self, endpoint_id: int, cluster_id: int, cluster_type: str = CLUSTER_TYPE_IN + ) -> dict[int, ZCLAttributeDef]: """Get zigbee attributes for specified cluster.""" return self.async_get_cluster(endpoint_id, cluster_id, cluster_type).attributes def async_get_cluster_commands( - self, endpoint_id, cluster_id, cluster_type=CLUSTER_TYPE_IN - ): + self, endpoint_id: int, cluster_id: int, cluster_type: str = CLUSTER_TYPE_IN + ) -> dict[str, dict[int, ZCLCommandDef]]: """Get zigbee commands for specified cluster.""" cluster = self.async_get_cluster(endpoint_id, cluster_id, cluster_type) return { @@ -1395,7 +1546,7 @@ async def write_zigbee_attribute( value: Any, cluster_type: str = CLUSTER_TYPE_IN, manufacturer: int | UndefinedType | None = UNDEFINED, - ) -> WriteAttributesResponse | None: + ) -> list[list[WriteAttributesStatusRecord]]: """Write a value to a zigbee attribute for a cluster in this entity.""" try: cluster: Cluster = self.async_get_cluster( @@ -1634,56 +1785,7 @@ 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.""" - self._primary_entity = None - - # First, check if any entity is explicitly primary - explicitly_primary = [entity for entity in entities if entity._attr_primary] - - if len(explicitly_primary) == 1: - self.debug( - "Device has a single explicitly primary entity," - " not performing weight matching" - ) - self._primary_entity = explicitly_primary[0] - 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. Entities disabled at runtime - # (via the entity registry in HA) deliberately stay candidates: the primary - # entity describes the main feature of the device, which does not change when - # its entity is disabled. - candidates = [ - e for e in entities if 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: - self._primary_entity = winner - return - - self.debug( - "Primary entity tie between %s and %s, no primary entity", winner, others[0] - ) - - def get_diagnostics_json(self): + def get_diagnostics_json(self) -> dict[str, Any]: """Get ZHA device information.""" info: dict[str, Any] = {} @@ -1702,7 +1804,11 @@ def get_diagnostics_json(self): info["power_source"] = self.power_source info["lqi"] = self.lqi info["rssi"] = self.rssi - info["last_seen"] = self.device._last_seen.isoformat() + info["last_seen"] = ( + self.device._last_seen.isoformat() + if self.device._last_seen is not None + else None + ) info["available"] = self.available info["device_type"] = self.device_type info["active_coordinator"] = self.is_active_coordinator @@ -1763,7 +1869,7 @@ def get_diagnostics_json(self): } } if isinstance(cluster, Ota) - and getattr(cluster, "last_query_cmd", None) is not None + and cluster.last_query_cmd is not None else {} ), } @@ -1845,7 +1951,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[PlatformEntity]: @@ -1872,3 +1978,127 @@ def discover_entities(self) -> Iterator[PlatformEntity]: sensor.DeviceCounterSensor.__name__, f"counter groups[{counter_groups}] counter group[{counter_group}] counter[{counter}]", ) + + +class GreenPowerDevice(BaseDevice[ZigpyGreenPowerDevice]): + """ZHA Green Power device object.""" + + 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 + + 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.""" + + async def async_configure(self) -> None: + """Configure the device: a GPD has nothing to bind or report.""" + self._discover_new_entities() + self.emit_reconfigure_done() + + def discover_entities(self) -> Iterator[PlatformEntity]: + """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.device._last_seen.isoformat() + if self.device._last_seen is not None + else None + ) + info["available"] = self.available + info["device_type"] = self.device_type + info["active_coordinator"] = self.is_active_coordinator + 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 01917fd54..79f2958f0 100644 --- a/zha/zigbee/group.py +++ b/zha/zigbee/group.py @@ -137,7 +137,7 @@ async def async_remove_from_group(self) -> None: str(ex), ) - def log(self, level: int, msg: str, *args: Any, **kwargs) -> None: + def log(self, level: int, msg: str, *args: Any, **kwargs: Any) -> None: """Log a message.""" msg = f"[%s](%s): {msg}" args = (f"0x{self._group.group_id:04x}", self.endpoint_id) + args @@ -169,7 +169,7 @@ def group_id(self) -> int: return self._zigpy_group.group_id @property - def endpoint(self) -> zigpy.endpoint.Endpoint: + def endpoint(self) -> GroupEndpoint: """Return the endpoint for this group.""" return self._zigpy_group.endpoint @@ -192,7 +192,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 ] @@ -236,7 +236,9 @@ def unregister_group_entity(self, group_entity: GroupEntity) -> None: self._group_entities.pop(group_entity.unique_id) self._entity_unsubs.pop(group_entity.unique_id)() - def _handle_maybe_update_group_members(self, event: EntityStateChangedEvent): + def _handle_maybe_update_group_members( + self, event: EntityStateChangedEvent + ) -> None: """Handle the maybe update group members event.""" self.gateway.async_create_task(self._maybe_update_group_members(event)) @@ -289,40 +291,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]: @@ -337,7 +339,7 @@ def get_platform_entities(self, platform: str) -> list[PlatformEntity]: return platform_entities - def log(self, level: int, msg: str, *args: Any, **kwargs) -> None: + def log(self, level: int, msg: str, *args: Any, **kwargs: Any) -> None: """Log a message.""" msg = f"[%s](%s): {msg}" args = (self.name, self.group_id) + args