Fix bootloader skip for Slaesh's stick

README.md

@@ -37,10 +37,11 @@ Launch Home Assistant with the `--skip-pip` command line option to prevent zigpy
  - Add https://github.com/home-assistant/hassio-addons-development as an addon repository.
  - Install the "Custom deps deployment" addon.
  - Add the following to your `configuration.yaml` file:
-	```yaml
-	pypi:
-	  - git+https://github.com/zigpy/zigpy-znp/
-	```
+   ```yaml
+   apk: []
+   pypi:
+     - git+https://github.com/zigpy/zigpy-znp/
+	 ```
 
 # Configuration
 Below are the defaults with the top-level Home Assistant `zha:` key.

tests/api/test_connect.py

@@ -102,17 +102,9 @@ async def test_connect_skip_bootloader_rts_dtr_pins(make_znp_server, mocker):
     mocker.patch.object(znp.nvram, "determine_alignment", new=CoroutineMock())
     mocker.patch.object(znp, "detect_zstack_version", new=CoroutineMock())
 
-    num_pings = 0
-
-    def ping_rsp(req):
-        nonlocal num_pings
-        num_pings += 1
-
-        if num_pings >= 3:
-            return c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)
-
-    # Ignore the first few pings so we trigger the pin toggling
-    znp_server.reply_to(c.SYS.Ping.Req(), responses=ping_rsp)
+    znp_server.reply_to(
+        c.SYS.Ping.Req(), responses=[c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)]
+    )
 
     await znp.connect(test_port=True)
 

zigpy_znp/api.py

@@ -38,7 +38,7 @@
 NETWORK_COMMISSIONING_TIMEOUT = 30
 BOOTLOADER_PIN_TOGGLE_DELAY = 0.15
 CONNECT_PING_TIMEOUT = 0.50
-CONNECT_PROBE_TIMEOUT = 5.0
+CONNECT_PROBE_TIMEOUT = 10
 
 
 class ZNP:
@@ -468,6 +468,16 @@ async def _skip_bootloader(self) -> c.SYS.Ping.Rsp:
         """
 
         async def ping_task():
+            LOGGER.debug("Toggling RTS/DTR pins to skip bootloader or reset chip")
+
+            # The default sequence is DTR=false and RTS toggling false/true/false
+            for dtr, rts in zip(
+                self._znp_config[conf.CONF_CONNECT_DTR_STATES],
+                self._znp_config[conf.CONF_CONNECT_RTS_STATES],
+            ):
+                self._uart.set_dtr_rts(dtr=dtr, rts=rts)
+                await asyncio.sleep(BOOTLOADER_PIN_TOGGLE_DELAY)
+
             # First, just try pinging
             try:
                 async with async_timeout.timeout(CONNECT_PING_TIMEOUT):
@@ -478,30 +488,18 @@ async def ping_task():
             # If that doesn't work, send the bootloader skip bytes and try again.
             # Sending a bunch at a time fixes UART issues when the radio was previously
             # probed with another library at a different baudrate.
+            LOGGER.debug("Sending CC253x bootloader skip bytes")
             self._uart.write(256 * bytes([c.ubl.BootloaderRunMode.FORCE_RUN]))
 
             await asyncio.sleep(AFTER_BOOTLOADER_SKIP_BYTE_DELAY)
 
-            try:
-                async with async_timeout.timeout(CONNECT_PING_TIMEOUT):
-                    return await self.request(c.SYS.Ping.Req())
-            except asyncio.TimeoutError:
-                pass
-
-            # This is normally done just for Slaesh's CC2652RB stick
-            LOGGER.debug("Toggling RTS/DTR pins to skip bootloader or reset chip")
-
-            # The default sequence is DTR=false and RTS toggling false/true/false
-            for dtr, rts in zip(
-                self._znp_config[conf.CONF_CONNECT_DTR_STATES],
-                self._znp_config[conf.CONF_CONNECT_RTS_STATES],
-            ):
-                self._uart.set_dtr_rts(dtr=dtr, rts=rts)
-                await asyncio.sleep(BOOTLOADER_PIN_TOGGLE_DELAY)
-
-            # At this point we have nothing else to try, don't catch the timeout
-            async with async_timeout.timeout(CONNECT_PING_TIMEOUT):
-                return await self.request(c.SYS.Ping.Req())
+            # At this point we have nothing left to try
+            while True:
+                try:
+                    async with async_timeout.timeout(2 * CONNECT_PING_TIMEOUT):
+                        return await self.request(c.SYS.Ping.Req())
+                except asyncio.TimeoutError:
+                    pass
 
         async with self.capture_responses([CatchAllResponse()]) as responses:
             ping_task = asyncio.create_task(ping_task())

zigpy_znp/zigbee/application.py

@@ -135,15 +135,15 @@ async def probe(cls, device_config: conf.ConfigType) -> bool:
         LOGGER.debug("Probing %s", znp._port_path)
 
         try:
-            async with async_timeout.timeout(PROBE_TIMEOUT):
-                await znp.connect()
-
-            return True
+            # `ZNP.connect` times out on its own
+            await znp.connect()
         except Exception as e:
             LOGGER.debug(
                 "Failed to probe ZNP radio with config %s", device_config, exc_info=e
             )
             return False
+        else:
+            return True
         finally:
             znp.close()