warpedpinball

 99def connect(
100    name_or_ip: str,
101    password: Optional[str] = None,
102    timeout: float = 20.0,
103    http_timeout: float = 10.0,
104) -> Machine:
105    """Connect to a machine by LAN name (via UDP discovery) or by IP address.
106
107    ``timeout`` is how long (seconds) to listen for the named board during UDP
108    discovery; it defaults to 20 s so slow-to-answer boards are still found,
109    and is ignored when connecting straight to an IP. Name matching is
110    case-insensitive: exact match first, then unique prefix/substring. Raises
111    :class:`MachineNotFoundError` (listing the names that *were* seen) or
112    :class:`AmbiguousMachineError` (listing candidates).
113    """
114    machine_name: Optional[str] = None
115    if _is_ip(name_or_ip):
116        host = name_or_ip
117    else:
118        found = discover(timeout=timeout, name=name_or_ip)
119        match = _match_by_name(name_or_ip, found)
120        host = match.ip
121        machine_name = match.name
122
123    transport = HttpTransport(host, password=password, timeout=http_timeout)
124    return Machine(transport, password=password, name=machine_name)

127def connect_usb(
128    port: Optional[str] = None,
129    timeout: float = 10.0,
130) -> Machine:
131    """Connect to a USB-attached machine.
132
133    With no ``port``, auto-picks when exactly one candidate serial port
134    (Raspberry Pi VID) is present. Authenticated routes need no password over
135    USB; the firmware trusts physical access.
136    """
137    from .transports.usb import UsbTransport
138    from .transports.usb import list_serial_ports as _lsp
139
140    if port is None:
141        candidates = _lsp()
142        if len(candidates) == 1:
143            port = candidates[0]
144        elif not candidates:
145            raise MachineNotFoundError("usb", seen_names=[])
146        else:
147            raise AmbiguousMachineError("usb", candidates)
148
149    return Machine(UsbTransport(port, timeout=timeout))

236def discover(
237    timeout: float = 20.0,
238    name: Optional[str] = None,
239) -> List[DiscoveredMachine]:
240    """Find Vector boards on the LAN; returns a list of :class:`DiscoveredMachine`.
241
242    Probes (PING + OFFLINE) go out every interface each ~1.5 s -- matching how
243    often boards service discovery -- and the first FULL reply ends discovery
244    immediately, since it is the registry's complete list. Boards heard only
245    directly (a HELLO or a unicast PONG) mean broadcasts toward us are being
246    filtered; after a short settle the complete list is fetched from a heard
247    board over HTTP instead. The usual result is everything within about two
248    seconds; ``timeout`` (default 20 s) is the worst-case cap for genuinely
249    quiet networks. ``name`` makes discovery return the instant that exact
250    board appears. Results are deduplicated by IP. Never registers this client
251    in the boards' own peer lists (that would take a HELLO, which is not sent).
252    """
253    sock = _open_socket()
254    local_ip = _local_ip()
255    local_ips = _local_ips()
256    probe_socks = _probe_sockets(local_ips)
257    targets = [BROADCAST_ADDR] + _directed_broadcasts(local_ips)
258    probes = [build_ping(), build_offline(local_ip)]
259
260    found: Dict[str, DiscoveredMachine] = {}  # via FULL frames (dedup by IP)
261    sightings: Dict[str, Optional[str]] = {}  # boards heard directly: ip -> name?
262    first_sighting: Optional[float] = None
263    target = name.lower() if name else None
264    deadline = time.monotonic() + timeout
265    next_broadcast = 0.0
266    try:
267        while True:
268            now = time.monotonic()
269            if now >= deadline:
270                break
271            if first_sighting is not None and now - first_sighting >= SETTLE_AFTER_SIGHTING:
272                break  # boards heard but their FULL isn't reaching us; go ask over HTTP
273            if now >= next_broadcast:
274                for frame in probes:
275                    for out in [sock] + probe_socks:
276                        for addr in targets:
277                            try:
278                                out.sendto(frame, (addr, DISCOVERY_PORT))
279                            except OSError:
280                                pass  # interface without a broadcast route
281                next_broadcast = now + REBROADCAST_INTERVAL
282            wait = min(0.5, deadline - now)
283            if first_sighting is not None:
284                wait = min(wait, first_sighting + SETTLE_AFTER_SIGHTING - now)
285            sock.settimeout(max(wait, 0.01))
286            try:
287                data, addr = sock.recvfrom(4096)
288            except socket.timeout:
289                continue
290            except ConnectionResetError:
291                # Windows only: a previously *sent* datagram bounced with ICMP
292                # port-unreachable, and Windows reports it on the next receive
293                # (WSAECONNRESET). Not a socket failure; keep listening.
294                continue
295            except OSError:
296                break
297
298            for machine in parse_full(data):
299                found[machine.ip] = machine
300                if target and machine.name.lower() == target:
301                    return list(found.values())
302            # A FULL frame is the registry's complete list of known boards, so
303            # once we have one there is nothing more to wait for -- return
304            # instead of burning the rest of the (deliberately long) timeout.
305            if len(data) >= 2 and data[0] == MSG_FULL:
306                return list(found.values())
307
308            src = addr[0]
309            if src == local_ip or src in local_ips:
310                continue  # our own probe looped back
311            hello_name = parse_hello(data)
312            if hello_name is not None:
313                sightings[src] = hello_name
314                if target and hello_name.lower() == target:
315                    found[src] = DiscoveredMachine(ip=src, name=hello_name)
316                    return list(found.values())
317            elif data[:1] == bytes([MSG_PONG]):
318                sightings.setdefault(src, None)
319            else:
320                continue
321            if first_sighting is None:
322                first_sighting = time.monotonic()
323    finally:
324        sock.close()
325        for out in probe_socks:
326            out.close()
327
328    # Boards were heard directly but no FULL broadcast made it through: get the
329    # complete list over unicast HTTP from the first heard board that answers.
330    for ip in sightings:
331        try:
332            for machine in _peers_over_http(ip):
333                found.setdefault(machine.ip, machine)
334            break  # one board's peer table is the whole picture
335        except Exception:
336            continue  # that board wouldn't talk HTTP; try the next sighting
337    for ip, seen_name in sightings.items():
338        found.setdefault(ip, DiscoveredMachine(ip=ip, name=seen_name or ""))
339    return list(found.values())

152def list_serial_ports(all_ports: bool = False) -> List[str]:
153    """List serial ports likely to be Vector boards (requires the usb extra)."""
154    from .transports.usb import list_serial_ports as _lsp
155
156    return _lsp(all_ports=all_ports)

 49class Machine:
 50    """A connected Vector board.
 51
 52    Usually built via :func:`warpedpinball.connect` /
 53    :func:`warpedpinball.connect_usb` rather than directly.
 54    """
 55
 56    def __init__(
 57        self,
 58        transport: Transport,
 59        password: Optional[str] = None,
 60        name: Optional[str] = None,
 61    ):
 62        self.transport = transport
 63        self._password = password
 64        self.name = name
 65        self._lock = threading.RLock()
 66        self._firmware_version: Optional[str] = None
 67
 68    # -- lifecycle -----------------------------------------------------------
 69
 70    def __enter__(self) -> "Machine":
 71        return self
 72
 73    def __exit__(self, *exc) -> None:
 74        self.close()
 75
 76    def close(self) -> None:
 77        self.transport.close()
 78
 79    def __repr__(self) -> str:
 80        label = self.name or "?"
 81        return f"<Machine {label!r} via {self.transport.description}>"
 82
 83    # -- credentials -----------------------------------------------------------
 84
 85    @property
 86    def password(self) -> Optional[str]:
 87        """Password for HMAC auth; falls back to $VECTOR_PASSWORD.
 88
 89        An empty string is a valid password (some boards have no password
 90        configured, which the firmware signs with an empty HMAC key) and is
 91        kept distinct from ``None``, which means no password has been set and
 92        triggers the ``$VECTOR_PASSWORD`` fallback.
 93        """
 94        if self._password is not None:
 95            return self._password
 96        return os.environ.get(PASSWORD_ENV_VAR)
 97
 98    @password.setter
 99    def password(self, value: Optional[str]) -> None:
100        self._password = value
101
102    def verify_password(self) -> bool:
103        """Validate credentials up front via ``/api/auth/password_check``."""
104        from .exceptions import AuthenticationError
105
106        try:
107            self.call("/api/auth/password_check", authenticated=True)
108            return True
109        except AuthenticationError:
110            return False
111
112    # -- raw escape hatch ------------------------------------------------------
113
114    def call(self, path: str, body: Any = None, authenticated: bool = False) -> Any:
115        """Perform one request and return the parsed response.
116
117        Handles auth signing, serialization, and error mapping. Every firmware
118        route is reachable through this, including ones with no wrapper yet.
119        """
120        with self._lock:
121            self._preflight_auth(path, authenticated)
122            self.transport.password = self.password
123            return self.transport.request(path, body=body, authenticated=authenticated)
124
125    def call_stream(
126        self, path: str, body: Any = None, authenticated: bool = False
127    ) -> Iterator[bytes]:
128        """Like :meth:`call` but returns an iterator of raw bytes chunks.
129
130        The device is fully consumed while iterating; the per-machine lock is
131        held until the iterator is exhausted or closed.
132        """
133        self._preflight_auth(path, authenticated)
134
135        def _locked_iter() -> Iterator[bytes]:
136            with self._lock:
137                self.transport.password = self.password
138                yield from self.transport.stream(
139                    path, body=body, authenticated=authenticated
140                )
141
142        return _locked_iter()
143
144    def _preflight_auth(self, path: str, authenticated: bool) -> None:
145        if (
146            authenticated
147            and self.transport.requires_password
148            and self.password is None
149        ):
150            raise AuthenticationRequiredError(
151                f"Route {path!r} requires authentication but no password is set; "
152                f"pass password= to connect(), set machine.password, or set "
153                f"${PASSWORD_ENV_VAR}"
154            )
155
156    # -- device info -----------------------------------------------------------
157
158    def version(self) -> Any:
159        """Firmware version from ``/api/version`` (also cached for error messages)."""
160        result = self.call("/api/version")
161        if isinstance(result, dict):
162            self._firmware_version = str(
163                result.get("version") or result.get("Version") or result
164            )
165        elif result is not None:
166            self._firmware_version = str(result)
167        return result
168
169    def machine_id(self) -> Any:
170        """Board identity from ``/api/machine_id``."""
171        return self.call("/api/machine_id")
172
173    def game_name(self) -> Any:
174        """Configured game/machine name from ``/api/game/name``."""
175        return self.call("/api/game/name")
176
177    def game_status(self) -> Any:
178        """Current gameplay status (balls, scores, active flag) from ``/api/game/status``."""
179        return self.call("/api/game/status")
180
181    def active_config(self) -> Any:
182        """Active machine configuration from ``/api/game/active_config``."""
183        return self.call("/api/game/active_config")
184
185    def wifi_status(self) -> Any:
186        """Wi-Fi connection status from ``/api/wifi/status``."""
187        return self.call("/api/wifi/status")
188
189    def faults(self) -> Any:
190        """Reported hardware/firmware faults from ``/api/fault``."""
191        return self.call("/api/fault")
192
193    def peers(self) -> Any:
194        """Peer table from ``GET /api/network/peers``: discovery without
195        broadcast (useful across VLANs when you know one IP)."""
196        return self.call("/api/network/peers")
197
198    # -- power -----------------------------------------------------------------
199
200    def reboot_game(self) -> Any:
201        """Power-cycle the pinball machine itself."""
202        return self._call_gated("/api/game/reboot", authenticated=True)
203
204    def reboot(self) -> Any:
205        """Reboot the Vector board."""
206        return self._call_gated("/api/settings/reboot", authenticated=True)
207
208    def wait_until_reachable(self, timeout: float = 120.0, interval: float = 2.0) -> Any:
209        """Poll ``/api/version`` until the board answers (after reboot/update)."""
210        deadline = time.monotonic() + timeout
211        last_error: Optional[Exception] = None
212        while time.monotonic() < deadline:
213            try:
214                return self.version()
215            except VectorError as exc:
216                last_error = exc
217                time.sleep(interval)
218        raise TransportError(
219            f"Machine did not become reachable within {timeout:g}s"
220        ) from last_error
221
222    # -- scores / players --------------------------------------------------------
223
224    def leaderboard(self) -> Any:
225        """High-score leaderboard from ``/api/leaders``."""
226        return self.call("/api/leaders")
227
228    def tournament(self) -> Any:
229        """Tournament standings from ``/api/tournament``."""
230        return self.call("/api/tournament")
231
232    def reset_leaderboard(self) -> Any:
233        """Clear the leaderboard via ``/api/leaders/reset`` (authenticated)."""
234        return self._call_gated("/api/leaders/reset", authenticated=True)
235
236    def reset_tournament(self) -> Any:
237        """Clear tournament standings via ``/api/tournament/reset`` (authenticated)."""
238        return self._call_gated("/api/tournament/reset", authenticated=True)
239
240    def claimable_scores(self) -> Any:
241        """Scores awaiting a player claim from ``/api/scores/claimable``."""
242        return self.call("/api/scores/claimable")
243
244    def claim_score(self, initials: str, player_index: int, score: int) -> Any:
245        """Claim a pending score for ``initials`` via ``/api/scores/claim``."""
246        return self.call(
247            "/api/scores/claim",
248            body={"initials": initials, "player_index": player_index, "score": score},
249        )
250
251    def players(self) -> Any:
252        """Registered players from ``/api/players``."""
253        return self.call("/api/players")
254
255    def update_player(
256        self, id: int, initials: str, full_name: Optional[str] = None
257    ) -> Any:
258        """Create/update a player's initials (and optional full name) via
259        ``/api/player/update`` (authenticated)."""
260        body: Dict[str, Any] = {"id": id, "initials": initials}
261        if full_name is not None:
262            body["full_name"] = full_name
263        return self._call_gated("/api/player/update", body=body, authenticated=True)
264
265    def export_scores(self) -> Any:
266        """Export all stored scores from ``/api/export/scores``."""
267        return self.call("/api/export/scores")
268
269    def import_scores(self, data: Any) -> Any:
270        """Import a previously exported score payload via ``/api/import/scores``
271        (authenticated)."""
272        return self._call_gated("/api/import/scores", body=data, authenticated=True)
273
274    # -- updates -----------------------------------------------------------------
275
276    def check_for_updates(self) -> Any:
277        """``/api/update/check``. Note the 10 s server-side cooldown."""
278        return self.call("/api/update/check")
279
280    def apply_update(
281        self,
282        url: Optional[str] = None,
283        progress: Optional[Callable[[dict], None]] = None,
284    ) -> List[dict]:
285        """Apply a firmware update, streaming progress.
286
287        When ``url`` is omitted it is taken from :meth:`check_for_updates`.
288        ``progress`` receives each ``{"log": ..., "percent": ...}`` line as it
289        streams. Returns the list of all progress records.
290        """
291        if url is None:
292            info = self.check_for_updates()
293            if isinstance(info, dict):
294                url = info.get("url")
295            if not url:
296                raise VectorError(
297                    "check_for_updates() did not report an update URL; "
298                    "pass url= explicitly"
299                )
300        stream = self.call_stream(
301            "/api/update/apply", body={"url": url}, authenticated=True
302        )
303        records: List[dict] = []
304        for line in _iter_lines(stream):
305            try:
306                record = json.loads(line)
307            except ValueError:
308                record = {"log": line}
309            records.append(record)
310            if progress is not None:
311                progress(record)
312        return records
313
314    # -- clock --------------------------------------------------------------------
315
316    def date(self) -> _dt.datetime:
317        """Read the device RTC as a :class:`datetime.datetime`."""
318        raw = self.call("/api/get_date")
319        return _parse_device_date(raw)
320
321    def set_date(self, when: Optional[_dt.datetime] = None) -> Any:
322        """Set the device RTC (defaults to the local clock now).
323
324        Sends ``[year, month, day, hour, minute, second]``; the firmware
325        derives the weekday itself via ``RTC.datetime()``.
326        """
327        when = when or _dt.datetime.now()
328        date_list = [
329            when.year,
330            when.month,
331            when.day,
332            when.hour,
333            when.minute,
334            when.second,
335        ]
336        return self._call_gated(
337            "/api/set_date", body={"date": date_list}, authenticated=True
338        )
339
340    # -- logs ------------------------------------------------------------------------
341
342    def logs(self) -> Iterator[bytes]:
343        """Stream the device log (authenticated; 10 s server-side cooldown)."""
344        return self.call_stream("/api/logs", authenticated=True)
345
346    # -- adjustments -------------------------------------------------------------------
347
348    def adjustments(self) -> Any:
349        """Saved adjustment profiles and their status from ``/api/adjustments/status``."""
350        return self.call("/api/adjustments/status")
351
352    def capture_adjustments(self, index: int) -> Any:
353        """Capture the machine's current adjustments into profile ``index``
354        via ``/api/adjustments/capture`` (authenticated)."""
355        return self._call_gated(
356            "/api/adjustments/capture", body={"index": index}, authenticated=True
357        )
358
359    def restore_adjustments(self, index: int) -> Any:
360        """Restore a captured adjustment profile (5 s server-side cooldown)."""
361        return self._call_gated(
362            "/api/adjustments/restore", body={"index": index}, authenticated=True
363        )
364
365    def name_adjustment(self, index: int, name: str) -> Any:
366        """Label adjustment profile ``index`` via ``/api/adjustments/name``
367        (authenticated)."""
368        return self._call_gated(
369            "/api/adjustments/name", body={"index": index, "name": name},
370            authenticated=True,
371        )
372
373    # -- memory ------------------------------------------------------------------------
374
375    def read(self, offset: int, count: int = 1, byteorder: str = "big") -> int:
376        """Read ``count`` bytes at ``offset`` and decode them as an unsigned int.
377
378        A convenience wrapper over :meth:`read_bytes` for the common case of
379        reading a small numeric value (a credit count, a hurry-up timer, a
380        flag). Defaults to a single byte; for a multi-byte value pass
381        ``byteorder="little"`` if the region is little-endian.
382        """
383        return int.from_bytes(self.read_bytes(offset, count), byteorder)
384
385    def read_bytes(self, offset: int, count: int) -> bytes:
386        """Bulk SRAM read, auto-chunked at 256 bytes per request."""
387        out = bytearray()
388        remaining = count
389        pos = offset
390        while remaining > 0:
391            chunk = min(remaining, ADDRESS_CHUNK)
392            result = self.call(
393                "/api/address/read",
394                body={"offset": pos, "count": chunk},
395                authenticated=True,
396            )
397            values = result["values"] if isinstance(result, dict) else result
398            out.extend(values)
399            pos += chunk
400            remaining -= chunk
401        return bytes(out)
402
403    def write_bytes(self, offset: int, data: Union[bytes, bytearray, List[int]]) -> None:
404        """Bulk SRAM write, auto-chunked at 256 bytes per request."""
405        data = bytes(data)
406        pos = 0
407        while pos < len(data):
408            chunk = data[pos : pos + ADDRESS_CHUNK]
409            self.call(
410                "/api/address/write",
411                body={"offset": offset + pos, "values": list(chunk)},
412                authenticated=True,
413            )
414            pos += len(chunk)
415
416    def memory_snapshot(self) -> bytes:
417        """Full SRAM dump via the streamed ``/api/memory-snapshot`` route."""
418        return b"".join(self.call_stream("/api/memory-snapshot"))
419
420    def set_memory_broadcast(
421        self, enabled: bool, frequency_ms: int = 100, ip: Optional[str] = None
422    ) -> Any:
423        """Toggle UDP streaming of memory snapshots to one client. Authenticated.
424
425        When enabled, the firmware sends the whole SRAM data region as UDP
426        packets to port 2040 on a single target IP every ``frequency_ms``
427        milliseconds (each packet is a 4-byte big-endian offset header
428        followed by up to 256 bytes of data) -- the live stream tools like
429        Warped Pinball's Memory Mapper listen to. The stream goes only to
430        that one address, never to the whole network, and the board keeps at
431        most one stream target at a time.
432
433        ``ip`` is the IPv4 address to stream to; when omitted the firmware
434        streams back to the requester (this machine), which is what you want
435        when the listener runs where this code runs. Over USB there is no
436        requester IP, so pass ``ip`` explicitly. ``frequency_ms`` is clamped
437        to the firmware's 10-60000 ms bounds; both extras are ignored when
438        disabling.
439        """
440        if enabled:
441            frequency_ms = max(10, min(60000, int(frequency_ms)))
442            body: Dict[str, Any] = {"enable": True, "frequency_ms": frequency_ms}
443            if ip is not None:
444                body["ip"] = ip
445        else:
446            body = {"enable": False}
447        return self._call_gated(
448            "/api/memory/toggle-broadcast", body=body, authenticated=True
449        )
450
451    @staticmethod
452    def diff_snapshots(a: bytes, b: bytes) -> List[Tuple[int, int, int]]:
453        """Compare two snapshots; returns ``(offset, a_value, b_value)`` per
454        changed byte (a length difference shows up as changes vs. -1)."""
455        changes: List[Tuple[int, int, int]] = []
456        for i in range(max(len(a), len(b))):
457            va = a[i] if i < len(a) else -1
458            vb = b[i] if i < len(b) else -1
459            if va != vb:
460                changes.append((i, va, vb))
461        return changes
462
463    # -- formats -----------------------------------------------------------------------
464
465    def formats(self) -> Any:
466        """Available game formats from ``/api/formats/available``."""
467        return self.call("/api/formats/available")
468
469    def active_format(self) -> Any:
470        """The format the machine is currently playing, from ``/api/formats/active``."""
471        return self.call("/api/formats/active")
472
473    def set_format(self, format_id: Any, options: Optional[Dict[str, Any]] = None) -> Any:
474        """Activate a game format via ``/api/formats/set`` (authenticated).
475
476        ``format_id`` is the numeric id (or the name) of one of the formats
477        :meth:`formats` returned. ``options`` are that format's configurable
478        settings, shaped like the ``Options`` block in the format's metadata.
479        """
480        body: Dict[str, Any] = {"format_id": format_id}
481        if options:
482            # The firmware reads this key capitalized, matching the casing it
483            # uses when it hands the options back out of /api/formats/available.
484            body["Options"] = options
485        return self._call_gated("/api/formats/set", body=body, authenticated=True)
486
487    # -- origin messages ---------------------------------------------------------------
488
489    def set_origin_target(self, secret: str, ip: Optional[str] = None) -> Any:
490        """Register where the board unicasts its Origin messages. Authenticated.
491
492        The board pushes live game events (game state, end of game, reset) as
493        UDP datagrams to port 6809. Until a target is registered it sends
494        nothing at all; once registered it sends *only* to this one address,
495        signing every datagram with ``secret`` (see :mod:`warpedpinball.origin`
496        for the frame layout and :func:`warpedpinball.origin.new_secret` for
497        generating one).
498
499        ``ip`` is the IPv4 address to send to; when omitted the board uses the
500        address this request arrived from, which is what you want whenever the
501        listener runs where this code runs -- including behind NAT, where the
502        board sees the translated address and the listener could not have
503        named it. Over USB there is no requester address, so pass ``ip``.
504
505        Registering again rotates the secret and resets the board's datagram
506        counter, so a listener that restarts simply re-registers.
507        """
508        body: Dict[str, Any] = {"enable": True, "secret": secret}
509        if ip is not None:
510            body["ip"] = ip
511        return self._call_gated("/api/origin/target", body=body, authenticated=True)
512
513    def clear_origin_target(self) -> Any:
514        """Stop the board sending Origin messages anywhere. Authenticated."""
515        return self._call_gated(
516            "/api/origin/target", body={"enable": False}, authenticated=True
517        )
518
519    # -- polling -----------------------------------------------------------------------
520
521    def watch_game(self, interval: float = 1.0) -> Iterator[GameEvent]:
522        """Poll ``/api/game/status`` and yield change events forever.
523
524        Keep ``interval`` >= 0.5 s to be kind to the device (enforced).
525        Change detection is heuristic over common status keys; anything else
526        that changes yields a generic ``status_changed`` event.
527        """
528        interval = max(interval, 0.5)
529        prev: Any = None
530        while True:
531            status = self.game_status()
532            if prev is not None and status != prev:
533                yield from _diff_status(prev, status)
534            prev = status
535            time.sleep(interval)
536
537    # -- internals -----------------------------------------------------------------------
538
539    def _call_gated(
540        self, path: str, body: Any = None, authenticated: bool = False
541    ) -> Any:
542        """call() that names the firmware version on 404 (route missing)."""
543        try:
544            return self.call(path, body=body, authenticated=authenticated)
545        except UnsupportedFirmwareError as exc:
546            raise UnsupportedFirmwareError(
547                exc.path, firmware_version=self._firmware_version
548            ) from None