Green Power support

pyproject.toml

@@ -15,7 +15,7 @@ license = {text = "GPL-3.0"}
 requires-python = ">=3.8"
 dependencies = [
     "voluptuous",
-    "zigpy>=0.80.0",
+    "zigpy>=2.4.0",
     'async-timeout; python_version<"3.11"',
 ]
 

tests/test_api.py

@@ -9,6 +9,7 @@
 import pytest
 import zigpy.config
 import zigpy.types as zigpy_t
+import zigpy.zgp
 
 if sys.version_info[:2] < (3, 11):
     from async_timeout import timeout as asyncio_timeout
@@ -335,6 +336,90 @@ async def test_command_not_connected(api):
         await api._command(cmd=deconz_api.CommandId.version, reserved=0)
 
 
+def green_power_indication(api, gpdf: bytes) -> bytes:
+    _, rx_schema = deconz_api.COMMAND_SCHEMAS[deconz_api.CommandId.zigbee_green_power]
+
+    return deconz_api.Command(
+        command_id=deconz_api.CommandId.zigbee_green_power,
+        seq=api._seq,
+        payload=t.serialize_dict(
+            {
+                "status": deconz_api.Status.SUCCESS,
+                "frame_length": t.uint16_t(7 + len(gpdf)),
+                "payload_length": t.uint16_t(len(gpdf)),
+                "gpdf": t.Bytes(gpdf),
+            },
+            rx_schema,
+        ),
+    ).serialize()
+
+
+async def test_green_power_indication(api, mock_command_rsp):
+    """A GPDF received directly is passed to zigpy unprocessed, with its MIC."""
+    await api.connect()
+
+    # The spec's test vector for SecurityLevel 0b10 (A.1.5.4.2.1), repeated by the GPD
+    gpdf = bytes.fromhex("8C 10 21 43 65 87 02 00 00 00 20 CF 78 7E 72")
+    api.data_received(green_power_indication(api, gpdf))
+    api.data_received(green_power_indication(api, gpdf))
+
+    await asyncio.sleep(0.01)
+
+    # Protected GPDFs are filtered by their frame counter in zigpy
+    assert (
+        api._app.gp_packet_received.mock_calls
+        == [
+            call(
+                zigpy_t.ZigbeeGpPacket(
+                    application_id=zigpy.zgp.ApplicationID.SrcID,
+                    src_id=zigpy.zgp.SrcID(0x87654321),
+                    command_id=zigpy.zgp.GPDCommandID.Off,
+                    payload=zigpy_t.SerializableBytes(b""),
+                    frame_counter=zigpy_t.uint32_t(2),
+                    security_level=zigpy.zgp.SecurityLevel.FullFrameCounterAndMIC,
+                    security_key_type=None,
+                    security_status=zigpy.zgp.SecurityStatus.Unprocessed,
+                    mic=zigpy_t.uint32_t(0x727E78CF),
+                    auto_commissioning=False,
+                    rx_after_tx=False,
+                    individual_key=False,
+                )
+            )
+        ]
+        * 2
+    )
+
+
+async def test_green_power_indication_unprotected_repeated(api, mock_command_rsp):
+    """Without a MAC sequence number, repetitions of an unprotected GPDF are dropped."""
+    await api.connect()
+
+    toggle = bytes.fromhex("0C 44 33 22 11 22")
+    on = bytes.fromhex("0C 44 33 22 11 21")
+
+    for gpdf in (toggle, toggle, on, toggle):
+        api.data_received(green_power_indication(api, gpdf))
+
+    await asyncio.sleep(0.01)
+
+    assert [c.args[0].command_id for c in api._app.gp_packet_received.mock_calls] == [
+        zigpy.zgp.GPDCommandID.Toggle,
+        zigpy.zgp.GPDCommandID.On,
+    ]
+
+
+async def test_green_power_indication_invalid(api, mock_command_rsp, caplog):
+    await api.connect()
+
+    # A Channel Configuration sent by another sink (Direction 0b1)
+    with caplog.at_level(logging.DEBUG):
+        api.data_received(green_power_indication(api, bytes.fromhex("8D 80 F3 1F")))
+        await asyncio.sleep(0.01)
+
+    assert "Ignoring GPDF" in caplog.text
+    assert len(api._app.gp_packet_received.mock_calls) == 0
+
+
 async def test_data_received(api, mock_command_rsp):
     await api.connect()
 

tests/test_send_receive.py

@@ -7,6 +7,7 @@
 import pytest
 import zigpy.exceptions
 import zigpy.types as zigpy_t
+import zigpy.zgp
 
 from zigpy_deconz.api import Status, TXStatus
 import zigpy_deconz.exception
@@ -207,6 +208,32 @@ async def test_send_packet_broadcast(app, tx_packet):  # noqa: F811
     assert req["radius"] == 12
 
 
+async def test_send_packet_green_power_endpoint(app, tx_packet):  # noqa: F811
+    """The Green Power endpoint is the one source endpoint that is kept as is."""
+    tx_packet.src_ep = zigpy.zgp.GP_ENDPOINT
+    tx_packet.dst_ep = zigpy.zgp.GP_ENDPOINT
+
+    with patch_data_request(app) as mock_req:
+        await app.send_packet(tx_packet)
+
+    assert mock_req.mock_calls[0].kwargs["src_ep"] == zigpy.zgp.GP_ENDPOINT
+
+
+async def test_send_packet_alias(app, tx_packet):  # noqa: F811
+    """The firmware cannot send on behalf of a Green Power alias."""
+    tx_packet.dst = zigpy_t.AddrModeAddress(
+        addr_mode=zigpy_t.AddrMode.Broadcast,
+        address=zigpy_t.BroadcastAddress.RX_ON_WHEN_IDLE,
+    )
+    tx_packet.alias_src = zigpy_t.NWK(0x1234)
+    tx_packet.alias_seq = zigpy_t.uint8_t(0)
+
+    with patch_data_request(app) as mock_req, pytest.raises(NotImplementedError):
+        await app.send_packet(tx_packet)
+
+    assert len(mock_req.mock_calls) == 0
+
+
 async def test_send_packet_enqueue_failure(app, tx_packet):  # noqa: F811
     with patch_data_request(app, fail_enqueue=True):  # noqa: F811
         with pytest.raises(zigpy.exceptions.DeliveryError) as e:

zigpy_deconz/api.py

@@ -14,7 +14,8 @@
 else:
     from asyncio import timeout as asyncio_timeout  # pragma: no cover
 
-from zigpy.datastructures import PriorityLock
+from zigpy.datastructures import Debouncer, PriorityLock
+from zigpy.device import GP_DUPLICATE_TIMEOUT
 from zigpy.types import (
     APSStatus,
     Bool,
@@ -25,6 +26,7 @@
     ZigbeePacket,
 )
 from zigpy.zdo.types import SimpleDescriptor
+from zigpy.zgp.util import parse_gpdf
 
 from zigpy_deconz.exception import CommandError, MismatchedResponseError, ParsingError
 import zigpy_deconz.types as t
@@ -404,7 +406,9 @@
             "status": Status,
             "frame_length": t.uint16_t,
             "payload_length": t.uint16_t,
-            "reserved": t.LongOctetString,
+            # The NWK frame of a received GPDF, raw: the firmware performs no security
+            # processing and strips the MAC header
+            "gpdf": t.Bytes,
         },
     ),
 }
@@ -433,6 +437,11 @@
         self._free_slots_available_event = asyncio.Event()
         self._free_slots_available_event.set()
 
+        # The firmware delivers every repetition of a GPDF and strips the MAC header,
+        # so the MAC sequence number of an unprotected GPDF is not available to filter
+        # them (spec A.3.6.1.2): its content is the only key
+        self._gp_debouncer = Debouncer()
+
         self._data_poller_event = asyncio.Event()
         self._data_poller_event.set()
         self._data_poller_task: asyncio.Task | None = None
@@ -581,7 +590,7 @@
 
             try:
                 LOGGER.debug("Sending %s%s (seq=%s)", cmd, kwargs, seq)
                 self._uart.send(command.replace(seq=seq).serialize())
 
                 async with asyncio_timeout(COMMAND_TIMEOUT):
                     return await fut
@@ -768,6 +777,27 @@
                     status=rsp["status"], device_state=rsp["device_state"]
                 )
 
+    def _handle_zigbee_green_power(self, status: Status, gpdf: t.Bytes) -> None:
+        try:
+            packet = parse_gpdf(gpdf)
+        except ValueError:
+            LOGGER.debug("Ignoring GPDF %r", gpdf, exc_info=True)
+            return
+
+        if packet.frame_counter is None and self._gp_debouncer.filter(
+            (
+                packet.src_id,
+                packet.command_id,
+                packet.endpoint,
+                packet.payload.serialize(),
+            ),
+            GP_DUPLICATE_TIMEOUT.total_seconds(),
+        ):
+            LOGGER.debug("Dropping repeated GPDF: %r", packet)
+            return
+
+        self._app.gp_packet_received(packet)
+
     def _handle_device_state_changed(
         self,
         status: t.Status,

zigpy_deconz/zigbee/application.py

@@ -23,6 +23,7 @@
 import zigpy.types
 import zigpy.util
 import zigpy.zdo.types as zdo_t
+import zigpy.zgp
 
 import zigpy_deconz
 from zigpy_deconz import types as t
@@ -545,6 +546,10 @@ async def add_endpoint(self, descriptor: zdo_t.SimpleDescriptor) -> None:
     async def send_packet(self, packet):
         LOGGER.debug("Sending packet: %r", packet)
 
+        if packet.alias_src is not None:
+            # The APS data request has no alias parameters
+            raise NotImplementedError("The firmware cannot send with an alias")
+
         if zigpy.types.TransmitOptions.APS_Encryption in packet.tx_options:
             # APS encryption is performed with the link key, not the network key, so
             # the "use NWK key" option must not be set
@@ -578,7 +583,13 @@ async def send_packet(self, packet):
                         ),
                         profile=packet.profile_id,
                         cluster=packet.cluster_id,
-                        src_ep=min(1, packet.src_ep),
+                        # The Green Power endpoint is registered on its own, every
+                        # other source endpoint collapses onto the first one
+                        src_ep=(
+                            packet.src_ep
+                            if packet.src_ep == zigpy.zgp.GP_ENDPOINT
+                            else min(1, packet.src_ep)
+                        ),
                         aps_payload=packet.data.serialize(),
                         tx_options=tx_options,
                         relays=packet.source_route,