Implement command priority (#240)

pyproject.toml

@@ -15,7 +15,7 @@ license = {text = "GPL-3.0"}
 requires-python = ">=3.8"
 dependencies = [
     "voluptuous",
-    "zigpy>=0.60.0",
+    "zigpy>=0.60.2",
     'async-timeout; python_version<"3.11"',
 ]
 

tests/test_api.py

@@ -1031,3 +1031,17 @@ async def test_firmware_responding_with_wrong_type_with_correct_seq(
         "Firmware responded incorrectly (Response is mismatched! Sent"
         " <CommandId.aps_data_confirm: 4>, received <CommandId.version: 13>), retrying"
     ) in caplog.text
+
+
+def test_get_command_priority(api):
+    assert (
+        api._get_command_priority(
+            deconz_api.Command(command_id=deconz_api.CommandId.write_parameter)
+        )
+        > api._get_command_priority(
+            deconz_api.Command(command_id=deconz_api.CommandId.update_neighbor)
+        )
+        > api._get_command_priority(
+            deconz_api.Command(command_id=deconz_api.CommandId.aps_data_request)
+        )
+    )

zigpy_deconz/api.py

@@ -15,6 +15,7 @@
     from asyncio import timeout as asyncio_timeout  # pragma: no cover
 
 from zigpy.config import CONF_DEVICE_PATH
+from zigpy.datastructures import PriorityLock
 from zigpy.types import (
     APSStatus,
     Bool,
@@ -419,7 +420,7 @@ def __init__(self, app: Callable, device_config: dict[str, Any]):
 
         # [seq][cmd_id] = [fut1, fut2, ...]
         self._awaiting = collections.defaultdict(lambda: collections.defaultdict(list))
-        self._command_lock = asyncio.Lock()
+        self._command_lock = PriorityLock()
         self._config = device_config
         self._device_state = DeviceState(
             network_state=NetworkState2.OFFLINE,
@@ -489,6 +490,16 @@ def close(self):
             self._uart.close()
             self._uart = None
 
+    def _get_command_priority(self, command: Command) -> int:
+        return {
+            # The watchdog is fed using `write_parameter` and `get_device_state` so they
+            # must take priority
+            CommandId.write_parameter: 999,
+            CommandId.device_state: 999,
+            # APS data requests are retried and can be deprioritized
+            CommandId.aps_data_request: -1,
+        }.get(command.command_id, 0)
+
     async def send_command(self, cmd, **kwargs) -> Any:
         while True:
             try:
@@ -557,7 +568,7 @@ async def _command(self, cmd, **kwargs):
             # connection was lost
             raise CommandError(Status.ERROR, "API is not running")
 
-        async with self._command_lock:
+        async with self._command_lock(priority=self._get_command_priority(command)):
             seq = self._seq
 
             LOGGER.debug("Sending %s%s (seq=%s)", cmd, kwargs, seq)

zigpy_deconz/zigbee/application.py

@@ -64,7 +64,7 @@ class ControllerApplication(zigpy.application.ControllerApplication):
         {zigpy.config.CONF_DEVICE_BAUDRATE: 115200},
     ]
 
-    _watchdog_period = 600 * 0.75
+    _watchdog_period = 30
 
     def __init__(self, config: dict[str, Any]):
         """Initialize instance."""
@@ -85,7 +85,7 @@ async def _watchdog_feed(self):
             and self._api.firmware_version <= 0x26450900
         ):
             await self._api.write_parameter(
-                NetworkParameter.watchdog_ttl, int(self._watchdog_period / 0.75)
+                NetworkParameter.watchdog_ttl, int(2 * self._watchdog_period)
             )
         else:
             await self._api.get_device_state()