Base ZGP device support

tests/test_green_power_device.py

@@ -0,0 +1,211 @@
+"""Tests for ZHA Green Power devices."""
+
+from __future__ import annotations
+
+from zigpy.device import GreenPowerDevice as ZigpyGreenPowerDevice
+from zigpy.types import EUI64
+from zigpy.zgp.types import ApplicationID, DeviceID, GPDCommandID, SrcID
+
+from zha.application.const import UNKNOWN_MANUFACTURER, UNKNOWN_MODEL
+from zha.application.gateway import Gateway
+from zha.quirks import (
+    DEVICE_REGISTRY,
+    QUIRK_REGISTRY_ENTRY_ATTR,
+    GreenPowerDeviceMatch,
+    GreenPowerQuirkRegistryEntry,
+)
+from zha.zigbee.device import GreenPowerDevice
+
+
+def make_zigpy_gpd(zha_gateway: Gateway) -> ZigpyGreenPowerDevice:
+    """Create a commissioned zigpy GPD on the test application."""
+    zigpy_gpd = ZigpyGreenPowerDevice(
+        zha_gateway.application_controller,
+        application_id=ApplicationID.SrcID,
+        src_id=SrcID(0x12345678),
+    )
+    zigpy_gpd.device_id = DeviceID.OnOffSwitch
+    zigpy_gpd.commands = [
+        GPDCommandID.Toggle,
+        GPDCommandID.Press1of1,
+        GPDCommandID.Release1of1,
+    ]
+
+    zha_gateway.application_controller.devices[zigpy_gpd.ieee] = zigpy_gpd
+    return zigpy_gpd
+
+
+async def test_green_power_device_creation(zha_gateway: Gateway) -> None:
+    """Test that a zigpy GPD is wrapped in the ZHA Green Power device class."""
+    zigpy_gpd = make_zigpy_gpd(zha_gateway)
+    zha_device = zha_gateway.get_or_create_device(zigpy_gpd)
+
+    assert isinstance(zha_device, GreenPowerDevice)
+    assert zha_device.device is zigpy_gpd
+    assert zha_gateway.devices[zigpy_gpd.ieee] is zha_device
+
+
+async def test_green_power_device_properties(zha_gateway: Gateway) -> None:
+    """Test the Green Power device property surface."""
+    zigpy_gpd = make_zigpy_gpd(zha_gateway)
+    zha_device = zha_gateway.get_or_create_device(zigpy_gpd)
+
+    assert zha_device.available is True
+    assert zha_device.is_mains_powered is False
+    assert zha_device.is_active_coordinator is False
+    assert zha_device.device_type == "GreenPower"
+    assert zha_device.manufacturer == UNKNOWN_MANUFACTURER
+    assert zha_device.model == UNKNOWN_MODEL
+    assert zha_device.manufacturer_code is None
+    assert zha_device.signature["device_id"] == DeviceID.OnOffSwitch
+    assert zha_device.signature["src_id"] == 0x12345678
+    assert zha_device.signature["commands"] == [
+        GPDCommandID.Toggle,
+        GPDCommandID.Press1of1,
+        GPDCommandID.Release1of1,
+    ]
+
+    # A GPD has no heartbeat: the availability check never flips it
+    await zha_device._check_available()
+    assert zha_device.available is True
+
+
+async def test_green_power_device_identifiers(zha_gateway: Gateway) -> None:
+    """Test manufacturer and model resolution from commissioning identifiers."""
+    zigpy_gpd = make_zigpy_gpd(zha_gateway)
+    zigpy_gpd.gpd_manufacturer_id = 0x1234
+    zigpy_gpd.gpd_model_id = 0x0007
+
+    zha_device = zha_gateway.get_or_create_device(zigpy_gpd)
+
+    assert zha_device.manufacturer == "0x1234"
+    assert zha_device.model == "0x0007"
+    assert zha_device.manufacturer_code == 0x1234
+    assert zha_device.name == "0x1234 0x0007"
+
+
+async def test_green_power_device_info(zha_gateway: Gateway) -> None:
+    """Test device info, extended device info, and diagnostics."""
+    zigpy_gpd = make_zigpy_gpd(zha_gateway)
+    zha_device = zha_gateway.get_or_create_device(zigpy_gpd)
+
+    device_info = zha_device.device_info
+    assert device_info.ieee == zigpy_gpd.ieee
+    assert device_info.device_type == "GreenPower"
+    assert device_info.signature["device_id"] == 0x02
+
+    extended_info = zha_device.extended_device_info
+    assert extended_info.active_coordinator is False
+    assert extended_info.entities == {}
+    assert extended_info.neighbors == []
+    assert extended_info.routes == []
+    assert extended_info.endpoint_names == []
+
+    diagnostics = zha_device.get_diagnostics_json()
+    assert diagnostics["ieee"] == str(zigpy_gpd.ieee)
+    assert diagnostics["device_type"] == "GreenPower"
+    assert diagnostics["signature"]["device_id"] == 0x02
+
+
+async def test_green_power_device_quirk_passthrough(zha_gateway: Gateway) -> None:
+    """Test that the quirk registry passes GP devices through untouched."""
+    zigpy_gpd = make_zigpy_gpd(zha_gateway)
+    assert DEVICE_REGISTRY.resolve(zigpy_gpd) is zigpy_gpd
+
+
+async def test_green_power_device_initialize(zha_gateway: Gateway) -> None:
+    """Test that a Green Power device initializes without entities."""
+    zigpy_gpd = make_zigpy_gpd(zha_gateway)
+    zha_device = zha_gateway.get_or_create_device(zigpy_gpd)
+
+    await zha_device.async_initialize(from_cache=True)
+    assert zha_device.platform_entities == {}
+
+
+async def test_green_power_device_match(zha_gateway: Gateway) -> None:
+    """Test the Green Power quirk matching criteria."""
+    zigpy_gpd = make_zigpy_gpd(zha_gateway)
+    zigpy_gpd.gpd_manufacturer_id = 0x1234
+
+    assert GreenPowerDeviceMatch().matches(zigpy_gpd)
+    assert GreenPowerDeviceMatch(device_id=DeviceID.OnOffSwitch).matches(zigpy_gpd)
+    assert not GreenPowerDeviceMatch(device_id=DeviceID.GenericSwitch).matches(
+        zigpy_gpd
+    )
+    assert GreenPowerDeviceMatch(manufacturer_id=0x1234).matches(zigpy_gpd)
+    assert not GreenPowerDeviceMatch(manufacturer_id=0x5678).matches(zigpy_gpd)
+    assert not GreenPowerDeviceMatch(model_id=0x0001).matches(zigpy_gpd)
+
+    assert GreenPowerDeviceMatch(src_id_ranges=((0x12000000, 0x12FFFFFF),)).matches(
+        zigpy_gpd
+    )
+    assert not GreenPowerDeviceMatch(src_id_ranges=((0x00, 0xFF),)).matches(zigpy_gpd)
+
+    # The synthetic IEEE of a SrcID-addressed GPD carries no vendor prefix
+    assert not GreenPowerDeviceMatch(ieee_prefixes=(bytes([0x04, 0xCD]),)).matches(
+        zigpy_gpd
+    )
+
+    assert GreenPowerDeviceMatch(
+        filters=(lambda device: GPDCommandID.Toggle in device.commands,)
+    ).matches(zigpy_gpd)
+    assert not GreenPowerDeviceMatch(
+        filters=(lambda device: GPDCommandID.Off in device.commands,)
+    ).matches(zigpy_gpd)
+
+
+async def test_green_power_device_match_ieee(zha_gateway: Gateway) -> None:
+    """Test matching an IEEE-addressed GPD by address prefix."""
+    zigpy_gpd = ZigpyGreenPowerDevice(
+        zha_gateway.application_controller,
+        application_id=ApplicationID.IEEE,
+        ieee=EUI64.convert("04:cd:15:00:11:22:33:44"),
+        endpoint=1,
+    )
+
+    assert GreenPowerDeviceMatch(ieee_prefixes=(bytes([0x04, 0xCD, 0x15]),)).matches(
+        zigpy_gpd
+    )
+    assert not GreenPowerDeviceMatch(
+        ieee_prefixes=(bytes([0x04, 0xCD, 0x16]),)
+    ).matches(zigpy_gpd)
+
+    # Either identity criterion is sufficient when both are declared
+    assert GreenPowerDeviceMatch(
+        src_id_ranges=((0x00, 0xFF),),
+        ieee_prefixes=(bytes([0x04, 0xCD, 0x15]),),
+    ).matches(zigpy_gpd)
+
+
+async def test_green_power_quirk_resolution(zha_gateway: Gateway) -> None:
+    """Test that a registered Green Power quirk resolves and builds the ZHA device."""
+    zigpy_gpd = make_zigpy_gpd(zha_gateway)
+
+    class QuirkedGreenPowerDevice(GreenPowerDevice):
+        """Quirk-supplied device class."""
+
+    generic_entry = GreenPowerQuirkRegistryEntry(
+        device_match=GreenPowerDeviceMatch(device_id=DeviceID.OnOffSwitch),
+    )
+    entry = GreenPowerQuirkRegistryEntry(
+        device_match=GreenPowerDeviceMatch(
+            device_id=DeviceID.OnOffSwitch,
+            src_id_ranges=((0x12000000, 0x12FFFFFF),),
+        ),
+        zha_device_factory=QuirkedGreenPowerDevice,
+    )
+
+    with DEVICE_REGISTRY.preserve_state():
+        DEVICE_REGISTRY.register(generic_entry)
+        DEVICE_REGISTRY.register(entry)
+
+        # The most recently registered matching entry wins
+        assert DEVICE_REGISTRY.match_green_power_entry(zigpy_gpd) is entry
+
+        resolved = DEVICE_REGISTRY.resolve(zigpy_gpd)
+        assert resolved is zigpy_gpd
+        assert getattr(resolved, QUIRK_REGISTRY_ENTRY_ATTR) is entry
+
+        zha_device = zha_gateway.get_or_create_device(resolved)
+        assert isinstance(zha_device, QuirkedGreenPowerDevice)
+        assert zha_device.quirk_applied

tests/test_platform_event.py

@@ -16,7 +16,7 @@
 )
 from zha.application import Platform
 from zha.application.gateway import Gateway
-from zha.application.platforms import EntityStateChangedEvent
+from zha.application.platforms import EntityStateChangedEvent, ZclPlatformEntity
 from zha.application.platforms.event import (
     BaseEvent,
     EntityEventTriggeredEvent,
@@ -31,8 +31,8 @@
 from zha.zigbee.device import Device
 
 
-class FakeEvent(BaseEvent):
-    """Event entity with a fixed set of event types."""
+class FakeEvent(BaseEvent, ZclPlatformEntity):
+    """ZCL-backed event entity with a fixed set of event types."""
 
     _unique_id_suffix = "fake"
     _attr_device_class = EventDeviceClass.BUTTON

zha/application/discovery.py

@@ -19,8 +19,8 @@
     BaseEntity,
     ClusterConfig,
     ClusterMatch,
-    PlatformEntity,
     PlatformFeatureGroup,
+    ZclPlatformEntity,
     alarm_control_panel,
     binary_sensor,
     button,
@@ -159,7 +159,7 @@ def _is_renamed_cluster(cluster: Cluster) -> bool:
     return cluster.ep_attribute != standard.ep_attribute
 
 
-def discover_entities_for_endpoint(endpoint: Endpoint) -> Iterator[PlatformEntity]:  # noqa: C901
+def discover_entities_for_endpoint(endpoint: Endpoint) -> Iterator[ZclPlatformEntity]:  # noqa: C901
     """Discover entities for an endpoint using the new registry-based discovery."""
     device = endpoint.device
 
@@ -178,7 +178,7 @@ def discover_entities_for_endpoint(endpoint: Endpoint) -> Iterator[PlatformEntit
         PlatformFeatureGroup | None,
         defaultdict[
             int,  # Weight
-            list[tuple[ClusterMatch, type[PlatformEntity]]],
+            list[tuple[ClusterMatch, type[ZclPlatformEntity]]],
         ],
     ] = defaultdict(lambda: defaultdict(list))
 
@@ -290,7 +290,7 @@ def discover_entities_for_endpoint(endpoint: Endpoint) -> Iterator[PlatformEntit
         if platform_override is not None and feature is not None:
             override_by_priority: defaultdict[
                 int,
-                list[tuple[ClusterMatch, type[PlatformEntity]]],
+                list[tuple[ClusterMatch, type[ZclPlatformEntity]]],
             ] = defaultdict(list)
 
             for priority, priority_matches in matches_by_priority.items():

zha/application/gateway.py

@@ -54,7 +54,13 @@
 )
 from zha.event import EventBase
 from zha.quirks import DEVICE_REGISTRY, QUIRK_REGISTRY_ENTRY_ATTR
-from zha.zigbee.device import Device, DeviceInfo, DeviceStatus, ExtendedDeviceInfo
+from zha.zigbee.device import (
+    BaseDevice,
+    Device,
+    DeviceInfo,
+    DeviceStatus,
+    ExtendedDeviceInfo,
+)
 from zha.zigbee.group import Group, GroupInfo, GroupMemberReference
 
 BLOCK_LOG_TIMEOUT: Final[int] = 60
@@ -176,7 +182,7 @@ def __init__(self, config: ZHAData) -> None:
         """Initialize the gateway."""
         super().__init__()
         self.config: ZHAData = config
-        self._devices: dict[EUI64, Device] = {}
+        self._devices: dict[EUI64, BaseDevice] = {}
         self._groups: dict[int, Group] = {}
         self.application_controller: ControllerApplication = None
         self.coordinator_zha_device: Device | None = None
@@ -488,6 +494,9 @@ async def _async_device_reinterviewed(
             )
             return
 
+        # Only Zigbee devices can be re-interviewed
+        assert isinstance(zha_device, Device)
+
         old_entry = getattr(zha_device.device, QUIRK_REGISTRY_ENTRY_ATTR, None)
         old_factory = (
             old_entry.zha_device_factory
@@ -521,6 +530,7 @@ async def _async_device_reinterviewed(
             await zha_device.async_teardown(emit_entity_events=True)
 
             zha_device = Device.new(new_zigpy_device, self)
+            assert isinstance(zha_device, Device)
             self._devices[new_zigpy_device.ieee] = zha_device
 
             zha_device.available = True
@@ -701,7 +711,7 @@ def state(self) -> State:
         return self.application_controller.state
 
     @property
-    def devices(self) -> dict[EUI64, Device]:
+    def devices(self) -> dict[EUI64, BaseDevice]:
         """Return devices."""
         return self._devices
 
@@ -710,7 +720,13 @@ def groups(self) -> dict[int, Group]:
         """Return groups."""
         return self._groups
 
-    def get_or_create_device(self, zigpy_device: zigpy.device.Device) -> Device:
+    def get_zigbee_device(self, ieee: EUI64) -> Device:
+        """Look up a Zigbee device by IEEE address."""
+        device = self._devices[ieee]
+        assert isinstance(device, Device)
+        return device
+
+    def get_or_create_device(self, zigpy_device: zigpy.device.BaseDevice) -> BaseDevice:
         """Get or create a ZHA device."""
         if (zha_device := self._devices.get(zigpy_device.ieee)) is None:
             zha_device = Device.new(zigpy_device, self)
@@ -847,8 +863,10 @@ async def async_create_zigpy_group(
                         name,
                         group_id,
                     )
+                    member_device = self.devices[member.ieee]
+                    assert isinstance(member_device, Device)
                     tasks.append(
-                        self.devices[member.ieee].async_add_endpoint_to_group(
+                        member_device.async_add_endpoint_to_group(
                             member.endpoint_id, group_id
                         )
                     )
@@ -866,6 +884,8 @@ async def async_remove_device(self, ieee: EUI64) -> None:
         for group_id, group in self.groups.items():
             for member_ieee_endpoint_id in list(group.zigpy_group.members.keys()):
                 if member_ieee_endpoint_id[0] == ieee:
+                    # Only Zigbee devices can be group members
+                    assert isinstance(device, Device)
                     await device.async_remove_from_group(group_id)
 
         await self.application_controller.remove(ieee)

zha/application/helpers.py

@@ -548,7 +548,7 @@ async def check_device_availability(self):
                 *(
                     dev._check_available()
                     for dev in self._gateway.devices.values()
-                    if not dev.is_coordinator
+                    if not dev.is_active_coordinator
                 ),
             )
             _LOGGER.debug("Device availability checker interval finished")