Class RpcClient

ci/rpc_client.py:101–740

Stateful JSON-RPC client for serial communication. Accepts a SerialInterface for serial I/O. Use create_serial_interface() to create the appropriate backend (pyserial or fbuild). If no serial_interface is provided, an fbuild backend is created by default. To use pyserial, create th

class RpcClient:
    """Stateful JSON-RPC client for serial communication.

    Accepts a SerialInterface for serial I/O. Use create_serial_interface()
    to create the appropriate backend (pyserial or fbuild).

    If no serial_interface is provided, an fbuild backend is created by default.
    To use pyserial, create the interface externally and pass it in:

        from ci.util.serial_interface import create_serial_interface
        iface = create_serial_interface(port, use_pyserial=True)
        client = RpcClient(port, serial_interface=iface)

    Attributes:
        port: Serial port path
        baudrate: Serial baud rate (default: 115200)
        timeout: Default RPC response timeout in seconds
        response_prefix: Prefix for RPC responses (default: "REMOTE: ")
    """

    RESPONSE_PREFIX = "REMOTE: "

    # Post-ACK timeout floor for async (Phase 3) RPCs. After the device sends
    # `{"acknowledged": true}` the client restarts its wait with at least this
    # many seconds for the final response (or the caller-supplied timeout,
    # whichever is larger). Picked generously so a 100-LED runSingleTest on a
    # slow MCU has headroom; the typical run on Teensy 4 finishes in 80-90s.
    # See FastLED#3060.
    ASYNC_POST_ACK_TIMEOUT: float = 600.0

    def __init__(
        self,
        port: str,
        baudrate: int = 115200,
        timeout: float = 10.0,
        read_timeout: float = 0.5,
        serial_interface: SerialInterface | None = None,
        verbose: bool = False,
        crash_decoder: CrashTraceDecoder | None = None,
    ):
        """Initialize RPC client.

        Args:
            port: Serial port path (e.g., "/dev/ttyUSB0", "COM3")
            baudrate: Serial baud rate (default: 115200)
            timeout: Default timeout for RPC operations in seconds
            read_timeout: Timeout for individual serial reads (unused with fbuild)
            serial_interface: Pre-created SerialInterface. If None, an fbuild
                backend is created automatically on connect().
            verbose: If True, enable detailed RPC logging
            crash_decoder: Optional CrashTraceDecoder for inline crash decoding.
        """
        self.port = port
        self.baudrate = baudrate
        self.timeout = timeout
        self.read_timeout = read_timeout
        self.verbose = verbose
        self._serial: SerialInterface | None = serial_interface