Function run_sweep

ci/autoresearch/driver_sweep.py:175–424

Run the full sweep test matrix on the device. Args: port: Serial port path driver: Driver name (e.g. "PARLIO", "RMT", "SPI") verbose: Enable verbose RPC output timeout: Per-test timeout in seconds Returns: List of TestResult objects

(
    port: str,
    driver: str,
    verbose: bool = False,
    timeout: float = 120.0,
)

async def run_sweep(
    port: str,
    driver: str,
    verbose: bool = False,
    timeout: float = 120.0,
) -> list[TestResult]:
    """Run the full sweep test matrix on the device.

    Args:
        port: Serial port path
        driver: Driver name (e.g. "PARLIO", "RMT", "SPI")
        verbose: Enable verbose RPC output
        timeout: Per-test timeout in seconds

    Returns:
        List of TestResult objects
    """
    test_matrix = build_test_matrix(driver)
    results: list[TestResult] = []

    print(f"\n{'=' * 70}")
    print(f"  {driver} Sweep Test")
    print(f"  {len(test_matrix)} test configurations")
    print(f"  Port: {port}")
    print(f"{'=' * 70}")
    print()

    # Print test matrix
    print("Test Matrix:")
    print(f"  {'#':<3} {'Config':<40} {'Sizes':<20} {'API':<10}")
    print(f"  {'-' * 3} {'-' * 40} {'-' * 20} {'-' * 10}")
    for i, tc in enumerate(test_matrix, 1):
        sizes_str = ",".join(str(s) for s in tc.lane_sizes)
        api = "legacy" if tc.use_legacy_api else "channel"
        print(
            f"  {i:<3} {tc.driver} {tc.lane_count}L base={tc.base_led_count:<8} [{sizes_str}]{'':<{15 - len(sizes_str)}} {api}"
        )
    print()

    # Connect to device using pyserial (avoids fbuild daemon port locking issues)
    print(f"Connecting to {port}...")
    iface = create_serial_interface(port, use_pyserial=True)
    client = RpcClient(
        port,
        timeout=timeout,
        verbose=verbose,
        serial_interface=iface,
    )

    try:
        await client.connect(boot_wait=3.0, drain_boot=True)
        print("Connected\n")

        # Ping to verify device is alive
        try:
            ping_response = await client.send("ping", timeout=10.0)
            print(f"Device alive: {ping_response.data}")
        except KeyboardInterrupt as ki: