Add Switch 2 wake capture and replay

tools/configure_switch2_wake.py

@@ -0,0 +1,213 @@
+#!/usr/bin/env python3
+"""Capture a Switch 2 wake record and generate the AIO firmware header."""
+
+import argparse
+import json
+import re
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_OUTPUT = (
+    ROOT / "src" / "firmware" / "platform" / "pico" /
+    "switch2_wake_config.h"
+)
+CAPTURE_PREFIX = "SWITCH2_WAKE_CAPTURE "
+ADDRESS_PATTERN = re.compile(r"(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}")
+EXPECTED_PREFIX = bytes.fromhex("0201061BFF53050100037E05")
+EXPECTED_SUFFIX = bytes.fromhex("0F00000000000000")
+
+
+class CaptureError(ValueError):
+    pass
+
+
+def normalize_address(value, field):
+    if not isinstance(value, str) or ADDRESS_PATTERN.fullmatch(value) is None:
+        raise CaptureError(f"{field} must be a six-byte Bluetooth address.")
+    normalized = value.upper()
+    if normalized == "00:00:00:00:00:00":
+        raise CaptureError(f"{field} must not be all zeroes.")
+    return normalized
+
+
+def parse_capture_line(line):
+    line = line.strip()
+    if line.startswith(CAPTURE_PREFIX):
+        line = line[len(CAPTURE_PREFIX):]
+    if not line.startswith("{"):
+        return None
+    try:
+        record = json.loads(line)
+    except json.JSONDecodeError:
+        return None
+    if not isinstance(record, dict) or "raw_hex" not in record:
+        return None
+    return validate_capture(record)
+
+
+def validate_capture(record):
+    if record.get("version", 1) != 1:
+        raise CaptureError("unsupported capture version")
+    if record.get("address_type") != 0:
+        raise CaptureError("capture must use a public advertiser address")
+    if record.get("event_type") != 0:
+        raise CaptureError("capture must be an ADV_IND wake advertisement")
+    advertiser = normalize_address(record.get("advertiser"), "advertiser")
+    console = normalize_address(record.get("console"), "console")
+    raw_hex = record.get("raw_hex")
+    if not isinstance(raw_hex, str) or re.fullmatch(r"[0-9A-Fa-f]{62}", raw_hex) is None:
+        raise CaptureError("raw_hex must contain exactly 31 bytes")
+    raw = bytes.fromhex(raw_hex)
+    if raw[:12] != EXPECTED_PREFIX:
+        raise CaptureError("capture is not a Nintendo Switch 2 advertisement")
+    if raw[14:16] != b"\x00\x01":
+        raise CaptureError("capture has an unexpected reconnect marker")
+    if raw[16] != 0x81:
+        raise CaptureError("capture does not contain the Switch 2 wake flag")
+    if raw[23:] != EXPECTED_SUFFIX:
+        raise CaptureError("capture has an unexpected reserved suffix")
+    decoded_console = ":".join(f"{byte:02X}" for byte in reversed(raw[17:23]))
+    if decoded_console != console:
+        raise CaptureError("console address does not match raw wake data")
+    pid = int.from_bytes(raw[12:14], "little")
+    if record.get("pid", f"{pid:04X}").upper() != f"{pid:04X}":
+        raise CaptureError("PID does not match raw wake data")
+    payload = raw[7:].hex().upper()
+    supplied_payload = record.get("esphome_payload_hex", payload)
+    if supplied_payload.upper() != payload:
+        raise CaptureError("manufacturer payload does not match raw wake data")
+    return {
+        "advertiser": advertiser,
+        "console": console,
+        "pid": pid,
+        "raw": raw,
+    }
+
+
+def byte_initializer(data):
+    return ", ".join(f"0x{byte:02X}" for byte in data)
+
+
+def render_header(capture):
+    source = bytes.fromhex(capture["advertiser"].replace(":", ""))
+    return (
+        "#pragma once\n\n"
+        "// Generated by tools/configure_switch2_wake.py. Do not commit this file.\n"
+        f"// Captured source controller: {capture['advertiser']}\n"
+        f"// Target Switch 2: {capture['console']}; PID: 0x{capture['pid']:04X}\n"
+        "#define SWITCH2_WAKE_CONFIGURED 1\n"
+        "#define SWITCH2_WAKE_SOURCE_ADDRESS_BYTES \\\n"
+        f"    {{{byte_initializer(source)}}}\n"
+        "#define SWITCH2_WAKE_ADVERTISEMENT_DATA_BYTES \\\n"
+        f"    {{{byte_initializer(capture['raw'])}}}\n"
+    )
+
+
+def capture_from_stream(stream, timeout):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        line = stream.readline()
+        if not line:
+            continue
+        if isinstance(line, bytes):
+            line = line.decode("utf-8", errors="replace")
+        print(line.rstrip())
+        capture = parse_capture_line(line)
+        if capture is not None:
+            return capture
+    raise CaptureError("timed out waiting for a Switch 2 wake capture")
+
+
+def pico_serial_ports():
+    try:
+        from serial.tools import list_ports
+    except ImportError as exc:
+        raise CaptureError(
+            "pyserial is required; install project dependencies first"
+        ) from exc
+    return [port.device for port in list_ports.comports() if port.vid == 0x2E8A]
+
+
+def capture_from_pico(port, timeout):
+    try:
+        import serial
+    except ImportError as exc:
+        raise CaptureError(
+            "pyserial is required; install project dependencies first"
+        ) from exc
+    if port is None:
+        ports = pico_serial_ports()
+        if len(ports) != 1:
+            detail = ", ".join(ports) if ports else "none"
+            raise CaptureError(
+                f"expected one Pico USB serial port, found {detail}; use --port"
+            )
+        port = ports[0]
+    print(f"Waiting for automatic wake capture on {port}...")
+    with serial.Serial(port, 115200, timeout=0.25, write_timeout=1) as device:
+        deadline = time.monotonic() + timeout
+        next_request = 0.0
+        while time.monotonic() < deadline:
+            now = time.monotonic()
+            if now >= next_request:
+                device.write(b"p\n")
+                next_request = now + 1.0
+            line = device.readline()
+            if not line:
+                continue
+            text = line.decode("utf-8", errors="replace").strip()
+            print(text)
+            capture = parse_capture_line(text)
+            if capture is not None:
+                return capture
+    raise CaptureError("timed out waiting for a Switch 2 wake capture")
+
+
+def write_header(path, capture):
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(render_header(capture), encoding="utf-8")
+    temporary.replace(path)
+
+
+def parse_args(argv=None):
+    parser = argparse.ArgumentParser(
+        description=(
+            "Read the one-shot Pico wake capture and configure the AIO build."
+        )
+    )
+    parser.add_argument("--port", help="Pico USB serial port; auto-detected by default")
+    parser.add_argument(
+        "--input",
+        type=Path,
+        help="read a saved capture log instead of a live Pico",
+    )
+    parser.add_argument("--timeout", type=float, default=90.0)
+    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
+    return parser.parse_args(argv)
+
+
+def main(argv=None):
+    args = parse_args(argv)
+    try:
+        if args.timeout <= 0:
+            raise CaptureError("timeout must be positive")
+        if args.input is None:
+            capture = capture_from_pico(args.port, args.timeout)
+        else:
+            with args.input.open("r", encoding="utf-8") as stream:
+                capture = capture_from_stream(stream, args.timeout)
+        write_header(args.output, capture)
+    except (CaptureError, OSError) as exc:
+        print(f"Error: {exc}", file=sys.stderr)
+        return 1
+    print(f"Configured Switch 2 wake: {args.output}")
+    print("Build and flash it with: python3 build.py --aio")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())