warpedpinball.transports.usb

  1"""USB serial transport for Vector boards.
  2
  3The firmware tunnels the same HTTP routes over a pipe-delimited line protocol
  4at 115200 baud:
  5
  6- Request: one line ``route|header_text|body\\n``. ``header_text`` is
  7  HTTP-style ``Name: value`` lines separated by ``\\n``; literal ``|`` in
  8  headers/body is escaped as ``\\|``.
  9- Response: a line prefixed ``USB API RESPONSE-->`` followed by JSON:
 10  ``{"route": ..., "status": int, "headers": {...}, "body": "<string>"}``.
 11  The firmware prints console logs to the same port, so unrelated lines must
 12  be skipped.
 13
 14Requests over USB bypass HMAC entirely (the firmware trusts physical access),
 15so this transport never signs and needs no password.
 16
 17Streaming responses arrive fully rendered in the ``body`` field (the firmware
 18joins generators before sending), so a USB "stream" is one large chunk; large
 19transfers like ``/api/memory-snapshot`` are buffered in memory on both ends.
 20"""
 21
 22from __future__ import annotations
 23
 24import json
 25import time
 26from typing import Any, Iterator, List, Optional
 27
 28from ..exceptions import TransportError
 29from . import Transport, parse_body, raise_for_status, serialize_body
 30
 31BAUD_RATE = 115200
 32READ_TIMEOUT = 10.0
 33#: The device may reset when the port opens; give it time to come back.
 34SETTLE_SECONDS = 2.0
 35RESPONSE_PREFIX = "USB API RESPONSE-->"
 36#: Raspberry Pi USB vendor ID (the Pico 2W on the Vector board).
 37RASPBERRY_PI_VID = 0x2E8A
 38
 39
 40def _require_pyserial():
 41    try:
 42        import serial  # noqa: F401
 43        import serial.tools.list_ports  # noqa: F401
 44    except ImportError as exc:  # pragma: no cover - import guard
 45        raise ImportError(
 46            "pyserial is required for USB support; "
 47            "install with: pip install warpedpinball[usb]"
 48        ) from exc
 49    return serial
 50
 51
 52def list_serial_ports(all_ports: bool = False) -> List[str]:
 53    """List serial ports likely to be Vector boards.
 54
 55    Filters to the Raspberry Pi USB vendor ID (0x2E8A) unless ``all_ports``
 56    is true or no port carries VID information.
 57    """
 58    _require_pyserial()
 59    from serial.tools import list_ports
 60
 61    ports = list(list_ports.comports())
 62    if all_ports:
 63        return [p.device for p in ports]
 64    matches = [p.device for p in ports if getattr(p, "vid", None) == RASPBERRY_PI_VID]
 65    return matches
 66
 67
 68def escape_field(text: str) -> str:
 69    """Escape literal ``|`` as ``\\|`` for the request frame."""
 70    return text.replace("|", "\\|")
 71
 72
 73def build_frame(route: str, headers: Optional[dict] = None, body: str = "") -> bytes:
 74    """Build one request line: ``route|header_text|body\\n``."""
 75    header_text = "\n".join(f"{k}: {v}" for k, v in (headers or {}).items())
 76    line = "|".join(
 77        (escape_field(route), escape_field(header_text), escape_field(body or ""))
 78    )
 79    return (line + "\n").encode("utf-8")
 80
 81
 82def parse_response_line(line: str) -> dict:
 83    """Decode the JSON envelope after the ``USB API RESPONSE-->`` prefix."""
 84    payload = line.split(RESPONSE_PREFIX, 1)[1].strip()
 85    try:
 86        envelope = json.loads(payload)
 87    except ValueError as exc:
 88        raise TransportError(f"Malformed USB response: {payload[:200]!r}") from exc
 89    if not isinstance(envelope, dict) or "status" not in envelope:
 90        raise TransportError(f"Unexpected USB response envelope: {payload[:200]!r}")
 91    return envelope
 92
 93
 94class UsbTransport(Transport):
 95    """Talks to a USB-attached Vector over its serial console."""
 96
 97    requires_password = False  # firmware trusts physical access; no HMAC
 98
 99    def __init__(
100        self,
101        port: str,
102        timeout: float = READ_TIMEOUT,
103        settle: float = SETTLE_SECONDS,
104        _serial: Any = None,
105    ):
106        self.port = port
107        self.timeout = timeout
108        if _serial is not None:
109            self._serial = _serial  # injected fake for tests
110        else:  # pragma: no cover - opens a real serial port (needs hardware)
111            serial = _require_pyserial()
112            try:
113                self._serial = serial.Serial(port, BAUD_RATE, timeout=timeout)
114            except serial.SerialException as exc:
115                raise TransportError(f"Failed to open {port}: {exc}") from exc
116            if settle:
117                time.sleep(settle)
118
119    @property
120    def description(self) -> str:
121        return f"usb:{self.port}"
122
123    def close(self) -> None:
124        try:
125            self._serial.close()
126        except Exception:  # noqa: BLE001 - closing best-effort
127            pass
128
129    # -- internals ---------------------------------------------------------
130
131    def _exchange(self, path: str, body_str: Optional[str]) -> dict:
132        headers = {}
133        if body_str is not None:
134            # On-device body parsing requires this header for JSON bodies.
135            headers["Content-Type"] = "application/json"
136        frame = build_frame(path, headers, body_str or "")
137        try:
138            self._serial.reset_input_buffer()
139        except Exception:  # noqa: BLE001 - not all fakes/ports support it
140            pass
141        try:
142            self._serial.write(frame)
143        except Exception as exc:  # noqa: BLE001
144            raise TransportError(f"USB write to {self.port} failed: {exc}") from exc
145
146        deadline = time.monotonic() + self.timeout
147        while time.monotonic() < deadline:
148            try:
149                raw = self._serial.readline()
150            except Exception as exc:  # noqa: BLE001
151                raise TransportError(f"USB read from {self.port} failed: {exc}") from exc
152            if not raw:
153                continue  # read timeout tick; keep waiting until deadline
154            line = raw.decode("utf-8", errors="replace")
155            if RESPONSE_PREFIX in line:
156                return parse_response_line(line)
157            # otherwise: firmware console noise on the shared port, so skip it
158        raise TransportError(
159            f"Timed out waiting for USB response to {path} on {self.port}"
160        )
161
162    # -- Transport interface -------------------------------------------------
163
164    def request(self, path: str, body: Any = None, authenticated: bool = False) -> Any:
165        # ``authenticated`` is accepted for interface parity but ignored:
166        # the firmware skips HMAC for requests arriving over USB.
167        body_str = serialize_body(body)
168        envelope = self._exchange(path, body_str)
169        raise_for_status(int(envelope.get("status", 500)), envelope.get("body", ""), path)
170        return parse_body(envelope.get("body", ""))
171
172    def stream(
173        self, path: str, body: Any = None, authenticated: bool = False
174    ) -> Iterator[bytes]:
175        # Streams arrive fully rendered in the body field over USB; yield the
176        # whole thing as a single chunk (memory implication documented above).
177        body_str = serialize_body(body)
178        envelope = self._exchange(path, body_str)
179        raise_for_status(int(envelope.get("status", 500)), envelope.get("body", ""), path)
180        raw = envelope.get("body", "")
181        if isinstance(raw, str):
182            raw = raw.encode("utf-8")
183
184        def _iter() -> Iterator[bytes]:
185            if raw:
186                yield raw
187
188        return _iter()