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

Add selectable Wii IR aiming and DolphinBar keepalive

Browse source
This commit is contained in:
Joey Yakimowich-Payne 2026-09-09 17:13:27 -06:00
commit 5ecfca105a
27 changed files with 4079 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

792
tools/dolphinbar_keepalive.py Executable file
View file

@ -0,0 +1,792 @@
#!/usr/bin/env python3
"""Temporarily emulate a neutral Wii Remote for a DolphinBar on Linux.
Run with sudo and press SYNC on the DolphinBar after the READY message.
Runs until Ctrl+C by default; --seconds N optionally sets a runtime limit.
No MAC spoofing, firmware changes, or persistent Bluetooth pairing keys.
The normal Bluetooth daemon is temporarily runtime-masked; settings and service
are restored on exit. --restore recovers an interrupted run from its journal.
"""
from __future__ import annotations
import argparse
import fcntl
import json
import os
import selectors
import signal
import socket
import struct
import subprocess
import sys
import time
from collections import deque
from contextlib import ExitStack, contextmanager
from pathlib import Path
from typing import ClassVar, TypedDict, cast
# Keep this executable directly as well as through python -m tools....
if __package__ in (None, ""):
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
AdapterInfo = TypedDict(
"AdapterInfo",
{
"address": str,
"version": int,
"manufacturer": int,
"supported": int,
"settings": int,
"class": str,
"name": str,
},
)
class RecoveryState(TypedDict):
schema: int
index: int
info: AdapterInfo
service_active: bool
hci: dict[str, str]
POWERED = 1 << 0
CONNECTABLE = 1 << 1
DISCOVERABLE = 1 << 3
BONDABLE = 1 << 4
SSP = 1 << 6
BREDR = 1 << 7
ADVERTISING = 1 << 10
SECURE_CONN = 1 << 11
stop_requested = False
def log(event: str, **fields) -> None:
print(
json.dumps({"time": round(time.monotonic(), 3), "event": event, **fields}),
flush=True,
)
def run(*args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
return subprocess.run(args, check=check, text=True, capture_output=True, timeout=20)
def address(raw: bytes) -> str:
return ":".join(f"{b:02X}" for b in raw[::-1])
class Management:
def __init__(self, index: int):
self.index = index
self.socket = socket.socket(
socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI
)
self.socket.bind((0xFFFF, 3)) # HCI_CHANNEL_CONTROL
self.socket.settimeout(5)
def close(self):
self.socket.close()
def request(self, opcode: int, payload: bytes = b"") -> bytes:
self.socket.send(
struct.pack("<HHH", opcode, self.index, len(payload)) + payload
)
deadline = time.monotonic() + 5
while time.monotonic() < deadline:
packet = self.socket.recv(4096)
if len(packet) < 9:
continue
event, index, size = struct.unpack_from("<HHH", packet)
if index != self.index or size + 6 != len(packet) or event not in (1, 2):
continue
command, status = struct.unpack_from("<HB", packet, 6)
if command != opcode:
continue
if status:
raise RuntimeError(
f"MGMT command 0x{opcode:04x}: status 0x{status:02x}"
)
return packet[9:]
raise TimeoutError(f"MGMT command 0x{opcode:04x} timed out")
def info(self) -> AdapterInfo:
data = self.request(4)
if len(data) != 280:
raise RuntimeError(f"Unexpected MGMT ReadInfo size {len(data)}")
return {
"address": address(data[:6]),
"version": data[6],
"manufacturer": struct.unpack_from("<H", data, 7)[0],
"supported": struct.unpack_from("<I", data, 9)[0],
"settings": struct.unpack_from("<I", data, 13)[0],
"class": data[17:20].hex(),
"name": data[20:269].split(b"\0", 1)[0].decode(errors="replace"),
}
def connections(self) -> int:
data = self.request(0x15)
if len(data) < 2:
raise RuntimeError("Short MGMT connection list")
count = struct.unpack_from("<H", data)[0]
if len(data) != 2 + count * 7:
raise RuntimeError("Malformed MGMT connection list")
return count
def setting(self, opcode: int, enabled: bool) -> None:
self.request(opcode, bytes([int(enabled)]))
def discoverable(self, mode: int, timeout: int = 0) -> None:
self.request(6, struct.pack("<BH", mode, timeout))
class Hci:
# read opcode -> (write opcode, expected payload length; -1 is variable)
SETTINGS: ClassVar[dict[int, tuple[int, int]]] = {
0x0C14: (0x0C13, 248),
0x0C23: (0x0C24, 3),
0x0C19: (0x0C1A, 1),
0x0C39: (0x0C3A, -1),
0x0C51: (0x0C52, 241),
0x0C09: (0x0C0A, 1),
}
def __init__(self, index: int):
self.socket = socket.socket(
socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI
)
self.socket.bind((index,))
self.socket.setsockopt(
0, 2, struct.pack("<IIIH2x", 1 << 4, 0xFFFFFFFF, 0xFFFFFFFF, 0)
)
self.socket.settimeout(4)
self.pending_events: deque[bytes] = deque()
def close(self):
self.socket.close()
def send(self, opcode: int, payload: bytes = b"") -> None:
if len(payload) > 255:
raise ValueError("HCI command too large")
self.socket.send(
bytes([1]) + struct.pack("<HB", opcode, len(payload)) + payload
)
def command(self, opcode: int, payload: bytes = b"") -> bytes:
self.send(opcode, payload)
deadline = time.monotonic() + 4
while time.monotonic() < deadline:
packet = self.socket.recv(1024)
if len(packet) < 7 or packet[0] != 4 or len(packet) != packet[2] + 3:
continue
if packet[1] == 0x0E and struct.unpack_from("<H", packet, 4)[0] == opcode:
if packet[6]:
raise RuntimeError(
f"HCI command 0x{opcode:04x}: status 0x{packet[6]:02x}"
)
return packet[7:]
if (
packet[1] == 0x0F
and struct.unpack_from("<H", packet, 5)[0] == opcode
and packet[3]
):
raise RuntimeError(
f"HCI command 0x{opcode:04x}: status 0x{packet[3]:02x}"
)
if packet[1] in (3, 4, 5, 6, 0x16, 0x18):
self.pending_events.append(packet)
raise TimeoutError(f"HCI command 0x{opcode:04x} timed out")
def snapshot(self) -> dict[str, str]:
result = {}
for read_opcode, (write_opcode, size) in self.SETTINGS.items():
data = self.command(read_opcode)
if (size >= 0 and len(data) != size) or (
size < 0 and (not data or len(data) != 1 + data[0] * 3)
):
raise RuntimeError(
f"Unexpected HCI setting length for 0x{read_opcode:04x}"
)
result[str(write_opcode)] = data.hex()
return result
class AdapterSession:
def __init__(self, index: int, journal: Path):
self.index = index
self.journal = journal
self.management = Management(index)
self.hci = Hci(index)
self.state: RecoveryState | None = None
self.lock_resources = ExitStack()
def acquire_lock(self) -> None:
@contextmanager
def held_lock():
with open(
f"/run/lock/switch-pico-dolphinbar-hci{self.index}.lock", "a"
) as lock:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
yield
self.lock_resources.enter_context(held_lock())
def prepare(self) -> None:
self.acquire_lock()
if self.journal.exists():
raise RuntimeError(
f"Recovery journal exists. Run --restore {self.journal} first"
)
info = self.management.info()
if not info["settings"] & POWERED or not info["settings"] & BREDR:
raise RuntimeError(
"Enable the adapter and Bluetooth Classic before running this tool"
)
if self.management.connections():
raise RuntimeError(
"Refusing to interrupt connected laptop Bluetooth devices"
)
if info["settings"] & (ADVERTISING | DISCOVERABLE):
raise RuntimeError(
"Stop existing laptop Bluetooth advertising/discovery before this test"
)
active = (
run("systemctl", "is-active", "bluetooth.service", check=False).returncode
== 0
)
masked = run(
"systemctl", "is-enabled", "bluetooth.service", check=False
).stdout.strip()
if masked not in ("enabled", "disabled", "static", "indirect", "alias"):
raise RuntimeError(
f"Unexpected Bluetooth service state {masked!r}; not changing it"
)
self.state = {
"schema": 1,
"index": self.index,
"info": info,
"service_active": active,
"hci": self.hci.snapshot(),
}
self.journal.parent.mkdir(parents=True, exist_ok=True)
fd = os.open(self.journal, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, "w") as stream:
json.dump(self.state, stream, indent=2)
stream.flush()
os.fsync(stream.fileno())
log(
"adapter_saved",
adapter=f"hci{self.index}",
address=info["address"],
journal=str(self.journal),
)
run("systemctl", "mask", "--runtime", "bluetooth.service")
run("systemctl", "stop", "bluetooth.service")
self.management.setting(5, False)
if info["supported"] & SECURE_CONN:
self.management.setting(0x2D, False)
self.management.setting(0x0B, False) # Classic legacy PIN, not SSP.
self.management.setting(9, True)
self.management.setting(7, True)
self.management.setting(5, True)
# Discovery starts only after all L2CAP listeners are bound.
self.management.discoverable(0)
self.hci.command(0x0C13, b"Nintendo RVL-CNT-01".ljust(248, b"\0"))
self.hci.command(0x0C0A, b"\0") # Variable PIN.
def advertise(self, seconds: int | None) -> None:
assert self.state is not None
# Discovery is bounded independently of an established connection.
self.management.discoverable(
2, min(seconds, 180) if seconds is not None else 180
)
self.hci.command(0x0C24, bytes.fromhex("042500"))
supported_iac = self.hci.command(0x0C38)
laps = (
bytes.fromhex("008b9e338b9e")
if supported_iac and supported_iac[0] >= 2
else bytes.fromhex("008b9e")
)
self.hci.command(0x0C3A, bytes([len(laps) // 3]) + laps)
name = b"Nintendo RVL-CNT-01"
eir = bytes([len(name) + 1, 9]) + name + bytes.fromhex("050324110012")
self.hci.command(0x0C52, b"\0" + eir.ljust(240, b"\0"))
self.hci.command(0x0C1A, b"\x03")
log(
"identity_ready",
address=self.state["info"]["address"],
name=name.decode(),
device_class="002504",
address_spoofed=False,
)
def restore(self) -> None:
if self.state is None:
return
# Recovery can be invoked after a partial cleanup restarted BlueZ.
# Prevent it from racing these controller commands in either path.
run("systemctl", "mask", "--runtime", "bluetooth.service")
run("systemctl", "stop", "bluetooth.service")
self.hci.socket.settimeout(4)
errors = []
settings = self.state["info"]["settings"]
def attempt(label, fn):
try:
fn()
except (
OSError,
RuntimeError,
ValueError,
subprocess.SubprocessError,
) as exc:
errors.append(f"{label}: {exc}")
def restore_discoverable():
wanted = bool(settings & DISCOVERABLE)
current = bool(self.management.info()["settings"] & DISCOVERABLE)
# Clearing connectable already clears discoverable. The kernel
# rejects a redundant discoverable request when not connectable.
if current != wanted:
self.management.discoverable(int(wanted))
attempt("power down", lambda: self.management.setting(5, False))
attempt("SSP", lambda: self.management.setting(0x0B, bool(settings & SSP)))
if self.state["info"]["supported"] & SECURE_CONN:
attempt(
"secure connections",
lambda: self.management.setting(0x2D, bool(settings & SECURE_CONN)),
)
attempt(
"bondable", lambda: self.management.setting(9, bool(settings & BONDABLE))
)
attempt(
"connectable",
lambda: self.management.setting(7, bool(settings & CONNECTABLE)),
)
attempt("power up", lambda: self.management.setting(5, True))
attempt("discoverable", restore_discoverable)
for opcode, value in self.state["hci"].items():
attempt(
f"HCI {opcode}",
lambda opcode=opcode, value=value: self.hci.command(
int(opcode), bytes.fromhex(value)
),
)
attempt(
"original power",
lambda: self.management.setting(5, bool(settings & POWERED)),
)
attempt(
"unmask service",
lambda: run("systemctl", "unmask", "--runtime", "bluetooth.service"),
)
if self.state["service_active"]:
attempt(
"restart service",
lambda: run("systemctl", "start", "bluetooth.service"),
)
if errors:
log(
"restore_failed",
errors=errors,
recovery=f"sudo {sys.executable} {__file__} --restore {self.journal}",
)
raise RuntimeError("Restoration incomplete; recovery journal retained")
self.journal.unlink(missing_ok=True)
log(
"adapter_restored",
adapter=f"hci{self.index}",
service_active=self.state["service_active"],
)
self.state = None
def close(self):
self.hci.close()
self.management.close()
self.lock_resources.close()
class Peripheral:
def __init__(self, session: AdapterSession, pin_mode: str, peer: str | None):
from tools.dolphinbar_sdp import SdpResponder
from tools.dolphinbar_wiimote import HID_DESCRIPTOR, Wiimote
self.session = session
self.pin_mode = pin_mode
self.peer = peer.upper() if peer else None
self.sdp_factory = lambda: SdpResponder(HID_DESCRIPTOR)
self.sdp_sessions: dict[socket.socket, SdpResponder] = {}
self.peer_mtus: dict[socket.socket, int] = {}
self.wiimote = Wiimote()
self.selector = selectors.DefaultSelector()
self.listeners: list[socket.socket] = []
self.clients: dict[socket.socket, int] = {}
self.pending: dict[socket.socket, deque[bytes]] = {}
self.waiting_interrupt: deque[bytes] = deque()
self.interrupt: socket.socket | None = None
self.commands = 0
self.sent = 0
self.pins = 0
self.authenticated: bool | None = None
self.generated_link_key = False
self.last_ir = False
assert session.state is not None
local = session.state["info"]["address"]
self.local_address = local
try:
for psm in (1, 0x11, 0x13):
listener = socket.socket(
socket.AF_BLUETOOTH, socket.SOCK_SEQPACKET, socket.BTPROTO_L2CAP
)
self.listeners.append(listener)
listener.setsockopt(274, 4, bytes([1, 0])) # BT_SECURITY_LOW
listener.bind((local, psm))
listener.listen(1)
listener.setblocking(False)
self.selector.register(listener, selectors.EVENT_READ, ("listen", psm))
session.hci.socket.setblocking(False)
self.selector.register(session.hci.socket, selectors.EVENT_READ, ("hci", 0))
except BaseException:
self.close()
raise
def close_client(self, client):
psm = self.clients.pop(client, None)
self.pending.pop(client, None)
self.sdp_sessions.pop(client, None)
self.peer_mtus.pop(client, None)
try:
self.selector.unregister(client)
except (KeyError, ValueError):
pass
if client is self.interrupt:
self.interrupt = None
client.close()
if psm is not None:
log("channel_closed", psm=hex(psm))
def close(self):
for client in list(self.clients):
self.close_client(client)
for listener in self.listeners:
listener.close()
self.selector.close()
def queue(self, client, packet: bytes):
if client is None:
if len(self.waiting_interrupt) >= 512:
raise RuntimeError("Peer exhausted pre-interrupt response queue")
self.waiting_interrupt.append(packet)
return
queue = self.pending[client]
if len(queue) >= 512:
raise RuntimeError("Peer exhausted bounded response queue")
queue.append(packet)
self.selector.modify(
client,
selectors.EVENT_READ | selectors.EVENT_WRITE,
("client", self.clients[client]),
)
def hci_event(self, packet: bytes):
if len(packet) < 3 or packet[0] != 4 or len(packet) != packet[2] + 3:
return
event, data = packet[1], packet[3:]
if event == 0x16 and len(data) == 6: # PIN_CODE_REQUEST
remote = address(data)
if self.peer and remote != self.peer:
self.session.management.request(0x0017, data + b"\0")
log("pin_rejected_other_peer", peer=remote)
return
self.peer = remote
pin = (
data
if self.pin_mode == "host"
else bytes.fromhex(self.local_address.replace(":", ""))[::-1]
)
# MGMT accepts binary PINs and updates the kernel's pairing state.
self.session.management.request(0x0016, data + b"\0\x06" + pin + bytes(10))
self.pins += 1
log("legacy_pin_replied", peer=remote, mode=self.pin_mode)
elif event == 3 and len(data) >= 11:
log(
"acl_connection",
status=data[0],
peer=address(data[3:9]),
encryption=data[10],
)
elif event == 6 and len(data) == 3:
self.authenticated = data[0] == 0
log(
"authentication",
status=data[0],
handle=struct.unpack_from("<H", data, 1)[0],
)
elif event == 5 and len(data) == 4:
log("acl_disconnected", reason=data[3])
elif event == 0x18 and len(data) == 23:
self.generated_link_key = True
log("link_key_generated", peer=address(data[:6]), persisted=False)
elif event == 4 and len(data) == 10:
log(
"connection_requested",
peer=address(data[:6]),
device_class=data[6:9].hex(),
)
def receive(self, client, psm: int, packet: bytes):
if psm == 1:
reply = self.sdp_sessions[client].reply(packet, self.peer_mtus[client])
self.queue(client, reply)
log("sdp", request=packet.hex(), response_bytes=len(reply))
return
if not packet:
return
prefix = packet[0]
if prefix in (0xA2, 0x52):
log("wiimote_command", psm=hex(psm), report=packet[1:].hex())
replies = self.wiimote.handle_output(packet[1:])
self.commands += 1
if prefix == 0x52 and psm == 0x11:
self.queue(client, b"\0")
for reply in replies:
self.queue(self.interrupt, reply)
if self.wiimote.ir_enabled != self.last_ir:
self.last_ir = self.wiimote.ir_enabled
log(
"emulated_camera",
enabled=self.last_ir,
mode=hex(self.wiimote.report_mode),
)
elif psm == 0x11:
if prefix == 0x15: # Virtual cable unplug.
self.close_client(client)
elif prefix == 0x60:
self.queue(client, b"\xa0\x01")
elif prefix == 0x71 or prefix & 0xF0 == 0x90:
self.queue(client, b"\0")
elif prefix == 0x80:
self.queue(client, b"\xa0\0")
elif prefix == 0x41 and len(packet) >= 2 and packet[1] == 0x20:
self.queue(client, self.wiimote.status_report())
else:
self.queue(client, b"\x03")
log("unsupported_hid_control", packet=packet.hex())
def serve(self, seconds: int | None):
self.session.hci.socket.settimeout(4)
self.session.advertise(seconds)
self.session.hci.socket.setblocking(False)
while self.session.hci.pending_events:
self.hci_event(self.session.hci.pending_events.popleft())
log(
"READY",
message="Listening; press only the DolphinBar SYNC button now",
seconds=seconds,
lifetime="until Ctrl+C" if seconds is None else f"{seconds} seconds",
pin_mode=self.pin_mode,
)
start = time.monotonic()
next_report = start
next_status = start + 10
while not stop_requested and (
seconds is None or time.monotonic() - start < seconds
):
for key, events in self.selector.select(0.01):
category, psm = key.data
channel_socket = cast(socket.socket, key.fileobj)
if category == "hci":
try:
self.hci_event(channel_socket.recv(1024))
except BlockingIOError:
pass
elif category == "listen":
client, remote = channel_socket.accept()
remote_address = remote[0].upper()
if self.peer and self.peer != remote_address:
client.close()
log("rejected_other_peer", peer=remote_address)
continue
if psm != 1:
self.peer = remote_address
if any(value == psm for value in self.clients.values()):
client.close()
continue
client.setblocking(False)
self.clients[client] = psm
self.pending[client] = deque()
options = client.getsockopt(6, 1, 12) # L2CAP_OPTIONS
self.peer_mtus[client] = struct.unpack_from("<H", options)[0]
if psm == 1:
self.sdp_sessions[client] = self.sdp_factory()
self.selector.register(
client, selectors.EVENT_READ, ("client", psm)
)
log("channel_connected", psm=hex(psm), peer=remote_address)
if psm == 0x13:
self.interrupt = client
while self.waiting_interrupt:
self.queue(client, self.waiting_interrupt.popleft())
self.queue(client, self.wiimote.status_report())
else:
client = channel_socket
try:
if events & selectors.EVENT_READ:
packet = client.recv(4096)
if not packet:
self.close_client(client)
continue
try:
self.receive(client, psm, packet)
except ValueError as exc:
log(
"malformed_peer_request",
psm=hex(psm),
error=str(exc),
)
self.close_client(client)
continue
if client in self.clients and events & selectors.EVENT_WRITE:
queue = self.pending[client]
if queue:
packet = queue[0]
if client.send(packet) != len(packet):
raise RuntimeError(
"Partial L2CAP sequenced-packet write"
)
queue.popleft()
self.sent += 1
if not queue:
self.selector.modify(
client, selectors.EVENT_READ, ("client", psm)
)
except BlockingIOError:
pass
except OSError as exc:
log("channel_error", psm=hex(psm), error=str(exc))
self.close_client(client)
now = time.monotonic()
if now >= next_report:
next_report = now + 0.01
if self.interrupt and not self.pending[self.interrupt]:
report = self.wiimote.periodic_report()
if report:
self.queue(self.interrupt, report)
if now >= next_status:
next_status = now + 10
log(
"status",
elapsed=round(now - start, 1),
peer=self.peer,
commands=self.commands,
sent=self.sent,
channels=sorted(self.clients.values()),
ir_enabled=self.wiimote.ir_enabled,
)
log(
"finished",
peer=self.peer,
commands=self.commands,
sent=self.sent,
pin_requests=self.pins,
authenticated=self.authenticated,
link_key_generated=self.generated_link_key,
ir_enabled=self.wiimote.ir_enabled,
)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--adapter", default="hci0")
parser.add_argument(
"--seconds",
type=int,
help="Stop after N seconds; omitted means run until Ctrl+C",
)
parser.add_argument("--pin-mode", choices=("host", "device"), default="host")
parser.add_argument(
"--peer", help="Restrict the experiment to one Bluetooth address"
)
parser.add_argument("--state-file", type=Path)
parser.add_argument(
"--inspect",
action="store_true",
help="Read adapter information without changing it",
)
parser.add_argument(
"--restore",
type=Path,
help="Restore settings from an interrupted run's journal",
)
args = parser.parse_args()
if not sys.platform.startswith("linux") or os.geteuid() != 0:
parser.error("Linux and root/CAP_NET_ADMIN access are required; run with sudo")
if not args.adapter.startswith("hci") or not args.adapter[3:].isdigit():
parser.error("--adapter must be hciN")
index = int(args.adapter[3:])
if index > 65534 or (args.seconds is not None and args.seconds < 1):
parser.error("Invalid adapter index or duration (must be positive)")
if args.peer:
try:
if len(bytes.fromhex(args.peer.replace(":", ""))) != 6:
raise ValueError
except ValueError:
parser.error("--peer must be a Bluetooth address")
journal = args.state_file or Path("build-dolphinbar") / f"hci{index}-state.json"
state: RecoveryState | None = None
if args.restore:
journal = args.restore
state = cast(RecoveryState, json.loads(journal.read_text()))
if state.get("schema") != 1:
parser.error("Unsupported recovery journal")
index = state["index"]
if type(index) is not int or not 0 <= index <= 65534:
parser.error("Invalid adapter index in recovery journal")
session = AdapterSession(index, journal)
peripheral = None
try:
if args.restore:
assert state is not None
session.acquire_lock()
if session.management.info()["address"] != state["info"]["address"]:
raise RuntimeError(
"Recovery journal belongs to a different Bluetooth adapter"
)
session.state = state
session.restore()
elif args.inspect:
log(
"inspection",
**session.management.info(),
connections=session.management.connections(),
hci=session.hci.snapshot(),
)
else:
def stop(_signal, _frame):
global stop_requested
stop_requested = True
for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
signal.signal(sig, stop)
session.prepare()
peripheral = Peripheral(session, args.pin_mode, args.peer)
peripheral.serve(args.seconds)
return 0
except (OSError, RuntimeError, ValueError, subprocess.SubprocessError) as exc:
log("error", error=str(exc))
return 1
finally:
if peripheral:
peripheral.close()
try:
session.restore()
finally:
session.close()
if __name__ == "__main__":
raise SystemExit(main())

382
tools/dolphinbar_sdp.py Normal file
View file

@ -0,0 +1,382 @@
"""Small, in-memory SDP server for an original Wii Remote identity.
Wire-format reference: Bluetooth Core, Vol 3, Part B. HID attribute values
are factual values from the original remote record documented in
https://github.com/rnconrad/WiimoteEmulator/blob/master/sdp.c and the identity
and channel assignments at https://wiibrew.org/wiki/Wiimote . No upstream
response arrays or implementation are used here.
"""
from __future__ import annotations
import struct
from collections import OrderedDict
from typing import NamedTuple
_MAX_REQUEST = 4096
_MAX_DESCRIPTOR = 4096
_MAX_ELEMENTS = 512
_MAX_DEPTH = 8
_MAX_CONTINUATIONS = 64
_TOKEN_BYTES = 8
_UUID_BASE_SUFFIX = bytes.fromhex("00001000800000805f9b34fb")
_VALID_SIZES = (
(0,), # Nil
(0, 1, 2, 3, 4), # Unsigned integer
(0, 1, 2, 3, 4), # Signed integer
(1, 2, 4), # UUID
(5, 6, 7), # Text string
(0,), # Boolean
(5, 6, 7), # Sequence
(5, 6, 7), # Alternative
(5, 6, 7), # URL
)
class _SdpError(Exception):
def __init__(self, code: int):
self.code = code
super().__init__(code)
def _encode(kind: int, payload: bytes) -> bytes:
"""Encode a data element, choosing a legal shortest length header."""
size = len(payload)
if kind == 0:
if size:
raise ValueError("Nil data elements have no payload")
return b"\x00"
if kind in (1, 2, 3, 5):
sizes = {1: 0, 2: 1, 4: 2, 8: 3, 16: 4}
index = sizes.get(size, -1)
if index not in _VALID_SIZES[kind]:
raise ValueError("Invalid scalar data element size")
return bytes([(kind << 3) | index]) + payload
if kind not in (4, 6, 7, 8):
raise ValueError("Invalid data element type")
for width, index in ((1, 5), (2, 6), (4, 7)):
if size < 1 << (8 * width):
return bytes([(kind << 3) | index]) + size.to_bytes(width, "big") + payload
raise ValueError("Data element is too large")
def _uint(value: int, width: int = 2) -> bytes:
return _encode(1, value.to_bytes(width, "big"))
def _uuid(value: int) -> bytes:
return _encode(3, value.to_bytes(2, "big"))
def _boolean(value: bool) -> bytes:
return _encode(5, bytes([int(value)]))
def _text(value: str) -> bytes:
return _encode(4, value.encode("utf-8"))
def _sequence(*elements: bytes) -> bytes:
return _encode(6, b"".join(elements))
class _Element(NamedTuple):
kind: int
size: int
value: int | bytes | tuple[_Element, ...] | None
class _Reader:
def __init__(self, data: bytes):
self.data = data
self.position = 0
self.elements = 0
def integer(self, width: int, end: int | None = None) -> int:
if end is None:
end = len(self.data)
if self.position + width > end:
raise _SdpError(0x0003)
value = int.from_bytes(self.data[self.position : self.position + width], "big")
self.position += width
return value
def element(self, end: int | None = None, depth: int = 0) -> _Element:
if end is None:
end = len(self.data)
self.elements += 1
if depth > _MAX_DEPTH or self.elements > _MAX_ELEMENTS:
raise _SdpError(0x0006)
header = self.integer(1, end)
kind, index = header >> 3, header & 7
if kind >= len(_VALID_SIZES) or index not in _VALID_SIZES[kind]:
raise _SdpError(0x0003)
if kind == 0:
return _Element(kind, 0, None)
size = (1 << index) if index < 5 else self.integer(1 << (index - 5), end)
stop = self.position + size
if stop > end:
raise _SdpError(0x0003)
if kind in (6, 7):
children = []
while self.position < stop:
children.append(self.element(stop, depth + 1))
return _Element(kind, size, tuple(children))
raw = self.data[self.position : stop]
self.position = stop
if kind in (1, 2, 5):
value = int.from_bytes(raw, "big", signed=kind == 2)
if kind == 5:
value = int(bool(value))
return _Element(kind, size, value)
return _Element(kind, size, raw)
def search_pattern(self) -> tuple[bytes, ...]:
element = self.element()
if element.kind != 6 or not 1 <= len(element.value) <= 12:
raise _SdpError(0x0003)
uuids = []
for child in element.value:
if child.kind != 3:
raise _SdpError(0x0003)
uuid = child.value
if child.size != 16:
uuid = uuid.rjust(4, b"\x00") + _UUID_BASE_SUFFIX
uuids.append(uuid)
return tuple(uuids)
def attributes(self) -> tuple[tuple[int, int], ...]:
element = self.element()
if element.kind != 6:
raise _SdpError(0x0003)
ranges = []
previous = -1
for child in element.value:
if child.kind != 1 or child.size not in (2, 4):
raise _SdpError(0x0003)
if child.size == 2:
first = last = child.value
else:
first, last = child.value >> 16, child.value & 0xFFFF
if first > last or first <= previous:
raise _SdpError(0x0003)
ranges.append((first, last))
previous = last
return tuple(ranges)
def continuation(self) -> bytes:
size = self.integer(1)
if size > 16:
raise _SdpError(0x0005)
if self.position + size != len(self.data):
raise _SdpError(0x0003)
return self.data[self.position :]
def _records(descriptor: bytes) -> dict[int, dict[int, bytes]]:
language = _sequence(_uint(0x656E), _uint(106), _uint(0x0100))
browse = _sequence(_uuid(0x1002))
def hid_protocol(psm: int) -> bytes:
return _sequence(_sequence(_uuid(0x0100), _uint(psm)), _sequence(_uuid(0x0011)))
hid = {
0x0000: _uint(0x00010000, 4),
0x0001: _sequence(_uuid(0x1124)),
0x0004: hid_protocol(0x0011),
0x0005: browse,
0x0006: language,
0x0009: _sequence(_sequence(_uuid(0x1124), _uint(0x0100))),
0x000D: _sequence(hid_protocol(0x0013)),
0x0100: _text("Nintendo RVL-CNT-01"),
0x0101: _text("Nintendo RVL-CNT-01"),
0x0102: _text("Nintendo"),
0x0200: _uint(0x0100), # HIDDeviceReleaseNumber
0x0201: _uint(0x0111), # HIDParserVersion
0x0202: _uint(0x04, 1), # HIDDeviceSubclass
0x0203: _uint(0x33, 1), # HIDCountryCode in the original record
0x0204: _boolean(False), # HIDVirtualCable
0x0205: _boolean(True), # HIDReconnectInitiate
0x0206: _sequence(_sequence(_uint(0x22, 1), _encode(4, descriptor))),
0x0207: _sequence(_sequence(_uint(0x0409), _uint(0x0100))),
0x0208: _boolean(False), # HIDSDPDisable
0x0209: _boolean(True), # HIDBatteryPower
0x020A: _boolean(True), # HIDRemoteWake
0x020B: _uint(0x0100), # HIDProfileVersion
0x020C: _uint(0x0C80), # HIDSupervisionTimeout
0x020D: _boolean(False), # HIDNormallyConnectable
0x020E: _boolean(False), # HIDBootDevice
}
pnp = {
0x0000: _uint(0x00010001, 4),
0x0001: _sequence(_uuid(0x1200)),
0x0005: browse,
0x0006: language,
0x0009: _sequence(_sequence(_uuid(0x1200), _uint(0x0103))),
0x0100: _text("Nintendo RVL-CNT-01"),
0x0102: _text("Nintendo"),
0x0200: _uint(0x0103), # Device ID specification version
0x0201: _uint(0x057E), # USB vendor: Nintendo
0x0202: _uint(0x0306), # Original Wii Remote
0x0203: _uint(0x0100), # Device version
0x0204: _boolean(True), # PrimaryRecord
0x0205: _uint(0x0002), # VendorIDSource: USB-IF
}
return {0x00010000: hid, 0x00010001: pnp}
def _collect_uuids(element: _Element, result: set[bytes]) -> None:
if element.kind == 3:
uuid = element.value
if element.size != 16:
uuid = uuid.rjust(4, b"\x00") + _UUID_BASE_SUFFIX
result.add(uuid)
elif element.kind in (6, 7):
for child in element.value:
_collect_uuids(child, result)
def _attribute_list(
record: dict[int, bytes], ranges: tuple[tuple[int, int], ...]
) -> bytes:
return _encode(
6,
b"".join(
_uint(attribute) + value
for attribute, value in sorted(record.items())
if any(first <= attribute <= last for first, last in ranges)
),
)
class SdpResponder:
"""Respond to one complete SDP request PDU at a time, without socket IO.
``mtu`` is the peer's L2CAP receive MTU, including the five-byte SDP
header, but not an L2CAP header. Keep this instance for the lifetime of
an SDP connection. At most 64 recently issued continuation states are
retained; evicted or mismatched states receive Invalid Continuation
State. Transaction IDs and attribute byte limits may change between
fragments; the actual query may not.
"""
def __init__(self, report_descriptor: bytes):
if not isinstance(report_descriptor, bytes):
raise TypeError("report_descriptor must be bytes")
if not 1 <= len(report_descriptor) <= _MAX_DESCRIPTOR:
raise ValueError("report_descriptor must contain 1 to 4096 bytes")
self._records = _records(report_descriptor)
self._record_uuids = {}
for handle, record in self._records.items():
uuids = set()
for value in record.values():
_collect_uuids(_Reader(value).element(), uuids)
self._record_uuids[handle] = uuids
self._continuations: OrderedDict[bytes, tuple[tuple, bytes, int]] = (
OrderedDict()
)
self._serial = 0
def _save(self, query: tuple, data: bytes, offset: int) -> bytes:
if self._serial == (1 << (8 * _TOKEN_BYTES)) - 1:
raise _SdpError(0x0006)
self._serial += 1
token = self._serial.to_bytes(_TOKEN_BYTES, "big")
self._continuations[token] = (query, data, offset)
if len(self._continuations) > _MAX_CONTINUATIONS:
self._continuations.popitem(last=False)
return token
def _data(
self,
pdu: int,
target: int | tuple[bytes, ...],
ranges: tuple[tuple[int, int], ...],
maximum: int,
) -> bytes:
if pdu == 0x04:
record = self._records.get(target)
if record is None:
raise _SdpError(0x0002)
return _attribute_list(record, ranges)
handles = [
handle
for handle, uuids in self._record_uuids.items()
if all(uuid in uuids for uuid in target)
]
if pdu == 0x02:
return b"".join(handle.to_bytes(4, "big") for handle in handles[:maximum])
return _sequence(
*(_attribute_list(self._records[handle], ranges) for handle in handles)
)
def _response(self, pdu: int, reader: _Reader, mtu: int) -> bytes:
if pdu not in (0x02, 0x04, 0x06):
raise _SdpError(0x0003)
target = reader.integer(4) if pdu == 0x04 else reader.search_pattern()
maximum = reader.integer(2)
if maximum < (1 if pdu == 0x02 else 7):
raise _SdpError(0x0003)
ranges = () if pdu == 0x02 else reader.attributes()
token = reader.continuation()
query = (pdu, target, ranges, maximum if pdu == 0x02 else None)
if token:
state = self._continuations.get(token)
if state is None or state[0] != query:
raise _SdpError(0x0005)
_, data, offset = state
self._continuations.move_to_end(token)
else:
data = self._data(pdu, target, ranges, maximum)
offset = 0
# Attribute fragments may split any data element, but must each
# contain at least two bytes; search fragments contain whole handles.
search = pdu == 0x02
overhead = 10 if search else 8
limit = 0xFFFF if search else maximum
remaining = len(data) - offset
final_capacity = min(limit, mtu - overhead)
if remaining <= final_capacity:
count = remaining
continuation = b"\x00"
else:
count = min(limit, mtu - overhead - _TOKEN_BYTES)
if search:
count -= count % 4
elif remaining - count == 1:
count -= 1
if count < (4 if search else 2):
raise _SdpError(0x0006)
next_token = self._save(query, data, offset + count)
continuation = bytes([len(next_token)]) + next_token
fragment = data[offset : offset + count]
if search:
prefix = struct.pack(">HH", len(data) // 4, count // 4)
else:
prefix = struct.pack(">H", count)
return prefix + fragment + continuation
def reply(self, packet: bytes, mtu: int = 672) -> bytes:
"""Return the matching response or an SDP ErrorResponse.
Invalid local MTUs below seven bytes raise ValueError: even an SDP
error cannot fit. Requests over 4096 bytes or overly complex data
elements receive Insufficient Resources. A truncated header lacking
a complete transaction ID is answered with transaction ID zero.
"""
if mtu < 7:
raise ValueError("MTU cannot fit an SDP ErrorResponse")
mtu = min(mtu, 0xFFFF + 5)
transaction = int.from_bytes(packet[1:3], "big") if len(packet) >= 3 else 0
try:
if len(packet) < 5 or int.from_bytes(packet[3:5], "big") != len(packet) - 5:
raise _SdpError(0x0004)
if len(packet) > _MAX_REQUEST:
raise _SdpError(0x0006)
response = self._response(packet[0], _Reader(packet[5:]), mtu)
response_pdu = packet[0] + 1
except _SdpError as error:
response_pdu = 0x01
response = struct.pack(">H", error.code)
return struct.pack(">BHH", response_pdu, transaction, len(response)) + response

314
tools/dolphinbar_wiimote.py Normal file
View file

@ -0,0 +1,314 @@
"""RAM-only, neutral RVL-CNT-01 protocol for the DolphinBar experiment.
Wire formats: https://wiibrew.org/wiki/Wiimote . Report handling facts were
cross-checked against rnconrad/WiimoteEmulator and Dolphin's WiimoteEmu.
This module contains no transport, physical inputs, audio output, or file I/O.
"""
from __future__ import annotations
_OUTPUT_LENGTHS = {
0x10: 1,
0x11: 1,
0x12: 2,
0x13: 1,
0x14: 1,
0x15: 1,
0x16: 21,
0x17: 6,
0x18: 21,
0x19: 1,
0x1A: 1,
}
_INPUT_LENGTHS = {
0x20: 6,
0x21: 21,
0x22: 4,
0x30: 2,
0x31: 5,
0x32: 10,
0x33: 17,
0x34: 21,
0x35: 21,
0x36: 21,
0x37: 21,
0x3D: 21,
0x3E: 21,
0x3F: 21,
}
def _hid_descriptor() -> bytes:
# Generic Desktop/Game Pad application; vendor-defined, opaque byte arrays.
descriptor = bytearray.fromhex("05 01 09 05 a1 01 15 00 26 ff 00 75 08 06 00 ff")
for lengths, item in ((_OUTPUT_LENGTHS, 0x91), (_INPUT_LENGTHS, 0x81)):
for report_id, size in lengths.items():
descriptor.extend((0x85, report_id, 0x95, size, 0x09, 0x01, item, 0x00))
descriptor.append(0xC0)
return bytes(descriptor)
HID_DESCRIPTOR = _hid_descriptor()
# High eight accelerometer bits: zero-g = 0x80, one-g = 0x9a.
# A stationary, face-up remote measures (0g, 0g, +1g); low bits are zero.
_ACCEL_ZERO = 0x80
_ACCEL_ONE = 0x9A
_ACCEL = bytes((_ACCEL_ZERO, _ACCEL_ZERO, _ACCEL_ONE))
_BUTTONS = b"\x00\x00"
_ABSENT = b"\xff" * 21
_NEUTRAL_REPORTS = {
0x30: b"\xa1\x30" + _BUTTONS,
0x31: b"\xa1\x31" + _BUTTONS + _ACCEL,
0x32: b"\xa1\x32" + _BUTTONS + _ABSENT[:8],
0x33: b"\xa1\x33" + _BUTTONS + _ACCEL + _ABSENT[:12],
0x34: b"\xa1\x34" + _BUTTONS + _ABSENT[:19],
0x35: b"\xa1\x35" + _BUTTONS + _ACCEL + _ABSENT[:16],
0x36: b"\xa1\x36" + _BUTTONS + _ABSENT[:19],
0x37: b"\xa1\x37" + _BUTTONS + _ACCEL + _ABSENT[:16],
0x3D: b"\xa1\x3d" + _ABSENT,
# The interleaved button fields carry the high and low nibbles of Z's
# eight-bit value, not the low bits used by normal accelerometer reports.
0x3E: bytes(
(
0xA1,
0x3E,
((_ACCEL_ONE >> 4) & 3) << 5,
((_ACCEL_ONE >> 6) & 3) << 5,
_ACCEL_ZERO,
)
)
+ _ABSENT[:18],
0x3F: bytes(
(0xA1, 0x3F, (_ACCEL_ONE & 3) << 5, ((_ACCEL_ONE >> 2) & 3) << 5, _ACCEL_ZERO)
)
+ _ABSENT[:18],
}
def _calibration_block(data: bytes) -> bytes:
return data + bytes(((sum(data) + 0x55) & 0xFF,))
def _new_eeprom() -> bytearray:
eeprom = bytearray(0x1700)
# Symmetric factory reference points in all four camera quadrants. These
# are calibration constants only, never emitted as observed IR spots.
points = ((128, 128), (896, 128), (128, 640), (896, 640))
packed = bytearray()
for index in (0, 2):
x1, y1 = points[index]
x2, y2 = points[index + 1]
high = ((y1 >> 8) << 6) | ((x1 >> 8) << 4) | ((y2 >> 8) << 2) | (x2 >> 8)
packed.extend((x1 & 0xFF, y1 & 0xFF, high, x2 & 0xFF, y2 & 0xFF))
ir_calibration = _calibration_block(bytes(packed))
accel_calibration = _calibration_block(
bytes(
(
_ACCEL_ZERO,
_ACCEL_ZERO,
_ACCEL_ZERO,
0,
_ACCEL_ONE,
_ACCEL_ONE,
_ACCEL_ONE,
0,
0x40,
)
)
)
eeprom[0x00:0x0B] = ir_calibration
eeprom[0x0B:0x16] = ir_calibration
eeprom[0x16:0x20] = accel_calibration
eeprom[0x20:0x2A] = accel_calibration
return eeprom
class Wiimote:
"""An original remote with no buttons pressed, extension, or visible IR.
``handle_output`` consumes report ID + its exact descriptor-sized payload;
callers remove the Bluetooth 0xa2/0x52 header. Every returned packet already
includes 0xa1. Drain command replies before requesting periodic reports.
There is no internal reply queue: a read returns at most 368 packets.
EEPROM writes change only this instance's RAM, not the simulated sensor.
"""
def __init__(self) -> None:
self.ir_enabled = False # Report 0x13 also controls the status flag/I2C.
self.ir_secondary_enabled = False # Report 0x1a, the second camera gate.
self.report_mode = 0x30
self.rumble = False
self.leds = 0
self.speaker_enabled = False
self.speaker_muted = False
self.speaker_data = b"" # Last accepted FIFO packet only; never played.
self._continuous = False
self._report_pending = True
self._interleaved_next = 0x3E
self._pair_pending = False
self._eeprom = _new_eeprom()
self._speaker = bytearray(0x0A)
self._camera = bytearray(0x5B)
self._camera[0x37:0x5B] = b"\xff" * 36
@staticmethod
def _ack(report_id: int, error: int = 0) -> bytes:
return bytes((0xA1, 0x22, 0, 0, report_id, error))
@staticmethod
def _read_reply(address: int, data: bytes = b"", error: int = 0) -> bytes:
size = 16 if error else len(data)
return bytes(
(
0xA1,
0x21,
0,
0,
((size - 1) << 4) | error,
(address >> 8) & 0xFF,
address & 0xFF,
)
) + data.ljust(16, b"\x00")
def status_report(self) -> bytes:
"""Return full battery, no extension, and the actual LED/feature flags."""
flags = (
self.leds | (int(self.speaker_enabled) << 2) | (int(self.ir_enabled) << 3)
)
return bytes((0xA1, 0x20, 0, 0, flags, 0, 0, 0xC0))
def _memory(
self, flags: int, address: int, size: int, writing: bool = False
) -> tuple[bytearray | None, int, int]:
"""Resolve the whole transfer before mutation; never resize a bank."""
space = flags & 0x0C
if space == 0x0C:
return None, 0, 6 # Invalid address space, not EEPROM or I2C.
if space == 0:
offset = address & 0xFFFF # EEPROM mirrors every 64 KiB.
if offset + size <= len(self._eeprom):
return self._eeprom, offset, 0
return None, 0, 8
peripheral = (address >> 16) & 0xFE
offset = address & 0xFF # Peripheral register high address byte is ignored.
if peripheral in (0xA4, 0xA6):
return None, 0, 7 # No extension or MotionPlus on the I2C bus.
if peripheral == 0xA2:
if offset + size <= len(self._speaker):
return self._speaker, offset, 0
elif peripheral == 0xB0:
if not self.ir_enabled:
return None, 0, 7
if offset + size <= 0x34:
return self._camera, offset, 0
if 0x37 <= offset and offset + size <= len(self._camera):
if writing:
return None, 0, 7 # Sensor output is read-only.
return self._camera, offset, 0
return None, 0, 8
def handle_output(self, report: bytes) -> list[bytes]:
"""Apply a well-formed command atomically and return its wire replies.
Unsupported/malformed commands return error 3, absent I2C extensions
error 7, invalid addresses error 8, invalid address-space selection 6.
Malformed reports never even change rumble. Empty input has no report
ID to acknowledge and is ignored. Valid writes always ACK; feature/mode
commands ACK only when requested by bit 1. Status, reads, rumble and
speaker streaming have their own reply rules, as on the original remote.
"""
if not report:
return []
report_id = report[0]
expected = _OUTPUT_LENGTHS.get(report_id)
if expected is None or len(report) != expected + 1:
return [self._ack(report_id, 3)]
flags = report[1]
if report_id == 0x12 and report[2] not in _NEUTRAL_REPORTS:
return [self._ack(report_id, 3)]
if report_id == 0x18 and not 1 <= flags >> 3 <= 20:
return [self._ack(report_id, 3)]
if report_id in (0x16, 0x17):
address = int.from_bytes(report[2:5], "big")
size = (
report[5] if report_id == 0x16 else int.from_bytes(report[5:7], "big")
)
if size == 0 or (report_id == 0x16 and size > 16):
if report_id == 0x17:
return [self._read_reply(address, error=3)]
return [self._ack(report_id, 3)]
# Register zero of the speaker is a streaming FIFO, not a write
# spanning configuration registers. Keep only the latest samples.
speaker_fifo = (
report_id == 0x16
and flags & 0x0C in (4, 8)
and (address >> 16) & 0xFE == 0xA2
and address & 0xFF == 0
)
if speaker_fifo:
self.speaker_data = report[6 : 6 + size]
self.rumble = bool(flags & 1)
return [self._ack(report_id)]
bank, offset, error = self._memory(flags, address, size, report_id == 0x16)
if error:
if report_id == 0x17:
return [self._read_reply(address, error=error)]
return [self._ack(report_id, error)]
self.rumble = bool(flags & 1)
if report_id == 0x16:
bank[offset : offset + size] = report[6 : 6 + size]
return [self._ack(report_id)]
return [
self._read_reply(
address + index,
bytes(bank[offset + index : offset + min(index + 16, size)]),
)
for index in range(0, size, 16)
]
self.rumble = bool(flags & 1)
if report_id == 0x10:
return []
if report_id == 0x11:
self.leds = flags & 0xF0
elif report_id == 0x12:
self.report_mode = report[2]
self._continuous = bool(flags & 4)
self._report_pending = True
self._interleaved_next = 0x3E
self._pair_pending = False
elif report_id == 0x13:
self.ir_enabled = bool(flags & 4)
elif report_id == 0x14:
self.speaker_enabled = bool(flags & 4)
elif report_id == 0x15:
return [self.status_report()]
elif report_id == 0x18:
self.speaker_data = report[2 : 2 + (flags >> 3)]
return []
elif report_id == 0x19:
self.speaker_muted = bool(flags & 4)
elif report_id == 0x1A:
self.ir_secondary_enabled = bool(flags & 4)
return [self._ack(report_id)] if flags & 2 else []
def periodic_report(self) -> bytes | None:
"""Return one 100-Hz tick's neutral report, or None if unchanged.
Interleaved modes always finish their 0x3e/0x3f pair, even with continuous
reporting disabled. The requested mode remains stable for logging.
Immutable packets are shared; idle/continuous ticks do not allocate.
"""
if not (self._continuous or self._report_pending or self._pair_pending):
return None
self._report_pending = False
if self.report_mode in (0x3E, 0x3F):
report_id = self._interleaved_next
self._pair_pending = report_id == 0x3E
self._interleaved_next = 0x3F if self._pair_pending else 0x3E
return _NEUTRAL_REPORTS[report_id]
return _NEUTRAL_REPORTS[self.report_mode]