Refactor api commands. (#39)

* Refactor api._command() method.
* Allow calling of API commands through `__getattr__()`.

Author: Adminiuga

tests/test_api.py

@@ -33,8 +33,9 @@ def test_close(api):
 def test_commands():
     import string
     anum = string.ascii_letters + string.digits + '_'
+    commands = {**xbee_api.COMMAND_REQUESTS, **xbee_api.COMMAND_RESPONSES}
 
-    for cmd_name, cmd_opts in xbee_api.COMMANDS.items():
+    for cmd_name, cmd_opts in commands.items():
         assert isinstance(cmd_name, str) is True
         assert all([c in anum for c in cmd_name]), cmd_name
         assert len(cmd_opts) == 3
@@ -47,12 +48,12 @@ def test_commands():
 @pytest.mark.asyncio
 async def test_command(api):
     def mock_api_frame(name, *args):
-        c = xbee_api.COMMANDS[name]
+        c = xbee_api.COMMAND_REQUESTS[name]
         return mock.sentinel.api_frame_data, c[2]
     api._api_frame = mock.MagicMock(side_effect=mock_api_frame)
     api._uart.send = mock.MagicMock()
 
-    for cmd_name, cmd_opts in xbee_api.COMMANDS.items():
+    for cmd_name, cmd_opts in xbee_api.COMMAND_REQUESTS.items():
         cmd_id, schema, expect_reply = cmd_opts
         ret = api._command(cmd_name, mock.sentinel.cmd_data)
         if expect_reply:
@@ -62,28 +63,30 @@ def mock_api_frame(name, *args):
             assert ret is None
         assert api._api_frame.call_count == 1
         assert api._api_frame.call_args[0][0] == cmd_name
-        assert api._api_frame.call_args[0][1] == mock.sentinel.cmd_data
+        assert api._api_frame.call_args[0][1] == api._seq - 1
+        assert api._api_frame.call_args[0][2] == mock.sentinel.cmd_data
         assert api._uart.send.call_count == 1
         assert api._uart.send.call_args[0][0] == mock.sentinel.api_frame_data
         api._api_frame.reset_mock()
         api._uart.send.reset_mock()
 
-
-def test_seq_command(api):
-    api._command = mock.MagicMock()
-    api._seq = mock.sentinel.seq
-    api._seq_command(mock.sentinel.cmd_name, mock.sentinel.args)
-    assert api._command.call_count == 1
-    assert api._command.call_args[0][0] == mock.sentinel.cmd_name
-    assert api._command.call_args[0][1] == mock.sentinel.seq
-    assert api._command.call_args[0][2] == mock.sentinel.args
+        ret = api._command(cmd_name, mock.sentinel.cmd_data, mask_frame_id=True)
+        assert ret is None
+        assert api._api_frame.call_count == 1
+        assert api._api_frame.call_args[0][0] == cmd_name
+        assert api._api_frame.call_args[0][1] == 0
+        assert api._api_frame.call_args[0][2] == mock.sentinel.cmd_data
+        assert api._uart.send.call_count == 1
+        assert api._uart.send.call_args[0][0] == mock.sentinel.api_frame_data
+        api._api_frame.reset_mock()
+        api._uart.send.reset_mock()
 
 
 async def _test_at_or_queued_at_command(api, cmd, 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]
+        rsp = xbee_api.COMMAND_REQUESTS[name][2]
         ret = None
         if rsp:
             ret = asyncio.Future()
@@ -99,9 +102,8 @@ def mock_command(name, *args):
         assert t.serialize.call_count == 1
         assert api._command.call_count == 1
         assert api._command.call_args[0][0] in ('at', 'queued_at')
-        assert api._command.call_args[0][1] == mock.sentinel.seq
-        assert api._command.call_args[0][2] == at_cmd.encode('ascii')
-        assert api._command.call_args[0][3] == mock.sentinel.serialize
+        assert api._command.call_args[0][1] == at_cmd.encode('ascii')
+        assert api._command.call_args[0][2] == mock.sentinel.serialize
         assert res == mock.sentinel.at_result
         t.serialize.reset_mock()
         api._command.reset_mock()
@@ -129,7 +131,7 @@ 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]
+        rsp = xbee_api.COMMAND_REQUESTS[name][2]
         ret = None
         if rsp:
             ret = asyncio.Future()
@@ -147,12 +149,11 @@ def mock_command(name, *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 api._command.call_args[0][1] == mock.sentinel.ieee
+        assert api._command.call_args[0][2] == mock.sentinel.nwk
+        assert api._command.call_args[0][3] == mock.sentinel.opts
+        assert api._command.call_args[0][4] == at_cmd.encode('ascii')
+        assert api._command.call_args[0][5] == mock.sentinel.serialize
         assert res == mock.sentinel.at_result
         t.serialize.reset_mock()
         api._command.reset_mock()
@@ -172,7 +173,7 @@ async def test_remote_at_cmd_no_rsp(api, monkeypatch):
 
 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():
+    for cmd_name, cmd_opts in xbee_api.COMMAND_REQUESTS.items():
         cmd_id, schema, repl = cmd_opts
         if schema:
             args = [ieee if isinstance(a(), t.EUI64) else a() for a in schema]
@@ -186,7 +187,7 @@ def test_frame_received(api, monkeypatch):
         return_value=(mock.sentinel.deserialize_data, b'')))
     my_handler = mock.MagicMock()
 
-    for cmd, cmd_opts in xbee_api.COMMANDS.items():
+    for cmd, cmd_opts in xbee_api.COMMAND_RESPONSES.items():
         cmd_id = cmd_opts[0]
         payload = b'\x01\x02\x03\x04'
         data = cmd_id.to_bytes(1, 'big') + payload
@@ -206,10 +207,10 @@ def test_frame_received_no_handler(api, monkeypatch):
     my_handler = mock.MagicMock()
     cmd = 'no_handler'
     cmd_id = 0x00
-    xbee_api.COMMANDS[cmd] = (cmd_id, (), None)
+    xbee_api.COMMAND_RESPONSES[cmd] = (cmd_id, (), None)
     api._commands_by_id[cmd_id] = cmd
 
-    cmd_opts = xbee_api.COMMANDS[cmd]
+    cmd_opts = xbee_api.COMMAND_RESPONSES[cmd]
     cmd_id = cmd_opts[0]
     payload = b'\x01\x02\x03\x04'
     data = cmd_id.to_bytes(1, 'big') + payload

tests/test_application.py

@@ -250,17 +250,17 @@ async def test_broadcast(app):
         0x260, 1, 2, 3, 0x0100, 0x06, 210, b'\x02\x01\x00'
     )
 
-    app._api._seq_command = mock.MagicMock(
+    app._api._command = mock.MagicMock(
         side_effect=asyncio.coroutine(mock.MagicMock())
     )
 
     await app.broadcast(
         profile, cluster, src_ep, dst_ep, grpid, radius, tsn, data)
-    assert app._api._seq_command.call_count == 1
-    assert app._api._seq_command.call_args[0][0] == 'tx_explicit'
-    assert app._api._seq_command.call_args[0][3] == src_ep
-    assert app._api._seq_command.call_args[0][4] == dst_ep
-    assert app._api._seq_command.call_args[0][9] == data
+    assert app._api._command.call_count == 1
+    assert app._api._command.call_args[0][0] == 'tx_explicit'
+    assert app._api._command.call_args[0][3] == src_ep
+    assert app._api._command.call_args[0][4] == dst_ep
+    assert app._api._command.call_args[0][9] == data
 
 
 @pytest.mark.asyncio
@@ -440,13 +440,13 @@ async def _test_request(app, do_reply=True, expect_reply=True, **kwargs):
     ieee = EUI64(b'\x01\x02\x03\x04\x05\x06\x07\x08')
     app.add_device(ieee, nwk)
 
-    def _mock_seq_command(cmdname, ieee, nwk, src_ep, dst_ep, cluster,
-                          profile, radius, options, data):
+    def _mock_command(cmdname, ieee, nwk, src_ep, dst_ep, cluster,
+                      profile, radius, options, data):
         if expect_reply:
             if do_reply:
                 app._pending[seq].set_result(mock.sentinel.reply_result)
 
-    app._api._seq_command = mock.MagicMock(side_effect=_mock_seq_command)
+    app._api._command = mock.MagicMock(side_effect=_mock_command)
     return await app.request(nwk, 0x0260, 1, 2, 3, seq, [4, 5, 6], expect_reply=expect_reply, **kwargs)
 
 

tests/test_types.py

@@ -62,6 +62,9 @@ class undEnum(t.uint8_t, t.UndefinedEnum):
     i = undEnum(0xEE)
     assert i.name == 'UNDEFINED_VALUE'
 
+    i = undEnum()
+    assert i is undEnum.OK
+
 
 def test_undefined_enum_undefinede():
     class undEnum(t.uint8_t, t.UndefinedEnum):

zigpy_xbee/api.py

@@ -30,15 +30,16 @@ class ModemStatus(t.uint8_t, t.UndefinedEnum):
 
 
 # https://www.digi.com/resources/documentation/digidocs/PDFs/90000976.pdf
-COMMANDS = {
+COMMAND_REQUESTS = {
     '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.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),
-
+}
+COMMAND_RESPONSES = {
     'at_response': (0x88, (t.FrameId, t.ATCommand, t.uint8_t, t.Bytes), None),
     'modem_status': (0x8A, (ModemStatus, ), None),
     'tx_status': (0x8B, (t.FrameId, t.NWK, t.uint8_t, t.uint8_t, t.uint8_t), None),
@@ -192,7 +193,7 @@ class XBee:
     def __init__(self):
         self._uart = None
         self._seq = 1
-        self._commands_by_id = {v[0]: k for k, v in COMMANDS.items()}
+        self._commands_by_id = {v[0]: k for k, v in COMMAND_RESPONSES.items()}
         self._awaiting = {}
         self._app = None
         self._cmd_mode_future = None
@@ -221,27 +222,24 @@ async def connect(self, device, baudrate=115200):
     def close(self):
         return self._uart.close()
 
-    def _command(self, name, *args):
+    def _command(self, name, *args, mask_frame_id=False):
         LOGGER.debug("Command %s %s", name, args)
-        data, needs_response = self._api_frame(name, *args)
+        frame_id = 0 if mask_frame_id else self._seq
+        data, needs_response = self._api_frame(name, frame_id, *args)
         self._uart.send(data)
         future = None
-        if needs_response:
+        if needs_response and frame_id:
             future = asyncio.Future()
-            self._awaiting[self._seq] = (future, )
+            self._awaiting[frame_id] = (future, )
         self._seq = (self._seq % 255) + 1
         return future
 
-    def _seq_command(self, name, *args):
-        LOGGER.debug("Sequenced command: %s %s", name, args)
-        return self._command(name, self._seq, *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], ))
         try:
             return await asyncio.wait_for(
-                self._command('remote_at', self._seq, ieee, nwk, options,
+                self._command('remote_at', ieee, nwk, options,
                               name.encode('ascii'), data,),
                 timeout=REMOTE_AT_COMMAND_TIMEOUT)
         except asyncio.TimeoutError:
@@ -253,7 +251,7 @@ async def _at_partial(self, cmd_type, name, *args):
         data = t.serialize(args, (AT_COMMANDS[name], ))
         try:
             return await asyncio.wait_for(
-                self._command(cmd_type, self._seq, name.encode('ascii'), data),
+                self._command(cmd_type, name.encode('ascii'), data),
                 timeout=AT_COMMAND_TIMEOUT)
         except asyncio.TimeoutError:
             LOGGER.warning("%s: No response to %s command", cmd_type, name)
@@ -263,13 +261,13 @@ async def _at_partial(self, cmd_type, name, *args):
     _queued_at = functools.partialmethod(_at_partial, 'queued_at')
 
     def _api_frame(self, name, *args):
-        c = COMMANDS[name]
+        c = COMMAND_REQUESTS[name]
         return (bytes([c[0]]) + t.serialize(args, c[1])), c[2]
 
     def frame_received(self, data):
         command = self._commands_by_id[data[0]]
         LOGGER.debug("Frame received: %s", command)
-        data, rest = t.deserialize(data[1:], COMMANDS[command][1])
+        data, rest = t.deserialize(data[1:], COMMAND_RESPONSES[command][1])
         try:
             getattr(self, '_handle_%s' % (command, ))(data)
         except AttributeError:
@@ -401,3 +399,8 @@ async def init_api_mode(self):
         LOGGER.debug(("Couldn't enter AT command mode at any known baudrate."
                       "Configure XBee manually for escaped API mode ATAP2"))
         return False
+
+    def __getattr__(self, item):
+        if item in COMMAND_REQUESTS:
+            return functools.partial(self._command, item)
+        raise AttributeError("Unknown command {}".format(item))

zigpy_xbee/zigbee/application.py

@@ -126,8 +126,7 @@ async def request(self, nwk, profile, cluster, src_ep, dst_ep, sequence, data, e
             self._pending[sequence] = reply_fut
 
         dev = self.get_device(nwk=nwk)
-        self._api._seq_command(
-            'tx_explicit',
+        self._api.tx_explicit(
             dev.ieee,
             nwk,
             src_ep,
@@ -214,7 +213,7 @@ def handle_rx(self, src_ieee, src_nwk, src_ep, dst_ep, cluster_id, profile_id, r
         try:
             tsn, command_id, is_reply, args = self.deserialize(device, src_ep, cluster_id, data)
         except ValueError as e:
-            LOGGER.error("Failed to parse message (%s) on cluster %d, because %s", binascii.hexlify(data), cluster_id, e)
+            LOGGER.error("Failed to parse message (%s) on cluster %s, because %s", binascii.hexlify(data), cluster_id, e)
             return
 
         if is_reply:
@@ -246,8 +245,7 @@ async def broadcast(self, profile, cluster, src_ep, dst_ep, grpid, radius,
         broadcast_as_bytes = [
             zigpy.types.uint8_t(b) for b in broadcast_address.to_bytes(8, 'big')
         ]
-        self._api._seq_command(
-            'tx_explicit',
+        self._api.tx_explicit(
             zigpy.types.EUI64(broadcast_as_bytes),
             broadcast_address,
             src_ep,