Implement 'remote_at' XBee command (#37)

* FrameId class for commands schema.
* Remote AT command schema.
* Remote AT command api support.
* Remote_AT ControllerApplication support.
* Update test units.
* Use partialmethod for at/queued_at commands.

Author: Adminiuga

tests/test_api.py

@@ -125,6 +125,51 @@ async def test_queued_at_command(api, monkeypatch):
     await _test_at_or_queued_at_command(api, api._queued_at, monkeypatch)
 
 
+async def _test_remote_at_command(api, monkeypatch, do_reply=True):
+    monkeypatch.setattr(t, 'serialize', mock.MagicMock(return_value=mock.sentinel.serialize))
+
+    def mock_command(name, *args):
+        rsp = xbee_api.COMMANDS[name][2]
+        ret = None
+        if rsp:
+            ret = asyncio.Future()
+            if do_reply:
+                ret.set_result(mock.sentinel.at_result)
+        return ret
+
+    api._command = mock.MagicMock(side_effect=mock_command)
+    api._seq = mock.sentinel.seq
+
+    for at_cmd in xbee_api.AT_COMMANDS:
+        res = await api._remote_at_command(
+            mock.sentinel.ieee, mock.sentinel.nwk, mock.sentinel.opts, at_cmd,
+            mock.sentinel.args)
+        assert t.serialize.call_count == 1
+        assert api._command.call_count == 1
+        assert api._command.call_args[0][0] == 'remote_at'
+        assert api._command.call_args[0][1] == mock.sentinel.seq
+        assert api._command.call_args[0][2] == mock.sentinel.ieee
+        assert api._command.call_args[0][3] == mock.sentinel.nwk
+        assert api._command.call_args[0][4] == mock.sentinel.opts
+        assert api._command.call_args[0][5] == at_cmd.encode('ascii')
+        assert api._command.call_args[0][6] == mock.sentinel.serialize
+        assert res == mock.sentinel.at_result
+        t.serialize.reset_mock()
+        api._command.reset_mock()
+
+
+@pytest.mark.asyncio
+async def test_remote_at_cmd(api, monkeypatch):
+    await _test_remote_at_command(api, monkeypatch)
+
+
+@pytest.mark.asyncio
+async def test_remote_at_cmd_no_rsp(api, monkeypatch):
+    monkeypatch.setattr(xbee_api, 'REMOTE_AT_COMMAND_TIMEOUT', 0.1)
+    with pytest.raises(asyncio.TimeoutError):
+        await _test_remote_at_command(api, monkeypatch, do_reply=False)
+
+
 def test_api_frame(api):
     ieee = t.EUI64([t.uint8_t(a) for a in range(0, 8)])
     for cmd_name, cmd_opts in xbee_api.COMMANDS.items():
@@ -215,6 +260,18 @@ def test_handle_at_response_undef_error(api):
     assert fut.exception() is not None
 
 
+def test_handle_remote_at_rsp(api):
+    api._handle_at_response = mock.MagicMock()
+    s = mock.sentinel
+    api._handle_remote_at_response([s.frame_id, s.ieee, s.nwk, s.cmd,
+                                    s.status, s.data])
+    assert api._handle_at_response.call_count == 1
+    assert api._handle_at_response.call_args[0][0][0] == s.frame_id
+    assert api._handle_at_response.call_args[0][0][1] == s.cmd
+    assert api._handle_at_response.call_args[0][0][2] == s.status
+    assert api._handle_at_response.call_args[0][0][3] == s.data
+
+
 def _send_modem_event(api, event):
     api._app = mock.MagicMock(spec=ControllerApplication)
     api._handle_modem_status([event])

tests/test_application.py

@@ -524,3 +524,18 @@ async def test_shutdown(app):
     app._api.close = mock.MagicMock()
     await app.shutdown()
     assert app._api.close.call_count == 1
+
+
+def test_remote_at_cmd(app, device):
+    dev = device()
+    app.get_device = mock.MagicMock(return_value=dev)
+    app._api = mock.MagicMock(spec=XBee)
+    s = mock.sentinel
+    app.remote_at_command(s.nwk, s.cmd, s.data,
+                          apply_changes=True, encryption=True)
+    assert app._api._remote_at_command.call_count == 1
+    assert app._api._remote_at_command.call_args[0][0] is dev.ieee
+    assert app._api._remote_at_command.call_args[0][1] == s.nwk
+    assert app._api._remote_at_command.call_args[0][2] == 0x22
+    assert app._api._remote_at_command.call_args[0][3] == s.cmd
+    assert app._api._remote_at_command.call_args[0][4] == s.data

zigpy_xbee/api.py

@@ -1,6 +1,7 @@
 import asyncio
 import binascii
 import enum
+import functools
 import logging
 
 from zigpy.types import LVList
@@ -11,6 +12,7 @@
 LOGGER = logging.getLogger(__name__)
 
 AT_COMMAND_TIMEOUT = 2
+REMOTE_AT_COMMAND_TIMEOUT = 30
 
 
 class ModemStatus(t.uint8_t, t.UndefinedEnum):
@@ -29,22 +31,22 @@ class ModemStatus(t.uint8_t, t.UndefinedEnum):
 
 # https://www.digi.com/resources/documentation/digidocs/PDFs/90000976.pdf
 COMMANDS = {
-    'at': (0x08, (t.uint8_t, t.ATCommand, t.Bytes), 0x88),
-    'queued_at': (0x09, (t.uint8_t, t.ATCommand, t.Bytes), 0x88),
-    'remote_at': (0x17, (), None),
+    'at': (0x08, (t.FrameId, t.ATCommand, t.Bytes), 0x88),
+    'queued_at': (0x09, (t.FrameId, t.ATCommand, t.Bytes), 0x88),
+    'remote_at': (0x17, (t.FrameId, t.EUI64, t.NWK, t.uint8_t, t.ATCommand, t.Bytes), 0x97),
     'tx': (0x10, (), None),
-    'tx_explicit': (0x11, (t.uint8_t, t.EUI64, t.NWK, t.uint8_t, t.uint8_t, t.uint16_t, t.uint16_t, t.uint8_t, t.uint8_t, t.Bytes), None),
-    'create_source_route': (0x21, (t.uint8_t, t.EUI64, t.NWK, t.uint8_t, LVList(t.NWK)), None),
+    'tx_explicit': (0x11, (t.FrameId, t.EUI64, t.NWK, t.uint8_t, t.uint8_t, t.uint16_t, t.uint16_t, t.uint8_t, t.uint8_t, t.Bytes), None),
+    'create_source_route': (0x21, (t.FrameId, t.EUI64, t.NWK, t.uint8_t, LVList(t.NWK)), None),
     'register_joining_device': (0x24, (), None),
 
-    'at_response': (0x88, (t.uint8_t, t.ATCommand, t.uint8_t, t.Bytes), None),
+    'at_response': (0x88, (t.FrameId, t.ATCommand, t.uint8_t, t.Bytes), None),
     'modem_status': (0x8A, (ModemStatus, ), None),
-    'tx_status': (0x8B, (t.uint8_t, t.NWK, t.uint8_t, t.uint8_t, t.uint8_t), None),
+    'tx_status': (0x8B, (t.FrameId, t.NWK, t.uint8_t, t.uint8_t, t.uint8_t), None),
     'route_information': (0x8D, (), None),
     'rx': (0x90, (), None),
     'explicit_rx_indicator': (0x91, (t.EUI64, t.NWK, t.uint8_t, t.uint8_t, t.uint16_t, t.uint16_t, t.uint8_t, t.Bytes), None),
     'rx_io_data_long_addr': (0x92, (), None),
-    'remote_at_response': (0x97, (), None),
+    'remote_at_response': (0x97, (t.FrameId, t.EUI64, t.NWK, t.ATCommand, t.uint8_t, t.Bytes), None),
     'extended_status': (0x98, (), None),
     'route_record_indicator': (0xA1, (t.EUI64, t.NWK, t.uint8_t, LVList(t.NWK)), None),
     'many_to_one_rri': (0xA3, (t.EUI64, t.NWK, t.uint8_t), None),
@@ -183,6 +185,7 @@ class ATCommandResult(enum.IntEnum):
     ERROR = 1
     INVALID_COMMAND = 2
     INVALID_PARAMETER = 3
+    TX_FAILURE = 4
 
 
 class XBee:
@@ -233,27 +236,32 @@ def _seq_command(self, name, *args):
         LOGGER.debug("Sequenced command: %s %s", name, args)
         return self._command(name, self._seq, *args)
 
-    def _queued_at(self, name, *args):
-        LOGGER.debug("Queue AT command: %s %s", name, args)
+    async def _remote_at_command(self, ieee, nwk, options, name, *args):
+        LOGGER.debug("Remote AT command: %s %s", name, args)
         data = t.serialize(args, (AT_COMMANDS[name], ))
-        return self._command(
-            'queued_at',
-            self._seq,
-            name.encode('ascii'),
-            data,
-        )
-
-    async def _at_command(self, name, *args):
-        LOGGER.debug("AT command: %s %s", name, args)
+        try:
+            return await asyncio.wait_for(
+                self._command('remote_at', self._seq, ieee, nwk, options,
+                              name.encode('ascii'), data,),
+                timeout=REMOTE_AT_COMMAND_TIMEOUT)
+        except asyncio.TimeoutError:
+            LOGGER.warning("No response to %s command", name)
+            raise
+
+    async def _at_partial(self, cmd_type, name, *args):
+        LOGGER.debug("%s command: %s %s", cmd_type, name, args)
         data = t.serialize(args, (AT_COMMANDS[name], ))
         try:
             return await asyncio.wait_for(
-                self._command('at', self._seq, name.encode('ascii'), data,),
+                self._command(cmd_type, self._seq, name.encode('ascii'), data),
                 timeout=AT_COMMAND_TIMEOUT)
         except asyncio.TimeoutError:
-            LOGGER.warning("No response to %s command", name)
+            LOGGER.warning("%s: No response to %s command", cmd_type, name)
             raise
 
+    _at_command = functools.partialmethod(_at_partial, 'at')
+    _queued_at = functools.partialmethod(_at_partial, 'queued_at')
+
     def _api_frame(self, name, *args):
         c = COMMANDS[name]
         return (bytes([c[0]]) + t.serialize(args, c[1])), c[2]
@@ -288,6 +296,11 @@ def _handle_at_response(self, data):
         response, remains = response_type.deserialize(data[3])
         fut.set_result(response)
 
+    def _handle_remote_at_response(self, data):
+        """Remote AT command response."""
+        LOGGER.debug("Remote AT command response: %s", data)
+        return self._handle_at_response((data[0], data[3], data[4], data[5]))
+
     def _handle_many_to_one_rri(self, data):
         LOGGER.debug("_handle_many_to_one_rri: %s", data)
 

zigpy_xbee/types.py

@@ -149,6 +149,10 @@ class UndefinedEnum(enum.Enum, metaclass=UndefinedEnumMeta):
     pass
 
 
+class FrameId(uint8_t):
+    pass
+
+
 class NWK(uint16_t):
     def __repr__(self):
         return '0x{:04x}'.format(self)

zigpy_xbee/zigbee/application.py

@@ -147,6 +147,18 @@ async def request(self, nwk, profile, cluster, src_ep, dst_ep, sequence, data, e
             self._pending.pop(sequence, None)
             raise
 
+    @zigpy.util.retryable_request
+    def remote_at_command(self, nwk, cmd_name, *args, apply_changes=True,
+                          encryption=True):
+        LOGGER.debug("Remote AT%s command: %s", cmd_name, args)
+        options = zigpy.types.uint8_t(0)
+        if apply_changes:
+            options |= 0x02
+        if encryption:
+            options |= 0x20
+        dev = self.get_device(nwk=nwk)
+        return self._api._remote_at_command(dev.ieee, nwk, options, cmd_name, *args)
+
     async def permit_ncp(self, time_s=60):
         assert 0 <= time_s <= 254
         await self._api._at_command('NJ', time_s)