warpedpinball.transports

  1"""Transport layer: the abstract interface shared by HTTP and USB transports."""
  2
  3from __future__ import annotations
  4
  5import json
  6from abc import ABC, abstractmethod
  7from typing import Any, Iterator, Optional
  8
  9from ..exceptions import (
 10    AuthenticationError,
 11    CooldownError,
 12    UnsupportedFirmwareError,
 13    VectorServerError,
 14)
 15
 16#: Documented server-side cooldowns (seconds) used to enrich CooldownError.
 17ROUTE_COOLDOWNS = {
 18    "/api/logs": 10.0,
 19    "/api/update/check": 10.0,
 20    "/api/adjustments/restore": 5.0,
 21}
 22
 23
 24def serialize_body(body: Any) -> Optional[str]:
 25    """Serialize a request body to the exact string sent (and signed).
 26
 27    Strings pass through untouched; dicts/lists are JSON-encoded compactly,
 28    exactly once: the same string must be both signed and transmitted.
 29    """
 30    if body is None:
 31        return None
 32    if isinstance(body, str):
 33        return body
 34    return json.dumps(body, separators=(",", ":"))
 35
 36
 37def parse_body(text: Any) -> Any:
 38    """Parse a response body: JSON when it looks like JSON, else the raw text."""
 39    if not isinstance(text, str):
 40        return text
 41    stripped = text.strip()
 42    if not stripped:
 43        return None
 44    if stripped[0] in "{[" or stripped in ("true", "false", "null") or _is_number(stripped):
 45        try:
 46            return json.loads(stripped)
 47        except ValueError:
 48            pass
 49    return text
 50
 51
 52def _is_number(s: str) -> bool:
 53    try:
 54        float(s)
 55        return True
 56    except ValueError:
 57        return False
 58
 59
 60def extract_error_detail(body: Any) -> str:
 61    """Pull the device's error string out of a response body."""
 62    parsed = parse_body(body)
 63    if isinstance(parsed, dict):
 64        for key in ("error", "message", "detail"):
 65            if key in parsed:
 66                return str(parsed[key])
 67    if parsed is None:
 68        return ""
 69    return str(parsed)
 70
 71
 72def raise_for_status(status: int, body: Any, path: str) -> None:
 73    """Map a non-2xx device response to a typed exception.
 74
 75    Note: 429 on the challenge route is handled separately by the HTTP
 76    transport (RateLimitedError with internal retry); a 429 that reaches here
 77    is a route cooldown.
 78    """
 79    if 200 <= status < 300:
 80        return
 81    detail = extract_error_detail(body)
 82    if status == 401:
 83        raise AuthenticationError(detail or "Unauthorized")
 84    if status == 404:
 85        raise UnsupportedFirmwareError(path)
 86    if status in (409, 429):
 87        hint = ROUTE_COOLDOWNS.get(path.split("?", 1)[0])
 88        msg = detail or ("Already running" if status == 409 else "Rate limited")
 89        raise CooldownError(f"{path}: {msg}", retry_after=hint)
 90    if status >= 500:
 91        raise VectorServerError(detail or f"Device error on {path}", status=status)
 92    raise VectorServerError(
 93        detail or f"Unexpected status {status} from {path}", status=status
 94    )
 95
 96
 97class Transport(ABC):
 98    """Abstract transport. HTTP and USB implement the same interface, so all
 99    ``Machine`` methods work identically over both."""
100
101    #: True when authenticated routes need a password on this transport.
102    #: (USB bypasses HMAC entirely; the firmware trusts physical access.)
103    requires_password: bool = True
104
105    #: Password used for HMAC signing (ignored by transports that don't sign).
106    password: Optional[str] = None
107
108    @abstractmethod
109    def request(
110        self, path: str, body: Any = None, authenticated: bool = False
111    ) -> Any:
112        """Perform one request; return the parsed response body.
113
114        Raises a typed exception from ``warpedpinball.exceptions`` on error.
115        """
116
117    @abstractmethod
118    def stream(
119        self, path: str, body: Any = None, authenticated: bool = False
120    ) -> Iterator[bytes]:
121        """Perform a streaming request; yield raw body chunks as bytes."""
122
123    @abstractmethod
124    def close(self) -> None:
125        """Release sockets / serial ports."""
126
127    @property
128    @abstractmethod
129    def description(self) -> str:
130        """Human-readable target, e.g. ``http://192.168.1.42`` or ``/dev/ttyACM0``."""
131
132    def __enter__(self) -> "Transport":
133        return self
134
135    def __exit__(self, *exc) -> None:
136        self.close()