Fix some EZSP v14 commands (#652)

* Fix some EZSP v14 commands * Remove unnecessary `from_ember_status` * Fix multicast broadcast address * Fix data type for sequence * Match old stack behavior * Add some unit tests * Use the correct calls in tests
zigpy · Oct 21, 2024 · 7e1008e · 7e1008e
1 parent 03dc152
commit 7e1008e
Show file tree

Hide file tree

Showing 3 changed files with 102 additions and 2 deletions.
diff --git a/bellows/ezsp/v14/__init__.py b/bellows/ezsp/v14/__init__.py
@@ -87,6 +87,43 @@ async def get_tc_link_key(self) -> zigpy.state.Key:
 
         return zigpy.state.Key(key=tc_link_key_data)
 
+    async def send_unicast(
+        self,
+        nwk: t.NWK,
+        aps_frame: t.EmberApsFrame,
+        message_tag: t.uint8_t,
+        data: bytes,
+    ) -> tuple[t.sl_Status, t.uint8_t]:
+        status, sequence = await self.sendUnicast(
+            message_type=t.EmberOutgoingMessageType.OUTGOING_DIRECT,
+            nwk=nwk,
+            aps_frame=aps_frame,
+            message_tag=message_tag,
+            message=data,
+        )
+
+        return status, sequence
+
+    async def send_multicast(
+        self,
+        aps_frame: t.EmberApsFrame,
+        radius: t.uint8_t,
+        non_member_radius: t.uint8_t,
+        message_tag: t.uint8_t,
+        data: bytes,
+    ) -> tuple[t.sl_Status, t.uint8_t]:
+        status, sequence = await self.sendMulticast(
+            aps_frame=aps_frame,
+            hops=radius,
+            broadcast_addr=t.BroadcastAddress.RX_ON_WHEN_IDLE,
+            alias=0x0000,
+            sequence=aps_frame.sequence,
+            message_tag=message_tag,
+            message=data,
+        )
+
+        return status, sequence
+
     async def send_broadcast(
         self,
         address: t.BroadcastAddress,

diff --git a/bellows/ezsp/v14/commands.py b/bellows/ezsp/v14/commands.py
@@ -113,7 +113,7 @@ class GetTokenDataRsp(Struct):
         },
     ),
     "sendBroadcast": (
-        0x0034,
+        0x0036,
         {
             "alias": t.uint16_t,
             "destination": BroadcastAddress,
@@ -128,6 +128,22 @@ class GetTokenDataRsp(Struct):
             "sequence": t.uint8_t,
         },
     ),
+    "sendMulticast": (
+        0x0038,
+        {
+            "aps_frame": t.EmberApsFrame,
+            "hops": t.uint8_t,
+            "broadcast_addr": t.BroadcastAddress,
+            "alias": t.uint16_t,
+            "sequence": t.uint8_t,
+            "message_tag": t.uint16_t,
+            "message": t.LVBytes,
+        },
+        {
+            "status": t.sl_Status,
+            "sequence": t.uint8_t,
+        },
+    ),
     "launchStandaloneBootloader": (
         0x008F,
         {

diff --git a/tests/test_ezsp_v14.py b/tests/test_ezsp_v14.py
@@ -155,8 +155,55 @@ async def test_get_network_key_without_network(ezsp_f):
         await ezsp_f.get_network_key()
 
 
+async def test_send_unicast(ezsp_f) -> None:
+    ezsp_f.sendUnicast.return_value = (t.sl_Status.OK, 0x0042)
+    status, message_tag = await ezsp_f.send_unicast(
+        nwk=0x1234,
+        aps_frame=t.EmberApsFrame(),
+        message_tag=0x42,
+        data=b"hello",
+    )
+
+    assert status == t.sl_Status.OK
+    assert message_tag == 0x42
+    assert ezsp_f.sendUnicast.mock_calls == [
+        call(
+            message_type=t.EmberOutgoingMessageType.OUTGOING_DIRECT,
+            nwk=0x1234,
+            aps_frame=t.EmberApsFrame(),
+            message_tag=0x42,
+            message=b"hello",
+        )
+    ]
+
+
+async def test_send_multicast(ezsp_f) -> None:
+    ezsp_f.sendMulticast.return_value = (t.sl_Status.OK, 0x0042)
+    status, message_tag = await ezsp_f.send_multicast(
+        aps_frame=t.EmberApsFrame(sequence=0x34),
+        radius=12,
+        non_member_radius=34,
+        message_tag=0x42,
+        data=b"hello",
+    )
+
+    assert status == t.sl_Status.OK
+    assert message_tag == 0x42
+    assert ezsp_f.sendMulticast.mock_calls == [
+        call(
+            aps_frame=t.EmberApsFrame(sequence=0x34),
+            hops=12,
+            broadcast_addr=t.BroadcastAddress.RX_ON_WHEN_IDLE,
+            alias=0x0000,
+            sequence=0x34,
+            message_tag=0x0042,
+            message=b"hello",
+        )
+    ]
+
+
 async def test_send_broadcast(ezsp_f) -> None:
-    ezsp_f.sendBroadcast.return_value = (t.sl_Status.OK, 0x42)
+    ezsp_f.sendBroadcast.return_value = (t.sl_Status.OK, 0x0042)
     status, message_tag = await ezsp_f.send_broadcast(
         address=t.BroadcastAddress.ALL_ROUTERS_AND_COORDINATOR,
         aps_frame=t.EmberApsFrame(),