pbio/drv/usb: Convert to serial

[laurensvalk]

The previous WebUSB driver was a promising start, but it was not reliable. Messages could get stuck and the hub would not know if the app was disconnected. Bidirectional traffic was slow and prone to lockups.

This replaces the USB drivers on all platforms with a standard serial USB device. We will be able to use this with Web Serial on most systems. Instead of manually subscribing and unsubscribing to keep track of the app connection state, we can use the DTR signal which is asserted on connect and deasserted on disconnect, even when the browser tab is abrubtly closed.

Since serial is handled by the OS rather than our host application, in-flight messages don't get stuck if the host app is not reading them, which was part of the reason we had lockups before. This should also make it possible to use it with RFCOMM so we can add Bluetooth support for EV3 and NXT with relatively little changes.

Although the medium is a serial stream, we keep the same packetized event messages as before, much like we do for BLE. Frames are encoded with COBS with a 0x00 delimiter between messages, which makes it easy to get back in sync.

Summary

The most important is the change in protocol.h. Subscribe is no longer needed and we add a read request and reply for the device information.

typedef enum {
    /** Reply to a ::PBIO_PYBRICKS_OUT_EP_MSG_COMMAND. */
    PBIO_PYBRICKS_IN_EP_MSG_RESPONSE = 1,
    /** Analog to BLE notification. Emitted while a host is connected. */
    PBIO_PYBRICKS_IN_EP_MSG_EVENT = 2,
+   /** Reply to a ::PBIO_PYBRICKS_OUT_EP_MSG_READ.  */
+   PBIO_PYBRICKS_IN_EP_MSG_READ_REPLY = 3,
} pbio_pybricks_usb_in_ep_msg_t;

typedef enum {
-   PBIO_PYBRICKS_OUT_EP_MSG_SUBSCRIBE = 1,
+   /** A characteristic read request.  */
+   PBIO_PYBRICKS_OUT_EP_MSG_READ = 1,
    /** A Pybricks command  */
    PBIO_PYBRICKS_OUT_EP_MSG_COMMAND = 2,
} pbio_pybricks_usb_out_ep_msg_t;

Next steps

• At the moment, we preserve the logic in pbdrv_usb_process_thread that effectively awaits one pbdrv_usb_tx_chunk per message that we send. This keeps this commit to the point, with the main focus on the driver update. But since the outgoing path is a serial stream, this common driver could just append it to an outgoing ring buffer, equivalent to the rx path. So we could probably do some refactoring with the various buffers we have today, and drop some assumptions about whole-packets in various places.
• Generalize information service: This is already an improvement in that the individual drivers (stm32, ev3, ...) no longer call into pbsys for hub version information. But it still lives in the common driver. This belongs in sys/host. We could do the same clean up for the bluetooth drivers which hardwire this too, but that touches too much code to do it right now.

Testing
This PR needs a complementary change to Pybricks Code, which is currently in progress. The following script can be used to start an existing slot, like the REPL:

~/git-pybricks/pybricks-micropython$ python run_stop_test.py --slot 128

The output should be:

Found MINDSTORMS EV3 on /dev/ttyACM0
Connected to /dev/ttyACM0 (DTR asserted)
--- sending flush delimiter ---
--- reading hub info ---
--- starting program in slot 128 ---
[read] hub name: Pybricks Hub
[status] 00 10 00 00 80 00
[read] software version: 1.5.0
[response] error=0
[status] 40 10 00 00 80 00
Pybricks MicroPython local-build-v4.0.0-3-g50bddc14d on 2026-06-18; MINDSTORMS EV3 Brick with TI Sitara AM1808
Type "help()" for more information.
>>> --- stopping program ---
[response] error=0

SystemExit: 
[status] 00 10 00 00 80 00
Done.

This is also a good overview that this encoding is very minimal, and far simpler than trying to stay in sync because we don't need to understand the context of each message at the driver level.

#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 The Pybricks Authors

"""
Manual test for the Pybricks USB CDC-ACM serial protocol.

This connects to a Pybricks hub over its virtual serial port, starts the user
program in slot 0, prints any stdout the program produces for a few seconds, then
stops the program.

It mirrors what a Pybricks app does over the byte stream:

  * We can assert DTR, which the firmware treats as "an app
    connected" (the USB analog of subscribing to BLE notifications).
  * Host -> hub: COBS-framed messages terminated with a 0x00 byte. The first
    decoded byte selects the type: 1 = characteristic read request, 2 = command.
  * Hub -> host: COBS-framed messages. The first decoded byte selects the
    message type: 1 = command response, 2 = event, 3 = read reply.

Requires pyserial, so run this in the Pybricks MicroPython virtual environment (venv).
"""

import argparse
import struct
import sys
import threading
import time

import serial
from serial.tools import list_ports

# USB identifiers (see lib/lego/lego/usb.h).
LEGO_USB_VID = 0x0694
HUB_PIDS = {
    0x0009: "SPIKE Prime",
    0x000D: "SPIKE Essential",
    0x0010: "MINDSTORMS Robot Inventor",
    0x0005: "MINDSTORMS EV3",
    0x0002: "MINDSTORMS NXT",
}

# Host -> hub command types (see pbio_pybricks_command_t in protocol.h).
COMMAND_STOP_USER_PROGRAM = 0
COMMAND_START_USER_PROGRAM = 1

# Host -> hub message types (see pbio_pybricks_usb_out_ep_msg_t in protocol.h).
OUT_EP_MSG_READ = 1
OUT_EP_MSG_COMMAND = 2

# Hub -> host message types (see pbio_pybricks_usb_in_ep_msg_t in protocol.h).
IN_EP_MSG_RESPONSE = 1
IN_EP_MSG_EVENT = 2
IN_EP_MSG_READ_REPLY = 3

# Hub -> host event types (see pbio_pybricks_event_t in protocol.h).
EVENT_STATUS_REPORT = 0
EVENT_WRITE_STDOUT = 1
EVENT_WRITE_APP_DATA = 2
EVENT_WRITE_TELEMETRY = 3

# Characteristic namespaces for read requests
# (see PBIO_PYBRICKS_USB_INTERFACE_READ_CHARACTERISTIC_* in protocol.h).
READ_CHARACTERISTIC_GATT = 0x01
READ_CHARACTERISTIC_PYBRICKS = 0x02

# Characteristics to read. Strings are UTF-8; capabilities is a binary blob
# (see pbio_pybricks_hub_capabilities in protocol.h).
READS = [
    (READ_CHARACTERISTIC_GATT, 0x2A00, "hub name", True),
    (READ_CHARACTERISTIC_GATT, 0x2A26, "firmware version", True),
    (READ_CHARACTERISTIC_GATT, 0x2A28, "software version", True),
    (READ_CHARACTERISTIC_PYBRICKS, 0x0003, "hub capabilities", False),
]
READ_INFO = {(service, char_id): (name, text) for service, char_id, name, text in READS}

COBS_DELIMITER = 0x00


def cobs_encode(data: bytes) -> bytes:
    """COBS-encodes ``data``. The result never contains a zero byte."""
    out = bytearray()
    code_idx = len(out)
    out.append(0)  # placeholder for the first code byte
    code = 1

    for byte in data:
        if byte == 0:
            out[code_idx] = code
            code_idx = len(out)
            out.append(0)
            code = 1
        else:
            out.append(byte)
            code += 1
            if code == 0xFF:
                out[code_idx] = code
                code_idx = len(out)
                out.append(0)
                code = 1

    out[code_idx] = code
    return bytes(out)


def cobs_decode(data: bytes) -> bytes:
    """COBS-decodes a single frame (delimiter already stripped)."""
    out = bytearray()
    idx = 0
    n = len(data)

    while idx < n:
        code = data[idx]
        idx += 1
        for _ in range(1, code):
            if idx >= n:
                return b""  # malformed
            out.append(data[idx])
            idx += 1
        if code < 0xFF and idx < n:
            out.append(0)

    return bytes(out)


def find_hub_port(explicit_port: str | None) -> str:
    """Returns the serial device path for a connected hub."""
    if explicit_port:
        return explicit_port

    for port in list_ports.comports():
        if port.vid == LEGO_USB_VID and port.pid in HUB_PIDS:
            print(f"Found {HUB_PIDS[port.pid]} on {port.device}")
            return port.device

    sys.exit(
        "No Pybricks hub found. Plug in the hub, or pass --port explicitly.\n"
        "Available ports:\n  "
        + "\n  ".join(
            f"{p.device} (vid={p.vid:#06x} pid={p.pid:#06x})"
            for p in list_ports.comports()
            if p.vid
        )
    )


def send_frame(ser: serial.Serial, message: bytes) -> None:
    """COBS-encodes ``message`` and writes it as a delimited frame."""
    ser.write(cobs_encode(message) + bytes([COBS_DELIMITER]))
    ser.flush()


def send_command(ser: serial.Serial, payload: bytes) -> None:
    """Sends a Pybricks command (prefixed with the OUT command discriminator)."""
    send_frame(ser, bytes([OUT_EP_MSG_COMMAND]) + payload)


def send_read(ser: serial.Serial, service: int, char_id: int) -> None:
    """Sends a characteristic read request; the hub replies with a READ_REPLY."""
    send_frame(
        ser, bytes([OUT_EP_MSG_READ, service, char_id & 0xFF, (char_id >> 8) & 0xFF])
    )


def handle_message(message: bytes) -> None:
    """Decodes and prints a single hub -> host message."""
    if not message:
        return

    msg_type = message[0]
    body = message[1:]

    if msg_type == IN_EP_MSG_RESPONSE:
        (error,) = struct.unpack_from("<I", body) if len(body) >= 4 else (None,)
        print(f"[response] error={error}")

    elif msg_type == IN_EP_MSG_EVENT:
        event_type = body[0] if body else None
        data = body[1:]
        if event_type == EVENT_WRITE_STDOUT:
            sys.stdout.write(data.decode("utf-8", errors="replace"))
            sys.stdout.flush()
        elif event_type == EVENT_STATUS_REPORT:
            print(f"[status] {data.hex(' ')}")
        else:
            print(f"[event {event_type}] {data.hex(' ')}")

    elif msg_type == IN_EP_MSG_READ_REPLY:
        service = body[0] if len(body) >= 1 else None
        char_id = body[1] | (body[2] << 8) if len(body) >= 3 else None
        value = body[3:]
        name, text = READ_INFO.get((service, char_id), (None, False))
        label = name or f"service {service} char {char_id}"
        if not value:
            print(f"[read] {label}: <unknown characteristic>")
        elif text:
            print(f"[read] {label}: {value.decode('utf-8', errors='replace')}")
        else:
            print(f"[read] {label}: {value.hex(' ')}")

    else:
        print(f"[unknown msg type {msg_type}] {body.hex(' ')}")


def reader_loop(ser: serial.Serial, stop_event: threading.Event) -> None:
    """Reads bytes, reassembles COBS frames, and dispatches messages."""
    buffer = bytearray()

    while not stop_event.is_set():
        chunk = ser.read(256)
        if not chunk:
            continue

        for byte in chunk:
            if byte == COBS_DELIMITER:
                if buffer:
                    handle_message(cobs_decode(bytes(buffer)))
                    buffer.clear()
            else:
                buffer.append(byte)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--port", help="serial port (autodetected if omitted)")
    parser.add_argument("--slot", type=int, default=0, help="program slot to run")
    parser.add_argument(
        "--duration", type=float, default=2.0, help="seconds to run before stopping"
    )
    args = parser.parse_args()

    port = find_hub_port(args.port)

    # Opening the port asserts DTR, which signals the hub that an app connected.
    with serial.Serial(port, baudrate=115200, timeout=0.1) as ser:

        # Starts asserted, but we want to assert it later.
        ser.dtr = False
        print(f"Connected to {port} (DTR asserted)")

        stop_event = threading.Event()
        reader = threading.Thread(
            target=reader_loop, args=(ser, stop_event), daemon=True
        )
        reader.start()

        # Send a lone delimiter to resync the hub's COBS framing on connect.
        # This discards any partial frame left in the hub's assembler by a
        # previous port opener (e.g. ModemManager's AT probe writes bytes with
        # no frame delimiter), so our first real command can't be corrupted.
        print("--- sending flush delimiter ---")
        ser.write(bytes([COBS_DELIMITER]))
        ser.flush()

        # Read hub info. The replies arrive asynchronously and are printed by
        # the reader thread.
        print("--- reading hub info ---")
        for service, char_id, _name, _text in READS:
            send_read(ser, service, char_id)


        # Hub uses this signal to start sending events.
        ser.dtr = True

        print(f"--- starting program in slot {args.slot} ---")
        send_command(ser, bytes([COMMAND_START_USER_PROGRAM, args.slot]))

        try:
            time.sleep(args.duration)
        except KeyboardInterrupt:
            pass

        print("--- stopping program ---")
        send_command(ser, bytes([COMMAND_STOP_USER_PROGRAM]))

        # Let final output / response arrive.
        time.sleep(0.5)
        stop_event.set()
        reader.join(timeout=1.0)

    print("Done.")


if __name__ == "__main__":
    main()