#!/usr/bin/env python3 """Bounded, non-pairing qualification of the switch-pico native Joy-Con USB hub. Requires Linux, PyUSB/libusb, and the existing sudo -n setfacl permission policy. Live checks use already-paired R/L donors or one full gamepad per virtual pair; they cannot wake or pair them. For two pairs, deliberately exercise independent buttons/sticks on BOTH controllers throughout the check (including controls on each R/L half); IMU mode also needs distinct deliberate motion on both sources. Neutral or unassigned pairs are not live input. Shared R/L motion is expected only within each full-gamepad pair, not across pairs. This cannot prove console gameplay, physical source isolation, or physical latency. --neutral explicitly selects standalone transport-only qualification with one or two pairs; it cannot prove live input, IMU, gameplay, or rumble. The JSON capture is created exclusively before USB access and retains failures. Qualification sends no reset, configuration change, pairing exchange, profile access, flash write, or HID output. Motor playback requires --rumble-sample. --capture-trace-on-error optionally asks a TRACE-enabled root to retain its current child EP0 context on the first child control-transfer error, before cleanup. It never retries the failed request or establishes that its SETUP reached the child. --reboot-bootsel is a separate, explicit root-only recovery operation: it sends the private BOOTSEL request and confirms ROM USB enumeration at the same port, without claiming interfaces or qualifying transport, live input, or gameplay. """ from __future__ import annotations import argparse import json import math import os import signal import struct import subprocess import time from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from typing import Any from switch2_native_imu import decode_block, native_block VID = 0x057E ROOT_PID = 0x2068 SERIAL_PREFIX = "switch-pico-" BOOTSEL_VID = 0x2E8A BOOTSEL_PIDS = (0x0003, 0x000F) # RP2040 and RP2350 ROM USB boot devices. NATIVE_HUB_TRACE_REQUEST = 0x5E NATIVE_HUB_TRACE_VALUE = 0x5452 NATIVE_HUB_TRACE_REPLY_SIZE = 16 SIDES = ("R", "L") # Only protocol constants live in source. Device-specific factory/calibration # captures stay in the private build paths configured by CMake. MODELS = { "R": {"pid": 0x2066, "port": 1, "report": 0x08, "diagnostic_host": "020000000001"}, "L": {"pid": 0x2067, "port": 2, "report": 0x07, "diagnostic_host": "020000000002"}, } CAPTURE_PREFIXES = ( "SWITCH2_PROBE", "SWITCH2_PROBE_SECOND", "SWITCH2_PROBE_THIRD", "SWITCH2_PROBE_FOURTH", ) def child_models(pairs: int) -> dict[str, dict[str, Any]]: if pairs not in (1, 2): raise ValueError("pair count must be 1 or 2") models = {} for slot in range(pairs * 2): side = SIDES[slot % 2] pair = "AB"[slot // 2] child = side if pairs == 1 else f"{pair}_{side}" models[child] = { **MODELS[side], "side": side, "pair": pair, "port": slot + 1, "capture_prefix": CAPTURE_PREFIXES[slot], "diagnostic_host": f"0200000000{slot + 1:02x}", } return models def neutral_calibration(factory: bytes, user: bytes) -> tuple[bytes, str]: # Match probe_memory_stick_calibration: each child's primary record, valid # user override before factory; a nominal 0x800 center is not sufficient. def valid(data: bytes) -> bool: axes = [] for offset in (0, 3, 6): axes.append( ( data[offset] | ((data[offset + 1] & 15) << 8), (data[offset + 1] >> 4) | (data[offset + 2] << 4), ) ) center, positive, negative = axes return all( 0 < center[axis] < 4095 and 0 < positive[axis] <= 4095 - center[axis] and 0 < negative[axis] <= center[axis] for axis in (0, 1) ) if user[0x40:0x42] == b"\xb2\xa1" and valid(user[0x42:0x4B]): return user[0x42:0x45], "user" if not valid(factory[0xA8:0xB1]): raise ValueError("no valid captured stick calibration for neutral reports") return factory[0xA8:0xAB], "factory" def model_references( build_dir: Path, *, require_imu: bool = True, pairs: int = 1, neutral: bool = False ) -> dict[str, dict[str, Any]]: cache = {} for line in (build_dir / "CMakeCache.txt").read_text().splitlines(): if line.startswith("SWITCH") and ":" in line and "=" in line: field, value = line.split("=", 1) cache[field.split(":", 1)[0]] = value def enabled(name: str) -> bool: return cache.get(name, "OFF").upper() not in ( "", "0", "OFF", "NO", "FALSE", "N", "IGNORE", "NOTFOUND", ) and not cache.get(name, "").upper().endswith("-NOTFOUND") if int(cache.get("SWITCH2_PROBE_PAIR_COUNT", "1")) != pairs: raise ValueError( "--pairs must match SWITCH2_PROBE_PAIR_COUNT in the build cache" ) if neutral != enabled("SWITCH2_PROBE_NEUTRAL_INPUT"): raise ValueError( "--neutral must match SWITCH2_PROBE_NEUTRAL_INPUT in the build cache" ) if ( pairs == 2 and not neutral and ( not enabled("SWITCH2_PROBE_HUB") or not enabled("SWITCH_PICO_SWITCH2_USB_BRIDGE") or cache.get("SWITCH2_BRIDGE_INPUT") not in ("DUALSENSE", "GAMEPAD") ) ): raise ValueError( "two-pair live references require a GAMEPAD or DUALSENSE Bluetooth bridge HUB" ) if neutral and ( not enabled("SWITCH2_PROBE_HUB") or not enabled("SWITCH2_PROBE_USB_INIT") or enabled("SWITCH_PICO_SWITCH2_USB_BRIDGE") ): raise ValueError( "neutral references require an initialized standalone HUB, not the Bluetooth bridge" ) if ( require_imu and not neutral and cache.get("SWITCH2_BRIDGE_INPUT") in ("DUALSENSE", "GAMEPAD") and cache.get("SWITCH2_BRIDGE_IMU_TARGET", "BOTH") != "BOTH" ): raise ValueError( "Full paired-IMU qualification requires SWITCH2_BRIDGE_IMU_TARGET=BOTH on every pair; " "use the USB-completion UART trace for LEFT/RIGHT routing comparisons" ) models = child_models(pairs) for child, model in models.items(): prefix = model["capture_prefix"] fields = ["IDENTITY_FILE", "VERSION_FILE", "FACTORY_FILE", "CONTROLLER_ADDRESS"] if neutral: fields.append("USER_CALIBRATION_FILE") missing = [ prefix + "_" + field for field in fields if not cache.get(prefix + "_" + field) ] if missing: raise ValueError(f"{child} missing build references: {', '.join(missing)}") identity = Path(cache[prefix + "_IDENTITY_FILE"]).read_bytes() version = Path(cache[prefix + "_VERSION_FILE"]).read_bytes() factory = Path(cache[prefix + "_FACTORY_FILE"]).read_bytes() address = bytes.fromhex(cache[prefix + "_CONTROLLER_ADDRESS"].replace(":", "")) if ( len(identity) != 64 or len(version) != 12 or len(factory) != 8192 or len(address) != 6 or address in (bytes(6), b"\xff" * 6) ): raise ValueError( f"{child} build references have invalid native lengths/address" ) if factory[:64] != identity or struct.unpack_from(" bytes: version = bytes.fromhex(model["version"]) return ( version[:3] + bytes(3) + version[4:7] + b"\0" + bytes.fromhex(model["mac_wire"]) ) def location(device: Any) -> dict[str, Any]: if device.bus is None or device.address is None or not device.port_numbers: raise RuntimeError("USB device has no usable physical bus/address/port path") return { "bus": int(device.bus), "address": int(device.address), "ports": list(device.port_numbers), "vid": int(device.idVendor), "pid": int(device.idProduct), } def command(group: int, subcommand: int, payload: bytes = b"") -> bytes: # An explicit allowlist, not a general-purpose command injection interface. if (group, subcommand) not in { (0x03, 0x0D), (0x03, 0x0A), (0x0C, 0x02), (0x0C, 0x04), (0x02, 0x04), (0x10, 0x01), (0x0A, 0x02), }: raise ValueError("command is outside the qualification allowlist") return bytes((group, 0x91, 0, subcommand, 0, len(payload), 0, 0)) + payload def reply(request: bytes, payload: bytes = b"") -> bytes: return bytes((request[0], 1, 0, request[3], 0, 0xF8, 0, 0)) + payload class Check: def __init__(self, args: argparse.Namespace, capture: Any) -> None: self.args = args self.models = child_models(args.pairs) self.children = tuple(self.models) self.capture = capture self.started = time.monotonic() self.deadline = self.started + args.timeout self.core: Any = None self.util: Any = None self.devices: dict[str, Any] = {} self.claimed: list[tuple[str, int]] = [] self.detached: list[tuple[str, int]] = [] self.current_stage = "dependencies" self.last_counter: dict[str, int] = {} self.last_controls: dict[str, tuple[bytes, bytes]] = {} self.imu_evidence: dict[str, set[bytes]] = { side: set() for side in self.children } self.motion_evidence: dict[str, set[tuple[Any, ...]]] = { child: set() for child in self.children } self.control_evidence: dict[str, set[tuple[bytes, bytes]]] = { child: set() for child in self.children } self.result: dict[str, Any] = { "schema_version": 1, "neutral_transport_only": args.neutral, "pair_count": args.pairs, "gameplay_proven": False, "physical_latency_proven": False, "physical_source_isolation_proven": False, "limitations": ( [ "No live controls, donor IMU, Bluetooth routing, console gameplay, or motor action is qualified.", "Neutral reports cannot distinguish same-side HID cross-routing when captured centers match; distinct EP0/bulk identities are checked separately.", "The build cache is a reference, not proof of which firmware is flashed.", ] if args.neutral else [ "Console gameplay, physical latency, and physical motor sensation are not qualified.", "Observed distinct input/motion samples do not prove physical source isolation; independently exercise every source throughout the check.", "Both virtual halves of every configured pair must show real activity; an unassigned or neutral pair cannot qualify.", "The build cache is a reference, not proof of which firmware is flashed.", ] ), "success": False, "exit_code": 2, "started_utc": datetime.now(timezone.utc).isoformat(), "parameters": { "timeout_seconds": args.timeout, "duration_seconds": args.duration, "usb_timeout_ms": args.usb_timeout_ms, "rumble_sample": args.rumble_sample, "require_imu": not args.input_only and not args.neutral, "input_only": args.input_only, "pairs": args.pairs, "neutral": args.neutral, "capture_trace_on_error": getattr( args, "capture_trace_on_error", False ), }, "safety": { "pairing_writes": False, "profile_access": False, "flash_writes": False, "usb_reset": False, "motor_requested": args.rumble_sample is not None, "trace_marker_requested": False, "trace_marker_note": "Optional root vendor IN retains volatile trace only; no native initialization, pairing/profile/flash access, or failed-request retry. Device context may predate the failed SETUP; endpoint/physical acceptance is not proven.", "initialization_note": "03/0d sets volatile diagnostic host/initialized state only; no 15/* pairing exchange or persistent pairing write is requested.", }, "scope": ( "Standalone neutral USB transport only; no live input/IMU/gameplay qualification" if args.neutral else "USB identity/protocol and observed live input/motion evidence, not console gameplay, physical source isolation, or physical latency qualification" ), "imu_decode_note": ( "No live IMU evidence is accepted or claimed; neutral reports must have empty IMU/mouse fields." if args.neutral else "Existing candidate codec; left uses its documented one-byte-earlier IMU boundary. Physical scales are not calibrated by this check." ), "stages": [], "errors": [], "seen_roots": [], "seen_children": [], "devices": {}, "acl": [], "interfaces": [], "controls": [], "failure_trace": None, "bulk": [], "active_rounds": [], "cleanup": [], "streams": { side: { "packets": 0, "valid_native_imu": 0, "valid_native_packets": 0, "imu_counter_changes": 0, "valid_neutral_packets": 0, "transport_counter_changes": 0, "last_transport_counter_change_seconds": None, "wrong_side": 0, "unexpected_report": 0, "invalid": 0, "zero_length_imu": 0, "timeouts": 0, "report_ids": {}, "imu_lengths": {}, "imu_formats": {}, "buttons_nonzero": 0, "control_changes": 0, "samples": [], "rejected_samples": [], "first_valid_seconds": None, "last_valid_seconds": None, "last_counter_change_seconds": None, "last_control_change_seconds": None, } for side in self.children }, } self.checkpoint() def elapsed(self) -> float: return round(time.monotonic() - self.started, 6) def checkpoint(self) -> None: self.result["elapsed_seconds"] = self.elapsed() self.capture.seek(0) json.dump(self.result, self.capture, indent=2, allow_nan=False) self.capture.write("\n") self.capture.truncate() self.capture.flush() def error(self, message: str, side: str | None = None) -> None: self.result["errors"].append( { "stage": self.current_stage, "side": side, "at_seconds": self.elapsed(), "message": message, } ) @contextmanager def stage(self, name: str): self.current_stage = name entry = {"name": name, "status": "running", "started_seconds": self.elapsed()} self.result["stages"].append(entry) self.checkpoint() print(f"[NATIVEHUB] stage={name}", flush=True) previous_errors = len(self.result["errors"]) try: yield except BaseException as error: entry["status"] = "failed" self.error(f"{type(error).__name__}: {error}") raise else: entry["status"] = ( "passed" if len(self.result["errors"]) == previous_errors else "failed" ) finally: entry["finished_seconds"] = self.elapsed() self.checkpoint() def timeout_ms(self, maximum: int | None = None, until: float | None = None) -> int: remaining = ( min(self.deadline, until if until is not None else self.deadline) - time.monotonic() ) if remaining <= 0: raise TimeoutError(f"bounded deadline reached during {self.current_stage}") return max( 1, min( self.args.usb_timeout_ms if maximum is None else maximum, int(remaining * 1000), ), ) def control( self, owner: str, name: str, request_type: int, request: int, value: int, index: int, length: int, ) -> bytes: entry = { "stage": self.current_stage, "side": owner, "name": name, "setup": [request_type, request, value, index, length], "at_seconds": self.elapsed(), } self.result["controls"].append(entry) transfer_attempted = False try: timeout = self.timeout_ms() device = self.devices[owner] transfer_attempted = True data = bytes( device.ctrl_transfer( request_type, request, value, index, length, timeout=timeout, ) ) entry["response_hex"] = data.hex() entry["length"] = len(data) return data except Exception as error: entry["error"] = str(error) entry["error_type"] = type(error).__name__ if ( transfer_attempted and isinstance(error, OSError) and getattr(self.args, "capture_trace_on_error", False) and owner != "root" and owner in self.models and self.result["failure_trace"] is None ): self.capture_failure_trace(owner, entry) raise def capture_failure_trace(self, owner: str, failed_control: dict[str, Any]) -> None: slot = self.models[owner]["port"] trace: dict[str, Any] = { "status": "pending", "request_attempted": False, "captured": False, "side": owner, "slot": slot, "failed_control": failed_control.copy(), "marker_setup": [ 0xC0, NATIVE_HUB_TRACE_REQUEST, NATIVE_HUB_TRACE_VALUE, slot, NATIVE_HUB_TRACE_REPLY_SIZE, ], "at_seconds": self.elapsed(), "response_hex": None, "receipt": None, } # Latch before USB access: root errors cannot recurse or cause a retry. self.result["failure_trace"] = trace self.result["parameters"]["capture_trace_on_error"] = True try: try: timeout = self.timeout_ms() except TimeoutError as error: trace["status"] = "deadline_expired" trace["error"] = str(error) return root = self.devices.get("root") if root is None: trace["status"] = "root_unavailable" trace["error"] = "no selected root available for the trace marker" return trace["request_attempted"] = True trace["timeout_ms"] = timeout self.result["safety"]["trace_marker_requested"] = True # Deliberately bypass control(): this is one diagnostic IN, not recovery. data = bytes(root.ctrl_transfer(*trace["marker_setup"], timeout=timeout)) trace["response_hex"] = data.hex() trace["status"] = "malformed" if len(data) != NATIVE_HUB_TRACE_REPLY_SIZE: raise ValueError("trace reply must be exactly 16 bytes") magic, version, status, echoed_slot, reserved, time_us, generation = ( struct.unpack("<4sBBBBII", data) ) if magic != b"NHTR" or version != 1 or reserved != 0: raise ValueError("invalid trace reply header") if echoed_slot != slot or status not in (0, 1): raise ValueError("invalid trace reply slot or status") if status == 1 and (time_us != 0 or generation != 0): raise ValueError("busy trace reply must have zero time and generation") trace["receipt"] = { "version": version, "status": status, "slot": echoed_slot, "time_us": time_us, "control_generation": generation, } trace["captured"] = status == 0 trace["status"] = "captured" if status == 0 else "busy" except (OSError, RuntimeError, ValueError, TypeError, struct.error) as error: if trace["status"] != "malformed": trace["status"] = "error" trace["error"] = f"{type(error).__name__}: {error}" finally: # Audit failures must not replace the original child-transfer exception. try: self.checkpoint() except (OSError, ValueError, TypeError) as error: trace["checkpoint_error"] = f"{type(error).__name__}: {error}" def discover(self, *, root_only: bool = False) -> None: until = min(self.deadline, self.started + 30) while time.monotonic() < until: devices = list(self.core.find(find_all=True) or []) roots = [] for device in devices: if (device.idVendor, device.idProduct) != (VID, ROOT_PID): continue seen = location(device) # Read kernel-cached serials before granting access, so genuine # Nintendo hubs and unrelated devices receive no USB requests/ACL. sysname = f"{seen['bus']}-" + ".".join(map(str, seen["ports"])) try: serial = ( (Path("/sys/bus/usb/devices") / sysname / "serial") .read_text() .strip() ) except OSError as error: seen["serial_error"] = str(error) serial = "" seen["serial"] = serial if seen not in self.result["seen_roots"]: self.result["seen_roots"].append(seen) if serial.startswith(SERIAL_PREFIX): roots.append((device, seen)) if len(roots) > 1: raise RuntimeError( "multiple switch-pico 057e:2068 hubs; refusing ambiguous target" ) if roots: root, root_info = roots[0] if root_only: self.devices = {"root": root} self.result["devices"] = {"root": root_info} return selected = {} direct_children = [] for device in devices: ports = tuple(device.port_numbers or ()) if device.bus != root.bus or ports[:-1] != tuple(root.port_numbers): continue seen = location(device) direct_children.append(seen) if seen not in self.result["seen_children"]: self.result["seen_children"].append(seen) for side in self.children: model = self.models[side] if (device.idVendor, device.idProduct, ports[-1]) == ( VID, model["pid"], model["port"], ): if side in selected: raise RuntimeError( f"duplicate {side} child on the expected hub port" ) selected[side] = device if len(selected) == len(self.children): if len(direct_children) != len(self.children): raise RuntimeError( "target hub has unexpected additional direct children" ) self.devices = {"root": root, **selected} if ( len({device.address for device in self.devices.values()}) != len(self.children) + 1 ): raise RuntimeError( "root and children do not have distinct USB addresses" ) self.result["devices"] = { "root": root_info, **{ side: { **location(selected[side]), "side": self.models[side]["side"], "pair": self.models[side]["pair"], "capture_prefix": self.models[side]["capture_prefix"], } for side in self.children }, } return time.sleep(min(0.1, max(0, until - time.monotonic()))) if root_only: raise TimeoutError( "discovery: expected one switch-pico 057e:2068 root; " "no child enumeration is required; inspect seen_roots" ) raise TimeoutError( "discovery: expected one switch-pico 057e:2068 with " + ", ".join( f"{child} {model['pid']:04x} at port {model['port']}" for child, model in self.models.items() ) + " on the same hub path; inspect seen_roots/seen_children" ) def permissions(self) -> None: for owner, device in self.devices.items(): node = f"/dev/bus/usb/{device.bus:03d}/{device.address:03d}" entry = {"owner": owner, "node": node, "granted": False} self.result["acl"].append(entry) try: subprocess.run( ["sudo", "-n", "setfacl", "-m", f"u:{os.getuid()}:rw", node], check=True, capture_output=True, text=True, timeout=self.timeout_ms(3000) / 1000, ) entry["granted"] = True except subprocess.CalledProcessError as error: entry["error"] = error.stderr.strip() raise RuntimeError( f"cannot grant access to {owner} {node}: {error.stderr.strip()}" ) from error data = self.control("root", "device_descriptor", 0x80, 6, 0x0100, 0, 18) if ( len(data) != 18 or data[:2] != b"\x12\x01" or data[4] != 9 or struct.unpack_from(" None: for side in self.children: # GET_CONFIGURATION only: do not reset USB or set a configuration. if self.control(side, "active_configuration", 0x80, 8, 0, 0, 1) != b"\x01": raise RuntimeError( f"{side} is not already configured as configuration 1" ) device = self.devices[side] for interface in (0, 1): entry = { "side": side, "interface": interface, "detached": False, "claimed": False, } self.result["interfaces"].append(entry) self.timeout_ms() if device.is_kernel_driver_active(interface): device.detach_kernel_driver(interface) self.detached.append((side, interface)) entry["detached"] = True self.util.claim_interface(device, interface) self.claimed.append((side, interface)) entry["claimed"] = True def descriptors(self, side: str, *, report: bool = True) -> None: model = self.models[side] device = self.control(side, "device_descriptor", 0x80, 6, 0x0100, 0, 18) if ( len(device) != 18 or device[:2] != b"\x12\x01" or device[7] != 64 or device[17] != 1 or struct.unpack_from(" len(config): raise RuntimeError(f"{side} malformed configuration item at {offset}") item = config[offset : offset + size] if item[1] == 4: if size != 9 or item[3] != 0 or item[2] in interfaces: raise RuntimeError(f"{side} duplicate or non-native interface") current = item[2] interfaces[current] = (item[5], item[4]) endpoints[current] = [] elif item[1] == 5: if size != 7 or current is None: raise RuntimeError(f"{side} malformed endpoint") endpoints[current].append( (item[2], item[3], int.from_bytes(item[4:6], "little")) ) elif item[1] == 0x21: if size != 9 or current != 0 or item[5:7] != b"\x01\x22": raise RuntimeError(f"{side} malformed HID descriptor") hid_length = int.from_bytes(item[7:9], "little") offset += size expected_endpoints = { 0: [(0x81, 3, 64), (0x01, 3, 64)], 1: [(0x02, 2, 64), (0x82, 2, 64)], } if ( interfaces != {0: (3, 2), 1: (255, 2)} or endpoints != expected_endpoints or hid_length != 100 ): raise RuntimeError(f"{side} is not HID0/vendor1 with native EP81/01/82/02") if not report: return hid = self.control( side, "hid_report_descriptor", 0x81, 6, 0x2200, 0, hid_length ) if len(hid) != hid_length: raise RuntimeError(f"{side} truncated HID report descriptor") inputs: dict[int, int] = {} outputs: dict[int, int] = {} report_id = report_size = report_count = 0 offset = 0 while offset < len(hid): prefix = hid[offset] offset += 1 size = 4 if prefix & 3 == 3 else prefix & 3 if prefix == 0xFE or offset + size > len(hid): raise RuntimeError(f"{side} malformed/unsupported HID item") value = int.from_bytes(hid[offset : offset + size], "little") offset += size kind = prefix & 0xFC if kind == 0x74: report_size = value elif kind == 0x94: report_count = value elif kind == 0x84: report_id = value elif kind in (0x80, 0x90): target = inputs if kind == 0x80 else outputs target[report_id] = ( target.get(report_id, 0) + report_size * report_count ) if inputs != {5: 504, model["report"]: 504} or outputs != {1: 504}: raise RuntimeError( f"{side} HID advertises the wrong side/report lengths: {inputs}, {outputs}" ) def identities(self, side: str, round_number: int = 0) -> None: # Alternate device and explicit interface recipients; both child-local # indexes must resolve to this USB address, never to the sibling. request_type, index = (0xC0, 0) if round_number % 2 == 0 else (0xC1, 1) identity = self.control( side, "vendor_identity03", request_type, 3, 0, index, 64 ) status = self.control(side, "vendor_version02", request_type, 2, 0, index, 64) self.result["devices"][side]["identity03_hex"] = identity.hex() self.result["devices"][side]["version02_hex"] = status.hex() if identity != bytes.fromhex(self.models[side]["identity"]): raise RuntimeError( f"{side} vendor 03 does not match its distinct captured factory identity" ) if status != expected_status(self.models[side]): raise RuntimeError( f"{side} vendor 02 must be 16 bytes with wire MAC tail {self.models[side]['mac_wire']}" ) # Full/short EP0 reads check transfer length/context teardown. short_length = (1, 7, 15)[round_number % 3] short = self.control( side, "vendor_version02_short", request_type, 2, 0, index, short_length ) if short != status[:short_length]: raise RuntimeError(f"{side} short vendor read leaked/truncated incorrectly") def interleaved_identities(self, round_number: int) -> None: # Complete short reads on every address before full reads in reverse # order, including while all child bulk replies are pending. request_type, index = (0xC0, 0) if round_number % 2 == 0 else (0xC1, 1) order = ( self.children if round_number % 2 == 0 else tuple(reversed(self.children)) ) for slot, side in enumerate(order): length = (1, 7, 15)[(round_number + slot) % 3] for request, expected in ( (3, bytes.fromhex(self.models[side]["identity"])), (2, expected_status(self.models[side])), ): actual = self.control( side, f"vendor{request:02x}_interleaved_short", request_type, request, 0, index, length, ) if actual != expected[:length]: raise RuntimeError( f"{side} interleaved short EP0 identity/status leaked" ) for side in reversed(order): self.identities(side, round_number) def exchange_children( self, requests: dict[str, tuple[bytes, bytes]], round_number: int = 0 ) -> None: order = ( self.children if round_number % 2 == 0 else tuple(reversed(self.children)) ) entries = {} # All devices have pending commands before any IN is consumed. Reverse # completion order to expose global reply-buffer reuse. for side in order: request, expected = requests[side] if request[:4] == b"\x0a\x91\x00\x02" and ( self.args.neutral or self.args.rumble_sample is None ): raise RuntimeError( "motor command requires --rumble-sample and live mode" ) entry = { "stage": self.current_stage, "side": side, "at_seconds": self.elapsed(), "request_hex": request.hex(), "expected_hex": expected.hex(), "segments_hex": [], "response_hex": "", "matched": False, } self.result["bulk"].append(entry) entries[side] = entry written = self.devices[side].write(0x02, request, timeout=self.timeout_ms()) entry["written"] = int(written) if written != len(request): raise RuntimeError( f"{side} short native bulk OUT ({written}/{len(request)})" ) if self.args.neutral: self.interleaved_identities(round_number) for side in reversed(order): expected = requests[side][1] actual = bytearray() entry = entries[side] while len(actual) < len(expected): segment = bytes( self.devices[side].read(0x82, 64, timeout=self.timeout_ms()) ) entry["segments_hex"].append(segment.hex()) remaining = len(expected) - len(actual) actual.extend(segment) entry["response_hex"] = actual.hex() if len(segment) != min(64, remaining): raise RuntimeError( f"{side} bulk segment length {len(segment)}, expected {min(64, remaining)}" ) if actual != expected: raise RuntimeError( f"{side} bulk header/length/content mismatch; possible cross-device reply leakage" ) entry["matched"] = True def initialize(self) -> None: for operation in ( "initialize", "select_report", "feature_mask", "feature_enable", ): requests = {} for side in self.children: if operation == "initialize": request = command( 3, 0x0D, b"\x01\x00" + bytes.fromhex(self.models[side]["diagnostic_host"]), ) response = reply(request, b"\x01\x00\x00\x00") elif operation == "select_report": request = command( 3, 0x0A, bytes((self.models[side]["report"], 0, 0, 0)) ) response = reply(request) else: # 0x01 buttons, 0x02 sticks, 0x04 IMU; not vibration/mouse/NFC. request = command( 0x0C, 2 if operation == "feature_mask" else 4, b"\x07\x00\x00\x00", ) response = reply(request, bytes(4)) requests[side] = (request, response) self.exchange_children(requests) def queries(self, round_number: int) -> None: request = command(0x10, 1) self.exchange_children( { side: ( request, reply(request, bytes.fromhex(self.models[side]["version"])), ) for side in self.children }, round_number, ) for region in ("factory", "user") if self.args.neutral else ("factory",): requests = {} for side_index, side in enumerate(self.children): # Distinct offsets isolate the echoed address even when same-side # firmware versions or captured calibration bytes are identical. offset = (round_number + side_index) % len(self.children) address = (0x13000 if region == "factory" else 0x1FC000) + offset request = command( 2, 4, b"\x50\x7e\x00\x00" + struct.pack(" str | None: stream = self.result["streams"][side] center = bytes.fromhex(self.models[side]["stick_center"]) if any(payload[2:4]): rejection = "neutral transport emitted nonzero buttons" elif payload[5:8] != center: rejection = "neutral transport stick does not match the selected captured calibration center" elif any(payload[8:]): rejection = "neutral transport emitted nonzero reserved/mouse/IMU fields" else: rejection = None if rejection: stream["invalid"] += 1 return rejection now = self.elapsed() counter = payload[0] sample["transport_counter"] = counter stream["valid_neutral_packets"] += 1 stream["zero_length_imu"] += 1 stream["last_valid_seconds"] = now if stream["first_valid_seconds"] is None: stream["first_valid_seconds"] = now if side in self.last_counter and self.last_counter[side] != counter: stream["transport_counter_changes"] += 1 stream["last_transport_counter_change_seconds"] = now self.last_counter[side] = counter samples = stream["samples"] if len(samples) < 4 or ( len(samples) < 32 and now - samples[-1]["at_seconds"] >= 0.5 ): samples.append(sample) stream["last_sample"] = sample return None def poll(self, side: str, until: float) -> None: stream = self.result["streams"][side] remaining = min(self.deadline, until) - time.monotonic() if remaining <= 0: return try: packet = bytes( self.devices[side].read( 0x81, 64, timeout=max(1, min(10, int(remaining * 1000))) ) ) except self.core.USBTimeoutError: stream["timeouts"] += 1 return stream["packets"] += 1 report_id = packet[0] if packet else -1 report_key = f"{report_id:02x}" if report_id >= 0 else "empty" stream["report_ids"][report_key] = stream["report_ids"].get(report_key, 0) + 1 sample = { "at_seconds": self.elapsed(), "stage": self.current_stage, "packet_hex": packet.hex(), } rejection = None if ( report_id == MODELS["L" if self.models[side]["side"] == "R" else "R"]["report"] ): stream["wrong_side"] += 1 rejection = "wrong-side native report on this device's HID pipe" elif report_id != self.models[side]["report"]: stream["unexpected_report"] += 1 rejection = ( "unexpected report ID; only this side's native 07/08 is accepted" ) elif len(packet) != 64: stream["invalid"] += 1 rejection = f"native report is {len(packet)} bytes, expected 64" else: payload = packet[1:] stream["valid_native_packets"] += 1 controls = (payload[2:4], payload[5:8]) sample["buttons_hex"], sample["stick_hex"] = ( part.hex() for part in controls ) if any(controls[0]): stream["buttons_nonzero"] += 1 if side in self.last_controls and self.last_controls[side] != controls: if any(controls[0]) and len(self.control_evidence[side]) < 512: self.control_evidence[side].add(controls) stream["control_changes"] += 1 stream["last_control_change_seconds"] = self.elapsed() self.last_controls[side] = controls length_offset = 14 if self.models[side]["side"] == "L" else 15 length = payload[length_offset] key = str(length) stream["imu_lengths"][key] = stream["imu_lengths"].get(key, 0) + 1 if self.args.neutral: rejection = self.neutral_report(side, payload, sample) elif length == 0: stream["zero_length_imu"] += 1 if self.args.input_only: if len(stream["samples"]) < 4: stream["samples"].append(sample) stream["last_sample"] = sample else: rejection = "zero-length IMU: inactive donor/neutral fallback is not live motion" else: try: block = ( native_block({"native_hex": packet.hex()}) if self.models[side]["side"] == "R" else payload[length_offset + 1 : length_offset + 1 + length] ) decoded = decode_block(block) if not any( value for vector in decoded["accelerations"] for value in vector["raw"] ): raise ValueError( "all-zero acceleration is not donor IMU evidence" ) if not all( math.isfinite(value) for value in decoded["quaternion_wire"] ): raise ValueError("non-finite decoded quaternion") sample["imu"] = decoded except (ValueError, AssertionError) as error: stream["invalid"] += 1 rejection = f"native IMU decode failed: {error}" else: now = self.elapsed() stream["valid_native_imu"] += 1 stream["last_valid_seconds"] = now if stream["first_valid_seconds"] is None: stream["first_valid_seconds"] = now counter = decoded["counter_ticks"] if side in self.last_counter and self.last_counter[side] != counter: stream["imu_counter_changes"] += 1 stream["last_counter_change_seconds"] = now self.last_counter[side] = counter if len(self.imu_evidence[side]) < 512: self.imu_evidence[side].add(block) if len(self.motion_evidence[side]) < 512: # Counters, elapsed ticks and temperature are not motion. self.motion_evidence[side].add( ( tuple(decoded["quaternion_wire"]), tuple( tuple(vector["raw"]) for vector in decoded["accelerations"] ), tuple( tuple(vector["raw"]) for vector in decoded["rotation_triplets"] ), ) ) format_key = f"{decoded['format']:02x}" first_format = format_key not in stream["imu_formats"] stream["imu_formats"][format_key] = ( stream["imu_formats"].get(format_key, 0) + 1 ) samples = stream["samples"] if ( len(samples) < 4 or first_format or ( len(samples) < 32 and now - samples[-1]["at_seconds"] >= 0.5 ) ): samples.append(sample) stream["last_sample"] = sample if rejection: sample["reason"] = rejection if len(stream["rejected_samples"]) < 8: stream["rejected_samples"].append(sample) def poll_children(self, until: float, reverse: bool = False) -> None: for side in reversed(self.children) if reverse else self.children: if time.monotonic() >= until: break self.poll(side, until) def counts(self) -> dict[str, int]: return { side: self.result["streams"][side][ "valid_neutral_packets" if self.args.neutral else "valid_native_packets" if self.args.input_only else "valid_native_imu" ] for side in self.children } def streams_ready(self) -> None: until = min(self.deadline, time.monotonic() + 60) iteration = 0 while time.monotonic() < until: self.poll_children(until, bool(iteration % 2)) iteration += 1 if all( ( stream["valid_neutral_packets"] >= 3 and stream["transport_counter_changes"] >= 2 ) if self.args.neutral else (stream["buttons_nonzero"] > 0 and stream["control_changes"] >= 2) if self.args.input_only else ( stream["valid_native_imu"] >= 3 and stream["imu_counter_changes"] >= 2 and ( self.args.pairs == 1 or ( stream["buttons_nonzero"] > 0 and stream["control_changes"] >= 2 ) ) ) for stream in self.result["streams"].values() ): return details = "; ".join( f"{side}: neutral={stream['valid_neutral_packets']}, transport_changes={stream['transport_counter_changes']}, imu={stream['valid_native_imu']}, counter_changes={stream['imu_counter_changes']}, buttons={stream['buttons_nonzero']}, control_changes={stream['control_changes']}, zero_imu={stream['zero_length_imu']}, invalid={stream['invalid']}, timeouts={stream['timeouts']}" for side, stream in self.result["streams"].items() ) raise RuntimeError( f"neutral reports/counters did not become ready on every child; {details}" if self.args.neutral else f"sources did not satisfy live-input evidence on every child during the manual-input window; both pairs must be assigned and independently exercised for --pairs 2; {details}; no pairing/wake/reset was attempted" ) def active(self) -> None: until = min(self.deadline, time.monotonic() + self.args.duration) initial_counts = self.counts() initial_controls = { child: self.result["streams"][child]["control_changes"] for child in self.children } for evidence in ( self.imu_evidence, self.motion_evidence, self.control_evidence, ): for samples in evidence.values(): samples.clear() change_key = ( "transport_counter_changes" if self.args.neutral else "control_changes" if self.args.input_only else "imu_counter_changes" ) initial_changes = { side: self.result["streams"][side][change_key] for side in self.children } next_query = time.monotonic() round_number = 0 pending = None while time.monotonic() < until: if time.monotonic() >= next_query and until - time.monotonic() >= 0.25: if pending is not None: pending["after"] = self.counts() if self.args.neutral: pending["after_changes"] = { side: self.result["streams"][side][change_key] for side in self.children } pending = { "round": round_number, "before": self.counts(), "started_seconds": self.elapsed(), } if self.args.neutral: pending["before_changes"] = { side: self.result["streams"][side][change_key] for side in self.children } self.result["active_rounds"].append(pending) for side in ( self.children if round_number % 2 == 0 else reversed(self.children) ): self.descriptors(side, report=round_number % 2 == 0) self.identities(side, round_number) self.queries(round_number) pending["reads_matched"] = True next_query = time.monotonic() + 0.25 round_number += 1 self.checkpoint() self.poll_children(until, bool(round_number % 2)) if pending is not None: pending["after"] = self.counts() if self.args.neutral: pending["after_changes"] = { side: self.result["streams"][side][change_key] for side in self.children } if round_number < 2: self.error( "fewer than two interleaved control/bulk rounds completed during streaming" ) if not any( all( entry.get("after", {}).get(side, 0) > entry["before"][side] and ( not self.args.neutral or entry.get("after_changes", {}).get(side, 0) > entry["before_changes"][side] ) for side in self.children ) for entry in self.result["active_rounds"] if entry.get("reads_matched") ): self.error( "no interleaved control/bulk round was bracketed by valid reports and advancing counters on every child" if self.args.neutral else "no interleaved control/bulk round was bracketed by valid input from every child" ) if self.args.neutral: self.result["imu_isolation"] = { "policy": "not_proven_neutral_transport_only" } else: pair_results = {} self.result["imu_isolation"] = { "sample_limit_per_child": 512, "policy": "not_required_input_only" if self.args.input_only else "pair_local_source_policy", "pairs": pair_results, } for offset in range(0, len(self.children), 2): pair_children = self.children[offset : offset + 2] right, left = pair_children shared = self.imu_evidence[right] & self.imu_evidence[left] shared_source = self.models[right].get("source_mode") in ( "DUALSENSE", "GAMEPAD", ) pair_results[self.models[right]["pair"]] = { "policy": "not_required_input_only" if self.args.input_only else "shared_physical_source" if shared_source else "independent_physical_sources", "identical_blocks_seen_on_both_sides": len(shared), "unique_blocks": { child: len(self.imu_evidence[child]) for child in pair_children }, "side_exclusive_blocks": { child: len(self.imu_evidence[child] - shared) for child in pair_children }, "unique_motion_samples": { child: len(self.motion_evidence[child]) for child in pair_children }, } if not self.args.input_only: for child in pair_children: evidence = self.imu_evidence[child] if not shared_source: evidence = evidence - shared if len(evidence) < 2 or len(self.motion_evidence[child]) < 2: self.error( "IMU evidence is frozen or lacks deliberate motion" if shared_source else "donor IMU evidence is frozen or duplicated within its virtual pair", child, ) if self.args.pairs == 2: comparison = { "policy": "distinct_exercised_samples_required_not_physical_source_isolation", "physical_source_isolation_proven": False, "children": {}, } self.result["inter_pair_evidence"] = comparison for child in self.children: other_pair = [ other for other in self.children if self.models[other]["pair"] != self.models[child]["pair"] ] other_side = next( other for other in other_pair if self.models[other]["side"] == self.models[child]["side"] ) controls = ( self.control_evidence[child] - self.control_evidence[other_side] ) motion = self.motion_evidence[child] - set().union( *(self.motion_evidence[other] for other in other_pair) ) comparison["children"][child] = { "pair_exclusive_pressed_control_states": len(controls), "pair_exclusive_motion_samples": len(motion), } if len(controls) < 2: self.error( "insufficient pair-distinct pressed control states; independently press buttons and vary sticks on both controllers, including every R/L half; neutral/static or mirrored input cannot qualify", child, ) if not self.args.input_only and len(motion) < 2: self.error( "insufficient pair-distinct motion; deliberately move both sources differently; shared/static samples or advancing counters alone cannot qualify", child, ) for side in self.children: stream = self.result["streams"][side] if self.args.pairs == 2 and not self.args.neutral: last_control = stream["last_control_change_seconds"] if ( stream["control_changes"] - initial_controls[side] < 2 or last_control is None or self.elapsed() - last_control > 2 ): self.error( "real controls must keep changing on every child of both pairs during active reads", side, ) if ( self.counts()[side] - initial_counts[side] < 3 or stream[change_key] - initial_changes[side] < 2 ): self.error( "insufficient fresh neutral reports/transport counters during active control/bulk reads" if self.args.neutral else "insufficient fresh controller transitions" if self.args.input_only else "insufficient fresh native source IMU during active control/bulk reads", side, ) last_change = stream[ "last_transport_counter_change_seconds" if self.args.neutral else "last_control_change_seconds" if self.args.input_only else "last_counter_change_seconds" ] if last_change is None or self.elapsed() - last_change > 2: self.error( "neutral transport counter stopped advancing before streaming finished" if self.args.neutral else "source stopped advancing before streaming finished", side, ) if stream["wrong_side"] or stream["unexpected_report"] or stream["invalid"]: self.error( f"rejected wrong-side={stream['wrong_side']}, unexpected={stream['unexpected_report']}, malformed={stream['invalid']} HID packets", side, ) if time.monotonic() >= self.deadline: raise TimeoutError( "overall timeout interrupted the required streaming duration" ) def cleanup(self) -> None: self.current_stage = "cleanup" for action, tracked in (("release", self.claimed), ("reattach", self.detached)): for side, interface in reversed(tracked): entry = { "action": action, "side": side, "interface": interface, "success": False, } self.result["cleanup"].append(entry) try: if action == "release": self.util.release_interface(self.devices[side], interface) else: self.devices[side].attach_kernel_driver(interface) entry["success"] = True except (OSError, RuntimeError, NotImplementedError) as error: entry["error"] = str(error) self.error(f"{action} interface {interface}: {error}", side) if self.util is not None: for owner, device in self.devices.items(): try: self.util.dispose_resources(device) except (OSError, RuntimeError, NotImplementedError) as error: self.error(f"dispose USB resources: {error}", owner) def run(self) -> int: completed = False try: with self.stage("dependencies"): import usb.core import usb.util with self.stage("references"): self.models = model_references( self.args.build_dir, require_imu=not self.args.input_only and not self.args.neutral, pairs=self.args.pairs, neutral=self.args.neutral, ) self.result["references"] = { "build_dir": str(self.args.build_dir.resolve()), "children": { side: { field: model[field] for field in ( "pair", "side", "port", "capture_prefix", "mac_wire", "source_mode", "stick_center", "calibration_source", ) if field in model } for side, model in self.models.items() }, } self.core, self.util = usb.core, usb.util with self.stage("discovery"): self.discover() with self.stage("permissions_and_root_identity"): self.permissions() with self.stage("claim_child_interfaces"): self.claim() with self.stage("descriptor_and_ep0_isolation"): for round_number in range(20): for side in ( self.children if round_number % 2 == 0 else reversed(self.children) ): self.descriptors(side, report=round_number in (0, 19)) self.identities(side, round_number) with self.stage("native_initialization"): self.initialize() with self.stage("bulk_isolation"): for round_number in range(2): self.queries(round_number) with self.stage( "neutral_stream_startup" if self.args.neutral else "donor_startup" ): self.streams_ready() with self.stage( "active_neutral_transport_and_read_isolation" if self.args.neutral else "active_input_and_read_isolation" ): self.active() if ( not self.args.neutral and self.args.rumble_sample is not None and not self.result["errors"] ): with self.stage("explicit_motor_sample_ack"): request = command( 0x0A, 2, bytes((self.args.rumble_sample, 0, 0, 0)) ) self.exchange_children( {side: (request, reply(request)) for side in self.children} ) self.result["motor_result"] = ( "native sample ACK received for each side; physical sensation is not measured" ) else: self.result["stages"].append( {"name": "explicit_motor_sample_ack", "status": "skipped"} ) if self.args.neutral: self.result["motor_result"] = ( "not_requested_neutral_transport_only; no motor action or ACK qualified" ) completed = True except KeyboardInterrupt: self.result["interrupted"] = True except ( OSError, RuntimeError, ValueError, TypeError, ImportError, subprocess.SubprocessError, struct.error, ) as error: # The stage records partial responses; retain the terminal cause too. self.result["failure"] = str(error) finally: try: with self.stage("cleanup"): self.cleanup() finally: self.result["success"] = completed and not self.result["errors"] self.result["exit_code"] = 0 if self.result["success"] else 2 failed = sorted({entry["stage"] for entry in self.result["errors"]}) streams = self.result["streams"] self.result["child_results"] = { side: { "pair": self.models[side]["pair"], "side": self.models[side]["side"], "port": self.models[side]["port"], "qualified": self.result["success"], "neutral_transport_only": self.args.neutral, "live_input_proven": self.result["success"] and not self.args.neutral, "live_imu_proven": self.result["success"] and not self.args.neutral and not self.args.input_only, "gameplay_proven": False, "physical_latency_proven": False, "physical_source_isolation_proven": False, "valid_reports": self.counts()[side], "transport_counter_changes": streams[side][ "transport_counter_changes" ], "control_changes": streams[side]["control_changes"], "imu_counter_changes": streams[side]["imu_counter_changes"], "last_sample": streams[side].get("last_sample"), "errors": [ entry for entry in self.result["errors"] if entry["side"] in (side, None, "root") ], } for side in self.children } if self.args.neutral: stream_summary = ( f"NEUTRAL_TRANSPORT_ONLY pairs={self.args.pairs} gameplay=not_proven " + " ".join( f"{side}={streams[side]['valid_neutral_packets']} {side}_transport_counter_changes={streams[side]['transport_counter_changes']}" for side in self.children ) + " " ) else: stream_summary = ( f"{'LIVE_INPUT_ONLY' if self.args.input_only else 'LIVE_INPUT_AND_IMU'} " f"pairs={self.args.pairs} gameplay=not_proven physical_latency=not_proven " + " ".join( f"{child}={self.counts()[child]} {child}_control_changes={streams[child]['control_changes']} {child}_imu_counter_changes={streams[child]['imu_counter_changes']}" for child in self.children ) + " " ) self.result["summary"] = ( f"{'PASS' if self.result['success'] else 'FAIL'} " f"{stream_summary}" f"active_rounds={len(self.result['active_rounds'])} " f"errors={len(self.result['errors'])} failed_stages={','.join(failed) or 'none'}" ) self.checkpoint() print(f"[NATIVEHUB] {self.result['summary']}", flush=True) print(f"[NATIVEHUB] capture={self.args.output}", flush=True) return self.result["exit_code"] class BootselRecovery(Check): """Reuse bounded auditing/root identity checks, never the qualification run.""" def __init__(self, args: argparse.Namespace, capture: Any) -> None: self.args = args self.capture = capture self.started = time.monotonic() self.deadline = self.started + args.timeout self.core: Any = None self.util: Any = None self.devices: dict[str, Any] = {} self.claimed: list[tuple[str, int]] = [] self.detached: list[tuple[str, int]] = [] self.current_stage = "dependencies" self.result: dict[str, Any] = { "schema_version": 1, "operation": "bootsel_recovery", "scope": "Root-only ROM BOOTSEL recovery; no controller or transport qualification", "success": False, "recovery_success": False, "qualification_success": False, "live_input_proven": False, "live_imu_proven": False, "gameplay_proven": False, "exit_code": 2, "started_utc": datetime.now(timezone.utc).isoformat(), "parameters": { "timeout_seconds": args.timeout, "usb_timeout_ms": args.usb_timeout_ms, "reboot_bootsel": True, }, "safety": { "pairing_writes": False, "profile_access": False, "flash_writes": False, "usb_reset": False, "motor_requested": False, "native_initialization": False, "interface_claims": False, }, "recovery": { "request_attempted": False, "request_acknowledged": False, "root_disappeared": False, "rom_confirmed": False, }, "stages": [], "errors": [], "seen_roots": [], "seen_bootsel": [], "devices": {}, "acl": [], "controls": [], "cleanup": [], } self.checkpoint() def ctrl_transfer( self, request_type: int, request: int, value: int, index: int, data: bytes, timeout: int, ) -> int: # config_manager owns the envelope/setup encoding; this transport adds # the scenario deadline, audit trail, and short-write detection. entry = { "stage": self.current_stage, "side": "root", "name": "bootsel_reboot", "setup": [request_type, request, value, index, len(data)], "request_hex": data.hex(), "at_seconds": self.elapsed(), } self.result["controls"].append(entry) self.result["recovery"]["request_attempted"] = True self.checkpoint() try: written = self.devices["root"].ctrl_transfer( request_type, request, value, index, data, timeout=self.timeout_ms(min(timeout, self.args.usb_timeout_ms)), ) entry["length"] = written if written != len(data): raise RuntimeError( f"short BOOTSEL control write: {written} of {len(data)} bytes; " "reboot outcome is unconfirmed" ) self.result["recovery"]["request_acknowledged"] = True return written except Exception as error: entry["error"] = str(error) raise def confirm_bootsel(self) -> None: root = self.result["devices"]["root"] recovery = self.result["recovery"] while time.monotonic() < self.deadline: at_port = [] for device in self.core.find(find_all=True) or []: same_port = device.bus == root["bus"] and tuple( device.port_numbers or () ) == tuple(root["ports"]) if same_port: at_port.append(device) if device.idVendor == BOOTSEL_VID and device.idProduct in BOOTSEL_PIDS: seen = location(device) if seen not in self.result["seen_bootsel"]: self.result["seen_bootsel"].append(seen) if not any( (device.idVendor, device.idProduct) == (VID, ROOT_PID) for device in at_port ): recovery["root_disappeared"] = True if len(at_port) == 1: device = at_port[0] if device.idVendor == BOOTSEL_VID and device.idProduct in BOOTSEL_PIDS: self.result["devices"]["bootsel"] = location(device) recovery["rom_confirmed"] = True return self.checkpoint() time.sleep(min(0.1, max(0, self.deadline - time.monotonic()))) raise TimeoutError( "BOOTSEL request acknowledged, but ROM USB did not replace the root " "at the same physical bus/port before the deadline; recovery is unconfirmed" ) def run(self) -> int: completed = False try: with self.stage("dependencies"): import usb.core import usb.util from switch_pico_bridge.config_manager import request_bootsel_reboot self.core, self.util = usb.core, usb.util with self.stage("root_only_discovery"): self.discover(root_only=True) with self.stage("permissions_and_root_identity"): self.permissions() with self.stage("explicit_bootsel_request"): request_bootsel_reboot(self) with self.stage("same_port_rom_enumeration"): self.confirm_bootsel() completed = True except KeyboardInterrupt: self.result["interrupted"] = True except ( OSError, RuntimeError, ValueError, TypeError, ImportError, subprocess.SubprocessError, struct.error, ) as error: self.result["failure"] = str(error) finally: try: with self.stage("cleanup"): self.cleanup() finally: success = completed and not self.result["errors"] self.result["success"] = self.result["recovery_success"] = success self.result["exit_code"] = 0 if success else 2 recovery = self.result["recovery"] self.result["summary"] = ( f"{'RECOVERY_CONFIRMED' if success else 'RECOVERY_INCOMPLETE'} " f"BOOTSEL request_acknowledged={recovery['request_acknowledged']} " f"same_port_rom_confirmed={recovery['rom_confirmed']} " "qualification=not_run live_input=not_proven gameplay=not_proven" ) self.checkpoint() print(f"[NATIVEHUB] {self.result['summary']}", flush=True) print(f"[NATIVEHUB] capture={self.args.output}", flush=True) return self.result["exit_code"] def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--build-dir", type=Path, default=Path(__file__).resolve().parents[1] / "build-switch2-native-hub", help="configured build containing private capture reference paths", ) parser.add_argument( "--output", required=True, type=Path, help="new JSON capture path; existing files are never overwritten", ) parser.add_argument( "--timeout", type=float, default=90, help="overall USB scenario deadline in seconds, at most 600 (default: 90)", ) parser.add_argument( "--duration", type=float, default=10, help="active input/read overlap duration, 2..120 seconds (default: 10)", ) parser.add_argument( "--usb-timeout-ms", type=int, default=500, help="per control/bulk transfer timeout, 20..3000 ms (default: 500)", ) input_mode = parser.add_mutually_exclusive_group() input_mode.add_argument( "--input-only", action="store_true", help="qualify without IMU; requires real button presses and continued control changes on every R/L half of every pair; use distinct independent controls for two pairs", ) input_mode.add_argument( "--neutral", action="store_true", help="OPT-IN: standalone neutral transport only; requires calibrated neutral controls and advancing USB counters, never proves live input/IMU/gameplay", ) input_mode.add_argument( "--reboot-bootsel", action="store_true", help="OPT-IN: reboot only the identified root into ROM BOOTSEL and confirm the same physical port; no qualification, children, build references, or interface claims", ) parser.add_argument( "--pairs", type=int, choices=(1, 2), default=1, help="hub pair count matching the build cache; two live pairs require GAMEPAD/DUALSENSE and independent activity on both controllers (default: 1)", ) parser.add_argument( "--rumble-sample", type=int, choices=range(8), help="OPT-IN: play native motor sample 0..7 once on each donor and require its ACK", ) parser.add_argument( "--capture-trace-on-error", action="store_true", help="OPT-IN: request one volatile child EP0 snapshot from a TRACE-enabled root after the first child control-transfer error, before cleanup; never retry the failed request", ) args = parser.parse_args() if args.reboot_bootsel and args.capture_trace_on_error: parser.error( "--reboot-bootsel forbids --capture-trace-on-error; recovery never captures child traces" ) if args.reboot_bootsel and args.rumble_sample is not None: parser.error( "--reboot-bootsel forbids --rumble-sample; recovery never actuates motors" ) if args.neutral and args.rumble_sample is not None: parser.error( "--neutral forbids --rumble-sample; transport qualification must not actuate motors" ) if not math.isfinite(args.timeout) or not 0 < args.timeout <= 600: parser.error("timeout must be finite and in (0,600]") if not args.reboot_bootsel: if not math.isfinite(args.duration) or not 2 <= args.duration <= 120: parser.error("duration must be finite and in [2,120]") if args.timeout < args.duration + 10: parser.error( "timeout must allow at least duration + 10 seconds for discovery/initialization" ) if not 20 <= args.usb_timeout_ms <= 3000: parser.error("usb-timeout-ms must be in [20,3000]") try: args.output.parent.mkdir(parents=True, exist_ok=True) capture = args.output.open("x+", encoding="utf-8") except OSError as error: parser.error(f"cannot create a new capture: {error}") def interrupted(signum: int, _frame: Any) -> None: raise KeyboardInterrupt(f"received signal {signum}") previous = signal.signal(signal.SIGTERM, interrupted) try: with capture: scenario = BootselRecovery if args.reboot_bootsel else Check return scenario(args, capture).run() finally: signal.signal(signal.SIGTERM, previous) if __name__ == "__main__": raise SystemExit(main())