diff --git a/CMakeLists.txt b/CMakeLists.txt index 61a69d9..947132c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -564,8 +564,14 @@ if(SWITCH_PICO_INPUT_BACKEND STREQUAL "BLUEPAD32") target_compile_definitions(switch-pico PRIVATE SWITCH_PICO_HD_RUMBLE=1) endif() else() + set(SWITCH_PICO_UART_CONTROLLERS 4 CACHE STRING + "Switch Pro controllers exposed by the UART firmware (1-4, one USB interface each)") + if(NOT SWITCH_PICO_UART_CONTROLLERS MATCHES "^[1-4]$") + message(FATAL_ERROR + "SWITCH_PICO_UART_CONTROLLERS must be 1, 2, 3 or 4 (got '${SWITCH_PICO_UART_CONTROLLERS}')") + endif() target_compile_definitions(switch-pico PRIVATE - SWITCH_PICO_HID_INSTANCE_COUNT=1 + SWITCH_PICO_HID_INSTANCE_COUNT=${SWITCH_PICO_UART_CONTROLLERS} ) endif() diff --git a/README.md b/README.md index 5549b8f..709f298 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ Raspberry Pi Pico firmware that emulates one or more Switch Pro controllers over USB. Input can come from the SDL3-to-UART computer bridge or, on Pico 2 W, directly from Bluetooth controllers through Bluepad32. ## What you get -- **Firmware** (`src/firmware/`): acts as a Switch Pro controller (one on standard Pico, four on Pico 2 W AIO), accepting either UART bridge reports or the optional Pico 2 W Bluepad32 backend. +- **Firmware** (`src/firmware/`): acts as up to four Switch Pro controllers on one USB device (regular Pico driven over UART, or Pico 2 W AIO), accepting either UART bridge reports or the optional Pico 2 W Bluepad32 backend. - **Python bridge** (`switch_pico_bridge.controller_uart_bridge` / CLI `controller-uart-bridge`): reads SDL3 controllers on the host, sends reports over UART, and applies rumble locally. Hot‑plug friendly and cross‑platform (macOS/Windows/Linux). - **Color configuration** (`src/firmware/platform/pico/controller_color_config.h`): compile-time RGB colors for emulated controller grips and supported Bluetooth controller LEDs. - **Pico 2 W AIO firmware** (`firmware/switch-pico-aio.uf2`): hosts four concurrent Bluetooth controllers and sends their controls, calibrated motion, rumble, and slot identity through four separate Switch Pro USB interfaces without a computer. @@ -2458,13 +2458,41 @@ RUMBLE (force feedback) Nintendo sends two stateful four-byte HD-rumble actuator words with full/relative low/high-band commands and up to three substeps. `SwitchHapticsDecoder` retains this timeline as well as conventional strong/weak magnitudes. The selected DualSense's native PCM backend uses the timeline; ordinary controller-parser and UART/SDL paths use the magnitudes. Preserving frequency intent is not a claim of identical force response across actuators. Native forwarding for genuine Switch-family controllers is [planned separately](ADAPTER_PARITY_PLAN.md#native-switch-family-hd-rumble--planned), not enabled by the DualSense implementation. -The UART return frame carries the decoded result rather than raw HD-rumble bytes: +### UART framing (v3, multi-controller) + +The regular Pico firmware exposes up to four Switch Pro controllers on one USB +device (`SWITCH_PICO_UART_CONTROLLERS`, default 4, CMake cache option). Every +frame in both directions names the slot it belongs to: ```text -0xBB, 0x02, low-frequency magnitude, high-frequency magnitude, checksum +Host -> Pico : 0xAA, 0x03, payload_len, slot, payload..., checksum +Pico -> Host : 0xBB, 0x03, slot, low-frequency magnitude, high-frequency magnitude, checksum ``` -The checksum is the sum of the first four bytes modulo 256. Firmware and Python bridge versions from before this change are not rumble-protocol compatible; controller input framing remains unchanged. +The checksum is the sum of all preceding bytes modulo 256; the payload is +unchanged from v2 (buttons, hat, sticks, IMU count and samples). The firmware +still accepts v2 input frames (`0xAA, 0x02, payload_len, payload..., checksum`) +as slot 0, so older `switch_pico_uart` scripts keep working, but it only emits +the slot-tagged rumble frame. The bridge accepts both the 0x03 slot frame and +the older 5-byte `0xBB, 0x02` frame (as slot 0). Firmware and bridge from before +this change are not compatible with each other beyond that: an old bridge sees +no rumble from new firmware, and an old firmware ignores v3 input. + +Bridge usage with several controllers on one Pico: + +```sh +# Auto-pairing fills slots 0-3 on a port once every available port has one controller. +controller-uart-bridge --ports COM11 +# Explicit slots (omitted slots are filled in order from 0). +controller-uart-bridge --map 0:COM11 --map 1:COM11 --map 2:COM11:3 +# Firmware built with SWITCH_PICO_UART_CONTROLLERS=1, or one controller per Pico: +controller-uart-bridge --slots-per-port 1 +``` + +All configured controllers are always present to the Switch, exactly like the +AIO firmware. Four controllers with IMU enabled at the default 500 Hz exceed the +921600-baud link (4 x 49 bytes x 500 Hz ~ 98 KB/s vs ~92 KB/s); use +`--frequency 250` or `--no-imu` when multiplexing all four. ## Hardware wiring (Pico) - UART1 pins (fixed in firmware): @@ -2698,12 +2726,12 @@ For simple scripts or tests you can skip SDL and drive the Pico directly with `s ```python from switch_pico_bridge import SwitchUARTClient, SwitchButton, SwitchDpad -with SwitchUARTClient("/dev/cu.usbserial-0001") as client: +with SwitchUARTClient("/dev/cu.usbserial-0001", slot=0) as client: # slot selects one of the Pico's four controllers client.press(SwitchButton.A) client.release(SwitchButton.A) client.move_left_stick(0.0, -1.0) # push up client.set_hat(SwitchDpad.UP_RIGHT) - print(client.poll_rumble()) # returns (left, right) amplitudes 0.0-1.0 or None + print(client.poll_rumble()) # returns (low, high) amplitudes 0.0-1.0 for this slot, or None ``` - `SwitchButton` is an `IntFlag` (bitwise friendly) and `SwitchDpad` is an `IntEnum` for the DPAD/hat values (alias `SwitchHat` remains for older scripts). - The helper only depends on `pyserial`; SDL is not required. diff --git a/examples/example_switch_macro.py b/examples/example_switch_macro.py index 87e8dfb..4429d9f 100644 --- a/examples/example_switch_macro.py +++ b/examples/example_switch_macro.py @@ -29,8 +29,9 @@ ALL_BUTTONS = [ def main() -> None: # auto_send keeps the current state flowing in the background, so we don't - # need to manually pump frames to the Pico. - with SwitchUARTClient(PORT, send_interval=SEND_INTERVAL, auto_send=True) as client: + # need to manually pump frames to the Pico. slot picks which of the Pico's + # four emulated controllers (0-3) this script drives. + with SwitchUARTClient(PORT, send_interval=SEND_INTERVAL, auto_send=True, slot=0) as client: client.neutral() # Press every button/DPAD direction one-by-one, holding each briefly. diff --git a/firmware/switch-pico.elf b/firmware/switch-pico.elf index 18e52eb..05d3caf 100755 Binary files a/firmware/switch-pico.elf and b/firmware/switch-pico.elf differ diff --git a/firmware/switch-pico.uf2 b/firmware/switch-pico.uf2 index ea65ae4..a28754b 100644 Binary files a/firmware/switch-pico.uf2 and b/firmware/switch-pico.uf2 differ diff --git a/src/firmware/main.cpp b/src/firmware/main.cpp index 2767cdf..7384974 100644 --- a/src/firmware/main.cpp +++ b/src/firmware/main.cpp @@ -33,7 +33,7 @@ #define UART_TX_PIN 4 #define UART_RX_PIN 5 #define UART_RUMBLE_HEADER 0xBB -#define UART_RUMBLE_TYPE 0x02 +#define UART_RUMBLE_TYPE_SLOT 0x03 #endif #ifdef SWITCH_PICO_BLUEPAD32 @@ -47,9 +47,8 @@ static bool g_last_ready[BLUEPAD32_INPUT_BACKEND_SLOT_COUNT]{}; static ControllerState g_user_states[BLUEPAD32_INPUT_BACKEND_SLOT_COUNT]{}; #else -static constexpr uint8_t SWITCH_HID_INSTANCE = 0; -static bool g_last_ready = false; -static ControllerState g_user_state; +static bool g_last_ready[SWITCH_PICO_HID_INSTANCE_COUNT]{}; +static ControllerState g_user_states[SWITCH_PICO_HID_INSTANCE_COUNT]{}; #endif static bool g_last_mounted = false; @@ -68,17 +67,20 @@ static ControllerState neutral_input() { } #ifndef SWITCH_PICO_BLUEPAD32 -static void send_rumble_uart_frame(const ControllerRumbleOutput& rumble) { - uint8_t frame[5] = { +// Pico -> host: 0xBB 0x03 slot low high checksum(sum of first 5 bytes). +static void send_rumble_uart_frame(uint8_t slot, + const ControllerRumbleOutput& rumble) { + uint8_t frame[6] = { UART_RUMBLE_HEADER, - UART_RUMBLE_TYPE, + UART_RUMBLE_TYPE_SLOT, + slot, rumble.low_frequency_magnitude, rumble.high_frequency_magnitude, 0, }; - for (uint8_t i = 0; i < 4; ++i) { - frame[4] = static_cast(frame[4] + frame[i]); + for (uint8_t i = 0; i < 5; ++i) { + frame[5] = static_cast(frame[5] + frame[i]); } uart_write_blocking(UART_ID, frame, sizeof(frame)); } @@ -96,10 +98,10 @@ static void on_rumble_from_usb(uint8_t instance, instance, controller_profile_runtime_scale_host_rumble( instance, snapshot, rumble)); #else - if (instance != SWITCH_HID_INSTANCE) { + if (instance >= SWITCH_PICO_HID_INSTANCE_COUNT) { return; } - send_rumble_uart_frame(rumble); + send_rumble_uart_frame(instance, rumble); #endif } @@ -137,7 +139,9 @@ static bool poll_uart_frames() { buffer[index++] = byte; if (index == 3) { - expected_len = static_cast(buffer[2] + 4u); + // v2: header(3) + payload + checksum; v3 adds a slot byte. + const uint8_t overhead = buffer[1] == 0x03 ? 5u : 4u; + expected_len = static_cast(buffer[2] + overhead); if (expected_len < 12 || expected_len > sizeof(buffer)) { index = 0; expected_len = 0; @@ -147,10 +151,12 @@ static bool poll_uart_frames() { if (expected_len > 0 && index >= expected_len) { ControllerState parsed{}; - if (switch_pro_apply_uart_packet(buffer, expected_len, parsed)) { - g_user_state = parsed; + uint8_t slot = 0; + if (switch_pro_apply_uart_packet(buffer, expected_len, parsed, slot)) { + g_user_states[slot] = parsed; new_data = true; - LOG_PRINTF("[UART] packet buttons=0x%04x hat=%u lx=%u ly=%u rx=%u ry=%u\n", + LOG_PRINTF("[UART] slot=%u buttons=0x%04x hat=%u lx=%u ly=%u rx=%u ry=%u\n", + slot, (parsed.button_east ? SWITCH_PRO_MASK_A : 0) | (parsed.button_south ? SWITCH_PRO_MASK_B : 0) | (parsed.button_north ? SWITCH_PRO_MASK_X : 0) | @@ -207,11 +213,14 @@ static void log_usb_state() { } } #else - const bool ready = usb_output_driver_is_ready(SWITCH_HID_INSTANCE); - if (ready != g_last_ready) { - g_last_ready = ready; - LOG_PRINTF("[SWITCH] driver %s\n", - ready ? "ready (handshake OK)" : "not ready"); + for (uint8_t instance = 0; + instance < SWITCH_PICO_HID_INSTANCE_COUNT; ++instance) { + const bool ready = usb_output_driver_is_ready(instance); + if (ready != g_last_ready[instance]) { + g_last_ready[instance] = ready; + LOG_PRINTF("[SWITCH %u] driver %s\n", instance, + ready ? "ready (handshake OK)" : "not ready"); + } } #endif } @@ -247,12 +256,14 @@ int main() { CONTROLLER_PROFILE_DEFAULT_DIGITAL_THRESHOLD); } #else - usb_output_driver_set_rumble_callback(SWITCH_HID_INSTANCE, - on_rumble_from_usb); - g_user_state = neutral_input(); - usb_output_driver_set_input(SWITCH_HID_INSTANCE, g_user_state, - SWITCH_PRO_DIGITAL_TRIGGER_THRESHOLD, - SWITCH_PRO_DIGITAL_TRIGGER_THRESHOLD); + for (uint8_t instance = 0; + instance < SWITCH_PICO_HID_INSTANCE_COUNT; ++instance) { + usb_output_driver_set_rumble_callback(instance, on_rumble_from_usb); + g_user_states[instance] = neutral_input(); + usb_output_driver_set_input(instance, g_user_states[instance], + SWITCH_PRO_DIGITAL_TRIGGER_THRESHOLD, + SWITCH_PRO_DIGITAL_TRIGGER_THRESHOLD); + } #endif #ifdef SWITCH_PICO_BLUEPAD32 @@ -261,8 +272,8 @@ int main() { usb_output_driver_mode_name()); #else LOG_PRINTF("[BOOT] switch-pico starting (UART0 log @ 115200)\n"); - LOG_PRINTF("[INFO] UART1 pins TX=%d RX=%d baud=%d\n", - UART_TX_PIN, UART_RX_PIN, BAUD_RATE); + LOG_PRINTF("[INFO] UART1 pins TX=%d RX=%d baud=%d slots=%d\n", + UART_TX_PIN, UART_RX_PIN, BAUD_RATE, SWITCH_PICO_HID_INSTANCE_COUNT); #endif while (true) { @@ -330,11 +341,13 @@ int main() { #else bool new_data = poll_uart_frames(); // Pull controller state from UART1 (void)new_data; - ControllerState state = g_user_state; - usb_output_driver_set_input(SWITCH_HID_INSTANCE, state, - SWITCH_PRO_DIGITAL_TRIGGER_THRESHOLD, - SWITCH_PRO_DIGITAL_TRIGGER_THRESHOLD); - (void)usb_output_driver_task(SWITCH_HID_INSTANCE); + for (uint8_t instance = 0; + instance < SWITCH_PICO_HID_INSTANCE_COUNT; ++instance) { + usb_output_driver_set_input(instance, g_user_states[instance], + SWITCH_PRO_DIGITAL_TRIGGER_THRESHOLD, + SWITCH_PRO_DIGITAL_TRIGGER_THRESHOLD); + (void)usb_output_driver_task(instance); + } #endif log_usb_state(); } diff --git a/src/firmware/usb/switch/switch_pro_driver.cpp b/src/firmware/usb/switch/switch_pro_driver.cpp index db293cc..09ceeb9 100644 --- a/src/firmware/usb/switch/switch_pro_driver.cpp +++ b/src/firmware/usb/switch/switch_pro_driver.cpp @@ -927,43 +927,56 @@ bool switch_pro_task(uint8_t instance) { } bool switch_pro_apply_uart_packet(const uint8_t* packet, uint8_t length, - ControllerState& out_state) { - if (packet == nullptr) { + ControllerState& out_state, + uint8_t& out_slot) { + if (packet == nullptr || length < 12 || packet[0] != 0xAA) { return false; } - // v2 format: 0xAA + 0x02 + payload_len + payload... + checksum - if (length < 12 || packet[0] != 0xAA || packet[1] != 0x02) { + // v2: 0xAA 0x02 payload_len payload... checksum (slot 0) + // v3: 0xAA 0x03 payload_len slot payload... checksum + uint8_t header_len; + uint8_t slot = 0; + if (packet[1] == 0x02) { + header_len = 3; + } else if (packet[1] == 0x03) { + header_len = 4; + slot = packet[3]; + } else { return false; } - uint8_t payload_len = packet[2]; - if ((uint16_t)payload_len + 4u != length) { + const uint8_t payload_len = packet[2]; + if ((uint16_t)payload_len + header_len + 1u != length) { return false; } uint16_t sum = 0; - for (uint16_t i = 0; i < (uint16_t)(3u + payload_len); ++i) { + for (uint16_t i = 0; i < (uint16_t)(length - 1u); ++i) { sum += packet[i]; } if ((sum & 0xFF) != packet[length - 1]) { return false; } + if (slot >= SWITCH_PICO_HID_INSTANCE_COUNT) { + return false; + } // payload: buttons(2 LE), hat, lx, ly, rx, ry, motion_count, // [motion_samples...] if (payload_len < 8) { return false; } + const uint8_t* payload = packet + header_len; SwitchProOutReport out{}; - out.buttons = static_cast(packet[3]) | - (static_cast(packet[4]) << 8); - out.hat = packet[5]; - out.lx = packet[6]; - out.ly = packet[7]; - out.rx = packet[8]; - out.ry = packet[9]; - uint8_t motion_count = packet[10]; + out.buttons = static_cast(payload[0]) | + (static_cast(payload[1]) << 8); + out.hat = payload[2]; + out.lx = payload[3]; + out.ly = payload[4]; + out.rx = payload[5]; + out.ry = payload[6]; + uint8_t motion_count = payload[7]; if (motion_count > CONTROLLER_MOTION_SAMPLE_CAPACITY) { motion_count = CONTROLLER_MOTION_SAMPLE_CAPACITY; } @@ -988,7 +1001,7 @@ bool switch_pro_apply_uart_packet(const uint8_t* packet, uint8_t length, ControllerState state = make_neutral_state(); state.motion_sample_count = motion_count; for (uint8_t i = 0; i < motion_count; ++i) { - const uint8_t* base = &packet[11 + i * 12]; + const uint8_t* base = &payload[8 + i * 12]; state.motion_samples[i].accel_x = read_int16(base + 0); state.motion_samples[i].accel_y = read_int16(base + 2); state.motion_samples[i].accel_z = read_int16(base + 4); @@ -1053,6 +1066,7 @@ bool switch_pro_apply_uart_packet(const uint8_t* packet, uint8_t length, state.right_stick_y = expand_axis(out.ry); out_state = state; + out_slot = slot; return true; } diff --git a/src/firmware/usb/switch/switch_pro_driver.h b/src/firmware/usb/switch/switch_pro_driver.h index a6525d5..31e53d1 100644 --- a/src/firmware/usb/switch/switch_pro_driver.h +++ b/src/firmware/usb/switch/switch_pro_driver.h @@ -38,9 +38,11 @@ void switch_pro_set_input(uint8_t instance, const ControllerState& state, // 0x30 input report was successfully queued. bool switch_pro_task(uint8_t instance); -// Convert a packed UART message into controller state (returns true if parsed). +// Convert a packed UART message (v2 = slot 0, v3 carries a slot byte) into +// controller state. Returns true if parsed; outputs are untouched on failure. bool switch_pro_apply_uart_packet(const uint8_t* packet, uint8_t length, - ControllerState& out_state); + ControllerState& out_state, + uint8_t& out_slot); // Driver state helpers bool switch_pro_is_ready(uint8_t instance); diff --git a/src/switch_pico_bridge/controller_uart_bridge.py b/src/switch_pico_bridge/controller_uart_bridge.py index ae03fb6..6cb8ec9 100644 --- a/src/switch_pico_bridge/controller_uart_bridge.py +++ b/src/switch_pico_bridge/controller_uart_bridge.py @@ -2,13 +2,14 @@ """ Bridge multiple SDL3 controllers to switch-pico over UART and mirror rumble back. -The framing matches ``switch-pico.cpp``: - - Host -> Pico : UART v2 controller report - - Pico -> Host : 0xBB, 0x02, low-frequency magnitude, +The framing matches ``src/firmware/main.cpp``: + - Host -> Pico : UART v3 controller report (0xAA, 0x03, len, slot, payload, checksum) + - Pico -> Host : 0xBB, 0x03, slot, low-frequency magnitude, high-frequency magnitude, checksum Features inspired by ``host/controller_bridge.py``: - - Multiple controllers paired to multiple UART ports + - Multiple controllers paired to multiple UART ports, or to the separate + controller slots (0-3) of one Pico sharing a single port - Rich-powered interactive pairing UI - Adjustable send frequency, deadzone, and trigger thresholds - Rumble feedback delivered to SDL3 controllers @@ -36,6 +37,7 @@ from rich.text import Text from .switch_pico_uart import ( UART_BAUD, + UART_SLOT_COUNT, MS2_PER_G, RAD_TO_DEG, ACCEL_LSB_PER_G, @@ -62,11 +64,14 @@ SDL_EVENT_GAMEPAD_SENSOR_UPDATE = getattr(sdl3, "SDL_EVENT_GAMEPAD_SENSOR_UPDATE GYRO_BIAS_SAMPLES = 200 -def parse_mapping(value: str) -> Tuple[int, str]: - """Parse 'index:serial_port' CLI mapping argument.""" - if ":" not in value: - raise argparse.ArgumentTypeError("Mapping must look like 'index:serial_port'") - idx_str, port = value.split(":", 1) +def parse_mapping(value: str) -> Tuple[int, str, Optional[int]]: + """Parse 'index:serial_port[:slot]' CLI mapping argument.""" + parts = value.split(":") + if len(parts) not in (2, 3): + raise argparse.ArgumentTypeError( + "Mapping must look like 'index:serial_port' or 'index:serial_port:slot'" + ) + idx_str, port = parts[0], parts[1].strip() try: idx = int(idx_str, 10) except ValueError as exc: @@ -75,7 +80,17 @@ def parse_mapping(value: str) -> Tuple[int, str]: ) from exc if not port: raise argparse.ArgumentTypeError("Serial port cannot be empty") - return idx, port.strip() + slot: Optional[int] = None + if len(parts) == 3: + try: + slot = int(parts[2], 10) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"Invalid slot '{parts[2]}'") from exc + if not 0 <= slot < UART_SLOT_COUNT: + raise argparse.ArgumentTypeError( + f"Slot must be 0-{UART_SLOT_COUNT - 1}, got {slot}" + ) + return idx, port, slot def download_controller_db(console: Console, destination: Path, url: str) -> bool: @@ -160,26 +175,41 @@ STICK_AXES = tuple(axis for axis, _ in STICK_AXIS_LABELS) def interactive_pairing( - console: Console, controller_info: Dict[int, str], ports: List[Dict[str, str]] -) -> List[Tuple[int, str]]: - """Prompt the user to pair controllers to UART ports via Rich UI.""" - available = ports.copy() - mappings: List[Tuple[int, str]] = [] - for controller_idx in controller_info: + console: Console, + controller_info: Dict[int, str], + ports: List[Dict[str, str]], + slots_per_port: int, +) -> List[Tuple[int, str, Optional[int]]]: + """Prompt the user to pair controllers to UART ports via Rich UI. + + A port stays selectable until all of its controller slots are taken. + """ + used_slots: Dict[str, int] = {} + mappings: List[Tuple[int, str, Optional[int]]] = [] + for controller_idx, name in controller_info.items(): + available = [ + port for port in ports if used_slots.get(port["device"], 0) < slots_per_port + ] if not available: console.print( - "[bold red]No more UART devices available for pairing.[/bold red]" + "[bold red]No more UART controller slots available for pairing.[/bold red]" ) break table = Table( - title=f"Available UART Devices for Controller {controller_idx} ({controller_info[controller_idx]})" + title=f"Available UART Devices for Controller {controller_idx} ({name})" ) table.add_column("Choice", justify="center") table.add_column("Port") table.add_column("Description") + table.add_column("Slots used", justify="center") for i, port in enumerate(available): - table.add_row(str(i), port["device"], port["description"]) + table.add_row( + str(i), + port["device"], + port["description"], + f"{used_slots.get(port['device'], 0)}/{slots_per_port}", + ) console.print(table) choices = [str(i) for i in range(len(available))] + ["q"] selection = Prompt.ask( @@ -189,11 +219,12 @@ def interactive_pairing( ) if selection == "q": break - idx = int(selection) - port = available.pop(idx) - mappings.append((controller_idx, port["device"])) + device = available[int(selection)]["device"] + slot = used_slots.get(device, 0) + used_slots[device] = slot + 1 + mappings.append((controller_idx, device, slot)) console.print( - f"[bold green]Paired controller {controller_idx} with {port['device']}[/bold green]" + f"[bold green]Paired controller {controller_idx} with {device} slot {slot}[/bold green]" ) return mappings @@ -227,6 +258,15 @@ def apply_rumble( return low != 0 or high != 0, accepted +@dataclass +class UartLink: + """One serial port shared by every controller mapped to its slots.""" + + port: str + uart: Optional[PicoUART] = None + last_reopen_attempt: float = 0.0 + + @dataclass class ControllerContext: controller: sdl3.SDL_Gamepad @@ -234,7 +274,7 @@ class ControllerContext: controller_index: int stable_id: str port: Optional[str] - uart: Optional[PicoUART] + slot: int = 0 report: SwitchReport = field(default_factory=SwitchReport) dpad: Dict[str, bool] = field( default_factory=lambda: { @@ -249,7 +289,6 @@ class ControllerContext: default_factory=lambda: {"left": False, "right": False} ) last_send: float = 0.0 - last_reopen_attempt: float = 0.0 last_rumble_at: float = 0.0 rumble_active: bool = False axis_offsets: Dict[int, int] = field(default_factory=dict) @@ -644,13 +683,29 @@ def build_arg_parser() -> argparse.ArgumentParser: action="append", type=parse_mapping, default=[], - help="Controller mapping 'index:serial_port'. Repeat per controller.", + help=( + "Controller mapping 'index:serial_port[:slot]'. Repeat per controller; " + "controllers sharing a serial port drive that Pico's separate emulated " + "controllers. Omitted slots are filled in order from 0." + ), ) parser.add_argument( "--ports", nargs="+", help="Serial ports to auto-pair with controllers in ascending index order.", ) + parser.add_argument( + "--slots-per-port", + type=int, + default=UART_SLOT_COUNT, + choices=range(1, UART_SLOT_COUNT + 1), + metavar="N", + help=( + "Emulated controllers per Pico for auto/interactive pairing (default: " + f"{UART_SLOT_COUNT}; match the firmware's SWITCH_PICO_UART_CONTROLLERS). " + "Auto-pairing spreads controllers across ports before reusing one." + ), + ) parser.add_argument( "--interactive", action="store_true", @@ -881,8 +936,11 @@ class DisplayIndexAllocator: @dataclass class PairingState: - mapping_by_index: Dict[int, str] + # controller display index -> (serial port, firmware slot) + mapping_by_index: Dict[int, Tuple[str, int]] + # Ports auto-pairing may hand out; a port stays here while it has free slots. available_ports: List[str] + slots_per_port: int = UART_SLOT_COUNT auto_assigned_indices: set[int] = field(default_factory=set) auto_pairing_enabled: bool = False auto_discover_ports: bool = False @@ -892,6 +950,17 @@ class PairingState: include_port_mfr: List[str] = field(default_factory=list) display_index_alloc: DisplayIndexAllocator = field(default_factory=DisplayIndexAllocator) + def used_slots(self, port: str) -> set[int]: + return {slot for mapped, slot in self.mapping_by_index.values() if mapped == port} + + def free_slot(self, port: str) -> Optional[int]: + """Lowest unused slot on ``port`` within the configured slot budget.""" + used = self.used_slots(port) + for slot in range(self.slots_per_port): + if slot not in used: + return slot + return None + def load_button_maps( console: Console, args: argparse.Namespace @@ -1078,6 +1147,31 @@ def list_serial_ports(console: Console, args: argparse.Namespace) -> None: console.print(table) +def resolve_mapping_slots( + mappings: List[Tuple[int, str, Optional[int]]], + slots_per_port: int, + parser: argparse.ArgumentParser, +) -> Dict[int, Tuple[str, int]]: + """Assign explicit or next-free slots to CLI/interactive mappings.""" + resolved: Dict[int, Tuple[str, int]] = {} + taken: Dict[str, set[int]] = {} + for index, port, slot in mappings: + if index in resolved: + parser.error(f"Controller {index} is mapped more than once.") + used = taken.setdefault(port, set()) + if slot is None: + slot = next((s for s in range(slots_per_port) if s not in used), None) + if slot is None: + parser.error( + f"{port} has no free controller slot (limit {slots_per_port}; see --slots-per-port)." + ) + elif slot in used: + parser.error(f"{port} slot {slot} is mapped to more than one controller.") + used.add(slot) + resolved[index] = (port, slot) + return resolved + + def prepare_pairing_state( args: argparse.Namespace, console: Console, @@ -1092,9 +1186,10 @@ def prepare_pairing_state( ignore_port_desc = [d.lower() for d in args.ignore_port_desc] include_port_desc = [d.lower() for d in args.include_port_desc] include_port_mfr = [m.lower() for m in args.include_port_manufacturer] + slots_per_port = int(args.slots_per_port) available_ports: List[str] = [] - mappings = list(args.map) + mappings: List[Tuple[int, str, Optional[int]]] = list(args.map) if args.interactive: if not controller_indices: parser.error("No controllers detected for interactive pairing.") @@ -1107,7 +1202,7 @@ def prepare_pairing_state( ) if not discovered: parser.error("No UART devices found for interactive pairing.") - mappings = interactive_pairing(console, controller_names, discovered) + mappings = interactive_pairing(console, controller_names, discovered, slots_per_port) if not mappings: parser.error("No controller-to-UART mappings were selected.") elif auto_pairing_enabled: @@ -1134,10 +1229,10 @@ def prepare_pairing_state( "[yellow]No UART devices detected yet; waiting for hotplug...[/yellow]" ) - mapping_by_index = {index: port for index, port in mappings} return PairingState( - mapping_by_index=mapping_by_index, + mapping_by_index=resolve_mapping_slots(mappings, slots_per_port, parser), available_ports=available_ports, + slots_per_port=slots_per_port, auto_pairing_enabled=auto_pairing_enabled, auto_discover_ports=auto_discover_ports, include_non_usb=include_non_usb, @@ -1149,32 +1244,75 @@ def prepare_pairing_state( def assign_port_for_index( pairing: PairingState, idx: int, console: Console -) -> Optional[str]: - """Return the UART assigned to a controller index, auto-pairing if allowed.""" +) -> Optional[Tuple[str, int]]: + """Return the (port, slot) for a controller index, auto-pairing if allowed. + + Auto-pairing prefers the port with the fewest controllers so several Picos + are filled evenly before any one of them multiplexes. + """ if idx in pairing.mapping_by_index: return pairing.mapping_by_index[idx] if not pairing.auto_pairing_enabled: return None - if not pairing.available_ports: + candidates = [ + (len(pairing.used_slots(port)), order, port) + for order, port in enumerate(pairing.available_ports) + if pairing.free_slot(port) is not None + ] + if not candidates: return None - port_choice = pairing.available_ports.pop(0) - pairing.mapping_by_index[idx] = port_choice + _, _, port_choice = min(candidates) + slot = pairing.free_slot(port_choice) + assert slot is not None + pairing.mapping_by_index[idx] = (port_choice, slot) pairing.auto_assigned_indices.add(idx) - console.print(f"[green]Auto-paired controller {idx} to {port_choice}[/green]") - return port_choice + console.print(f"[green]Auto-paired controller {idx} to {port_choice} slot {slot}[/green]") + return port_choice, slot def ports_in_use(pairing: PairingState, contexts: Dict[int, ControllerContext]) -> set[str]: """Return a set of UART paths currently reserved or mapped.""" - used = set(pairing.mapping_by_index.values()) + used = {port for port, _ in pairing.mapping_by_index.values()} used.update(ctx.port for ctx in contexts.values() if ctx.port) return used +def close_link(links: Dict[str, UartLink], port: str) -> None: + """Close and forget the shared UART for ``port`` if it is open.""" + link = links.pop(port, None) + if link and link.uart: + try: + link.uart.close() + except Exception: + pass + + +def ensure_link( + links: Dict[str, UartLink], port: str, baud: int, console: Console, now: float +) -> UartLink: + """Return the shared link for ``port``, opening the serial device if needed.""" + link = links.get(port) + if link is None: + link = UartLink(port=port) + links[port] = link + if link.uart is None: + link.last_reopen_attempt = now + link.uart = open_uart_or_warn(port, baud, console) + return link + + +def detach_context_from_port(ctx: ControllerContext) -> None: + sdl3.SDL_RumbleGamepad(ctx.controller, 0, 0, 0) + ctx.port = None + ctx.slot = 0 + ctx.rumble_active = False + + def handle_removed_port( path: str, pairing: PairingState, contexts: Dict[int, ControllerContext], + links: Dict[str, UartLink], console: Console, ) -> None: """Clear mappings/contexts for a UART path that disappeared.""" @@ -1184,31 +1322,26 @@ def handle_removed_port( f"[yellow]UART {path} removed; dropping from available pool[/yellow]" ) indices_to_clear = [ - idx for idx, mapped in pairing.mapping_by_index.items() if mapped == path + idx for idx, (mapped, _) in pairing.mapping_by_index.items() if mapped == path ] for idx in indices_to_clear: pairing.mapping_by_index.pop(idx, None) pairing.auto_assigned_indices.discard(idx) + close_link(links, path) for ctx in list(contexts.values()): if ctx.port != path: continue - if ctx.uart: - try: - ctx.uart.close() - except Exception: - pass - sdl3.SDL_RumbleGamepad(ctx.controller, 0, 0, 0) - ctx.uart = None - ctx.port = None - ctx.rumble_active = False - ctx.last_reopen_attempt = time.monotonic() + detach_context_from_port(ctx) console.print( f"[yellow]UART {path} removed; controller {ctx.controller_index} waiting for reassignment[/yellow]" ) def discover_new_ports( - pairing: PairingState, contexts: Dict[int, ControllerContext], console: Console + pairing: PairingState, + contexts: Dict[int, ControllerContext], + links: Dict[str, UartLink], + console: Console, ) -> None: """Scan for new serial ports and add unused ones to the available pool.""" if not pairing.auto_discover_ports: @@ -1221,12 +1354,11 @@ def discover_new_ports( ) current_paths = {info["device"] for info in discovered} known_paths = set(pairing.available_ports) - known_paths.update(pairing.mapping_by_index.values()) - known_paths.update(ctx.port for ctx in contexts.values() if ctx.port) + known_paths.update(ports_in_use(pairing, contexts)) # Drop any paths we previously knew about that are no longer present. removed_paths = [path for path in known_paths if path not in current_paths] for path in removed_paths: - handle_removed_port(path, pairing, contexts, console) + handle_removed_port(path, pairing, contexts, links, console) in_use = ports_in_use(pairing, contexts) for info in discovered: path = info["device"] @@ -1238,35 +1370,88 @@ def discover_new_ports( ) +def announce_pairing( + ctx: ControllerContext, link: UartLink, console: Console +) -> None: + who = f"Controller {ctx.controller_index} (id {ctx.stable_id}, inst {ctx.instance_id})" + if link.uart: + console.print(f"[green]{who} paired to {link.port} slot {ctx.slot}[/green]") + else: + console.print(f"[yellow]{who} waiting for UART {link.port} (slot {ctx.slot})[/yellow]") + + def pair_waiting_contexts( args: argparse.Namespace, pairing: PairingState, contexts: Dict[int, ControllerContext], - uarts: List[PicoUART], + links: Dict[str, UartLink], console: Console, ) -> None: - """Attach UARTs to contexts that are waiting for a port assignment/open.""" + """Attach ports to contexts that are waiting for a slot assignment.""" for ctx in list(contexts.values()): if ctx.port is not None: continue - # Try to grab a port for this controller; if none are available, leave it waiting. - port_choice = assign_port_for_index(pairing, ctx.controller_index, console) - if port_choice is None: + # Try to grab a slot for this controller; if none are available, leave it waiting. + assignment = assign_port_for_index(pairing, ctx.controller_index, console) + if assignment is None: continue - ctx.port = port_choice - uart = open_uart_or_warn(port_choice, args.baud, console) - ctx.last_reopen_attempt = time.monotonic() - if uart: - uarts.append(uart) - ctx.uart = uart - console.print( - f"[green]Controller {ctx.controller_index} (id {ctx.stable_id}, inst {ctx.instance_id}) paired to {port_choice}[/green]" - ) - else: - ctx.uart = None - console.print( - f"[yellow]Controller {ctx.controller_index} (id {ctx.stable_id}, inst {ctx.instance_id}) waiting for UART {port_choice}[/yellow]" - ) + ctx.port, ctx.slot = assignment + link = ensure_link(links, ctx.port, args.baud, console, time.monotonic()) + announce_pairing(ctx, link, console) + + +def attach_controller( + sdl_id: int, + args: argparse.Namespace, + pairing: PairingState, + contexts: Dict[int, ControllerContext], + links: Dict[str, UartLink], + console: Console, + config: BridgeConfig, +) -> None: + """Open an SDL gamepad, give it a slot if one is mapped or free, and track it.""" + if sdl_id in contexts: + return + if not sdl3.SDL_IsGamepad(sdl_id): + name = sdl3.SDL_GetJoystickNameForID(sdl_id) + name_str = name.decode() if isinstance(name, bytes) else str(name) if name else "Unknown" + console.print( + f"[yellow]Device {sdl_id} is not a GameController ({name_str}).[/yellow]" + ) + return + display_idx = pairing.display_index_alloc.allocate() + assignment = assign_port_for_index(pairing, display_idx, console) + if assignment is None and not pairing.auto_pairing_enabled: + pairing.display_index_alloc.release(display_idx) + return + try: + controller, instance_id, guid = open_controller(sdl_id) + except Exception as exc: + console.print(f"[red]Failed to open controller {display_idx}: {exc}[/red]") + pairing.display_index_alloc.release(display_idx) + return + should_swap = display_idx in config.swap_abxy_indices or guid in config.swap_abxy_ids + ctx = ControllerContext( + controller=controller, + instance_id=instance_id, + controller_index=display_idx, + stable_id=guid, + port=assignment[0] if assignment else None, + slot=assignment[1] if assignment else 0, + swap_abxy=should_swap, + ) + if assignment: + link = ensure_link(links, assignment[0], args.baud, console, time.monotonic()) + announce_pairing(ctx, link, console) + else: + console.print( + f"[yellow]Controller {display_idx} (id {guid}, inst {instance_id}) connected; waiting for an available UART slot[/yellow]" + ) + if not config.no_imu: + initialize_controller_sensors(ctx, console) + if config.zero_sticks: + zero_context_sticks(ctx, console) + contexts[instance_id] = ctx def open_initial_contexts( @@ -1275,62 +1460,13 @@ def open_initial_contexts( controller_indices: List[int], console: Console, config: BridgeConfig, -) -> Tuple[Dict[int, ControllerContext], List[PicoUART]]: +) -> Tuple[Dict[int, ControllerContext], Dict[str, UartLink]]: """Open initial controllers and UARTs for detected indices.""" contexts: Dict[int, ControllerContext] = {} - uarts: List[PicoUART] = [] + links: Dict[str, UartLink] = {} for instance_id in controller_indices: - if not sdl3.SDL_IsGamepad(instance_id): - name = sdl3.SDL_GetJoystickNameForID(instance_id) - name_str = name.decode() if isinstance(name, bytes) else str(name) if name else "Unknown" - console.print( - f"[yellow]ID {instance_id} is not a GameController ({name_str}). Trying raw open failed.[/yellow]" - ) - continue - display_idx = pairing.display_index_alloc.allocate() - port = assign_port_for_index(pairing, display_idx, console) - if port is None and not pairing.auto_pairing_enabled: - pairing.display_index_alloc.release(display_idx) - continue - try: - controller, opened_instance_id, guid = open_controller(instance_id) - except Exception as exc: - console.print(f"[red]Failed to open controller {display_idx}: {exc}[/red]") - pairing.display_index_alloc.release(display_idx) - continue - stable_id = guid - should_swap = ( - display_idx in config.swap_abxy_indices or stable_id in config.swap_abxy_ids - ) - uart = open_uart_or_warn(port, args.baud, console) if port else None - if uart: - uarts.append(uart) - console.print( - f"[green]Controller {display_idx} (id {stable_id}, inst {opened_instance_id}) paired to {port}[/green]" - ) - elif port: - console.print( - f"[yellow]Controller {display_idx} (id {stable_id}, inst {opened_instance_id}) waiting for UART {port}[/yellow]" - ) - else: - console.print( - f"[yellow]Controller {display_idx} (id {stable_id}, inst {opened_instance_id}) connected; waiting for an available UART[/yellow]" - ) - ctx = ControllerContext( - controller=controller, - instance_id=opened_instance_id, - controller_index=display_idx, - stable_id=stable_id, - port=port, - uart=uart, - swap_abxy=should_swap, - ) - if not config.no_imu: - initialize_controller_sensors(ctx, console) - if config.zero_sticks: - zero_context_sticks(ctx, console) - contexts[opened_instance_id] = ctx - return contexts, uarts + attach_controller(instance_id, args, pairing, contexts, links, console, config) + return contexts, links def handle_axis_motion( @@ -1506,72 +1642,22 @@ def handle_device_added( args: argparse.Namespace, pairing: PairingState, contexts: Dict[int, ControllerContext], - uarts: List[PicoUART], + links: Dict[str, UartLink], console: Console, config: BridgeConfig, ) -> None: - """Handle controller hotplug by opening and pairing UART if possible.""" - sdl_id = event.gdevice.which - if sdl_id in contexts: - return - if not sdl3.SDL_IsGamepad(sdl_id): - name = sdl3.SDL_GetJoystickNameForID(sdl_id) - name_str = name.decode() if isinstance(name, bytes) else str(name) if name else "Unknown" - console.print( - f"[yellow]Device {sdl_id} is not a GameController ({name_str}).[/yellow]" - ) - return - display_idx = pairing.display_index_alloc.allocate() - port = assign_port_for_index(pairing, display_idx, console) - if port is None and not pairing.auto_pairing_enabled: - pairing.display_index_alloc.release(display_idx) - return - try: - controller, instance_id, guid = open_controller(sdl_id) - except Exception as exc: - console.print(f"[red]Hotplug open failed for controller {display_idx}: {exc}[/red]") - pairing.display_index_alloc.release(display_idx) - return - stable_id = guid - should_swap = display_idx in config.swap_abxy_indices or stable_id in config.swap_abxy_ids - uart = open_uart_or_warn(port, args.baud, console) if port else None - if uart: - uarts.append(uart) - console.print( - f"[green]Controller {display_idx} (id {stable_id}, inst {instance_id}) paired to {port}[/green]" - ) - elif port: - console.print( - f"[yellow]Controller {display_idx} (id {stable_id}, inst {instance_id}) waiting for UART {port}[/yellow]" - ) - else: - console.print( - f"[yellow]Controller {display_idx} (id {stable_id}, inst {instance_id}) connected; waiting for an available UART[/yellow]" - ) - ctx = ControllerContext( - controller=controller, - instance_id=instance_id, - controller_index=display_idx, - stable_id=stable_id, - port=port, - uart=uart, - swap_abxy=should_swap, - ) - if not config.no_imu: - initialize_controller_sensors(ctx, console) - if config.zero_sticks: - zero_context_sticks(ctx, console) - contexts[instance_id] = ctx + """Handle controller hotplug by opening it and pairing a UART slot if possible.""" + attach_controller(event.gdevice.which, args, pairing, contexts, links, console, config) def handle_device_removed( event: sdl3.SDL_Event, pairing: PairingState, contexts: Dict[int, ControllerContext], - uarts: List[PicoUART], + links: Dict[str, UartLink], console: Console, ) -> None: - """Handle controller removal and release any auto-assigned UART.""" + """Handle controller removal and release any auto-assigned UART slot.""" instance_id = event.gdevice.which ctx = contexts.pop(instance_id, None) if not ctx: @@ -1579,57 +1665,37 @@ def handle_device_removed( console.print( f"[yellow]Controller {ctx.controller_index} (id {ctx.stable_id}) removed[/yellow]" ) - # Close the UART handle *before* returning the port to the pool so the - # next consumer can actually open it (Windows holds the port exclusively). - if ctx.uart: - try: - ctx.uart.close() - except Exception: - pass - if ctx.uart in uarts: - uarts.remove(ctx.uart) - ctx.uart = None sdl3.SDL_RumbleGamepad(ctx.controller, 0, 0, 0) + # Close the serial handle once no controller uses the port so the next + # consumer can actually open it (Windows holds the port exclusively). + if ctx.port and all(other.port != ctx.port for other in contexts.values()): + close_link(links, ctx.port) if ctx.controller_index in pairing.auto_assigned_indices: - # Return auto-paired UART back to the pool so a future device can use it. + # Return the slot to the pool so a future device can use it. freed = pairing.mapping_by_index.pop(ctx.controller_index, None) pairing.auto_assigned_indices.discard(ctx.controller_index) - if freed and freed not in pairing.available_ports: - pairing.available_ports.append(freed) - console.print(f"[cyan]Released UART {freed} back to pool[/cyan]") + if freed: + port, slot = freed + if port not in pairing.available_ports: + pairing.available_ports.append(port) + console.print(f"[cyan]Released {port} slot {slot} back to pool[/cyan]") pairing.display_index_alloc.release(ctx.controller_index) sdl3.SDL_CloseGamepad(ctx.controller) -def service_contexts( +def service_link( now: float, - args: argparse.Namespace, config: BridgeConfig, - contexts: Dict[int, ControllerContext], - uarts: List[PicoUART], + link: UartLink, + members: List[ControllerContext], console: Console, ) -> None: - """Poll controllers, reconnect UARTs, send reports, and apply rumble.""" - for ctx in list(contexts.values()): - current_button_map = ( - config.button_map_swapped - if (config.swap_abxy_global or ctx.swap_abxy) - else config.button_map_default - ) - poll_controller_buttons(ctx, current_button_map) - # Reconnect UART if needed. - if ctx.port and ctx.uart is None and (now - ctx.last_reopen_attempt) > 1.0: - ctx.last_reopen_attempt = now - uart = open_uart_or_warn(ctx.port, args.baud, console) - if uart: - uarts.append(uart) - console.print( - f"[green]Reconnected UART {ctx.port} for controller {ctx.controller_index}[/green]" - ) - ctx.uart = uart - if ctx.uart is None: - continue - try: + """Send due reports for every controller on one port and demux its rumble.""" + uart = link.uart + if uart is None: + return + try: + for ctx in members: if now - ctx.last_send >= config.interval: if ctx.sensors_enabled and not config.no_imu: # Keep publishing the latest complete sensor window. Draining @@ -1637,19 +1703,24 @@ def service_contexts( ctx.report.imu_samples = ctx.imu_samples else: ctx.report.imu_samples = [] - ctx.uart.send_report(ctx.report) + uart.send_report(ctx.report, ctx.slot) ctx.last_send = now - latest_rumble = None - while True: - rumble = ctx.uart.read_rumble() - if rumble is None: - break - latest_rumble = rumble - ctx.debug_rumble_frames += 1 + # Keep only the freshest rumble command per slot seen during this tick. + latest_by_slot: Dict[int, Tuple[float, float]] = {} + frames_by_slot: Dict[int, int] = {} + while True: + frame = uart.read_rumble() + if frame is None: + break + slot, low, high = frame + latest_by_slot[slot] = (low, high) + frames_by_slot[slot] = frames_by_slot.get(slot, 0) + 1 + for ctx in members: + ctx.debug_rumble_frames += frames_by_slot.get(ctx.slot, 0) + latest_rumble = latest_by_slot.get(ctx.slot) if latest_rumble is not None: - # Apply only the freshest rumble command seen during this tick. ctx.rumble_active, accepted = apply_rumble( ctx.controller, latest_rumble[0], @@ -1664,30 +1735,65 @@ def service_contexts( ctx.last_debug_rumble = latest_rumble error = "" if accepted else f" sdl_error={sdl3.SDL_GetError().decode(errors='ignore')!r}" print( - f"[RUMBLE idx={ctx.controller_index}] frame#{ctx.debug_rumble_frames} " + f"[RUMBLE idx={ctx.controller_index} slot={ctx.slot}] frame#{ctx.debug_rumble_frames} " f"low={latest_rumble[0]:.3f} high={latest_rumble[1]:.3f} " f"-> motor low={shape_rumble(latest_rumble[0], config.rumble_gain, config.rumble_curve):.3f} " f"high={shape_rumble(latest_rumble[1], config.rumble_gain, config.rumble_curve):.3f} " f"accepted={accepted}{error}" ) - elif ( - ctx.rumble_active - and (now - ctx.last_rumble_at) > RUMBLE_IDLE_TIMEOUT - ): + elif ctx.rumble_active and (now - ctx.last_rumble_at) > RUMBLE_IDLE_TIMEOUT: sdl3.SDL_RumbleGamepad(ctx.controller, 0, 0, 0) ctx.rumble_active = False - except SerialException as exc: - console.print(f"[yellow]UART {ctx.port} disconnected: {exc}[/yellow]") - try: - ctx.uart.close() - except Exception: - pass + except SerialException as exc: + console.print(f"[yellow]UART {link.port} disconnected: {exc}[/yellow]") + try: + uart.close() + except Exception: + pass + link.uart = None + link.last_reopen_attempt = now + for ctx in members: sdl3.SDL_RumbleGamepad(ctx.controller, 0, 0, 0) - ctx.uart = None ctx.rumble_active = False - ctx.last_reopen_attempt = now - except Exception as exc: - console.print(f"[red]UART error on {ctx.port}: {exc}[/red]") + except Exception as exc: + console.print(f"[red]UART error on {link.port}: {exc}[/red]") + + +def service_contexts( + now: float, + args: argparse.Namespace, + config: BridgeConfig, + contexts: Dict[int, ControllerContext], + links: Dict[str, UartLink], + console: Console, +) -> None: + """Poll controllers, reconnect UARTs, send reports, and apply rumble.""" + members_by_port: Dict[str, List[ControllerContext]] = {} + for ctx in list(contexts.values()): + current_button_map = ( + config.button_map_swapped + if (config.swap_abxy_global or ctx.swap_abxy) + else config.button_map_default + ) + poll_controller_buttons(ctx, current_button_map) + if ctx.port: + members_by_port.setdefault(ctx.port, []).append(ctx) + + for port, members in members_by_port.items(): + link = links.get(port) + if link is None: + link = UartLink(port=port) + links[port] = link + # Reconnect the shared UART if needed. + if link.uart is None and (now - link.last_reopen_attempt) > 1.0: + link.last_reopen_attempt = now + link.uart = open_uart_or_warn(port, args.baud, console) + if link.uart: + console.print( + f"[green]Reconnected UART {port} for controller(s) " + f"{', '.join(str(ctx.controller_index) for ctx in members)}[/green]" + ) + service_link(now, config, link, members, console) def run_bridge_loop( @@ -1696,7 +1802,7 @@ def run_bridge_loop( config: BridgeConfig, pairing: PairingState, contexts: Dict[int, ControllerContext], - uarts: List[PicoUART], + links: Dict[str, UartLink], hotkey: Optional[HotkeyMonitor] = None, ) -> None: """Main event loop for bridging controllers to UART and handling rumble.""" @@ -1721,20 +1827,18 @@ def run_bridge_loop( handle_sensor_update(event, contexts, config) elif event.type == sdl3.SDL_EVENT_GAMEPAD_ADDED: handle_device_added( - event, args, pairing, contexts, uarts, console, config + event, args, pairing, contexts, links, console, config ) elif event.type == sdl3.SDL_EVENT_GAMEPAD_REMOVED: - handle_device_removed(event, pairing, contexts, uarts, console) + handle_device_removed(event, pairing, contexts, links, console) now = time.monotonic() if now - last_port_scan > port_scan_interval: # Periodically rescan for new UARTs to auto-pair hotplugged devices. - discover_new_ports(pairing, contexts, console) + discover_new_ports(pairing, contexts, links, console) last_port_scan = now - pair_waiting_contexts(args, pairing, contexts, uarts, console) - else: - pair_waiting_contexts(args, pairing, contexts, uarts, console) - service_contexts(now, args, config, contexts, uarts, console) + pair_waiting_contexts(args, pairing, contexts, links, console) + service_contexts(now, args, config, contexts, links, console) if hotkey: for key in hotkey.poll_keys(): if key == config.zero_hotkey: @@ -1744,12 +1848,13 @@ def run_bridge_loop( sdl3.SDL_Delay(1) -def cleanup(contexts: Dict[int, ControllerContext], uarts: List[PicoUART]) -> None: +def cleanup(contexts: Dict[int, ControllerContext], links: Dict[str, UartLink]) -> None: """Gracefully close controllers, UARTs, and SDL subsystems.""" for ctx in contexts.values(): sdl3.SDL_CloseGamepad(ctx.controller) - for uart in uarts: - uart.close() + for link in links.values(): + if link.uart: + link.uart.close() sdl3.SDL_Quit() @@ -1764,7 +1869,7 @@ def main() -> None: config = build_bridge_config(console, args) initialize_sdl(parser) contexts: Dict[int, ControllerContext] = {} - uarts: List[PicoUART] = [] + links: Dict[str, UartLink] = {} hotkey_monitor: Optional[HotkeyMonitor] = None try: if args.list_controllers: @@ -1790,18 +1895,18 @@ def main() -> None: candidate = HotkeyMonitor(console, hotkey_messages) if candidate.start(): hotkey_monitor = candidate - contexts, uarts = open_initial_contexts( + contexts, links = open_initial_contexts( args, pairing, controller_indices, console, config ) if not contexts: console.print( "[yellow]No controllers opened; waiting for hotplug events...[/yellow]" ) - run_bridge_loop(args, console, config, pairing, contexts, uarts, hotkey_monitor) + run_bridge_loop(args, console, config, pairing, contexts, links, hotkey_monitor) finally: if hotkey_monitor: hotkey_monitor.stop() - cleanup(contexts, uarts) + cleanup(contexts, links) if __name__ == "__main__": diff --git a/src/switch_pico_bridge/switch_pico_uart.py b/src/switch_pico_bridge/switch_pico_uart.py index a74111c..3de62f2 100644 --- a/src/switch_pico_bridge/switch_pico_uart.py +++ b/src/switch_pico_bridge/switch_pico_uart.py @@ -4,11 +4,13 @@ Lightweight helpers for talking to the switch-pico firmware over UART. This module exposes the report structure plus a small convenience wrapper so other scripts can do things like "press a button" or "move a stick" without -depending on SDL. It mirrors the framing in ``switch-pico.cpp``: +depending on SDL. It mirrors the framing in ``src/firmware/main.cpp``: - Host -> Pico : UART v2 controller report - Pico -> Host : 0xBB, 0x02, low-frequency magnitude, high-frequency magnitude, - checksum (sum of the first 4 bytes) + Host -> Pico : UART v3 controller report + 0xAA, 0x03, payload_len, slot, payload..., checksum + (v2 frames without the slot byte are still accepted as slot 0) + Pico -> Host : 0xBB, 0x03, slot, low-frequency magnitude, + high-frequency magnitude, checksum (sum of the first 5 bytes) """ from __future__ import annotations @@ -25,10 +27,13 @@ import serial from serial.tools import list_ports, list_ports_common UART_HEADER = 0xAA -UART_PROTOCOL_VERSION = 0x02 +UART_PROTOCOL_VERSION = 0x03 RUMBLE_HEADER = 0xBB +# Legacy 5-byte frame (no slot) from firmware before multi-controller support. RUMBLE_TYPE_DECODED = 0x02 +RUMBLE_TYPE_SLOT = 0x03 UART_BAUD = 921600 +UART_SLOT_COUNT = 4 IMU_SAMPLES_PER_REPORT = 3 MS2_PER_G = 9.80665 @@ -242,8 +247,10 @@ class SwitchReport: ry: int = 128 imu_samples: List[IMUSample] = field(default_factory=list) - def to_bytes(self) -> bytes: - """Serialize the report into UART v2 framed packet format.""" + def to_bytes(self, slot: int = 0) -> bytes: + """Serialize the report into a UART v3 frame addressed to ``slot``.""" + if not 0 <= slot < UART_SLOT_COUNT: + raise ValueError(f"slot must be 0-{UART_SLOT_COUNT - 1}, got {slot}") count = min(len(self.imu_samples), IMU_SAMPLES_PER_REPORT) payload = struct.pack( " None: - """Send a controller report to the Pico.""" - self.serial.write(report.to_bytes()) + def send_report(self, report: SwitchReport, slot: int = 0) -> None: + """Send a controller report to one of the Pico's controller slots.""" + self.serial.write(report.to_bytes(slot)) - def read_rumble(self) -> Optional[Tuple[float, float]]: + def read_rumble(self) -> Optional[Tuple[int, float, float]]: """ - Extract one decoded rumble frame as normalized low/high magnitudes. + Extract one decoded rumble frame as (slot, low, high) with magnitudes + normalized to 0.0-1.0. Frame format: 0: 0xBB (RUMBLE_HEADER) - 1: type (0x02 for decoded rumble) - 2: low-frequency magnitude (0-255) - 3: high-frequency magnitude (0-255) - 4: checksum (sum of first 4 bytes) & 0xFF + 1: type (0x03 slot frame; legacy 0x02 has no slot byte and means slot 0) + 2: slot (0x03 only) + then low-frequency magnitude, high-frequency magnitude (0-255) + and checksum (sum of the preceding bytes) & 0xFF """ waiting = self.serial.in_waiting if waiting: @@ -317,19 +325,25 @@ class PicoUART: if start < 0: self._buffer.clear() return None + if len(self._buffer) - start < 2: + del self._buffer[:start] + return None - if len(self._buffer) - start < 5: + frame_type = self._buffer[start + 1] + length = 6 if frame_type == RUMBLE_TYPE_SLOT else 5 + if len(self._buffer) - start < length: if start > 0: del self._buffer[:start] return None - frame = self._buffer[start : start + 5] - checksum = compute_checksum(bytes(frame[:4])) - - if frame[1] == RUMBLE_TYPE_DECODED and checksum == frame[4]: - rumble = (frame[2] / 255.0, frame[3] / 255.0) - del self._buffer[: start + 5] - return rumble + frame = bytes(self._buffer[start : start + length]) + if compute_checksum(frame[:-1]) == frame[-1]: + if frame_type == RUMBLE_TYPE_SLOT and frame[2] < UART_SLOT_COUNT: + del self._buffer[: start + length] + return frame[2], frame[3] / 255.0, frame[4] / 255.0 + if frame_type == RUMBLE_TYPE_DECODED: + del self._buffer[: start + length] + return 0, frame[2] / 255.0, frame[3] / 255.0 del self._buffer[: start + 1] @@ -408,6 +422,7 @@ class SwitchUARTClient: baud: int = UART_BAUD, send_interval: float = 1.0 / 500.0, auto_send: bool = True, + slot: int = 0, ) -> None: """ Args: @@ -416,7 +431,11 @@ class SwitchUARTClient: send_interval: Minimum interval between sends in seconds (defaults to 500 Hz). auto_send: If True, keep sending the current state in a background thread so the Pico continuously sees the latest input (mirrors controller_uart_bridge). + slot: Which of the Pico's emulated controllers (0-3) this client drives. """ + if not 0 <= slot < UART_SLOT_COUNT: + raise ValueError(f"slot must be 0-{UART_SLOT_COUNT - 1}, got {slot}") + self.slot = slot self.uart = PicoUART(port, baud) self.state = SwitchControllerState() self.send_interval = max(0.0, send_interval) @@ -432,7 +451,7 @@ class SwitchUARTClient: now = time.monotonic() if self.send_interval and (now - self._last_send) < self.send_interval: return - self.uart.send_report(self.state.report) + self.uart.send_report(self.state.report, self.slot) self._last_send = now def _start_auto_send_thread(self) -> None: @@ -517,10 +536,17 @@ class SwitchUARTClient: def poll_rumble(self) -> Optional[Tuple[float, float]]: """ - Poll for decoded low/high rumble magnitudes normalized to 0.0-1.0. - Returns None if no rumble frame was available. + Poll for decoded low/high rumble magnitudes normalized to 0.0-1.0 for + this client's slot. Returns None if no rumble frame was available; + frames addressed to other slots are discarded. """ - return self.uart.read_rumble() + while True: + frame = self.uart.read_rumble() + if frame is None: + return None + slot, low, high = frame + if slot == self.slot: + return low, high def close(self) -> None: if self._auto_thread: diff --git a/tests/switch_pro_driver_context_test.cpp b/tests/switch_pro_driver_context_test.cpp index 937cf68..4900147 100644 --- a/tests/switch_pro_driver_context_test.cpp +++ b/tests/switch_pro_driver_context_test.cpp @@ -681,8 +681,10 @@ void test_uart_parser_is_pure() { packet.back() = static_cast(packet.back() + packet[i]); } ControllerState parsed{}; - expect(switch_pro_apply_uart_packet(packet.data(), packet.size(), parsed), + uint8_t slot = 0xff; + expect(switch_pro_apply_uart_packet(packet.data(), packet.size(), parsed, slot), "valid UART packet was rejected"); + expect(slot == 0, "v2 UART packet must map to slot 0"); expect(parsed.button_east && parsed.button_left_shoulder && parsed.dpad_down && parsed.dpad_left, "UART buttons or hat were parsed incorrectly"); @@ -707,12 +709,44 @@ void test_uart_parser_is_pure() { ControllerState unchanged{}; unchanged.button_system = true; unchanged.left_stick_x = 123; + uint8_t unchanged_slot = 0xff; packet.back() ^= 0xffu; expect(!switch_pro_apply_uart_packet(packet.data(), packet.size(), - unchanged), + unchanged, unchanged_slot), "invalid UART checksum was accepted"); - expect(unchanged.button_system && unchanged.left_stick_x == 123, - "failed UART parse modified its output reference"); + expect(unchanged.button_system && unchanged.left_stick_x == 123 && + unchanged_slot == 0xff, + "failed UART parse modified its output references"); + + // v3 inserts a slot byte between the length and the payload. + std::array slotted{}; + slotted[0] = 0xaa; + slotted[1] = 0x03; + slotted[2] = 8; + slotted[3] = 2; + std::copy(packet.begin() + 3, packet.begin() + 10, slotted.begin() + 4); + for (unsigned i = 0; i < slotted.size() - 1; ++i) { + slotted.back() = static_cast(slotted.back() + slotted[i]); + } + ControllerState slotted_state{}; + expect(switch_pro_apply_uart_packet(slotted.data(), slotted.size(), + slotted_state, slot), + "valid v3 UART packet was rejected"); + expect(slot == 2, "v3 slot byte was not reported"); + expect(slotted_state.button_east && slotted_state.button_left_shoulder && + slotted_state.dpad_down && slotted_state.dpad_left && + slotted_state.right_stick_y == + controller_axis_from_unsigned(0x7878), + "v3 payload offsets were parsed incorrectly"); + + slotted[3] = SWITCH_PICO_HID_INSTANCE_COUNT; + slotted.back() = 0; + for (unsigned i = 0; i < slotted.size() - 1; ++i) { + slotted.back() = static_cast(slotted.back() + slotted[i]); + } + expect(!switch_pro_apply_uart_packet(slotted.data(), slotted.size(), + slotted_state, slot), + "out-of-range v3 slot was accepted"); } void test_motion_backpressure_retries_without_advancing_state() { diff --git a/tests/test_controller_uart_bridge.py b/tests/test_controller_uart_bridge.py index ecdcb50..2b9e822 100644 --- a/tests/test_controller_uart_bridge.py +++ b/tests/test_controller_uart_bridge.py @@ -28,10 +28,10 @@ class RecordingUART: def __init__(self) -> None: self.sent_imu: list[tuple[IMUSample, ...]] = [] - def send_report(self, report: SwitchReport) -> None: + def send_report(self, report: SwitchReport, slot: int = 0) -> None: self.sent_imu.append(tuple(report.imu_samples)) - def read_rumble(self) -> tuple[float, float] | None: + def read_rumble(self) -> tuple[int, float, float] | None: return None @@ -101,9 +101,8 @@ def test_sensor_buffer_retains_latest_three_samples() -> None: def test_service_republishes_latest_imu_window(monkeypatch: MonkeyPatch) -> None: uart = RecordingUART() controller = cast(sdl3.SDL_Gamepad, object()) - ctx = bridge.ControllerContext( - controller, 7, 0, "dualsense", "/dev/null", cast(PicoUART, cast(object, uart)) - ) + ctx = bridge.ControllerContext(controller, 7, 0, "dualsense", "/dev/null") + links = {"/dev/null": bridge.UartLink("/dev/null", cast(PicoUART, cast(object, uart)))} ctx.sensors_enabled = True samples = [ IMUSample(1, 2, 3, 4, 5, 6), @@ -123,8 +122,8 @@ def test_service_republishes_latest_imu_window(monkeypatch: MonkeyPatch) -> None args = Namespace(baud=UART_BAUD) console = Console(file=StringIO()) - bridge.service_contexts(1.0, args, config, contexts, [], console) - bridge.service_contexts(2.0, args, config, contexts, [], console) + bridge.service_contexts(1.0, args, config, contexts, links, console) + bridge.service_contexts(2.0, args, config, contexts, links, console) assert uart.sent_imu == [tuple(samples), tuple(samples)] assert ctx.imu_samples == samples diff --git a/tests/test_uart_protocol.py b/tests/test_uart_protocol.py index 071193b..b828202 100644 --- a/tests/test_uart_protocol.py +++ b/tests/test_uart_protocol.py @@ -1,4 +1,4 @@ -"""Tests for UART v2 protocol serialization in switch_pico_uart.""" +"""Tests for UART v3 protocol serialization in switch_pico_uart.""" import struct import pytest @@ -9,8 +9,10 @@ from switch_pico_bridge.switch_pico_uart import ( PicoUART, UART_HEADER, UART_PROTOCOL_VERSION, + UART_SLOT_COUNT, RUMBLE_HEADER, RUMBLE_TYPE_DECODED, + RUMBLE_TYPE_SLOT, ACCEL_LSB_PER_G, GYRO_LSB_PER_RAD_S, MS2_PER_G, @@ -35,7 +37,12 @@ class BufferedSerial: self._data.extend(data) -def make_rumble_frame(low: int, high: int) -> bytes: +def make_rumble_frame(low: int, high: int, slot: int = 0) -> bytes: + frame = bytes([RUMBLE_HEADER, RUMBLE_TYPE_SLOT, slot, low, high]) + return frame + bytes([compute_checksum(frame)]) + + +def make_legacy_rumble_frame(low: int, high: int) -> bytes: frame = bytes([RUMBLE_HEADER, RUMBLE_TYPE_DECODED, low, high]) return frame + bytes([compute_checksum(frame)]) @@ -48,8 +55,8 @@ def make_uart(data: bytes = b"") -> tuple[PicoUART, BufferedSerial]: return uart, serial_port -def test_v2_frame_with_imu_samples(): - """V2 frame with 3 IMU samples should be 48 bytes with correct layout.""" +def test_v3_frame_with_imu_samples(): + """V3 frame with 3 IMU samples should be 49 bytes with correct layout.""" r = SwitchReport( buttons=0, imu_samples=[ @@ -59,35 +66,49 @@ def test_v2_frame_with_imu_samples(): ], ) data = r.to_bytes() - assert len(data) == 48, f"Expected 48 bytes, got {len(data)}" + assert len(data) == 49, f"Expected 49 bytes, got {len(data)}" assert data[0] == UART_HEADER # 0xAA - assert data[1] == UART_PROTOCOL_VERSION # 0x02 + assert data[1] == UART_PROTOCOL_VERSION # 0x03 assert data[2] == 44 # payload_len - assert data[10] == 3 # imu_count + assert data[3] == 0 # slot + assert data[11] == 3 # imu_count # Verify checksum assert data[-1] == compute_checksum(data[:-1]) - # Verify first sample accel_x (int16 LE at byte 11) - ax0 = struct.unpack_from(" None: - self.rumble: list[tuple[float, float]] = [] + self.rumble: list[tuple[int, float, float]] = [] + self.sent: list[tuple[int, int]] = [] - def send_report(self, _report: SwitchReport) -> None: - pass + def send_report(self, report: SwitchReport, slot: int = 0) -> None: + self.sent.append((slot, report.buttons)) - def read_rumble(self) -> tuple[float, float] | None: + def read_rumble(self) -> tuple[int, float, float] | None: if not self.rumble: return None return self.rumble.pop(0) +def make_links(port: str, uart: RecordingUART) -> dict[str, bridge.UartLink]: + return {port: bridge.UartLink(port, cast(PicoUART, cast(object, uart)))} + + def make_config() -> bridge.BridgeConfig: return bridge.BridgeConfig( interval=10.0, @@ -88,29 +93,87 @@ def test_repeated_constant_rumble_stays_active_until_idle_timeout( uart = RecordingUART() controller = cast(sdl3.SDL_Gamepad, object()) - ctx = bridge.ControllerContext( - controller, - 7, - 0, - "controller", - "/dev/null", - cast(PicoUART, cast(object, uart)), - ) + ctx = bridge.ControllerContext(controller, 7, 0, "controller", "/dev/null") + links = make_links("/dev/null", uart) contexts = {ctx.instance_id: ctx} args = Namespace(baud=UART_BAUD) console = Console(file=StringIO()) - magnitude = (64 / 255.0, 192 / 255.0) + magnitude = (0, 64 / 255.0, 192 / 255.0) uart.rumble.append(magnitude) - bridge.service_contexts(1.0, args, make_config(), contexts, [], console) + bridge.service_contexts(1.0, args, make_config(), contexts, links, console) uart.rumble.append(magnitude) - bridge.service_contexts(1.7, args, make_config(), contexts, [], console) - bridge.service_contexts(1.71, args, make_config(), contexts, [], console) + bridge.service_contexts(1.7, args, make_config(), contexts, links, console) + bridge.service_contexts(1.71, args, make_config(), contexts, links, console) assert calls == [(16448, 49344, 50), (16448, 49344, 50)] assert ctx.rumble_active - bridge.service_contexts(1.96, args, make_config(), contexts, [], console) + bridge.service_contexts(1.96, args, make_config(), contexts, links, console) assert calls[-1] == (0, 0, 0) assert not ctx.rumble_active + + +def test_shared_port_routes_reports_and_rumble_by_slot( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[tuple[object, int, int]] = [] + monkeypatch.setattr( + bridge.sdl3, + "SDL_RumbleGamepad", + lambda controller, low, high, _duration: calls.append((controller, low, high)) or True, + ) + monkeypatch.setattr(bridge, "poll_controller_buttons", lambda _ctx, _map: None) + + uart = RecordingUART() + pad_a = cast(sdl3.SDL_Gamepad, object()) + pad_b = cast(sdl3.SDL_Gamepad, object()) + ctx_a = bridge.ControllerContext(pad_a, 7, 0, "a", "COM11", slot=0) + ctx_b = bridge.ControllerContext(pad_b, 8, 1, "b", "COM11", slot=2) + ctx_a.report.buttons = 0x0001 + ctx_b.report.buttons = 0x0002 + contexts = {7: ctx_a, 8: ctx_b} + links = make_links("COM11", uart) + args = Namespace(baud=UART_BAUD) + console = Console(file=StringIO()) + + # Slot 2 rumbles, slot 0 is idle, slot 3 has no controller attached. + uart.rumble.extend([(0, 0.0, 0.0), (2, 1.0, 0.5), (3, 1.0, 1.0)]) + bridge.service_contexts(20.0, args, make_config(), contexts, links, console) + + assert sorted(uart.sent) == [(0, 0x0001), (2, 0x0002)] + assert calls == [(pad_a, 0, 0), (pad_b, 0xFFFF, 0x7FFF)] + assert not ctx_a.rumble_active + assert ctx_b.rumble_active + + +def test_auto_pairing_spreads_controllers_across_ports_then_fills_slots() -> None: + pairing = bridge.PairingState( + mapping_by_index={}, + available_ports=["COM11", "COM12"], + slots_per_port=2, + auto_pairing_enabled=True, + ) + console = Console(file=StringIO()) + + assignments = [bridge.assign_port_for_index(pairing, idx, console) for idx in range(5)] + + assert assignments == [("COM11", 0), ("COM12", 0), ("COM11", 1), ("COM12", 1), None] + # Releasing a slot makes exactly that slot reusable. + del pairing.mapping_by_index[2] + assert bridge.assign_port_for_index(pairing, 9, console) == ("COM11", 1) + + +def test_explicit_mappings_fill_omitted_slots_and_reject_conflicts() -> None: + parser = bridge.build_arg_parser() + resolved = bridge.resolve_mapping_slots( + [(0, "COM11", None), (1, "COM11", 3), (2, "COM11", None)], 4, parser + ) + assert resolved == {0: ("COM11", 0), 1: ("COM11", 3), 2: ("COM11", 1)} + with pytest.raises(SystemExit): + bridge.resolve_mapping_slots([(0, "COM11", 1), (1, "COM11", 1)], 4, parser) + with pytest.raises(SystemExit): + bridge.resolve_mapping_slots([(0, "COM11", None), (1, "COM11", None)], 1, parser) + assert bridge.parse_mapping("2:COM11:3") == (2, "COM11", 3) + assert bridge.parse_mapping("0:/dev/ttyUSB0") == (0, "/dev/ttyUSB0", None)