Skip the 60s bootloader delay on the CC2531

Author: puddly

tests/test_application.py

@@ -304,6 +304,41 @@ def on_endpoint_deletion(req):
     return app, znp_server
 
 
+@pytest_mark_asyncio_timeout(seconds=5)
+async def test_application_startup_skip_bootloader(application, mocker):
+    app, znp_server = application
+
+    first_uart_byte = None
+
+    def create_patched_write(original_write):
+        def patched_write(data):
+            nonlocal first_uart_byte
+
+            # Intercept the first byte if it's destined for the bootloader
+            is_for_bootloader = data[0] in c.ubl.BootloaderRunMode._value2member_map_
+
+            if first_uart_byte is None and is_for_bootloader:
+                first_uart_byte = data[0]
+                data = data[1:]
+
+            return original_write(data)
+
+        return patched_write
+
+    async def patched_uart_connect(config, api):
+        protocol = await uart_connect(config, api)
+        protocol.transport.write = create_patched_write(protocol.transport.write)
+
+        return protocol
+
+    mocker.patch("zigpy_znp.uart.connect", side_effect=patched_uart_connect)
+
+    app.update_config({conf.CONF_ZNP_CONFIG: {conf.CONF_SKIP_BOOTLOADER: True}})
+    await app.startup(auto_form=False)
+
+    assert first_uart_byte == c.ubl.BootloaderRunMode.FORCE_RUN
+
+
 @pytest_mark_asyncio_timeout(seconds=5)
 async def test_application_startup_nib(application):
     app, znp_server = application

zigpy_znp/api.py

@@ -155,12 +155,16 @@ def set_application(self, app):
     def _port_path(self) -> str:
         return self._config[conf.CONF_DEVICE][conf.CONF_DEVICE_PATH]
 
-    async def connect(self, *, test_port=True) -> None:
+    async def connect(self, *, test_port=True, skip_bootloader=False) -> None:
         assert self._uart is None
 
         try:
             self._uart = await uart.connect(self._config[conf.CONF_DEVICE], self)
 
+            if skip_bootloader:
+                LOGGER.debug("Sending first UART byte to the bootloader")
+                self._uart._transport_write(bytes([c.ubl.BootloaderRunMode.FORCE_RUN]))
+
             if test_port:
                 LOGGER.debug(
                     "Testing connection to %s", self._uart.transport.serial.name

zigpy_znp/commands/ubl.py

@@ -29,6 +29,12 @@ class BootloaderDeviceType(t.enum_uint8):
     CC2530 = 2
 
 
+class BootloaderRunMode(t.enum_uint8):
+    # Read the code, not the spec
+    FORCE_BOOT = 0x10
+    FORCE_RUN = FORCE_BOOT ^ 0xFF
+
+
 class UBL(t.CommandsBase, subsystem=t.Subsystem.UBL_FUNC):
     WriteReq = t.CommandDef(
         t.CommandType.AREQ,

zigpy_znp/config.py

@@ -40,6 +40,7 @@
 CONF_TX_POWER = "tx_power"
 CONF_SREQ_TIMEOUT = "sync_request_timeout"
 CONF_AUTO_RECONNECT_RETRY_DELAY = "auto_reconnect_retry_delay"
+CONF_SKIP_BOOTLOADER = "skip_bootloader"
 
 CONFIG_SCHEMA = CONFIG_SCHEMA.extend(
     {
@@ -53,6 +54,7 @@
                 vol.Optional(
                     CONF_AUTO_RECONNECT_RETRY_DELAY, default=5
                 ): VolPositiveNumber,
+                vol.Optional(CONF_SKIP_BOOTLOADER, default=True): cv_boolean,
             }
         ),
     }

zigpy_znp/uart.py

@@ -77,7 +77,9 @@ def data_received(self, data: bytes) -> None:
 
     def send(self, payload: frames.GeneralFrame) -> None:
         """Sends data taking care of framing."""
-        data = frames.TransportFrame(payload).serialize()
+        self._transport_write(frames.TransportFrame(payload).serialize())
+
+    def _transport_write(self, data: bytes) -> None:
         LOGGER.trace("Sending data: %s", Bytes.__repr__(data))
         self.transport.write(data)
 

zigpy_znp/zigbee/application.py

@@ -303,7 +303,9 @@ async def startup(self, auto_form=False):
         znp = ZNP(self.config)
         znp.set_application(self)
         self._bind_callbacks(znp)
-        await znp.connect()
+        await znp.connect(
+            skip_bootloader=self.config[conf.CONF_ZNP_CONFIG][conf.CONF_SKIP_BOOTLOADER]
+        )
 
         self._znp = znp