GPDF sending

bellows/zigbee/application.py

@@ -3,7 +3,7 @@
 import asyncio
 from asyncio import timeout as asyncio_timeout
 from collections.abc import AsyncGenerator
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 import importlib.metadata
 import logging
 import os
@@ -123,6 +123,10 @@ def __init__(self, config: dict) -> None:
         self._req_lock = asyncio.Lock()
         self._packet_capture_channel: int | None = None
 
+        # The NCP queues a single GPDF for all GPDs
+        self._gpdf_lock = asyncio.Lock()
+        self._gpdf_sent: asyncio.Future[None] | None = None
+
     @property
     def controller_event(self):
         """Return asyncio.Event for controller app."""
@@ -729,6 +733,11 @@ def ezsp_callback_handler(self, frame_name, args):
                 proxyTableIndex,
                 gpdCommandPayload,
             )
+        elif frame_name == "dGpSentHandler":
+            # The NCP reports neither the transmit status nor the GPEP handle: both are
+            # always zero. With one GPDF queued at a time, this is the queued one.
+            if self._gpdf_sent is not None and not self._gpdf_sent.done():
+                self._gpdf_sent.set_result(None)
         elif frame_name == "messageSentHandler":
             if self._ezsp.ezsp_version >= 14:
                 (
@@ -1383,6 +1392,89 @@ async def _unsubscribe_from_multicast_group(
 
         await self._multicast.unsubscribe(group_id=group_id, endpoint_id=endpoint_id)
 
+    async def _send_gpdf(
+        self,
+        *,
+        target: zgp_t.SrcID | tuple[zigpy.types.EUI64, int] | None,
+        command_id: zgp_t.GPDCommandID,
+        payload: bytes,
+        channel: int,
+        lifetime: timedelta,
+    ) -> None:
+        """Send a GPDF from the NCP in its GPD's next receive window."""
+        # The NCP only transmits on the operational channel, outside of a GP channel
+        # delivery scan
+        if channel != self.state.network_info.channel:
+            raise zigpy.exceptions.DeliveryError(
+                f"Cannot send a GPDF on channel {channel}"
+            )
+
+        # A Maintenance frame is queued for SrcID 0 (A.1.5.2.2)
+        match target:
+            case None:
+                addr = t.EmberGpAddress(
+                    applicationId=zgp_t.ApplicationID.SrcID,
+                    source_id=zgp_t.SrcID(0x00000000),
+                    source_id_padding=0,
+                    endpoint=0,
+                )
+            case (ieee, endpoint):
+                addr = t.EmberGpAddress(
+                    applicationId=zgp_t.ApplicationID.IEEE,
+                    gpd_eui64=ieee,
+                    endpoint=endpoint,
+                )
+            case _:
+                addr = t.EmberGpAddress(
+                    applicationId=zgp_t.ApplicationID.SrcID,
+                    source_id=target,
+                    source_id_padding=0,
+                    endpoint=0,
+                )
+
+        # The NCP never expires a queued GPDF, which would block every other GPD, so the
+        # GPDF is removed once `lifetime` is up
+        async with self._gpdf_lock:
+            self._gpdf_sent = asyncio.get_running_loop().create_future()
+
+            try:
+                (status,) = await self._ezsp.dGpSend(
+                    action=True,
+                    useCca=False,
+                    addr=addr,
+                    gpdCommandId=command_id,
+                    gpdAsdu=payload,
+                    gpepHandle=0,
+                    gpTxQueueEntryLifetimeMs=lifetime // timedelta(milliseconds=1),
+                )
+
+                if t.sl_Status.from_ember_status(status) != t.sl_Status.OK:
+                    raise zigpy.exceptions.DeliveryError(
+                        f"Failed to queue GPDF: {status!r}"
+                    )
+
+                try:
+                    async with asyncio_timeout(lifetime.total_seconds()):
+                        await self._gpdf_sent
+                except TimeoutError:
+                    (status,) = await self._ezsp.dGpSend(
+                        action=False,
+                        useCca=False,
+                        addr=addr,
+                        gpdCommandId=command_id,
+                        gpdAsdu=b"",
+                        gpepHandle=0,
+                        gpTxQueueEntryLifetimeMs=0,
+                    )
+
+                    # Not finding it means the GPD took it meanwhile
+                    if t.sl_Status.from_ember_status(status) == t.sl_Status.OK:
+                        raise zigpy.exceptions.DeliveryError(
+                            "GPD did not listen before the GPDF expired"
+                        ) from None
+            finally:
+                self._gpdf_sent = None
+
     def _handle_id_conflict(self, nwk: t.EmberNodeId) -> None:
         LOGGER.warning("NWK conflict is reported for 0x%04x", nwk)
         self.state.counters[COUNTERS_CTRL][COUNTER_NWK_CONFLICTS].increment()

tests/test_application.py

@@ -1,5 +1,6 @@
 import asyncio
 import contextlib
+from datetime import timedelta
 import importlib.metadata
 import logging
 from unittest.mock import AsyncMock, MagicMock, PropertyMock, call, patch, sentinel
@@ -3261,3 +3262,151 @@
         app.ezsp_callback_handler("gpepIncomingMessageHandler", list(args.values()))
 
     assert len(gp_packet_received.mock_calls) == 0
+
+
+GPD_IEEE = zigpy_t.EUI64.convert("00:11:22:33:44:55:66:77")
+
+
+async def send_gpdf(
+    app: ControllerApplication,
+    target: zgp_t.SrcID | tuple[zigpy_t.EUI64, int] | None = zgp_t.SrcID(0x12345678),
+    *,
+    channel: int = 25,
+    lifetime: timedelta = timedelta(seconds=20),
+) -> None:
+    await app.send_gpdf(
+        target=target,
+        command_id=zgp_t.GPDCommandID(0xFE),
+        payload=b"\x00",
+        channel=channel,
+        lifetime=lifetime,
+    )
+
+
+@pytest.mark.parametrize(
+    ("target", "addr"),
+    [
+        (
+            zgp_t.SrcID(0x12345678),
+            t.EmberGpAddress(
+                applicationId=zgp_t.ApplicationID.SrcID,
+                source_id=zgp_t.SrcID(0x12345678),
+                source_id_padding=0,
+                endpoint=0,
+            ),
+        ),
+        (
+            (GPD_IEEE, 0x01),
+            t.EmberGpAddress(
+                applicationId=zgp_t.ApplicationID.IEEE,
+                gpd_eui64=GPD_IEEE,
+                endpoint=0x01,
+            ),
+        ),
+        # A Maintenance frame is queued for SrcID 0
+        (
+            None,
+            t.EmberGpAddress(
+                applicationId=zgp_t.ApplicationID.SrcID,
+                source_id=zgp_t.SrcID(0x00000000),
+                source_id_padding=0,
+                endpoint=0,
+            ),
+        ),
+    ],
+)
+async def test_send_gpdf(
+    app: ControllerApplication,
+    target: zgp_t.SrcID | tuple[zigpy_t.EUI64, int] | None,
+    addr: t.EmberGpAddress,
+) -> None:
+    """A GPDF is queued in the NCP and sent once the NCP reports it."""
+    app.state.network_info.channel = 25
+    app._ezsp.dGpSend = AsyncMock(return_value=[t.EmberStatus.SUCCESS])
+
+    task = asyncio.create_task(send_gpdf(app, target))
+    await asyncio.sleep(0.01)
+    assert not task.done()
+
+    # The NCP reports neither a status nor the handle
+    app.ezsp_callback_handler("dGpSentHandler", [t.EmberStatus.SUCCESS, 0])
+    await task
+
+    assert app._ezsp.dGpSend.mock_calls == [
+        call(
+            action=True,
+            useCca=False,
+            addr=addr,
+            gpdCommandId=zgp_t.GPDCommandID(0xFE),
+            gpdAsdu=b"\x00",
+            gpepHandle=0,
+            gpTxQueueEntryLifetimeMs=20_000,
+        )
+    ]
+
+
+async def test_send_gpdf_one_at_a_time(app: ControllerApplication) -> None:
+    """The NCP queues one GPDF for all GPDs, so a second one waits for the first."""
+    app.state.network_info.channel = 25
+    app._ezsp.dGpSend = AsyncMock(return_value=[t.EmberStatus.SUCCESS])
+
+    first = asyncio.create_task(send_gpdf(app, zgp_t.SrcID(0x00000001)))
+    second = asyncio.create_task(send_gpdf(app, zgp_t.SrcID(0x00000002)))
+    await asyncio.sleep(0.01)
+    assert len(app._ezsp.dGpSend.mock_calls) == 1
+
+    app.ezsp_callback_handler("dGpSentHandler", [t.EmberStatus.SUCCESS, 0])
+    await first
+    await asyncio.sleep(0.01)
+    assert len(app._ezsp.dGpSend.mock_calls) == 2
+    assert not second.done()
+
+    app.ezsp_callback_handler("dGpSentHandler", [t.EmberStatus.SUCCESS, 0])
+    await second
+
+
+async def test_send_gpdf_expired(app: ControllerApplication) -> None:
+    """The NCP never expires a GPDF, so it is removed once its lifetime is up."""
+    app.state.network_info.channel = 25
+    app._ezsp.dGpSend = AsyncMock(return_value=[t.EmberStatus.SUCCESS])
+
+    with pytest.raises(zigpy.exceptions.DeliveryError, match="expired"):
+        await send_gpdf(app, lifetime=timedelta(milliseconds=10))
+
+    [_, removal] = app._ezsp.dGpSend.mock_calls
+    assert removal.kwargs["action"] is False
+    assert removal.kwargs["addr"].source_id == 0x12345678
+
+
+async def test_send_gpdf_sent_while_expiring(app: ControllerApplication) -> None:
+    """A GPDF the NCP no longer holds when it expires was taken by its GPD."""
+    app.state.network_info.channel = 25
+    app._ezsp.dGpSend = AsyncMock(
+        side_effect=[[t.EmberStatus.SUCCESS], [t.EmberStatus.ERR_FATAL]]
+    )
+
+    await send_gpdf(app, lifetime=timedelta(milliseconds=10))
+
+
+async def test_send_gpdf_not_queued(app: ControllerApplication) -> None:
+    app.state.network_info.channel = 25
+    app._ezsp.dGpSend = AsyncMock(return_value=[t.EmberStatus.ERR_FATAL])
+
+    with pytest.raises(zigpy.exceptions.DeliveryError, match="Failed to queue"):
+        await send_gpdf(app)
+
+
+async def test_send_gpdf_other_channel(app: ControllerApplication) -> None:
+    """The NCP cannot send a GPDF off the operational channel."""
+    app.state.network_info.channel = 25
+    app._ezsp.dGpSend = AsyncMock()
+
+    with pytest.raises(zigpy.exceptions.DeliveryError, match="channel 15"):
+        await send_gpdf(app, channel=15)
+
+    assert len(app._ezsp.dGpSend.mock_calls) == 0
+
+
+def test_gpdf_sent_without_pending(app: ControllerApplication) -> None:
+    """A GPDF sent after it was given up on is ignored."""
+    app.ezsp_callback_handler("dGpSentHandler", [t.EmberStatus.SUCCESS, 0])