jyapayne/switch-pico
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add stateful HD-rumble decoder with TinyUSB report normalization

Browse source 
Decode Nintendo HD-rumble words once in firmware into conventional low/high magnitudes and normalize stripped control SET_REPORT vs complete interrupt OUT reports through one path.
This commit is contained in:
Joey Yakimowich-Payne 2026-08-29 22:12:24 -06:00
commit bce94d38ce
21 changed files with 902 additions and 180 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/switch_pico_bridge/__init__.py
View file

@ -10,7 +10,6 @@ from .switch_pico_uart import ( # noqa: F401
SwitchDpad,
SwitchUARTClient,
axis_to_stick,
decode_rumble,
discover_serial_ports,
first_serial_port,
str_to_dpad,
@ -24,7 +23,6 @@ __all__ = [
"discover_serial_ports",
"first_serial_port",
"axis_to_stick",
"decode_rumble",
"str_to_dpad",
"trigger_to_button",
]

74
src/switch_pico_bridge/controller_uart_bridge.py
View file

@ -1,16 +1,17 @@
#!/usr/bin/env python3
"""
Bridge multiple SDL2 controllers to switch-pico over UART and mirror rumble back.
Bridge multiple SDL3 controllers to switch-pico over UART and mirror rumble back.
The framing matches ``switch-pico.cpp``:
- Host -> Pico : 0xAA, buttons (LE16), hat, lx, ly, rx, ry
- Pico -> Host : 0xBB, 0x01, 8 rumble bytes, checksum (sum of first 10 bytes)
- Host -> Pico : UART v2 controller report
- Pico -> Host : 0xBB, 0x02, low-frequency magnitude,
high-frequency magnitude, checksum
Features inspired by ``host/controller_bridge.py``:
- Multiple controllers paired to multiple UART ports
- Rich-powered interactive pairing UI
- Adjustable send frequency, deadzone, and trigger thresholds
- Rumble feedback delivered to SDL2 controllers
- Rumble feedback delivered to SDL3 controllers
"""
from __future__ import annotations
@ -48,15 +49,12 @@ from .switch_pico_uart import (
SwitchReport,
axis_to_stick,
str_to_dpad,
decode_rumble,
discover_serial_ports,
trigger_to_button,
)
RUMBLE_IDLE_TIMEOUT = 0.25 # seconds without packets before forcing rumble off
RUMBLE_STUCK_TIMEOUT = 0.60 # continuous same-energy rumble will be stopped after this
RUMBLE_MIN_ACTIVE = 0.40 # below this, rumble is treated as off/noise
RUMBLE_SCALE = 1.0
RUMBLE_DURATION_MS = 50
CONTROLLER_DB_URL_DEFAULT = "https://raw.githubusercontent.com/mdqinc/SDL_GameControllerDB/refs/heads/master/gamecontrollerdb.txt"
SDL_TRUE = True
SDL_EVENT_GAMEPAD_SENSOR_UPDATE = getattr(sdl3, "SDL_EVENT_GAMEPAD_SENSOR_UPDATE", 0x658)
@ -199,21 +197,16 @@ def interactive_pairing(
return mappings
def apply_rumble(controller: sdl3.SDL_Gamepad, payload: bytes) -> float:
"""Apply rumble payload to SDL controller and return max normalized energy."""
left_norm, right_norm = decode_rumble(payload)
max_norm = max(left_norm, right_norm)
# Treat small rumble as "off" to avoid idle buzz.
if max_norm < RUMBLE_MIN_ACTIVE:
sdl3.SDL_RumbleGamepad(controller, 0, 0, 0)
return 0.0
# Attenuate to feel closer to a real controller; cap at ~25% strength.
scale = RUMBLE_SCALE
low = int(min(1.0, left_norm * scale) * 0xFFFF) # SDL: low_frequency_rumble
high = int(min(1.0, right_norm * scale) * 0xFFFF) # SDL: high_frequency_rumble
duration = 10
sdl3.SDL_RumbleGamepad(controller, low, high, duration)
return max_norm
def apply_rumble(
controller: sdl3.SDL_Gamepad,
low_frequency: float,
high_frequency: float,
) -> bool:
"""Apply normalized low/high rumble magnitudes to an SDL controller."""
low = int(max(0.0, min(1.0, low_frequency)) * 0xFFFF)
high = int(max(0.0, min(1.0, high_frequency)) * 0xFFFF)
sdl3.SDL_RumbleGamepad(controller, low, high, RUMBLE_DURATION_MS)
return low != 0 or high != 0
@dataclass
@ -239,9 +232,7 @@ class ControllerContext:
)
last_send: float = 0.0
last_reopen_attempt: float = 0.0
last_rumble: float = 0.0
last_rumble_change: float = 0.0
last_rumble_energy: float = 0.0
last_rumble_at: float = 0.0
rumble_active: bool = False
axis_offsets: Dict[int, int] = field(default_factory=dict)
swap_abxy: bool = False
@ -1134,7 +1125,6 @@ def handle_removed_port(
ctx.uart = None
ctx.port = None
ctx.rumble_active = False
ctx.last_rumble_energy = 0.0
ctx.last_reopen_attempt = time.monotonic()
console.print(
f"[yellow]UART {path} removed; controller {ctx.controller_index} waiting for reassignment[/yellow]"
@ -1574,32 +1564,25 @@ def service_contexts(
ctx.uart.send_report(ctx.report)
ctx.last_send = now
last_payload = None
latest_rumble = None
while True:
p = ctx.uart.read_rumble_payload()
if not p:
rumble = ctx.uart.read_rumble()
if rumble is None:
break
last_payload = p
latest_rumble = rumble
if last_payload is not None:
# Apply only the freshest rumble payload seen during this tick.
energy = apply_rumble(ctx.controller, last_payload)
ctx.rumble_active = energy >= RUMBLE_MIN_ACTIVE
if ctx.rumble_active and energy != ctx.last_rumble_energy:
ctx.last_rumble_change = now
ctx.last_rumble_energy = energy
ctx.last_rumble = now
elif ctx.rumble_active and (now - ctx.last_rumble) > RUMBLE_IDLE_TIMEOUT:
sdl3.SDL_RumbleGamepad(ctx.controller, 0, 0, 0)
ctx.rumble_active = False
ctx.last_rumble_energy = 0.0
if latest_rumble is not None:
# Apply only the freshest rumble command seen during this tick.
ctx.rumble_active = apply_rumble(
ctx.controller, latest_rumble[0], latest_rumble[1]
)
ctx.last_rumble_at = now
elif (
ctx.rumble_active
and (now - ctx.last_rumble_change) > RUMBLE_STUCK_TIMEOUT
and (now - ctx.last_rumble_at) > RUMBLE_IDLE_TIMEOUT
):
sdl3.SDL_RumbleGamepad(ctx.controller, 0, 0, 0)
ctx.rumble_active = False
ctx.last_rumble_energy = 0.0
except SerialException as exc:
console.print(f"[yellow]UART {ctx.port} disconnected: {exc}[/yellow]")
try:
@ -1609,7 +1592,6 @@ def service_contexts(
sdl3.SDL_RumbleGamepad(ctx.controller, 0, 0, 0)
ctx.uart = None
ctx.rumble_active = False
ctx.last_rumble_energy = 0.0
ctx.last_reopen_attempt = now
except Exception as exc:
console.print(f"[red]UART error on {ctx.port}: {exc}[/red]")

56
src/switch_pico_bridge/switch_pico_uart.py
View file

@ -2,12 +2,13 @@
"""
Lightweight helpers for talking to the switch-pico firmware over UART.
This module exposes the raw report structure plus a small convenience wrapper
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``:
Host -> Pico : 0xAA, buttons (LE16), hat, lx, ly, rx, ry
Pico -> Host : 0xBB, 0x01, 8 rumble bytes, checksum (sum of first 10 bytes)
Host -> Pico : UART v2 controller report
Pico -> Host : 0xBB, 0x02, low-frequency magnitude, high-frequency magnitude,
checksum (sum of the first 4 bytes)
"""
from __future__ import annotations
@ -26,7 +27,7 @@ from serial.tools import list_ports, list_ports_common
UART_HEADER = 0xAA
UART_PROTOCOL_VERSION = 0x02
RUMBLE_HEADER = 0xBB
RUMBLE_TYPE_RUMBLE = 0x01
RUMBLE_TYPE_DECODED = 0x02
UART_BAUD = 921600
IMU_SAMPLES_PER_REPORT = 3
@ -300,15 +301,16 @@ class PicoUART:
"""Send a controller report to the Pico."""
self.serial.write(report.to_bytes())
def read_rumble_payload(self) -> Optional[bytes]:
def read_rumble(self) -> Optional[Tuple[float, float]]:
"""
Drain available UART bytes into an internal buffer, then extract one rumble frame.
Extract one decoded rumble frame as normalized low/high magnitudes.
Frame format:
0: 0xBB (RUMBLE_HEADER)
1: type (0x01 for rumble)
2-9: 8-byte rumble payload
10: checksum (sum of first 10 bytes) & 0xFF
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
"""
waiting = self.serial.in_waiting
if waiting:
@ -323,18 +325,18 @@ class PicoUART:
self._buffer.clear()
return None
if len(self._buffer) - start < 11:
if len(self._buffer) - start < 5:
if start > 0:
del self._buffer[:start]
return None
frame = self._buffer[start : start + 11]
checksum = compute_checksum(bytes(frame[:10]))
frame = self._buffer[start : start + 5]
checksum = compute_checksum(bytes(frame[:4]))
if frame[1] == RUMBLE_TYPE_RUMBLE and checksum == frame[10]:
payload = bytes(frame[2:10])
del self._buffer[: start + 11]
return payload
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
del self._buffer[: start + 1]
@ -343,21 +345,6 @@ class PicoUART:
self.serial.close()
def decode_rumble(payload: bytes) -> Tuple[float, float]:
"""Return normalized rumble amplitudes (0.0-1.0) for left/right."""
if len(payload) < 8:
return 0.0, 0.0
if payload == b"\x00\x01\x40\x40\x00\x01\x40\x40":
return 0.0, 0.0
right_raw = ((payload[1] & 0x03) << 8) | payload[0]
left_raw = ((payload[5] & 0x03) << 8) | payload[4]
if left_raw < 8 and right_raw < 8:
return 0.0, 0.0
left = min(max(left_raw / 1023.0, 0.0), 1.0)
right = min(max(right_raw / 1023.0, 0.0), 1.0)
return left, right
@dataclass
class SwitchControllerState:
"""Mutable controller state with helpers for building reports."""
@ -537,13 +524,10 @@ class SwitchUARTClient:
def poll_rumble(self) -> Optional[Tuple[float, float]]:
"""
Poll for the latest rumble payload and return normalized amplitudes.
Poll for decoded low/high rumble magnitudes normalized to 0.0-1.0.
Returns None if no rumble frame was available.
"""
payload = self.uart.read_rumble_payload()
if payload:
return decode_rumble(payload)
return None
return self.uart.read_rumble()
def close(self) -> None:
if self._auto_thread: