Implement transport.set_signals

assets/manifests/skyconnect.json

@@ -39,5 +39,6 @@
 			"version": "2.2.2.0"
 		}
 	],
-	"allow_custom_firmware_upload": true
+	"allow_custom_firmware_upload": true,
+	"bootloader_gpio_reset": null
 }

requirements.txt

@@ -1 +1 @@
-universal-silabs-flasher==0.0.26
+universal-silabs-flasher==0.0.28

src/const.ts

@@ -32,6 +32,10 @@ export enum FirmwareType {
   UNKNOWN = 'unknown',
 }
 
+export enum BootloaderGpioReset {
+  RTS_DTR = 'rts_dtr',
+}
+
 export const LegacyTypeToFirmwareType = {
   'ncp-uart-hw': FirmwareType.ZIGBEE_NCP,
   'ncp-uart-sw': FirmwareType.ZIGBEE_NCP,
@@ -91,5 +95,6 @@ export interface Manifest {
   baudrates: ManifestBaudrates;
   usb_filters: USBFilter[];
   firmwares: Firmware[];
+  bootloader_gpio_reset?: BootloaderGpioReset;
   allow_custom_firmware_upload: boolean;
 }

src/flashing-dialog.ts

@@ -280,11 +280,12 @@ export class FlashingDialog extends LitElement {
         `
       from universal_silabs_flasher.flasher import Flasher
 
-      def create_flasher(baudrates, probe_methods, device):
+      def create_flasher(baudrates, probe_methods, device, bootloader_reset):
           return Flasher(
               baudrates=baudrates.to_py(),
               probe_methods=probe_methods.to_py(),
               device=device,
+              bootloader_reset=bootloader_reset,
           )
 
       create_flasher
@@ -309,6 +310,7 @@ export class FlashingDialog extends LitElement {
           PyApplicationType.ROUTER,
         ],
         device: '/dev/webserial', // the device name is ignored
+        bootloader_reset: this.manifest.bootloader_gpio_reset,
       });
 
     await this.detectRunningFirmware();
@@ -320,6 +322,7 @@ export class FlashingDialog extends LitElement {
     try {
       await this.pyFlasher.probe_app_type();
     } catch (e) {
+      console.error('Probing failed: ', e);
       this.pyFlasher = undefined;
       this.serialPort = undefined;
 

src/requirements.txt

@@ -30,7 +30,7 @@ referencing==0.35.1
 rpds-py==0.20.1
 setuptools==75.3.0
 typing_extensions==4.12.2
-universal-silabs-flasher==0.0.26
+universal-silabs-flasher==0.0.28
 voluptuous==0.15.2
 yarl==1.17.1
 zigpy==0.71.0

src/setup-pyodide.ts

@@ -66,7 +66,16 @@ function parseRequirementsTxt(requirementsTxt: string): Map<string, string> {
   const lineEnding = requirementsTxt.includes('\r\n') ? '\r\n' : '\n';
 
   for (const line of requirementsTxt.trim().split(lineEnding)) {
-    const [pkg, version] = line.split('==');
+    let pkg, version;
+
+    if (!line.startsWith('./')) {
+      [pkg, version] = line.split('==');
+    } else {
+      // Local dependencies
+      pkg = line.trim();
+      version = '0.0.0';
+    }
+
     packages.set(pkg, version);
   }
 
@@ -106,7 +115,12 @@ export async function setupPyodide(
 
   for (const [pkg, version] of requirementsTxt) {
     if (!MOCKED_MODULES.find(m => m.package === pkg)) {
-      requirements.push(`${pkg}==${version}`);
+      if (!pkg.startsWith('./')) {
+        requirements.push(`${pkg}==${version}`);
+      } else {
+        const url = new URL(pkg, window.location.href);
+        requirements.push(url.href);
+      }
     }
   }
 

src/webserial_transport.py

@@ -83,6 +83,26 @@ async def _reader_loop(self) -> None:
 
             self._protocol.data_received(bytes(result.value))
 
+    async def set_signals(
+        self, rts: bool | None = None, dtr: bool | None = None, **kwargs: bool | None
+    ) -> None:
+        other_signals = {k: v for k, v in kwargs.items() if v is not None}
+        if other_signals:
+            _LOGGER.warning(
+                "Ignoring unsupported flow control signals: %s", other_signals
+            )
+
+        signals = {}
+
+        if rts is not None:
+            signals["requestToSend"] = rts
+
+        if dtr is not None:
+            signals["dataTerminalReady"] = dtr
+
+        if signals:
+            await self._port.setSignals(**signals)
+
     def write(self, data: bytes) -> None:
         self._write_queue.put_nowait(data)
 
@@ -153,6 +173,8 @@ async def create_serial_connection(
     rtscts=False,
     xonxoff=False,
 ) -> tuple[WebSerialTransport, asyncio.Protocol]:
+    _LOGGER.debug("Opening a serial connection at %d with rtscts=%s", baudrate, rtscts)
+
     while _SERIAL_PORT_CLOSING_TASKS:
         _LOGGER.warning(
             "Serial connection was not closed before a new one was opened!"
@@ -163,7 +185,7 @@ async def create_serial_connection(
     # `url` is ignored, `_SERIAL_PORT` is used instead
     await _SERIAL_PORT.open(
         baudRate=baudrate,
-        flowControl="hardware" if rtscts else None,
+        flowControl="hardware" if rtscts else "none",
     )
 
     protocol = protocol_factory()