warpedpinball.exceptions

  1"""Typed exceptions for the Warped Pinball Vector client."""
  2
  3from __future__ import annotations
  4
  5from typing import List, Optional
  6
  7
  8class VectorError(Exception):
  9    """Base class for all errors raised by this library."""
 10
 11
 12class TransportError(VectorError):
 13    """A connection, timeout, or protocol-level failure talking to the device."""
 14
 15
 16class DeviceUnreachableError(TransportError):
 17    """Could not open a connection to the device.
 18
 19    The address refused the connection, could not be resolved, or is not on
 20    the network at all (device powered off, wrong IP, different subnet). A
 21    subclass of :class:`TransportError`, so ``except TransportError`` still
 22    catches it. ``target`` is the host that was tried; ``cause`` is the
 23    underlying networking exception, if you want the gory details.
 24    """
 25
 26    def __init__(
 27        self,
 28        target: str,
 29        detail: Optional[str] = None,
 30        cause: Optional[BaseException] = None,
 31    ):
 32        self.target = target
 33        self.cause = cause
 34        message = (
 35            f"Could not reach the machine at {target} "
 36            f"(is it powered on and connected to the network?)"
 37        )
 38        if detail:
 39            message = f"{message}: {detail}"
 40        super().__init__(message)
 41
 42
 43class DeviceTimeoutError(TransportError):
 44    """The device accepted the connection but did not answer in time.
 45
 46    Usually the board is busy (mid-game, applying an update) or the link is
 47    flaky. A subclass of :class:`TransportError`. ``target`` is the host,
 48    ``timeout`` the seconds waited, and ``cause`` the underlying exception.
 49    """
 50
 51    def __init__(
 52        self,
 53        target: str,
 54        timeout: Optional[float] = None,
 55        cause: Optional[BaseException] = None,
 56    ):
 57        self.target = target
 58        self.timeout = timeout
 59        self.cause = cause
 60        if timeout is not None:
 61            message = (
 62                f"The machine at {target} did not respond within {timeout:g}s "
 63                f"(it may be busy or the connection is unstable)"
 64            )
 65        else:
 66            message = f"The machine at {target} did not respond in time"
 67        super().__init__(message)
 68
 69
 70class MachineNotFoundError(VectorError):
 71    """No machine matching the requested name was found during discovery.
 72
 73    ``seen_names`` lists the machine names that *were* discovered, so callers
 74    (and error messages) can show what is actually on the network.
 75    """
 76
 77    def __init__(self, name: str, seen_names: Optional[List[str]] = None):
 78        self.name = name
 79        self.seen_names = seen_names or []
 80        seen = ", ".join(self.seen_names) if self.seen_names else "none"
 81        super().__init__(
 82            f"No machine named {name!r} found on the network (machines seen: {seen})"
 83        )
 84
 85
 86class AmbiguousMachineError(VectorError):
 87    """More than one discovered machine matched the requested name."""
 88
 89    def __init__(self, name: str, candidates: List[str]):
 90        self.name = name
 91        self.candidates = candidates
 92        super().__init__(
 93            f"Machine name {name!r} is ambiguous; candidates: {', '.join(candidates)}"
 94        )
 95
 96
 97class AuthenticationRequiredError(VectorError):
 98    """An authenticated route was called with no password configured.
 99
100    Raised before any network traffic. Set ``machine.password``, pass
101    ``password=`` to ``connect()``, or set the ``VECTOR_PASSWORD`` env var.
102    """
103
104
105class AuthenticationError(VectorError):
106    """The device rejected the request's credentials (HTTP 401).
107
108    ``reason`` carries the device's ``{"error": ...}`` detail, e.g.
109    ``"Bad Credentials"`` or ``"Challenge expired"``.
110    """
111
112    def __init__(self, reason: str = "authentication failed"):
113        self.reason = reason
114        super().__init__(reason)
115
116
117class RateLimitedError(VectorError):
118    """The device returned 429 while fetching an auth challenge.
119
120    The device holds at most ~10 outstanding challenges; expired ones are
121    purged on each challenge request, so retrying after a short sleep is safe.
122    """
123
124
125class CooldownError(VectorError):
126    """The route is locked or in cooldown (HTTP 409 "Already running" / 429).
127
128    ``retry_after`` is a best-effort hint in seconds based on the route's
129    documented server-side cooldown (``/api/logs`` 10 s, ``/api/update/check``
130    10 s, ``/api/adjustments/restore`` 5 s), or ``None`` when unknown.
131    """
132
133    def __init__(self, message: str, retry_after: Optional[float] = None):
134        self.retry_after = retry_after
135        if retry_after is not None:
136            message = f"{message} (retry after ~{retry_after:g}s)"
137        super().__init__(message)
138
139
140class VectorServerError(VectorError):
141    """The device handler raised an error (HTTP 5xx). Body carries the detail."""
142
143    def __init__(self, message: str, status: int = 500):
144        self.status = status
145        super().__init__(message)
146
147
148class UnsupportedFirmwareError(VectorError):
149    """The route does not exist on this firmware version (HTTP 404)."""
150
151    def __init__(self, path: str, firmware_version: Optional[str] = None):
152        self.path = path
153        self.firmware_version = firmware_version
154        detail = f" (device firmware: {firmware_version})" if firmware_version else ""
155        super().__init__(
156            f"Route {path!r} is not supported by this firmware{detail}; "
157            "a firmware update may be required"
158        )