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

Initial gyro, swinging wildly

Browse source
This commit is contained in:
Joey Yakimowich-Payne 2025-11-24 12:37:54 -07:00
commit 4640b9fe54
No known key found for this signature in database
GPG key ID: 6BFE655FA5ABD1E1
4 changed files with 443 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

396
src/switch_pico_bridge/controller_uart_bridge.py
View file

@ -21,7 +21,7 @@ import sys
import time
import urllib.request
from dataclasses import dataclass, field
from ctypes import create_string_buffer
from ctypes import create_string_buffer, c_float
from pathlib import Path
from typing import Dict, List, Optional, Tuple
@ -34,6 +34,12 @@ from rich.table import Table
from .switch_pico_uart import (
UART_BAUD,
MS2_PER_G,
RAD_TO_DEG,
SENSOR_ACCEL,
SENSOR_GYRO,
IMU_SAMPLES_PER_REPORT,
IMUSample,
PicoUART,
SwitchButton,
SwitchDpad,
@ -49,9 +55,9 @@ 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
CONTROLLER_DB_URL_DEFAULT = (
"https://raw.githubusercontent.com/mdqinc/SDL_GameControllerDB/refs/heads/master/gamecontrollerdb.txt"
)
CONTROLLER_DB_URL_DEFAULT = "https://raw.githubusercontent.com/mdqinc/SDL_GameControllerDB/refs/heads/master/gamecontrollerdb.txt"
SDL_TRUE = getattr(sdl2, "SDL_TRUE", 1)
def parse_mapping(value: str) -> Tuple[int, str]:
@ -62,7 +68,9 @@ def parse_mapping(value: str) -> Tuple[int, str]:
try:
idx = int(idx_str, 10)
except ValueError as exc:
raise argparse.ArgumentTypeError(f"Invalid controller index '{idx_str}'") from exc
raise argparse.ArgumentTypeError(
f"Invalid controller index '{idx_str}'"
) from exc
if not port:
raise argparse.ArgumentTypeError("Serial port cannot be empty")
return idx, port.strip()
@ -84,9 +92,13 @@ def download_controller_db(console: Console, destination: Path, url: str) -> boo
try:
destination.write_bytes(data)
except Exception as exc:
console.print(f"[red]Failed to write controller database to {destination}: {exc}[/red]")
console.print(
f"[red]Failed to write controller database to {destination}: {exc}[/red]"
)
return False
console.print(f"[green]Updated controller database ({len(data)} bytes) at {destination}[/green]")
console.print(
f"[green]Updated controller database ({len(data)} bytes) at {destination}[/green]"
)
return True
@ -98,7 +110,9 @@ def parse_hotkey(value: str) -> str:
if not value:
return ""
if len(value) != 1:
raise argparse.ArgumentTypeError("Hotkeys must be a single character (or empty to disable).")
raise argparse.ArgumentTypeError(
"Hotkeys must be a single character (or empty to disable)."
)
return value
@ -151,7 +165,9 @@ def interactive_pairing(
mappings: List[Tuple[int, str]] = []
for controller_idx in controller_info:
if not available:
console.print("[bold red]No more UART devices available for pairing.[/bold red]")
console.print(
"[bold red]No more UART devices available for pairing.[/bold red]"
)
break
table = Table(
@ -174,7 +190,9 @@ def interactive_pairing(
idx = int(selection)
port = available.pop(idx)
mappings.append((controller_idx, port["device"]))
console.print(f"[bold green]Paired controller {controller_idx} with {port['device']}[/bold green]")
console.print(
f"[bold green]Paired controller {controller_idx} with {port['device']}[/bold green]"
)
return mappings
@ -188,7 +206,7 @@ def apply_rumble(controller: sdl2.SDL_GameController, payload: bytes) -> float:
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
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
sdl2.SDL_GameControllerRumble(controller, low, high, duration)
@ -204,9 +222,18 @@ class ControllerContext:
port: Optional[str]
uart: Optional[PicoUART]
report: SwitchReport = field(default_factory=SwitchReport)
dpad: Dict[str, bool] = field(default_factory=lambda: {"up": False, "down": False, "left": False, "right": False})
dpad: Dict[str, bool] = field(
default_factory=lambda: {
"up": False,
"down": False,
"left": False,
"right": False,
}
)
button_state: Dict[int, bool] = field(default_factory=dict)
last_trigger_state: Dict[str, bool] = field(default_factory=lambda: {"left": False, "right": False})
last_trigger_state: Dict[str, bool] = field(
default_factory=lambda: {"left": False, "right": False}
)
last_send: float = 0.0
last_reopen_attempt: float = 0.0
last_rumble: float = 0.0
@ -214,6 +241,10 @@ class ControllerContext:
last_rumble_energy: float = 0.0
rumble_active: bool = False
axis_offsets: Dict[int, int] = field(default_factory=dict)
sensors_supported: bool = False
sensors_enabled: bool = False
imu_samples: List[IMUSample] = field(default_factory=list)
last_sensor_poll: float = 0.0
def capture_stick_offsets(controller: sdl2.SDL_GameController) -> Dict[int, int]:
@ -226,7 +257,9 @@ def capture_stick_offsets(controller: sdl2.SDL_GameController) -> Dict[int, int]
def format_axis_offsets(offsets: Dict[int, int]) -> str:
"""Return a human-friendly summary of per-axis offsets (for logging)."""
return ", ".join(f"{label}={offsets.get(axis, 0):+d}" for axis, label in STICK_AXIS_LABELS)
return ", ".join(
f"{label}={offsets.get(axis, 0):+d}" for axis, label in STICK_AXIS_LABELS
)
def calibrate_axis_value(value: int, axis: int, ctx: ControllerContext) -> int:
@ -242,7 +275,9 @@ def calibrate_axis_value(value: int, axis: int, ctx: ControllerContext) -> int:
class HotkeyMonitor:
"""Platform-aware helper that watches for configured hotkeys without blocking the main loop."""
def __init__(self, console: Console, key_messages: Optional[Dict[str, str]] = None) -> None:
def __init__(
self, console: Console, key_messages: Optional[Dict[str, str]] = None
) -> None:
self.console = console
self._platform = os.name
self._fd: Optional[int] = None
@ -271,7 +306,9 @@ class HotkeyMonitor:
try:
import msvcrt # type: ignore
except ImportError:
self.console.print("[yellow]Hotkeys disabled: msvcrt unavailable.[/yellow]")
self.console.print(
"[yellow]Hotkeys disabled: msvcrt unavailable.[/yellow]"
)
return False
self._msvcrt = msvcrt
self._active = True
@ -296,7 +333,11 @@ class HotkeyMonitor:
def suspend(self) -> None:
if not self._active:
return
if self._platform != "nt" and self._fd is not None and self._orig_termios is not None:
if (
self._platform != "nt"
and self._fd is not None
and self._orig_termios is not None
):
import termios
termios.tcsetattr(self._fd, termios.TCSADRAIN, self._orig_termios)
@ -316,7 +357,11 @@ class HotkeyMonitor:
self._active = True
def stop(self) -> None:
if self._platform != "nt" and self._fd is not None and self._orig_termios is not None:
if (
self._platform != "nt"
and self._fd is not None
and self._orig_termios is not None
):
import termios
termios.tcsetattr(self._fd, termios.TCSADRAIN, self._orig_termios)
@ -339,7 +384,9 @@ class HotkeyMonitor:
def _print_instructions(self) -> None:
if not self._keys:
return
instructions = " | ".join(f"'{key.upper()}' to {message}" for key, message in self._keys.items())
instructions = " | ".join(
f"'{key.upper()}' to {message}" for key, message in self._keys.items()
)
self.console.print(f"[magenta]Hotkeys active: {instructions}[/magenta]")
def _read_key(self) -> Optional[str]:
@ -361,7 +408,11 @@ class HotkeyMonitor:
return ch
def zero_context_sticks(ctx: ControllerContext, console: Optional[Console] = None, reason: str = "Zeroed stick centers") -> None:
def zero_context_sticks(
ctx: ControllerContext,
console: Optional[Console] = None,
reason: str = "Zeroed stick centers",
) -> None:
"""Capture and store the current stick positions for a controller."""
offsets = capture_stick_offsets(ctx.controller)
ctx.axis_offsets = offsets
@ -371,7 +422,9 @@ def zero_context_sticks(ctx: ControllerContext, console: Optional[Console] = Non
)
def zero_all_context_sticks(contexts: Dict[int, ControllerContext], console: Console) -> None:
def zero_all_context_sticks(
contexts: Dict[int, ControllerContext], console: Console
) -> None:
"""Zero every connected controller's sticks."""
if not contexts:
console.print("[yellow]No controllers available to zero right now.[/yellow]")
@ -390,10 +443,14 @@ def controller_display_name(ctx: ControllerContext) -> str:
return str(name)
def toggle_abxy_for_context(ctx: ControllerContext, config: BridgeConfig, console: Console) -> None:
def toggle_abxy_for_context(
ctx: ControllerContext, config: BridgeConfig, console: Console
) -> None:
"""Toggle the ABXY layout for a single controller."""
if config.swap_abxy_global:
console.print("[yellow]Global --swap-abxy is enabled; disable it to use per-controller toggles.[/yellow]")
console.print(
"[yellow]Global --swap-abxy is enabled; disable it to use per-controller toggles.[/yellow]"
)
return
swapped = ctx.stable_id in config.swap_abxy_ids
action = "standard" if swapped else "swapped"
@ -414,9 +471,13 @@ def prompt_swap_abxy_controller(
) -> None:
"""Prompt the user to choose a controller whose ABXY layout should be toggled."""
if not contexts:
console.print("[yellow]No controllers connected to toggle ABXY layout.[/yellow]")
console.print(
"[yellow]No controllers connected to toggle ABXY layout.[/yellow]"
)
return
controllers = sorted(contexts.values(), key=lambda ctx: (ctx.controller_index, ctx.instance_id))
controllers = sorted(
contexts.values(), key=lambda ctx: (ctx.controller_index, ctx.instance_id)
)
table = Table(title="Toggle ABXY layout for a controller")
table.add_column("Choice", justify="center")
table.add_column("SDL Index", justify="center")
@ -461,7 +522,9 @@ def open_controller(index: int) -> Tuple[sdl2.SDL_GameController, int, str]:
"""Open an SDL GameController by index and return it with instance ID and GUID string."""
controller = sdl2.SDL_GameControllerOpen(index)
if not controller:
raise RuntimeError(f"Failed to open controller {index}: {sdl2.SDL_GetError().decode()}")
raise RuntimeError(
f"Failed to open controller {index}: {sdl2.SDL_GetError().decode()}"
)
joystick = sdl2.SDL_GameControllerGetJoystick(controller)
instance_id = sdl2.SDL_JoystickInstanceID(joystick)
guid_str = guid_string_from_joystick(joystick)
@ -503,7 +566,9 @@ def open_uart_or_warn(port: str, baud: int, console: Console) -> Optional[PicoUA
def build_arg_parser() -> argparse.ArgumentParser:
"""Construct the CLI argument parser for the bridge."""
parser = argparse.ArgumentParser(description="Bridge SDL2 controllers to switch-pico UART (with rumble)")
parser = argparse.ArgumentParser(
description="Bridge SDL2 controllers to switch-pico UART (with rumble)"
)
parser.add_argument(
"--map",
action="append",
@ -516,8 +581,16 @@ def build_arg_parser() -> argparse.ArgumentParser:
nargs="+",
help="Serial ports to auto-pair with controllers in ascending index order.",
)
parser.add_argument("--interactive", action="store_true", help="Launch an interactive pairing UI using Rich.")
parser.add_argument("--all-ports", action="store_true", help="Include non-USB serial ports when listing devices.")
parser.add_argument(
"--interactive",
action="store_true",
help="Launch an interactive pairing UI using Rich.",
)
parser.add_argument(
"--all-ports",
action="store_true",
help="Include non-USB serial ports when listing devices.",
)
parser.add_argument(
"--frequency",
type=float,
@ -627,7 +700,9 @@ def build_arg_parser() -> argparse.ArgumentParser:
return parser
def poll_controller_buttons(ctx: ControllerContext, button_map: Dict[int, SwitchButton]) -> None:
def poll_controller_buttons(
ctx: ControllerContext, button_map: Dict[int, SwitchButton]
) -> None:
"""Update button/hat state based on current SDL controller readings."""
changed = False
for sdl_button, switch_bit in button_map.items():
@ -682,7 +757,95 @@ class PairingState:
include_port_mfr: List[str] = field(default_factory=list)
def load_button_maps(console: Console, args: argparse.Namespace) -> Tuple[Dict[int, SwitchButton], Dict[int, SwitchButton], set[int]]:
def clamp_int16(value: float) -> int:
return int(max(-32768, min(32767, round(value))))
def read_sensor_triplet(
controller: sdl2.SDL_GameController, sensor_type: int
) -> Optional[Tuple[float, float, float]]:
if not hasattr(sdl2, "SDL_GameControllerGetSensorData"):
return None
data = (c_float * 3)()
result = sdl2.SDL_GameControllerGetSensorData(controller, sensor_type, data, 3)
if result != 0:
return None
return float(data[0]), float(data[1]), float(data[2])
def convert_accel_to_raw(accel_ms2: float) -> int:
g_units = accel_ms2 / MS2_PER_G
return clamp_int16(g_units * 4096.0)
def convert_gyro_to_raw(gyro_rad: float) -> int:
dps = gyro_rad * RAD_TO_DEG
return clamp_int16(dps / 0.070)
def initialize_controller_sensors(ctx: ControllerContext, console: Console) -> None:
if not all(
hasattr(sdl2, name)
for name in (
"SDL_GameControllerHasSensor",
"SDL_GameControllerSetSensorEnabled",
"SDL_GameControllerGetSensorData",
)
):
return
if SENSOR_ACCEL is None or SENSOR_GYRO is None:
return
accel_supported = bool(
sdl2.SDL_GameControllerHasSensor(ctx.controller, SENSOR_ACCEL)
)
gyro_supported = bool(sdl2.SDL_GameControllerHasSensor(ctx.controller, SENSOR_GYRO))
ctx.sensors_supported = accel_supported and gyro_supported
if not ctx.sensors_supported:
console.print(
f"[yellow]Controller {ctx.controller_index} has no accelerometer/gyro sensors[/yellow]"
)
return
accel_enabled = (
sdl2.SDL_GameControllerSetSensorEnabled(ctx.controller, SENSOR_ACCEL, SDL_TRUE)
== 0
)
gyro_enabled = (
sdl2.SDL_GameControllerSetSensorEnabled(ctx.controller, SENSOR_GYRO, SDL_TRUE)
== 0
)
ctx.sensors_enabled = accel_enabled and gyro_enabled
if not ctx.sensors_enabled:
console.print(
f"[yellow]Controller {ctx.controller_index} failed to enable sensors[/yellow]"
)
def collect_imu_sample(ctx: ControllerContext) -> None:
if not ctx.sensors_enabled or SENSOR_ACCEL is None or SENSOR_GYRO is None:
return
accel = read_sensor_triplet(ctx.controller, SENSOR_ACCEL)
gyro = read_sensor_triplet(ctx.controller, SENSOR_GYRO)
if not accel or not gyro:
return
sample = IMUSample(
accel_x=convert_accel_to_raw(accel[0]),
accel_y=convert_accel_to_raw(accel[1]),
accel_z=convert_accel_to_raw(accel[2]),
gyro_x=convert_gyro_to_raw(gyro[0]),
gyro_y=convert_gyro_to_raw(gyro[1]),
gyro_z=convert_gyro_to_raw(gyro[2]),
)
ctx.imu_samples.append(sample)
if len(ctx.imu_samples) > 6:
ctx.imu_samples = ctx.imu_samples[-6:]
def load_button_maps(
console: Console, args: argparse.Namespace
) -> Tuple[Dict[int, SwitchButton], Dict[int, SwitchButton], set[int]]:
"""Load SDL controller mappings and return button map variants."""
default_mapping = Path(__file__).parent / "controller_db" / "gamecontrollerdb.txt"
if args.update_controller_db or not default_mapping.exists():
@ -697,13 +860,19 @@ def load_button_maps(console: Console, args: argparse.Namespace) -> Tuple[Dict[i
button_map_swapped[sdl2.SDL_CONTROLLER_BUTTON_B] = SwitchButton.A
button_map_swapped[sdl2.SDL_CONTROLLER_BUTTON_X] = SwitchButton.Y
button_map_swapped[sdl2.SDL_CONTROLLER_BUTTON_Y] = SwitchButton.X
swap_abxy_indices = {idx for idx in args.swap_abxy_index if idx is not None and idx >= 0}
swap_abxy_indices = {
idx for idx in args.swap_abxy_index if idx is not None and idx >= 0
}
for mapping_path in mappings_to_load:
try:
loaded = sdl2.SDL_GameControllerAddMappingsFromFile(mapping_path.encode())
console.print(f"[green]Loaded {loaded} SDL mapping(s) from {mapping_path}[/green]")
console.print(
f"[green]Loaded {loaded} SDL mapping(s) from {mapping_path}[/green]"
)
except Exception as exc:
console.print(f"[red]Failed to load SDL mapping {mapping_path}: {exc}[/red]")
console.print(
f"[red]Failed to load SDL mapping {mapping_path}: {exc}[/red]"
)
return button_map_default, button_map_swapped, swap_abxy_indices
@ -712,7 +881,9 @@ def build_bridge_config(console: Console, args: argparse.Namespace) -> BridgeCon
interval = 1.0 / max(args.frequency, 1.0)
deadzone_raw = int(max(0.0, min(args.deadzone, 1.0)) * 32767)
trigger_threshold = int(max(0.0, min(args.trigger_threshold, 1.0)) * 32767)
button_map_default, button_map_swapped, swap_abxy_indices = load_button_maps(console, args)
button_map_default, button_map_swapped, swap_abxy_indices = load_button_maps(
console, args
)
swap_abxy_guids = {g.lower() for g in args.swap_abxy_guid}
return BridgeConfig(
interval=interval,
@ -736,7 +907,14 @@ def initialize_sdl(parser: argparse.ArgumentParser) -> None:
set_hint("SDL_JOYSTICK_HIDAPI_SWITCH", "1")
# Use controller button labels so Nintendo layouts (ABXY) map correctly on Linux.
set_hint("SDL_GAMECONTROLLER_USE_BUTTON_LABELS", "1")
if sdl2.SDL_Init(sdl2.SDL_INIT_GAMECONTROLLER | sdl2.SDL_INIT_JOYSTICK | sdl2.SDL_INIT_EVERYTHING) != 0:
if (
sdl2.SDL_Init(
sdl2.SDL_INIT_GAMECONTROLLER
| sdl2.SDL_INIT_JOYSTICK
| sdl2.SDL_INIT_EVERYTHING
)
!= 0
):
parser.error(f"SDL init failed: {sdl2.SDL_GetError().decode(errors='ignore')}")
@ -754,8 +932,12 @@ def detect_controllers(
if sdl2.SDL_IsGameController(index):
name = sdl2.SDL_GameControllerNameForIndex(index)
name_str = name.decode() if isinstance(name, bytes) else str(name)
if include_controller_name and all(substr not in name_str.lower() for substr in include_controller_name):
console.print(f"[yellow]Skipping controller {index} ({name_str}) due to name filter[/yellow]")
if include_controller_name and all(
substr not in name_str.lower() for substr in include_controller_name
):
console.print(
f"[yellow]Skipping controller {index} ({name_str}) due to name filter[/yellow]"
)
continue
console.print(f"[cyan]Detected controller {index}: {name_str}[/cyan]")
controller_indices.append(index)
@ -763,14 +945,22 @@ def detect_controllers(
else:
name = sdl2.SDL_JoystickNameForIndex(index)
name_str = name.decode() if isinstance(name, bytes) else str(name)
if include_controller_name and all(substr not in name_str.lower() for substr in include_controller_name):
console.print(f"[yellow]Skipping joystick {index} ({name_str}) due to name filter[/yellow]")
if include_controller_name and all(
substr not in name_str.lower() for substr in include_controller_name
):
console.print(
f"[yellow]Skipping joystick {index} ({name_str}) due to name filter[/yellow]"
)
continue
console.print(f"[yellow]Found joystick {index} (not a GameController): {name_str}[/yellow]")
console.print(
f"[yellow]Found joystick {index} (not a GameController): {name_str}[/yellow]"
)
return controller_indices, controller_names
def list_controllers_with_guids(console: Console, parser: argparse.ArgumentParser) -> None:
def list_controllers_with_guids(
console: Console, parser: argparse.ArgumentParser
) -> None:
"""Print detected controllers with their GUID strings and exit."""
count = sdl2.SDL_NumJoysticks()
if count < 0:
@ -785,10 +975,16 @@ def list_controllers_with_guids(console: Console, parser: argparse.ArgumentParse
table.add_column("GUID")
for idx in range(count):
is_gc = sdl2.SDL_IsGameController(idx)
name = sdl2.SDL_GameControllerNameForIndex(idx) if is_gc else sdl2.SDL_JoystickNameForIndex(idx)
name = (
sdl2.SDL_GameControllerNameForIndex(idx)
if is_gc
else sdl2.SDL_JoystickNameForIndex(idx)
)
name_str = name.decode() if isinstance(name, bytes) else str(name)
guid_str = guid_string_for_device_index(idx)
table.add_row(str(idx), "GameController" if is_gc else "Joystick", name_str, guid_str)
table.add_row(
str(idx), "GameController" if is_gc else "Joystick", name_str, guid_str
)
console.print(table)
@ -827,7 +1023,9 @@ def prepare_pairing_state(
elif auto_pairing_enabled:
if args.ports:
available_ports.extend(list(args.ports))
console.print(f"[green]Prepared {len(available_ports)} specified UART port(s) for auto-pairing.[/green]")
console.print(
f"[green]Prepared {len(available_ports)} specified UART port(s) for auto-pairing.[/green]"
)
else:
# Passive mode: grab whatever UARTs exist now, and keep looking later.
discovered = discover_serial_ports(
@ -842,7 +1040,9 @@ def prepare_pairing_state(
for info in discovered:
console.print(f" {info['device']} ({info['description']})")
else:
console.print("[yellow]No UART devices detected yet; waiting for hotplug...[/yellow]")
console.print(
"[yellow]No UART devices detected yet; waiting for hotplug...[/yellow]"
)
mapping_by_index = {index: port for index, port in mappings}
return PairingState(
@ -857,7 +1057,9 @@ def prepare_pairing_state(
)
def assign_port_for_index(pairing: PairingState, idx: int, console: Console) -> Optional[str]:
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."""
if idx in pairing.mapping_by_index:
return pairing.mapping_by_index[idx]
@ -879,12 +1081,21 @@ def ports_in_use(pairing: PairingState, contexts: Dict[int, ControllerContext])
return used
def handle_removed_port(path: str, pairing: PairingState, contexts: Dict[int, ControllerContext], console: Console) -> None:
def handle_removed_port(
path: str,
pairing: PairingState,
contexts: Dict[int, ControllerContext],
console: Console,
) -> None:
"""Clear mappings/contexts for a UART path that disappeared."""
if path in pairing.available_ports:
pairing.available_ports.remove(path)
console.print(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]
console.print(
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
]
for idx in indices_to_clear:
pairing.mapping_by_index.pop(idx, None)
pairing.auto_assigned_indices.discard(idx)
@ -902,10 +1113,14 @@ def handle_removed_port(path: str, pairing: PairingState, contexts: Dict[int, Co
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]")
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) -> None:
def discover_new_ports(
pairing: PairingState, contexts: Dict[int, ControllerContext], console: Console
) -> None:
"""Scan for new serial ports and add unused ones to the available pool."""
if not pairing.auto_discover_ports:
return
@ -929,7 +1144,9 @@ def discover_new_ports(pairing: PairingState, contexts: Dict[int, ControllerCont
if path in in_use or path in pairing.available_ports:
continue
pairing.available_ports.append(path)
console.print(f"[green]Discovered UART {path} ({info['description']}); available for pairing.[/green]")
console.print(
f"[green]Discovered UART {path} ({info['description']}); available for pairing.[/green]"
)
def pair_waiting_contexts(
@ -977,7 +1194,9 @@ def open_initial_contexts(
if index >= sdl2.SDL_NumJoysticks() or not sdl2.SDL_IsGameController(index):
name = sdl2.SDL_JoystickNameForIndex(index)
name_str = name.decode() if isinstance(name, bytes) else str(name)
console.print(f"[yellow]Index {index} is not a GameController ({name_str}). Trying raw open failed.[/yellow]")
console.print(
f"[yellow]Index {index} is not a GameController ({name_str}). Trying raw open failed.[/yellow]"
)
continue
port = assign_port_for_index(pairing, index, console)
if port is None and not pairing.auto_pairing_enabled:
@ -993,9 +1212,13 @@ def open_initial_contexts(
uart = open_uart_or_warn(port, args.baud, console) if port else None
if uart:
uarts.append(uart)
console.print(f"[green]Controller {index} (id {stable_id}, inst {instance_id}) paired to {port}[/green]")
console.print(
f"[green]Controller {index} (id {stable_id}, inst {instance_id}) paired to {port}[/green]"
)
elif port:
console.print(f"[yellow]Controller {index} (id {stable_id}, inst {instance_id}) waiting for UART {port}[/yellow]")
console.print(
f"[yellow]Controller {index} (id {stable_id}, inst {instance_id}) waiting for UART {port}[/yellow]"
)
else:
console.print(
f"[yellow]Controller {index} (id {stable_id}, inst {instance_id}) connected; waiting for an available UART[/yellow]"
@ -1010,11 +1233,14 @@ def open_initial_contexts(
)
if config.zero_sticks:
zero_context_sticks(ctx, console)
initialize_controller_sensors(ctx, console)
contexts[instance_id] = ctx
return contexts, uarts
def handle_axis_motion(event: sdl2.SDL_Event, contexts: Dict[int, ControllerContext], config: BridgeConfig) -> None:
def handle_axis_motion(
event: sdl2.SDL_Event, contexts: Dict[int, ControllerContext], config: BridgeConfig
) -> None:
"""Process axis motion event into stick/trigger state."""
ctx = contexts.get(event.caxis.which)
if not ctx:
@ -1095,7 +1321,9 @@ def handle_device_added(
if idx >= sdl2.SDL_NumJoysticks() or not sdl2.SDL_IsGameController(idx):
name = sdl2.SDL_JoystickNameForIndex(idx)
name_str = name.decode() if isinstance(name, bytes) else str(name)
console.print(f"[yellow]Index {idx} is not a GameController ({name_str}). Trying raw open failed.[/yellow]")
console.print(
f"[yellow]Index {idx} is not a GameController ({name_str}). Trying raw open failed.[/yellow]"
)
return
try:
controller, instance_id, guid = open_controller(idx)
@ -1109,9 +1337,13 @@ def handle_device_added(
uart = open_uart_or_warn(port, args.baud, console) if port else None
if uart:
uarts.append(uart)
console.print(f"[green]Controller {idx} (id {stable_id}, inst {instance_id}) paired to {port}[/green]")
console.print(
f"[green]Controller {idx} (id {stable_id}, inst {instance_id}) paired to {port}[/green]"
)
elif port:
console.print(f"[yellow]Controller {idx} (id {stable_id}, inst {instance_id}) waiting for UART {port}[/yellow]")
console.print(
f"[yellow]Controller {idx} (id {stable_id}, inst {instance_id}) waiting for UART {port}[/yellow]"
)
else:
console.print(
f"[yellow]Controller {idx} (id {stable_id}, inst {instance_id}) connected; waiting for an available UART[/yellow]"
@ -1126,6 +1358,7 @@ def handle_device_added(
)
if config.zero_sticks:
zero_context_sticks(ctx, console)
initialize_controller_sensors(ctx, console)
contexts[instance_id] = ctx
@ -1140,7 +1373,9 @@ def handle_device_removed(
ctx = contexts.pop(instance_id, None)
if not ctx:
return
console.print(f"[yellow]Controller {instance_id} (id {ctx.stable_id}) removed[/yellow]")
console.print(
f"[yellow]Controller {instance_id} (id {ctx.stable_id}) removed[/yellow]"
)
if ctx.controller_index in pairing.auto_assigned_indices:
# Return auto-paired UART back to the pool so a future device can use it.
freed = pairing.mapping_by_index.pop(ctx.controller_index, None)
@ -1167,18 +1402,25 @@ def service_contexts(
else config.button_map_default
)
poll_controller_buttons(ctx, current_button_map)
collect_imu_sample(ctx)
# 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]")
console.print(
f"[green]Reconnected UART {ctx.port} for controller {ctx.controller_index}[/green]"
)
ctx.uart = uart
if ctx.uart is None:
continue
try:
if now - ctx.last_send >= config.interval:
if ctx.imu_samples:
ctx.report.imu_samples = ctx.imu_samples[-IMU_SAMPLES_PER_REPORT:]
else:
ctx.report.imu_samples = []
ctx.uart.send_report(ctx.report)
ctx.last_send = now
@ -1201,7 +1443,10 @@ def service_contexts(
sdl2.SDL_GameControllerRumble(ctx.controller, 0, 0, 0)
ctx.rumble_active = False
ctx.last_rumble_energy = 0.0
elif ctx.rumble_active and (now - ctx.last_rumble_change) > RUMBLE_STUCK_TIMEOUT:
elif (
ctx.rumble_active
and (now - ctx.last_rumble_change) > RUMBLE_STUCK_TIMEOUT
):
sdl2.SDL_GameControllerRumble(ctx.controller, 0, 0, 0)
ctx.rumble_active = False
ctx.last_rumble_energy = 0.0
@ -1242,10 +1487,15 @@ def run_bridge_loop(
break
if event.type == sdl2.SDL_CONTROLLERAXISMOTION:
handle_axis_motion(event, contexts, config)
elif event.type in (sdl2.SDL_CONTROLLERBUTTONDOWN, sdl2.SDL_CONTROLLERBUTTONUP):
elif event.type in (
sdl2.SDL_CONTROLLERBUTTONDOWN,
sdl2.SDL_CONTROLLERBUTTONUP,
):
handle_button_event(event, config, contexts)
elif event.type == sdl2.SDL_CONTROLLERDEVICEADDED:
handle_device_added(event, args, pairing, contexts, uarts, console, config)
handle_device_added(
event, args, pairing, contexts, uarts, console, config
)
elif event.type == sdl2.SDL_CONTROLLERDEVICEREMOVED:
handle_device_removed(event, pairing, contexts, console)
@ -1291,7 +1541,9 @@ def main() -> None:
list_controllers_with_guids(console, parser)
return
controller_indices, controller_names = detect_controllers(console, args, parser)
pairing = prepare_pairing_state(args, console, parser, controller_indices, controller_names)
pairing = prepare_pairing_state(
args, console, parser, controller_indices, controller_names
)
hotkey_messages: Dict[str, str] = {}
if config.zero_hotkey:
hotkey_messages[config.zero_hotkey] = "re-zero controller sticks"
@ -1301,14 +1553,20 @@ def main() -> None:
hotkey_messages[config.swap_hotkey] + "; toggle ABXY layout"
)
else:
hotkey_messages[config.swap_hotkey] = "toggle ABXY layout for a controller"
hotkey_messages[config.swap_hotkey] = (
"toggle ABXY layout for a controller"
)
if hotkey_messages:
candidate = HotkeyMonitor(console, hotkey_messages)
if candidate.start():
hotkey_monitor = candidate
contexts, uarts = open_initial_contexts(args, pairing, controller_indices, console, config)
contexts, uarts = open_initial_contexts(
args, pairing, controller_indices, console, config
)
if not contexts:
console.print("[yellow]No controllers opened; waiting for hotplug events...[/yellow]")
console.print(
"[yellow]No controllers opened; waiting for hotplug events...[/yellow]"
)
run_bridge_loop(args, console, config, pairing, contexts, uarts, hotkey_monitor)
finally:
if hotkey_monitor:

19
switch-pico.cpp
View file

@ -62,8 +62,9 @@ static void on_rumble_from_switch(const uint8_t rumble[8]) {
// Consume UART bytes and forward complete frames to the Switch Pro driver.
static void poll_uart_frames() {
static uint8_t buffer[8];
static uint8_t buffer[64];
static uint8_t index = 0;
static uint8_t expected_len = 0;
static absolute_time_t last_byte_time = {0};
static bool has_last_byte = false;
@ -73,6 +74,7 @@ static void poll_uart_frames() {
uint64_t now = to_ms_since_boot(get_absolute_time());
if (has_last_byte && (now - to_ms_since_boot(last_byte_time)) > 20) {
index = 0; // stale data, restart frame
expected_len = 0;
}
last_byte_time = get_absolute_time();
has_last_byte = true;
@ -84,9 +86,19 @@ static void poll_uart_frames() {
}
buffer[index++] = byte;
if (index >= sizeof(buffer)) {
if (index == 3) {
// We just stored payload_len; compute expected frame length (payload + header/version/len/checksum).
expected_len = static_cast<uint8_t>(buffer[2] + 4);
if (expected_len > sizeof(buffer) || expected_len < 8) {
index = 0;
expected_len = 0;
continue;
}
}
if (expected_len && index >= expected_len) {
SwitchInputState parsed{};
if (switch_pro_apply_uart_packet(buffer, sizeof(buffer), &parsed)) {
if (switch_pro_apply_uart_packet(buffer, expected_len, &parsed)) {
g_user_state = parsed;
LOG_PRINTF("[UART] packet buttons=0x%04x hat=%u lx=%u ly=%u rx=%u ry=%u\n",
(parsed.button_a ? SWITCH_PRO_MASK_A : 0) |
@ -110,6 +122,7 @@ static void poll_uart_frames() {
parsed.lx >> 8, parsed.ly >> 8, parsed.rx >> 8, parsed.ry >> 8);
}
index = 0;
expected_len = 0;
}
}
}

96
switch_pro_driver.cpp
View file

@ -190,9 +190,40 @@ static SwitchInputState make_neutral_state() {
s.ly = SWITCH_PRO_JOYSTICK_MID;
s.rx = SWITCH_PRO_JOYSTICK_MID;
s.ry = SWITCH_PRO_JOYSTICK_MID;
s.imu_sample_count = 0;
return s;
}
static void fill_imu_report_data(const SwitchInputState& state) {
// Only include IMU data when the host explicitly enabled it.
if (!is_imu_enabled || state.imu_sample_count == 0) {
memset(switch_report.imuData, 0x00, sizeof(switch_report.imuData));
return;
}
uint8_t sample_count = state.imu_sample_count > 3 ? 3 : state.imu_sample_count;
uint8_t* dst = switch_report.imuData;
for (uint8_t i = 0; i < 3; ++i) {
SwitchImuSample sample{};
if (i < sample_count) {
sample = state.imu_samples[i];
}
dst[0] = static_cast<uint8_t>(sample.accel_x & 0xFF);
dst[1] = static_cast<uint8_t>((sample.accel_x >> 8) & 0xFF);
dst[2] = static_cast<uint8_t>(sample.accel_y & 0xFF);
dst[3] = static_cast<uint8_t>((sample.accel_y >> 8) & 0xFF);
dst[4] = static_cast<uint8_t>(sample.accel_z & 0xFF);
dst[5] = static_cast<uint8_t>((sample.accel_z >> 8) & 0xFF);
dst[6] = static_cast<uint8_t>(sample.gyro_x & 0xFF);
dst[7] = static_cast<uint8_t>((sample.gyro_x >> 8) & 0xFF);
dst[8] = static_cast<uint8_t>(sample.gyro_y & 0xFF);
dst[9] = static_cast<uint8_t>((sample.gyro_y >> 8) & 0xFF);
dst[10] = static_cast<uint8_t>(sample.gyro_z & 0xFF);
dst[11] = static_cast<uint8_t>((sample.gyro_z >> 8) & 0xFF);
dst += 12;
}
}
static void send_identify() {
memset(report_buffer, 0x00, sizeof(report_buffer));
report_buffer[0] = REPORT_USB_INPUT_81;
@ -485,6 +516,7 @@ static void update_switch_report_from_state() {
switch_report.inputs.rightStick.setX(std::min(std::max(scaleRightStickX,rightMinX), rightMaxX));
switch_report.inputs.rightStick.setY(-std::min(std::max(scaleRightStickY,rightMinY), rightMaxY));
fill_imu_report_data(g_input_state);
switch_report.rumbleReport = 0x09;
}
@ -617,18 +649,51 @@ void switch_pro_task() {
}
bool switch_pro_apply_uart_packet(const uint8_t* packet, uint8_t length, SwitchInputState* out_state) {
// Packet format: 0xAA, buttons(2 LE), hat, lx, ly, rx, ry
// Packet v2 format:
// 0:0xAA header
// 1:version (0x02)
// 2:payload_len (bytes 3..3+len-1)
// 3-4: buttons LE
// 5: hat
// 6-9: lx, ly, rx, ry (0-255)
// 10: imu_sample_count (0-3)
// 11-46: up to 3 samples of accel/gyro (int16 LE each axis)
// 47: checksum (sum of bytes 0..46) & 0xFF
if (length < 8 || packet[0] != 0xAA) {
return false;
}
if (packet[1] != 0x02) {
return false;
}
uint8_t payload_len = packet[2];
uint16_t expected_len = static_cast<uint16_t>(payload_len) + 4; // header+version+len+checksum
if (length < expected_len) {
return false;
}
uint16_t checksum_end = static_cast<uint16_t>(3 + payload_len - 1); // last payload byte
uint16_t checksum_index = static_cast<uint16_t>(3 + payload_len);
if (checksum_index >= length) {
return false;
}
uint16_t sum = 0;
for (uint16_t i = 0; i <= checksum_end; ++i) {
sum = static_cast<uint16_t>(sum + packet[i]);
}
if ((sum & 0xFF) != packet[checksum_index]) {
return false;
}
SwitchProOutReport out{};
out.buttons = static_cast<uint16_t>(packet[1]) | (static_cast<uint16_t>(packet[2]) << 8);
out.hat = packet[3];
out.lx = packet[4];
out.ly = packet[5];
out.rx = packet[6];
out.ry = packet[7];
out.buttons = static_cast<uint16_t>(packet[3]) | (static_cast<uint16_t>(packet[4]) << 8);
out.hat = packet[5];
out.lx = packet[6];
out.ly = packet[7];
out.rx = packet[8];
out.ry = packet[9];
auto expand_axis = [](uint8_t v) -> uint16_t {
return static_cast<uint16_t>(v) << 8 | v;
@ -636,6 +701,23 @@ bool switch_pro_apply_uart_packet(const uint8_t* packet, uint8_t length, SwitchI
SwitchInputState state = make_neutral_state();
state.imu_sample_count = std::min<uint8_t>(packet[10], 3);
auto read_int16 = [](const uint8_t* src) -> int16_t {
return static_cast<int16_t>(static_cast<uint16_t>(src[0]) | (static_cast<uint16_t>(src[1]) << 8));
};
const uint8_t* imu_base = &packet[11];
for (uint8_t i = 0; i < state.imu_sample_count; ++i) {
const uint8_t* sample_ptr = imu_base + (i * 12);
state.imu_samples[i].accel_x = read_int16(sample_ptr + 0);
state.imu_samples[i].accel_y = read_int16(sample_ptr + 2);
state.imu_samples[i].accel_z = read_int16(sample_ptr + 4);
state.imu_samples[i].gyro_x = read_int16(sample_ptr + 6);
state.imu_samples[i].gyro_y = read_int16(sample_ptr + 8);
state.imu_samples[i].gyro_z = read_int16(sample_ptr + 10);
}
switch (out.hat) {
case SWITCH_PRO_HAT_UP: state.dpad_up = true; break;
case SWITCH_PRO_HAT_UPRIGHT: state.dpad_up = true; state.dpad_right = true; break;

11
switch_pro_driver.h
View file

@ -10,6 +10,15 @@
#include <stdint.h>
#include "switch_pro_descriptors.h"
typedef struct {
int16_t accel_x;
int16_t accel_y;
int16_t accel_z;
int16_t gyro_x;
int16_t gyro_y;
int16_t gyro_z;
} SwitchImuSample;
typedef struct {
bool dpad_up;
bool dpad_down;
@ -35,6 +44,8 @@ typedef struct {
uint16_t ly;
uint16_t rx;
uint16_t ry;
uint8_t imu_sample_count;
SwitchImuSample imu_samples[3];
} SwitchInputState;
// Initialize USB state and calibration before entering the main loop.