Add a command queue so there is always only one pending command (#112)

On RaspBee II modules (Firmware 26520700) sending multiple concurrent commands may result in missing command responses and duplicate device_state_changed commands. Sending commands one at a time with waiting for their response, seems to fix this issue.

Author: jloehr

tests/test_api.py

@@ -10,7 +10,7 @@
 
 
 @pytest.fixture
-def api():
+def api(event_loop):
     api = deconz_api.Deconz()
     api._uart = mock.MagicMock()
     return api
@@ -76,6 +76,26 @@ async def mock_fut():
         api._uart.send.reset_mock()
 
 
+@pytest.mark.asyncio
+async def test_command_queue(api, monkeypatch):
+    def mock_api_frame(name, *args):
+        return mock.sentinel.api_frame_data, api._seq
+
+    api._api_frame = mock.MagicMock(side_effect=mock_api_frame)
+    api._uart.send = mock.MagicMock()
+
+    monkeypatch.setattr(deconz_api, "COMMAND_TIMEOUT", 0.1)
+
+    for cmd, cmd_opts in deconz_api.TX_COMMANDS.items():
+        async with api._command_lock:
+            with pytest.raises(asyncio.TimeoutError):
+                await asyncio.wait_for(api._command(cmd, mock.sentinel.cmd_data), 0.1)
+        assert api._api_frame.call_count == 0
+        assert api._uart.send.call_count == 0
+        api._api_frame.reset_mock()
+        api._uart.send.reset_mock()
+
+
 @pytest.mark.asyncio
 async def test_command_timeout(api, monkeypatch):
     def mock_api_frame(name, *args):

zigpy_deconz/api.py

@@ -193,11 +193,11 @@ def __init__(self):
         self._seq = 1
         self._awaiting = {}
         self._app = None
-        self._cmd_mode_future = None
-        self._conn_lost_task = None
+        self._command_lock = asyncio.Lock()
+        self._conn_lost_task: typing.Optional[asyncio.Task] = None
+        self._data_indication: bool = False
+        self._data_confirm: bool = False
         self._device_state = DeviceState(NetworkState.OFFLINE)
-        self._data_indication = False
-        self._data_confirm = False
         self._proto_ver = None
         self._aps_data_ind_flags = 0x01
 
@@ -263,20 +263,23 @@ def close(self):
             self._uart = None
 
     async def _command(self, cmd, *args):
-        LOGGER.debug("Command %s %s", cmd, args)
         if self._uart is None:
             # connection was lost
             raise CommandError(Status.ERROR, "API is not running")
-        data, seq = self._api_frame(cmd, *args)
-        self._uart.send(data)
-        fut = asyncio.Future()
-        self._awaiting[seq] = fut
-        try:
-            return await asyncio.wait_for(fut, timeout=COMMAND_TIMEOUT)
-        except asyncio.TimeoutError:
-            LOGGER.warning("No response to '%s' command", cmd)
-            self._awaiting.pop(seq)
-            raise
+        async with self._command_lock:
+            LOGGER.debug("Command %s %s", cmd, args)
+            data, seq = self._api_frame(cmd, *args)
+            self._uart.send(data)
+            fut = asyncio.Future()
+            self._awaiting[seq] = fut
+            try:
+                return await asyncio.wait_for(fut, timeout=COMMAND_TIMEOUT)
+            except asyncio.TimeoutError:
+                LOGGER.warning(
+                    "No response to '%s' command with seq id '0x%02x'", cmd, seq
+                )
+                self._awaiting.pop(seq)
+                raise
 
     def _api_frame(self, cmd, *args):
         schema = TX_COMMANDS[cmd]