From 70d7ff363cec6ba9d3146a28c5ec0ae9b7299cb5 Mon Sep 17 00:00:00 2001 From: btrippcsci Date: Fri, 9 Oct 2026 11:45:35 -0400 Subject: [PATCH 1/2] feat(phone): add network shot stream, club API and reconnect catch-up Adds the network transport for phone apps and catch-up of missed shots after a reconnect, on top of the BLE transport from #285. - GET /api/shots/stream serves the schema 2 phone messages as Server-Sent Events: current club, profiles and power status, then the session's shots, then live shots and events, with a 15 s heartbeat. Up to eight clients; 503 beyond that. - GET/POST /api/club reads and sets the club through apply_club_selection, and club changes are broadcast on the stream. - A reconnecting phone names the newest shot it has (last_event_id in BLE hello, or Last-Event-ID on the stream) and is sent that shot and every later current-session shot, oldest first, up to 20. Replays are the exact bytes sent live; cleared and deleted shots are never replayed. hello advertises the shot_catch_up feature. Co-Authored-By: Claude Opus 5.5 --- docs/changelog.md | 13 + docs/ios-ble.md | 160 ++++++++-- src/openflight/ble/protocol.py | 1 + src/openflight/ble/publisher.py | 101 ++++++- src/openflight/phone_catch_up.py | 97 ++++++ src/openflight/server.py | 120 +++++++- src/openflight/shot_stream.py | 228 ++++++++++++++ tests/ble_harness.py | 5 + .../ble_goldens/v2_response_hello.json | 50 ++-- tests/test_ble_catch_up.py | 277 ++++++++++++++++++ tests/test_control_commands.py | 29 +- tests/test_phone_catch_up.py | 113 +++++++ tests/test_phone_transport.py | 174 ++++++++--- tests/test_phone_transport_server.py | 39 ++- tests/test_server.py | 134 +++++++++ tests/test_shot_stream.py | 207 +++++++++++++ tests/test_shot_stream_catch_up.py | 215 ++++++++++++++ 17 files changed, 1860 insertions(+), 103 deletions(-) create mode 100644 src/openflight/phone_catch_up.py create mode 100644 src/openflight/shot_stream.py create mode 100644 tests/test_ble_catch_up.py create mode 100644 tests/test_phone_catch_up.py create mode 100644 tests/test_shot_stream.py create mode 100644 tests/test_shot_stream_catch_up.py diff --git a/docs/changelog.md b/docs/changelog.md index f95d9a6eb..ed9101f02 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -70,6 +70,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 full horizontal speed, overstating attack angle on any shot with club path. ### Added +- **Network shot stream and club API for phone apps, and catch-up after a + reconnect.** `GET /api/shots/stream` sends the same schema 2 messages as BLE + as Server-Sent Events: a stream opens with the current club, profiles and + (with `--battery`) power status, then the session's shots, then live shots + and events, with a heartbeat every 15 s. It needs no flag and accepts up to + eight clients (`503` beyond that). `GET`/`POST /api/club` reads and sets the + club through the same server operation as the kiosk and BLE, and club + changes are broadcast on the stream too. A reconnecting phone names the + newest shot it has (`last_event_id` in BLE `hello`, or `Last-Event-ID` on the + stream) and is sent that shot again plus every later current-session shot, + oldest first, up to 20; replays are the exact bytes sent live, and cleared or + deleted shots are never replayed. `hello` advertises the `shot_catch_up` + feature. See [phone app connection](ios-ble.md#catch-up-after-a-reconnect). - **Bluetooth LE connection for phone apps.** `--ble` (with the optional `ble` extra, `bless==0.3.0` on Linux) advertises one GATT service with a shot and a control characteristic, speaking schema version 2; without BlueZ or the diff --git a/docs/ios-ble.md b/docs/ios-ble.md index d3ca0f4b5..e811dc53c 100644 --- a/docs/ios-ble.md +++ b/docs/ios-ble.md @@ -1,12 +1,13 @@ -# Phone app connection (Bluetooth LE) +# Phone app connection (Bluetooth LE and network) > **BLE blocker — check the Raspberry Pi kernel first:** Raspberry Pi kernel > `6.18.34+rpt-rpi-2712` has a confirmed regression that rejects every BLE -> advertisement. Run `uname -r` on the Pi. If it reports that version, boot a -> working kernel such as 6.12.x; there is no userspace workaround. See the [full diagnosis](#known-bad-raspberry-pi-kernel-61834rpt-rpi-2712). +> advertisement. Run `uname -r` on the Pi. If it reports that version, use the +> network transport or boot a working kernel such as 6.12.x; there is no userspace workaround. See the [full diagnosis](#known-bad-raspberry-pi-kernel-61834rpt-rpi-2712). OpenFlight sends each completed shot from a Raspberry Pi to a phone app over -Bluetooth LE, using the versioned payload described below (schema version 2). +one of two local transports. Both carry the identical versioned payload +described below (schema version 2), so an app behaves the same either way. The Kotlin Multiplatform companion for Android and iOS, [`btripp/openflight-mobile-kmp`](https://github.com/btripp/openflight-mobile-kmp), speaks it. jake-fishtech's SwiftUI app on the @@ -14,6 +15,14 @@ speaks it. jake-fishtech's SwiftUI app on the speaks an earlier, unreleased version one that the Pi does not serve; it needs updating to schema 2. +| Transport | Pi setup | Use it when | +|---|---|---| +| **Bluetooth** | start with `--ble` | The phone cannot reach the Pi over a network | +| **Network** | always on | The phone can reach the Pi over IP (Wi-Fi, Ethernet or any other link), or Bluetooth advertising is unavailable | + +The network transport needs no flag: it streams from the same HTTP server that serves the +browser UI, and exposes nothing the browser UI does not already broadcast. + ## Requirements - Raspberry Pi running Raspberry Pi OS (working Bluetooth only for the BLE transport) @@ -46,21 +55,58 @@ scripts/start-kiosk.sh --ble BLE startup and delivery errors are isolated from shot recording. If Bluetooth is unavailable, the browser UI and session logger continue to work. +## Network transport + +The server streams shots as [Server-Sent Events](https://developer.mozilla.org/docs/Web/API/Server-sent_events) +at `/api/shots/stream`. Check it from any machine on the network before +involving a phone: + +```bash +curl -N http://raspberrypi.local:8080/api/shots/stream +``` + +A connection opens with a `: ping` comment, the current state, and the +session's shots (see [Network stream](#network-stream)), then emits one message +per shot or event with a heartbeat every 15 seconds while idle: + +```text +: ping + +event: shot +id: 05dd37ec-49ed-596b-b1a4-953d54e4f239 +data: {"ball_speed_mph":151.4,"club":"driver",...,"schema_version":2,"type":"shot"} +``` + +In the app, pick **Wi-Fi** (the app's label; it works over any IP network, so +the Pi can be on Ethernet) and enter the Pi's address. `raspberrypi.local:8080` +is the default and works on a stock Raspberry Pi OS install, which publishes its +hostname over mDNS; if you renamed the Pi, use `.local:8080` or its IP. +The port defaults to 8080 when you leave it off. The app reconnects on its own +with backoff, and iOS asks once for permission to talk to devices on the local +network. + +The server accepts up to eight simultaneous stream clients and answers `503` +beyond that, so a forgotten `curl` cannot crowd out a phone. + ## Connect the phone app The phone apps are not part of this repository; build and install one from its -own repository. Start OpenFlight on the Pi with `--ble`. The app scans only for -the OpenFlight service, connects automatically, and subscribes to shot and -control notifications. Hit a shot and its metrics should replace the empty -dashboard. The most recent shot is replayed when a phone connects, so a newly +own repository. Start OpenFlight on the Pi, adding `--ble` for the Bluetooth +transport. Over Bluetooth the app scans only for the OpenFlight service, +connects automatically, and subscribes to shot and control notifications. +Over the network it opens the shot stream and keeps it open. Either way, hit a +shot and its metrics should replace the empty dashboard. The most recent shot is replayed when a phone connects, so a newly connected phone does not have to wait for another shot. ## Select the club from the iPhone Use **Club for next shot** on the dashboard to select any supported wood, hybrid, iron, or wedge. OpenFlight applies the club to subsequent shots and -confirms the change before the app updates its saved selection. The change -goes over the framed control characteristic described below. +confirms the change before the app updates its saved selection. The app sends +the change over the currently selected transport: + +- Bluetooth uses the framed control characteristic described below. +- The network transport sends `POST /api/club` with `{"club":"7-iron"}`. The browser UI and simulator integrations use the same server operation, so a phone club change affects the same launch, spin, and carry processing state. @@ -132,7 +178,7 @@ the same broadcasts. | Command | Payload | Result | Also broadcasts | |---|---|---|---| -| `hello` | `{"client_schema_max":2}` | see [Negotiation](#negotiation) | | +| `hello` | `{"client_schema_max":2}`, optional `last_event_id` | see [Negotiation](#negotiation) | | | `get_club` | `{}` | `{"status":"current","club":"7-iron"}`: the Pi-owned club, unchanged | | | `set_club` | `{"club":"7-iron"}` | `{"status":"applied","club":"7-iron"}` | `club_changed` | | `get_profiles` | `{}` | `{"status":"sent"}` | `profiles`: the roster arrives as the event, not in the result | @@ -152,22 +198,67 @@ timeouts. ### Negotiation 1. Discover the service and subscribe to the control characteristic. -2. Write `hello`: +2. Write `hello`. Add `last_event_id`, the `event_id` of the newest shot the app + already has, to [catch up](#catch-up-after-a-reconnect) on shots missed while + disconnected; leave it out on a first connection: ```json - {"payload":{"client_schema_max":2},"request_id":"","schema_version":2,"type":"hello"} + {"payload":{"client_schema_max":2,"last_event_id":"05dd37ec-49ed-596b-b1a4-953d54e4f239"},"request_id":"","schema_version":2,"type":"hello"} ``` The result names the schema, the features and the characteristics: ```json - {"ok":true,"request_id":"","result":{"characteristics":{"control":"7BA96E63-12C2-4CE0-BB84-3513C7FD1474","shot":"ED365FE6-3ABF-4FC3-8E44-D9525A22DABD"},"features":["provisional_shots","shot_processing","profiles","power_status","shot_deleted","club"],"schema_version":2},"schema_version":2} + {"ok":true,"request_id":"","result":{"characteristics":{"control":"7BA96E63-12C2-4CE0-BB84-3513C7FD1474","shot":"ED365FE6-3ABF-4FC3-8E44-D9525A22DABD"},"features":["provisional_shots","shot_processing","profiles","power_status","shot_deleted","club","shot_catch_up"],"schema_version":2},"schema_version":2} ``` -3. Subscribe to the shot characteristic. The latest shot is replayed. +3. Subscribe to the shot characteristic. The catch-up shots arrive. A client + that skipped `hello` gets only the latest shot. 4. Ask for state: `get_club`, `get_profiles` and, if wanted, `get_power_status`. A `client_schema_max` below 2 fails with `ok:false`: the Pi speaks schema 2 only. +### Catch-up after a reconnect + +A phone that leaves the app, walks out of range or loses the network misses the +shots taken meanwhile. It is caught up on reconnect, with one rule for both +transports: + +| | BLE | Network | +|---|---|---| +| Name the newest shot you have | `last_event_id` in the `hello` payload | `Last-Event-ID` request header, or `?last_event_id=` (the header wins) | +| Catch-up arrives | On the shot characteristic, after the `hello` response; held until the phone subscribes to it | Seeded after the state events, before live events | +| Without catch-up | A client that skips `hello` gets the latest shot | — (every connection gets catch-up) | + +- The Pi sends the named shot again, then every current-session shot after + it, oldest first. Resending the named shot means a phone that only had its + provisional version (it disconnected before the final arrived) ends up with + the final. +- No `last_event_id`, or one the session no longer holds (the session was + cleared, that shot was deleted, or the Pi restarted), means the whole + current session. +- At most the 20 most recent shots are sent. Use `shot_number` gaps to tell + that older shots were not synced (deleted shots also leave gaps). +- Replayed shots are the exact bytes last sent live, including `final` and + `enrichment`. A shot no longer cached is rebuilt as a final shot with + `enrichment: null`. +- Cleared and deleted shots are never replayed: the Pi's session decides what + exists. +- Upsert by `event_id` as for live shots; replays of shots the app already has + are harmless. Invalid `last_event_id` values are treated as absent and never + fail `hello`. +- Every network `shot` frame carries `id: `, so an `EventSource` + resends `Last-Event-ID` on its own when it reconnects. Other events carry no + `id`, which leaves the last shot id in place. +- Over BLE a notification reaches every subscribed phone, so another connected + phone receives the catch-up too and upserts it. +- A shot taken while a BLE catch-up is still being sent can push the oldest + queued catch-up shot out of the eight-message delivery queue. The phone then + lacks that one shot until its session is replayed in full (for example after + it reconnects without `last_event_id`). + +The Pi advertises support with the `shot_catch_up` feature in the `hello` +result. + ### Shot -Sent on the shot characteristic: +Sent on the shot characteristic, and as `event: shot` on the network stream: | Field | Type | Meaning | |---|---|---| @@ -195,7 +286,8 @@ hardware configured is sent once, final. The contract fixture is ### Events -Notified on the control characteristic. Each has `schema_version: 2` and a `type`, and never a `request_id`: +Notified on the control characteristic, and as named events on the network +stream. Each has `schema_version: 2` and a `type`, and never a `request_id`: | `type` | Fields | When | |---|---|---| @@ -219,6 +311,17 @@ Profiles carry only `id` and `name`. `created_at` and the open-ended `settings` stay on Socket.IO, because the phone only selects profiles here and an unbounded `settings` object could not be guaranteed to fit in one BLE message. +### Network stream + +`GET /api/shots/stream` streams schema 2 as Server-Sent Events. `?schema=2` is +accepted for clients that name it; any other `schema` value returns `400`. A +stream opens with `: ping`, then the current state as `club_changed`, +`profiles` and (with a battery monitor) `power_status`, then the +[catch-up](#catch-up-after-a-reconnect) shots (the whole session unless +`Last-Event-ID` names a shot). Each `shot` frame carries `id: `. +Event names match the `type` of the payload. Commands stay on `/api/club` and +Socket.IO. + ### Size budget A BLE message is at most 255 fragments × 15 bytes = 3,825 bytes. Tests encode @@ -229,21 +332,24 @@ message that would not fit instead of sending a truncated one. ## Delivery behavior -- Shot processing never waits for Bluetooth, and a BLE failure cannot affect - the browser UI or session logging. +- Shot processing never waits for either transport, and a failure in one cannot + affect the other, the browser UI, or session logging. - Each connected client gets a bounded queue of eight unsent events; the oldest queued event is dropped if that client cannot keep up. One stalled phone cannot slow down another. -- Disconnecting clears that client's queue; the latest completed shot is - replayed on the next connection. +- Disconnecting clears that client's queue. On the next connection the client + is [caught up](#catch-up-after-a-reconnect) on the session shots it missed + (up to 20). - Clients upsert shots by `event_id`, which ignores replays of shots they already have and merges a provisional shot with its final version. ## Security and scope The phone protocol intentionally has no application authentication or -encryption layer. Enable BLE only where nearby Bluetooth devices receiving -shots and issuing club and profile selections is acceptable. +encryption layer on either transport. Enable BLE only where nearby Bluetooth +devices receiving shots and issuing club and profile selections is acceptable, +and treat the network API as accessible to anything on the same network — the +same assumption the browser UI already makes. BLE is unauthenticated: any nearby device can connect and write the control characteristic. The protocol therefore exposes only reading state, selecting @@ -283,15 +389,17 @@ Bluetooth adapter or `bless` install: ```bash uv run pytest tests/test_ble_protocol.py \ tests/test_ble_publisher.py tests/test_ble_loopback.py tests/test_ble_goldens.py \ - tests/test_phone_transport_server.py tests/test_phone_transport.py \ - tests/test_control_commands.py tests/test_configure_bluetooth.py -v + tests/test_shot_stream.py tests/test_phone_transport_server.py \ + tests/test_phone_transport.py tests/test_control_commands.py \ + tests/test_phone_catch_up.py tests/test_ble_catch_up.py \ + tests/test_shot_stream_catch_up.py tests/test_configure_bluetooth.py -v uv run python scripts/ble/generate_goldens.py --check ``` What still needs a Pi and phones: BlueZ advertising, discovery and connection from iOS and Android, pairing and permission prompts, fragment pacing over a real link, reconnects after a Pi restart, background behaviour, -and coexistence with Socket.IO clients. +and coexistence with SSE and Socket.IO clients. ## Troubleshooting diff --git a/src/openflight/ble/protocol.py b/src/openflight/ble/protocol.py index 5e7f9e4ae..c1cd80b1a 100644 --- a/src/openflight/ble/protocol.py +++ b/src/openflight/ble/protocol.py @@ -67,6 +67,7 @@ "power_status", "shot_deleted", "club", + "shot_catch_up", ) EVENT_TYPES = ( diff --git a/src/openflight/ble/publisher.py b/src/openflight/ble/publisher.py index cf667c8b9..50e36a145 100644 --- a/src/openflight/ble/publisher.py +++ b/src/openflight/ble/publisher.py @@ -7,8 +7,8 @@ import logging import threading import uuid -from collections.abc import Callable -from typing import Mapping +from collections.abc import Callable, Sequence +from typing import Any, Mapping from .protocol import ( ACCEPTED_REQUEST_SCHEMAS, @@ -27,6 +27,9 @@ logger = logging.getLogger(__name__) CommandHandler = Callable[[str, Mapping], tuple[dict, int]] +# Given a client's ``last_event_id`` (or ``None``), the ``(event_id, shot +# payload)`` pairs it missed, oldest first. See ``openflight.phone_catch_up``. +CatchUpProvider = Callable[[Any], Sequence[tuple[str, bytes]]] _SHOT = SHOT_CHARACTERISTIC_UUID.lower() _CONTROL = CONTROL_CHARACTERISTIC_UUID.lower() @@ -56,13 +59,14 @@ class BleShotPublisher: value, so every connected phone receives every message. """ - def __init__( + def __init__( # pylint: disable=too-many-arguments self, *, name: str = "OpenFlight", queue_size: int = 8, fragment_interval_s: float = 0.01, command_handler: CommandHandler | None = None, + catch_up_provider: CatchUpProvider | None = None, ): if queue_size < 1: raise ValueError("BLE queue size must be at least one") @@ -70,6 +74,7 @@ def __init__( self.queue_size = queue_size self.fragment_interval_s = fragment_interval_s self.command_handler = command_handler + self.catch_up_provider = catch_up_provider self._state_lock = threading.Lock() self._thread: threading.Thread | None = None @@ -87,6 +92,9 @@ def __init__( self._control_sequence = 0 self._control_reassembler = FragmentReassembler() self._control_send_lock: asyncio.Lock | None = None + # Missed shots computed at a ``hello`` that arrived before the phone + # subscribed to the shot characteristic; sent when it does. + self._pending_catch_up: list[bytes] | None = None @property def subscribed(self) -> bool: @@ -186,6 +194,7 @@ def _run_thread(self) -> None: self._subscribed = False self._subscriptions = set() self._control_send_lock = None + self._pending_catch_up = None async def _run(self) -> None: # Bless is an optional dependency and must not affect non-BLE installs. @@ -309,6 +318,12 @@ def _set_subscriptions(self, subscriptions: set[str]) -> None: latest_payload = self._latest_payload queue = self._queue shot_started = _SHOT in subscriptions and _SHOT not in previous + pending_catch_up = None + if shot_started or not subscribed: + # Consumed by the subscription it was waiting for, or discarded + # with the connection that asked for it. + pending_catch_up = self._pending_catch_up if shot_started else None + self._pending_catch_up = None if subscribed != was_subscribed: logger.info("[BLE] Client %s", "subscribed" if subscribed else "unsubscribed") @@ -317,8 +332,13 @@ def _set_subscriptions(self, subscriptions: set[str]) -> None: # Replay the latest shot when the shot characteristic gains a # subscriber, so a phone that subscribes to control first still gets it. - if shot_started and latest_payload is not None: - self._enqueue_payload(latest_payload) + # A phone that already sent ``hello`` gets its catch-up instead, which + # holds the latest shot unless the phone already had it. + if shot_started: + if pending_catch_up is not None: + self._start_catch_up(pending_catch_up) + elif latest_payload is not None: + self._enqueue_payload(latest_payload) def _enqueue_payload(self, payload: bytes) -> None: with self._state_lock: @@ -327,6 +347,55 @@ def _enqueue_payload(self, payload: bytes) -> None: if subscribed: self._offer(queue, payload) + async def _load_catch_up(self, last_event_id) -> list[bytes] | None: + """The shots a client missed, or ``None`` when catch-up is unavailable.""" + provider = self.catch_up_provider + if provider is None: + return None + try: + entries = await asyncio.to_thread(provider, last_event_id) + return [payload for _event_id, payload in entries] + except Exception: # pylint: disable=broad-exception-caught + # Never fail ``hello`` over catch-up: the phone keeps the + # latest-shot replay. + logger.warning("[BLE] Could not load missed shots for catch-up", exc_info=True) + return None + + def _schedule_catch_up(self, payloads: list[bytes] | None) -> None: + """Send catch-up now if the shot characteristic is subscribed, else on subscribe. + + ``None`` (no catch-up for this command) does nothing. + """ + if payloads is None: + return + with self._state_lock: + ready = _SHOT in self._subscriptions + if not ready: + self._pending_catch_up = payloads + logger.info( + "[BLE] Catch-up: %d missed shot(s)%s", + len(payloads), + "" if ready else " (sent when the client subscribes to shots)", + ) + if ready: + self._start_catch_up(payloads) + + def _start_catch_up(self, payloads: list[bytes]) -> None: + with self._state_lock: + loop = self._loop + if loop is not None and payloads: + asyncio.run_coroutine_threadsafe(self._deliver_catch_up(payloads), loop) + + async def _deliver_catch_up(self, payloads: list[bytes]) -> None: + """Queue catch-up shots in order, waiting for room instead of dropping any.""" + for payload in payloads: + with self._state_lock: + queue = self._queue + subscribed = self._subscribed + if queue is None or not subscribed: + return + await queue.put(payload) + @staticmethod def _offer(queue: asyncio.Queue[bytes] | None, payload: bytes) -> None: if queue is None: @@ -416,6 +485,7 @@ def _on_write_request(self, characteristic, value, **_kwargs) -> None: async def _process_control_payload(self, payload: bytes) -> None: request_id = "unknown" + catch_up: list[bytes] | None = None try: command = json.loads(payload) if not isinstance(command, dict): @@ -433,10 +503,7 @@ async def _process_control_payload(self, payload: bytes) -> None: raise ValueError("Control command payload must be an object") if command_type == "hello": - # Answered by the publisher itself: ``hello`` is about which - # characteristics exist, not server state. - result = build_hello_result(command_payload.get("client_schema_max")) - response = build_control_response(request_id, result=result) + response, catch_up = await self._answer_hello(request_id, command_payload) else: if self.command_handler is None: raise ValueError("Phone controls are not configured on this OpenFlight server") @@ -458,6 +525,22 @@ async def _process_control_payload(self, payload: bytes) -> None: ) await self._send_control_response(encode_message(response)) + # After the answer, so a client sees ``hello`` succeed before shots arrive. + self._schedule_catch_up(catch_up) + + async def _answer_hello( + self, + request_id: str, + command_payload: Mapping, + ) -> tuple[dict, list[bytes] | None]: + """Negotiate the transport and load the client's catch-up shots. + + ``hello`` is answered by the publisher itself: it is about which + characteristics exist, not server state. + """ + result = build_hello_result(command_payload.get("client_schema_max")) + response = build_control_response(request_id, result=result) + return response, await self._load_catch_up(command_payload.get("last_event_id")) async def _send_control_response(self, payload: bytes) -> None: """Send one complete control message without interleaving fragments.""" diff --git a/src/openflight/phone_catch_up.py b/src/openflight/phone_catch_up.py new file mode 100644 index 000000000..60d3a6edf --- /dev/null +++ b/src/openflight/phone_catch_up.py @@ -0,0 +1,97 @@ +"""Catch a reconnecting phone up on the session shots it missed. + +Both phone transports share this rule. A client names the last shot it +has (``last_event_id`` in the BLE ``hello`` payload, ``Last-Event-ID`` on the +network stream) and receives that shot again plus every current-session shot +after it, oldest first. The named shot is resent because the client may hold +only its provisional version if it disconnected before the final one arrived. +A client that names no shot, or one the session no longer holds (cleared, +deleted, or from before a Pi restart), receives the whole session. Either way +the most recent ``CATCH_UP_LIMIT`` shots are sent, and clients upsert them by +``event_id`` so a shot they already have is harmless. + +The session itself (``monitor.get_shots()``) decides which shots exist, so +clears, deletes and profile changes need no bookkeeping here. ``PhoneShotCache`` +only remembers the exact bytes last published per ``event_id``, so a replayed +shot carries the same ``final`` and ``enrichment`` state the phone would have +received live. +""" + +from __future__ import annotations + +import threading +from collections import OrderedDict +from typing import Sequence + +CATCH_UP_LIMIT = 20 +# Far above any realistic session; bounds memory on a Pi left running for days. +DEFAULT_CACHE_ENTRIES = 512 +# Event ids are UUIDs (36 characters); anything much longer is not one of ours. +MAX_EVENT_ID_LENGTH = 64 + +CatchUpEntry = tuple[str, bytes] + + +def normalize_last_event_id(value) -> str | None: + """Return a usable ``last_event_id``, or ``None`` when the client sent none. + + Invalid values are treated as "no anchor" rather than rejected: a client + that sends garbage still gets the whole session instead of a failed + ``hello`` that would push it back to version one. + """ + if not isinstance(value, str): + return None + value = value.strip() + if not value or len(value) > MAX_EVENT_ID_LENGTH: + return None + return value + + +def select_catch_up( + entries: Sequence[CatchUpEntry], + last_event_id: str | None, + *, + limit: int = CATCH_UP_LIMIT, +) -> list[CatchUpEntry]: + """The ``last_event_id`` shot and those after it (all shots if it is unknown). + + ``entries`` are ``(event_id, payload)`` pairs in session order. The result + keeps that order and holds at most the most recent ``limit`` shots. + """ + if limit < 1: + raise ValueError("Catch-up limit must be at least one") + start = 0 + if last_event_id is not None: + for index, (event_id, _payload) in enumerate(entries): + if event_id == last_event_id: + start = index + break + return list(entries[start:])[-limit:] + + +class PhoneShotCache: + """The latest encoded shot per ``event_id``, bounded and thread-safe.""" + + def __init__(self, max_entries: int = DEFAULT_CACHE_ENTRIES): + if max_entries < 1: + raise ValueError("Phone shot cache must hold at least one entry") + self.max_entries = max_entries + self._lock = threading.Lock() + self._payloads: OrderedDict[str, bytes] = OrderedDict() + + def __len__(self) -> int: + with self._lock: + return len(self._payloads) + + def remember(self, event_id: str, payload: bytes) -> None: + """Store the newest payload for a shot (a final replaces its provisional).""" + with self._lock: + self._payloads[event_id] = payload + self._payloads.move_to_end(event_id) + while len(self._payloads) > self.max_entries: + self._payloads.popitem(last=False) + + def get(self, event_id: str) -> bytes | None: + """The last published payload for ``event_id``, if still cached.""" + with self._lock: + return self._payloads.get(event_id) diff --git a/src/openflight/server.py b/src/openflight/server.py index 1c8ca3eb9..76efd7efa 100644 --- a/src/openflight/server.py +++ b/src/openflight/server.py @@ -26,11 +26,14 @@ from .ballistics import resolve_launch, simulate from .ble.protocol import ( + build_club_event, build_power_status_event, build_profiles_event, build_session_cleared_event, build_shot_deleted_event, build_shot_processing_event, + encode_shot_event, + stable_shot_event_id, ) from .clubs import ClubType from .clubs.physics import ( @@ -46,10 +49,12 @@ SpeedReading, set_show_raw_readings, ) +from .phone_catch_up import PhoneShotCache, normalize_last_event_id, select_catch_up from .power import SUPPORTED_BATTERY_PROVIDERS, PowerMonitor, PowerStatus from .profiles import ProfileStore from .rolling_buffer.monitor import estimate_carry_with_spin, get_optimal_spin_for_ball_speed from .session_logger import get_session_logger, init_session_logger, log_session_error +from .shot_stream import SSE_MIMETYPE, ShotStreamBroker, ShotStreamFull from .sim import ( IncompleteShotError, PlayerState as SimPlayerState, @@ -173,6 +178,15 @@ def get_profile_store() -> ProfileStore: active_club = ClubType.DRIVER club_selection_lock = threading.Lock() +# Network shot delivery for phone apps. Always available: it exposes the same +# shots the browser UI already broadcasts over WebSocket, so it adds no reach +# beyond the existing HTTP server. +shot_stream = ShotStreamBroker() + +# The exact bytes last sent per shot, so a reconnecting phone's catch-up +# (BLE and network alike) replays what it would have received live. +phone_shot_cache = PhoneShotCache() + shutdown_lock = threading.Lock() shutdown_cleanup_started = False # One active hardware job plus two waiting shots is enough for normal golf @@ -1026,6 +1040,10 @@ def _broadcast_club_selection(club: ClubType) -> None: socketio.emit("club_changed", club_data) except Exception: # pylint: disable=broad-exception-caught logger.warning("[SERVER] Failed to broadcast club over WebSocket", exc_info=True) + try: + shot_stream.publish_club(club.value) + except Exception: # pylint: disable=broad-exception-caught + logger.warning("[SERVER] Failed to broadcast club over network stream", exc_info=True) if ble_publisher is not None: try: ble_publisher.publish_club(club.value) @@ -1057,6 +1075,28 @@ def dispatch_phone_control_command(command_type, payload): return handler(payload) +def _phone_state_events() -> list[dict]: + """Current club, profiles and power, as seeded to a new stream client.""" + with club_selection_lock: + club_value = active_club.value + events = [build_club_event(club_value)] + try: + events.append(build_profiles_event(get_profile_store().snapshot())) + except Exception: # pylint: disable=broad-exception-caught + logger.warning("[SERVER] Could not read profiles for the shot stream", exc_info=True) + if power_monitor is not None and power_monitor.status is not None: + events.append(build_power_status_event(power_monitor.status.to_dict())) + return events + + +@app.route("/api/club", methods=["GET", "POST"]) +def api_club_selection(): + """Read or set the active club over the network (HTTP).""" + if request.method == "GET": + return current_club_selection() + return apply_club_selection(request.get_json(silent=True)) + + @app.route("/") def static_files(path): """Serve static files.""" @@ -1573,6 +1613,45 @@ def handle_get_camera_capture_settings(): socketio.emit("camera_capture_settings", _camera_capture_settings_payload()) +def _stream_catch_up(last_event_id) -> list[tuple[str, bytes]] | None: + """Catch-up for a stream client, or ``None`` (latest-shot replay) on failure.""" + try: + return phone_catch_up(last_event_id) + except Exception: # pylint: disable=broad-exception-caught + logger.warning("[SERVER] Could not load missed shots for the stream", exc_info=True) + return None + + +@app.route("/api/shots/stream") +def shots_stream(): + """Stream completed shots to phones as Server-Sent Events. + + Speaks schema 2, the only phone schema; ``?schema=2`` is accepted for + clients that name it. A client resumes with ``Last-Event-ID`` (or + ``?last_event_id=``) and is seeded with the session shots it missed, as + BLE ``hello`` does. + """ + if request.args.get("schema", "2") != "2": + return {"error": "Unsupported schema; use 2"}, 400 + last_event_id = request.headers.get("Last-Event-ID") or request.args.get("last_event_id") + try: + subscriber = shot_stream.subscribe( + initial_events=_phone_state_events(), + catch_up=_stream_catch_up(last_event_id), + ) + except ShotStreamFull as exc: + logger.warning("[SERVER] Refused shot stream client: %s", exc) + return str(exc), 503 + + response = Response(shot_stream.frames(subscriber), mimetype=SSE_MIMETYPE) + response.headers["Cache-Control"] = "no-cache" + response.headers["X-Accel-Buffering"] = "no" + # Covers the case where the response is discarded without ever being + # iterated; unsubscribing twice is a no-op. + response.call_on_close(lambda: shot_stream.unsubscribe(subscriber)) + return response + + @socketio.on("set_camera_capture_settings") def handle_set_camera_capture_settings(data): """Apply live-safe camera controls and alignment-guide position.""" @@ -3628,8 +3707,15 @@ def _publish_phone_shot( final: bool, enrichment: dict | None, ) -> None: - """Hand one shot to the phone transports; never raises.""" - transports = [] + """Hand one shot to the BLE and SSE phone transports; never raises.""" + try: + phone_shot_cache.remember( + stable_shot_event_id(shot_data), + encode_shot_event(shot_data, final=final, enrichment=enrichment), + ) + except (KeyError, TypeError, ValueError): + logger.warning("[SERVER] Could not cache shot for phone catch-up", exc_info=True) + transports = [("network stream", shot_stream)] if ble_publisher is not None: transports.append(("BLE", ble_publisher)) for name, transport in transports: @@ -3639,9 +3725,34 @@ def _publish_phone_shot( logger.warning("[SERVER] Failed to queue shot over %s", name, exc_info=True) +def phone_catch_up(last_event_id=None) -> list[tuple[str, bytes]]: + """The current-session shots a reconnecting phone missed, oldest first. + + Shared by BLE ``hello`` and the network stream's ``Last-Event-ID``; see + ``openflight.phone_catch_up`` for the rule. The session decides which + shots exist, so cleared and deleted shots are never replayed. A shot is + replayed as the bytes last published for it, or rebuilt as a final shot + when that is no longer cached. + """ + session = monitor + if session is None or not hasattr(session, "get_shots"): + return [] + entries = [] + for shot in session.get_shots(): + try: + shot_data = shot_to_dict(shot) + event_id = stable_shot_event_id(shot_data) + payload = phone_shot_cache.get(event_id) or encode_shot_event(shot_data, final=True) + except (AttributeError, KeyError, TypeError, ValueError): + logger.warning("[SERVER] Skipped a shot that could not be replayed", exc_info=True) + continue + entries.append((event_id, payload)) + return select_catch_up(entries, normalize_last_event_id(last_event_id)) + + def _publish_phone_event(event: dict) -> None: - """Hand one event to the phone transports; never raises.""" - transports = [] + """Hand one event to the BLE and SSE phone transports; never raises.""" + transports = [("network stream", shot_stream)] if ble_publisher is not None: transports.append(("BLE", ble_publisher)) for name, transport in transports: @@ -5288,6 +5399,7 @@ def main(): ble_publisher = BleShotPublisher( command_handler=dispatch_phone_control_command, + catch_up_provider=phone_catch_up, ) ble_publisher.start() print("Bluetooth LE enabled (advertising as OpenFlight)") diff --git a/src/openflight/shot_stream.py b/src/openflight/shot_stream.py new file mode 100644 index 000000000..33f7d81cd --- /dev/null +++ b/src/openflight/shot_stream.py @@ -0,0 +1,228 @@ +"""Fan out shots and phone events to HTTP clients as Server-Sent Events. + +This is the network sibling of the BLE publisher: the same payloads, the same +bounded-queue delivery policy, the same isolation from shot recording. It +exists so a phone can receive shots over the network on hardware where BLE +advertising is unavailable, and so the payload contract can be exercised with +nothing but ``curl``. +""" + +from __future__ import annotations + +import logging +import queue +import threading +from typing import Iterable, Iterator, Mapping, Sequence + +# The wire payloads are the BLE transport's, so both transports are validated +# against one contract and one set of fixtures. +from .ble.protocol import build_club_event, build_shot_event, encode_message + +logger = logging.getLogger(__name__) + +SSE_MIMETYPE = "text/event-stream" +DEFAULT_HEARTBEAT_INTERVAL_S = 15.0 +DEFAULT_MAX_SUBSCRIBERS = 8 + +# An SSE comment. Clients ignore it, but it keeps idle connections from being +# reaped and lets the server notice a vanished client between shots. +HEARTBEAT_FRAME = ": ping\n\n" + + +class ShotStreamFull(RuntimeError): + """Raised when the broker already serves the maximum number of clients.""" + + +class StreamEvent(bytes): + """Encoded JSON carrying its SSE name while remaining bytes-compatible. + + Shots also carry their ``event_id`` as the SSE ``id``, so a client's + ``Last-Event-ID`` names the last shot it received (see ``phone_catch_up``). + """ + + name: str + event_id: str | None + + def __new__(cls, name: str, payload: bytes, event_id: str | None = None): + event = super().__new__(cls, payload) + event.name = name + event.event_id = event_id + return event + + +def format_event(event: StreamEvent | bytes) -> str: + """Frame one encoded event for the SSE wire protocol. + + ``encode_message`` emits compact single-line JSON, so the payload never + needs to be split across multiple ``data:`` lines. + """ + name = event.name if isinstance(event, StreamEvent) else "shot" + event_id = event.event_id if isinstance(event, StreamEvent) else None + id_line = f"id: {event_id}\n" if event_id else "" + return f"event: {name}\n{id_line}data: {event.decode('utf-8')}\n\n" + + +class ShotStreamBroker: + """Deliver encoded shots and events to every connected SSE subscriber.""" + + def __init__( + self, + *, + queue_size: int = 8, + heartbeat_interval_s: float = DEFAULT_HEARTBEAT_INTERVAL_S, + max_subscribers: int = DEFAULT_MAX_SUBSCRIBERS, + ): + if queue_size < 1: + raise ValueError("Shot stream queue size must be at least one") + if heartbeat_interval_s <= 0: + raise ValueError("Shot stream heartbeat interval must be positive") + if max_subscribers < 1: + raise ValueError("Shot stream must allow at least one subscriber") + + self.queue_size = queue_size + self.heartbeat_interval_s = heartbeat_interval_s + self.max_subscribers = max_subscribers + + self._lock = threading.Lock() + self._subscribers: list[queue.Queue[StreamEvent]] = [] + self._latest_shot: StreamEvent | None = None + + @property + def subscriber_count(self) -> int: + """How many clients are currently streaming.""" + with self._lock: + return len(self._subscribers) + + def publish_shot( + self, + shot_data: Mapping, + *, + final: bool, + enrichment: Mapping | None = None, + ) -> bool: + """Send a provisional or final shot to every subscriber; never raises upward.""" + try: + shot_event = build_shot_event(shot_data, final=final, enrichment=enrichment) + event = StreamEvent("shot", encode_message(shot_event), shot_event["event_id"]) + except (KeyError, TypeError, ValueError): + logger.warning("[STREAM] Failed to encode shot payload", exc_info=True) + return False + + with self._lock: + self._latest_shot = event + subscribers = list(self._subscribers) + for subscriber in subscribers: + self._offer(subscriber, event) + return True + + def publish_event(self, event: Mapping) -> bool: + """Send one event, named by its ``type``, to every subscriber.""" + try: + stream_event = StreamEvent(str(event["type"]), encode_message(event)) + except (KeyError, TypeError, ValueError): + logger.warning("[STREAM] Failed to encode event", exc_info=True) + return False + + with self._lock: + subscribers = list(self._subscribers) + for subscriber in subscribers: + self._offer(subscriber, stream_event) + return True + + def publish_club(self, club: str) -> bool: + """Broadcast an authoritative club change to every subscriber.""" + try: + event = build_club_event(club) + except (TypeError, ValueError): + logger.warning("[STREAM] Failed to encode club payload", exc_info=True) + return False + return self.publish_event(event) + + def subscribe( + self, + *, + initial_events: Iterable[Mapping] = (), + catch_up: Sequence[tuple[str, bytes]] | None = None, + ) -> queue.Queue[StreamEvent]: + """Register a subscriber, seeded with current state and missed shots. + + The seed is ``initial_events`` (current state such as club and + profiles), then the ``catch_up`` shots (``(event_id, payload)`` pairs, + oldest first) or, when catch-up is unavailable (``None``), the latest + shot. + """ + seed = [] + for event in initial_events: + try: + seed.append(StreamEvent(str(event["type"]), encode_message(event))) + except (KeyError, TypeError, ValueError): + logger.warning("[STREAM] Failed to encode initial event", exc_info=True) + with self._lock: + if len(self._subscribers) >= self.max_subscribers: + raise ShotStreamFull(f"Shot stream already has {self.max_subscribers} clients") + if catch_up is not None: + seed.extend( + StreamEvent("shot", payload, event_id) for event_id, payload in catch_up + ) + elif self._latest_shot is not None: + seed.append(self._latest_shot) + subscriber: queue.Queue[StreamEvent] = queue.Queue( + maxsize=max(self.queue_size, len(seed)) + ) + for event in seed: + subscriber.put_nowait(event) + self._subscribers.append(subscriber) + count = len(self._subscribers) + logger.info("[STREAM] Client subscribed (%d streaming)", count) + return subscriber + + def unsubscribe(self, subscriber: queue.Queue[StreamEvent]) -> None: + """Drop a subscriber. Unsubscribing twice is not an error.""" + with self._lock: + if subscriber not in self._subscribers: + return + self._subscribers.remove(subscriber) + count = len(self._subscribers) + logger.info("[STREAM] Client unsubscribed (%d streaming)", count) + + def frames(self, subscriber: queue.Queue[StreamEvent]) -> Iterator[str]: + """Yield SSE frames for an already-registered subscriber. + + Registration is deliberately not folded in here: a generator body does + not run until first iteration, so a caller that needs to reject a client + before sending response headers has to call ``subscribe`` itself. + """ + try: + # Open with a heartbeat so the WSGI server flushes response headers + # at once. Without it a client learns nothing -- not even that it + # connected -- until the first shot or heartbeat, because headers + # are not written until the first chunk of the body. + yield HEARTBEAT_FRAME + while True: + try: + event = subscriber.get(timeout=self.heartbeat_interval_s) + except queue.Empty: + yield HEARTBEAT_FRAME + continue + yield format_event(event) + finally: + self.unsubscribe(subscriber) + + def _offer(self, subscriber: queue.Queue[StreamEvent], event: StreamEvent) -> None: + """Queue an event, dropping the oldest unsent event when full.""" + try: + subscriber.put_nowait(event) + return + except queue.Full: + pass + + try: + subscriber.get_nowait() + logger.warning("[STREAM] Delivery queue full; dropped oldest unsent shot") + except queue.Empty: + pass + + try: + subscriber.put_nowait(event) + except queue.Full: + logger.warning("[STREAM] Delivery queue still full; dropped shot") diff --git a/tests/ble_harness.py b/tests/ble_harness.py index d657b7db6..f04d6ecb3 100644 --- a/tests/ble_harness.py +++ b/tests/ble_harness.py @@ -349,7 +349,9 @@ def server_loopback(monkeypatch, tmp_path): """ from openflight import server as server_module # pylint: disable=import-outside-toplevel from openflight.launch_monitor import ClubType # pylint: disable=import-outside-toplevel + from openflight.phone_catch_up import PhoneShotCache # pylint: disable=import-outside-toplevel from openflight.profiles import ProfileStore # pylint: disable=import-outside-toplevel + from openflight.shot_stream import ShotStreamBroker # pylint: disable=import-outside-toplevel emitted = [] lock = threading.Lock() @@ -361,6 +363,7 @@ def emit(event, payload=None, **_kwargs): monkeypatch.setattr(server_module.socketio, "emit", emit) monkeypatch.setattr(server_module, "monitor", None) monkeypatch.setattr(server_module, "profile_store", ProfileStore(tmp_path / "profiles.json")) + monkeypatch.setattr(server_module, "shot_stream", ShotStreamBroker()) monkeypatch.setattr(server_module, "active_club", ClubType.DRIVER) monkeypatch.setattr(server_module, "iwr6843_runtime", None) monkeypatch.setattr(server_module, "power_monitor", None) @@ -373,10 +376,12 @@ def emit(event, payload=None, **_kwargs): monkeypatch.setattr(server_module, "debug_mode", False) monkeypatch.setattr(server_module, "sim_connectors", []) monkeypatch.setattr(server_module, "get_session_logger", lambda: None) + monkeypatch.setattr(server_module, "phone_shot_cache", PhoneShotCache()) loopback = BleLoopback( monkeypatch, command_handler=server_module.dispatch_phone_control_command, + catch_up_provider=server_module.phone_catch_up, ) with loopback: monkeypatch.setattr(server_module, "ble_publisher", loopback.publisher) diff --git a/tests/fixtures/ble_goldens/v2_response_hello.json b/tests/fixtures/ble_goldens/v2_response_hello.json index 26298b2c1..a018b6264 100644 --- a/tests/fixtures/ble_goldens/v2_response_hello.json +++ b/tests/fixtures/ble_goldens/v2_response_hello.json @@ -17,7 +17,8 @@ "profiles", "power_status", "shot_deleted", - "club" + "club", + "shot_catch_up" ], "characteristics": { "shot": "ED365FE6-3ABF-4FC3-8E44-D9525A22DABD", @@ -25,29 +26,30 @@ } } }, - "payload_hex": "7b226f6b223a747275652c22726571756573745f6964223a2235453046324334412d384231442d344333452d394636412d374432423143304539413834222c22726573756c74223a7b22636861726163746572697374696373223a7b22636f6e74726f6c223a2237424139364536332d313243322d344345302d424238342d333531334337464431343734222c2273686f74223a2245443336354645362d334142462d344643332d384534342d443935323541323244414244227d2c226665617475726573223a5b2270726f766973696f6e616c5f73686f7473222c2273686f745f70726f63657373696e67222c2270726f66696c6573222c22706f7765725f737461747573222c2273686f745f64656c65746564222c22636c7562225d2c22736368656d615f76657273696f6e223a327d2c22736368656d615f76657273696f6e223a327d", + "payload_hex": "7b226f6b223a747275652c22726571756573745f6964223a2235453046324334412d384231442d344333452d394636412d374432423143304539413834222c22726573756c74223a7b22636861726163746572697374696373223a7b22636f6e74726f6c223a2237424139364536332d313243322d344345302d424238342d333531334337464431343734222c2273686f74223a2245443336354645362d334142462d344643332d384534342d443935323541323244414244227d2c226665617475726573223a5b2270726f766973696f6e616c5f73686f7473222c2273686f745f70726f63657373696e67222c2270726f66696c6573222c22706f7765725f737461747573222c2273686f745f64656c65746564222c22636c7562222c2273686f745f63617463685f7570225d2c22736368656d615f76657273696f6e223a327d2c22736368656d615f76657273696f6e223a327d", "frames_hex": [ - "01000000167b226f6b223a747275652c22726571", - "0100000116756573745f6964223a223545304632", - "01000002164334412d384231442d344333452d39", - "01000003164636412d3744324231433045394138", - "010000041634222c22726573756c74223a7b2263", - "0100000516686172616374657269737469637322", - "01000006163a7b22636f6e74726f6c223a223742", - "01000007164139364536332d313243322d344345", - "0100000816302d424238342d3335313343374644", - "010000091631343734222c2273686f74223a2245", - "0100000a16443336354645362d334142462d3446", - "0100000b1643332d384534342d44393532354132", - "0100000c163244414244227d2c22666561747572", - "0100000d166573223a5b2270726f766973696f6e", - "0100000e16616c5f73686f7473222c2273686f74", - "0100000f165f70726f63657373696e67222c2270", - "0100001016726f66696c6573222c22706f776572", - "01000011165f737461747573222c2273686f745f", - "010000121664656c65746564222c22636c756222", - "01000013165d2c22736368656d615f7665727369", - "01000014166f6e223a327d2c22736368656d615f", - "010000151676657273696f6e223a327d" + "01000000177b226f6b223a747275652c22726571", + "0100000117756573745f6964223a223545304632", + "01000002174334412d384231442d344333452d39", + "01000003174636412d3744324231433045394138", + "010000041734222c22726573756c74223a7b2263", + "0100000517686172616374657269737469637322", + "01000006173a7b22636f6e74726f6c223a223742", + "01000007174139364536332d313243322d344345", + "0100000817302d424238342d3335313343374644", + "010000091731343734222c2273686f74223a2245", + "0100000a17443336354645362d334142462d3446", + "0100000b1743332d384534342d44393532354132", + "0100000c173244414244227d2c22666561747572", + "0100000d176573223a5b2270726f766973696f6e", + "0100000e17616c5f73686f7473222c2273686f74", + "0100000f175f70726f63657373696e67222c2270", + "0100001017726f66696c6573222c22706f776572", + "01000011175f737461747573222c2273686f745f", + "010000121764656c65746564222c22636c756222", + "01000013172c2273686f745f63617463685f7570", + "0100001417225d2c22736368656d615f76657273", + "0100001517696f6e223a327d2c22736368656d61", + "01000016175f76657273696f6e223a327d" ] } diff --git a/tests/test_ble_catch_up.py b/tests/test_ble_catch_up.py new file mode 100644 index 000000000..4a2b4fe48 --- /dev/null +++ b/tests/test_ble_catch_up.py @@ -0,0 +1,277 @@ +"""BLE schema v2 catch-up: a reconnecting phone receives the session shots it missed. + +Driven over the loopback harness with the real publisher and server dispatch. +The phone follows the documented negotiation: subscribe to v2 control, write +``hello`` (optionally naming ``last_event_id``), then subscribe to v2 shot. +""" + +import json +from datetime import datetime + +import pytest +from ble_harness import PHONE_UUIDS, server_loopback, settle + +from openflight import server as server_module +from openflight.ble.protocol import ( + CONTROL_CHARACTERISTIC_UUID, + SHOT_CHARACTERISTIC_UUID, + stable_shot_event_id, +) +from openflight.launch_monitor import ClubType, Shot +from openflight.phone_catch_up import CATCH_UP_LIMIT + + +@pytest.fixture +def pi(monkeypatch, tmp_path): + with server_loopback(monkeypatch, tmp_path) as loopback: + yield loopback + + +def _session(monkeypatch, count, *, profile_id=""): + """A mock monitor holding ``count`` shots; returns their event ids in order.""" + monitor = server_module.MockLaunchMonitor() + for index in range(count): + monitor._shots.append( + Shot( + ball_speed_mph=100.0 + index, + timestamp=datetime(2026, 9, 28, 12, index // 60, index % 60, 1000), + club=ClubType.DRIVER, + shot_number=index + 1, + profile_id=profile_id, + ) + ) + monkeypatch.setattr(server_module, "monitor", monitor) + return [stable_shot_event_id(server_module.shot_to_dict(shot)) for shot in monitor.get_shots()] + + +def _publish_latest_live(): + """Publish the session's last shot as the live path would (sets the latest replay).""" + shot_data = server_module.shot_to_dict(server_module.monitor.get_shots()[-1]) + server_module._publish_phone_shot(shot_data, final=True, enrichment=None) + + +def _hello(phone, **payload): + return phone.request( + "hello", + {"client_schema_max": 2, **payload}, + control=CONTROL_CHARACTERISTIC_UUID, + schema_version=2, + ) + + +def _connect(pi, name="v2-app", **hello_payload): + """Negotiate as the documented flow does: control, hello, then shot.""" + phone = pi.central(name) + phone.subscribe(CONTROL_CHARACTERISTIC_UUID) + answer = _hello(phone, **hello_payload) + assert answer["ok"] is True, answer + phone.subscribe(SHOT_CHARACTERISTIC_UUID) + return phone + + +def _received_ids(phone): + return [shot["event_id"] for shot in phone.decoded(SHOT_CHARACTERISTIC_UUID)] + + +def test_hello_advertises_catch_up(pi): + phone = pi.central() + phone.subscribe(CONTROL_CHARACTERISTIC_UUID) + + assert "shot_catch_up" in _hello(phone)["result"]["features"] + + +def test_hello_without_anchor_replays_whole_session_in_order(pi, monkeypatch): + ids = _session(monkeypatch, 3) + + phone = _connect(pi) + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, 3) + settle(pi) + assert _received_ids(phone) == ids + assert all(shot["final"] for shot in phone.decoded(SHOT_CHARACTERISTIC_UUID)) + + +def test_hello_with_known_anchor_resends_it_and_sends_missed_shots(pi, monkeypatch): + ids = _session(monkeypatch, 5) + + phone = _connect(pi, last_event_id=ids[1]) + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, 4) + settle(pi) + assert _received_ids(phone) == ids[1:] + + +def test_hello_with_latest_anchor_resends_only_that_shot_once(pi, monkeypatch): + ids = _session(monkeypatch, 3) + _publish_latest_live() + + phone = _connect(pi, last_event_id=ids[-1]) + phone.wait_for(SHOT_CHARACTERISTIC_UUID) + settle(pi, rounds=10) + + # The catch-up replaces the latest-shot replay rather than adding to it. + assert _received_ids(phone) == [ids[-1]] + + +def test_anchor_held_as_provisional_is_brought_up_to_final(pi, monkeypatch): + """The phone saw the provisional, disconnected, and missed the final.""" + ids = _session(monkeypatch, 2) + shot_data = server_module.shot_to_dict(server_module.monitor.get_shots()[-1]) + server_module._publish_phone_shot(shot_data, final=False, enrichment={"status": "pending"}) + server_module._publish_phone_shot(shot_data, final=True, enrichment={"status": "complete"}) + + phone = _connect(pi, last_event_id=ids[-1]) + + shot = phone.wait_for(SHOT_CHARACTERISTIC_UUID) + assert shot["event_id"] == ids[-1] + assert shot["final"] is True + assert shot["enrichment"] == {"status": "complete"} + + +def test_unknown_anchor_replays_whole_session(pi, monkeypatch): + ids = _session(monkeypatch, 2) + + phone = _connect(pi, last_event_id="11111111-2222-3333-4444-555555555555") + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, 2) + settle(pi) + assert _received_ids(phone) == ids + + +@pytest.mark.parametrize("anchor", [7, "", None, {"id": "x"}]) +def test_invalid_anchor_does_not_fail_hello(pi, monkeypatch, anchor): + ids = _session(monkeypatch, 2) + + phone = _connect(pi, last_event_id=anchor) + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, 2) + settle(pi) + assert _received_ids(phone) == ids + + +def test_more_missed_shots_than_the_queue_holds_are_all_delivered(pi, monkeypatch): + count = pi.publisher.queue_size + 4 + ids = _session(monkeypatch, count) + + phone = _connect(pi) + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, count, timeout=15) + settle(pi) + assert _received_ids(phone) == ids + + +def test_catch_up_is_capped_to_most_recent_shots(pi, monkeypatch): + ids = _session(monkeypatch, CATCH_UP_LIMIT + 3) + + phone = _connect(pi) + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, CATCH_UP_LIMIT, timeout=20) + settle(pi, rounds=10) + assert _received_ids(phone) == ids[-CATCH_UP_LIMIT:] + + +def test_hello_after_shot_subscription_sends_catch_up_immediately(pi, monkeypatch): + ids = _session(monkeypatch, 3) + phone = pi.central() + phone.subscribe(*PHONE_UUIDS) + settle(pi) + + assert _hello(phone, last_event_id=ids[1])["ok"] is True + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, 2) + settle(pi) + assert _received_ids(phone) == ids[1:] + + +def test_replayed_shot_is_byte_identical_to_last_published_version(pi, monkeypatch): + """A provisional-then-final shot replays as the final, enrichment included.""" + _session(monkeypatch, 1) + shot_data = server_module.shot_to_dict(server_module.monitor.get_shots()[0]) + server_module._publish_phone_shot(shot_data, final=False, enrichment={"status": "pending"}) + server_module._publish_phone_shot(shot_data, final=True, enrichment={"status": "complete"}) + published = server_module.phone_shot_cache.get(stable_shot_event_id(shot_data)) + + phone = _connect(pi) + + shot = phone.wait_for(SHOT_CHARACTERISTIC_UUID) + settle(pi) + assert published is not None + assert shot == json.loads(published) + assert shot["final"] is True + assert shot["enrichment"] == {"status": "complete"} + assert phone.decoded(SHOT_CHARACTERISTIC_UUID) == [shot] + + +def test_deleted_shots_are_not_replayed(pi, monkeypatch): + ids = _session(monkeypatch, 3) + first = server_module.monitor.get_shots()[0] + server_module.apply_delete_shot({"timestamp": first.timestamp.isoformat()}) + + phone = _connect(pi) + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, 2) + settle(pi) + assert _received_ids(phone) == ids[1:] + + +def test_cleared_profile_shots_are_not_replayed(pi, monkeypatch): + active_id = server_module.get_profile_store().get_active().id + _session(monkeypatch, 2, profile_id=active_id) + server_module.apply_clear_session({}) + + phone = _connect(pi) + settle(pi, rounds=10) + + assert _received_ids(phone) == [] + + +def test_disconnect_before_shot_subscription_discards_pending_catch_up(pi, monkeypatch): + ids = _session(monkeypatch, 3) + _publish_latest_live() + phone = pi.central() + phone.subscribe(CONTROL_CHARACTERISTIC_UUID) + assert _hello(phone)["ok"] is True + phone.disconnect() + settle(pi) + + # Reconnects without a new hello: only the latest-shot replay, as before. + phone.subscribe(SHOT_CHARACTERISTIC_UUID) + phone.wait_for(SHOT_CHARACTERISTIC_UUID) + settle(pi, rounds=10) + assert _received_ids(phone) == [ids[-1]] + + +def test_rejected_hello_does_not_trigger_catch_up(pi, monkeypatch): + _session(monkeypatch, 3) + phone = pi.central() + phone.subscribe(CONTROL_CHARACTERISTIC_UUID) + + answer = _hello(phone, client_schema_max=1) + phone.subscribe(SHOT_CHARACTERISTIC_UUID) + settle(pi, rounds=10) + + assert answer["ok"] is False + assert _received_ids(phone) == [] + + +def test_catch_up_failure_keeps_hello_and_falls_back_to_latest_replay(pi, monkeypatch): + ids = _session(monkeypatch, 2) + _publish_latest_live() + + def broken(_last_event_id): + raise RuntimeError("session unavailable") + + monkeypatch.setattr(pi.publisher, "catch_up_provider", broken) + + phone = _connect(pi) + + phone.wait_for(SHOT_CHARACTERISTIC_UUID) + settle(pi, rounds=10) + assert _received_ids(phone) == [ids[-1]] + + +def test_no_monitor_means_no_catch_up(pi): + phone = _connect(pi) + settle(pi, rounds=10) + + assert _received_ids(phone) == [] diff --git a/tests/test_control_commands.py b/tests/test_control_commands.py index 79c38531a..dca8900f4 100644 --- a/tests/test_control_commands.py +++ b/tests/test_control_commands.py @@ -28,18 +28,21 @@ def _isolate_club_state(monkeypatch): """Keep club changes and their broadcasts from leaking into other tests. ``apply_club_selection`` writes the module-global ``active_club`` and fans - out over Socket.IO and BLE, so every test gets its own. + out over Socket.IO, the SSE broker and BLE, so every test gets its own. """ monkeypatch.setattr(server_module, "active_club", ClubType.DRIVER) + monkeypatch.setattr(server_module, "shot_stream", _ClubPublisher()) monkeypatch.setattr(server_module, "ble_publisher", None) monkeypatch.setattr(server_module.socketio, "emit", lambda *_args, **_kwargs: None) def test_apply_club_selection_updates_monitor_and_broadcasts(monkeypatch): monitor = _Monitor() + stream = _ClubPublisher() ble = _ClubPublisher() emitted = [] monkeypatch.setattr(server_module, "monitor", monitor) + monkeypatch.setattr(server_module, "shot_stream", stream) monkeypatch.setattr(server_module, "ble_publisher", ble) monkeypatch.setattr(server_module, "active_club", ClubType.DRIVER) monkeypatch.setattr( @@ -53,6 +56,7 @@ def test_apply_club_selection_updates_monitor_and_broadcasts(monkeypatch): assert monitor.clubs == [ClubType.IRON_7] assert emitted == [("club_changed", {"club": "7-iron"})] assert server_module.active_club is ClubType.IRON_7 + assert stream.clubs == ["7-iron"] assert ble.clubs == ["7-iron"] @@ -67,6 +71,29 @@ def test_apply_club_selection_rejects_unknown_club(monkeypatch): assert monitor.clubs == [] +def test_network_club_endpoint_uses_shared_selection_logic(monkeypatch): + monitor = _Monitor() + monkeypatch.setattr(server_module, "monitor", monitor) + + response = server_module.app.test_client().post( + "/api/club", + json={"club": "pw"}, + ) + + assert response.status_code == 200 + assert response.get_json() == {"status": "applied", "club": "pw"} + assert monitor.clubs == [ClubType.PW] + + +def test_network_club_endpoint_returns_authoritative_selection(monkeypatch): + monkeypatch.setattr(server_module, "active_club", ClubType.WOOD_3) + + response = server_module.app.test_client().get("/api/club") + + assert response.status_code == 200 + assert response.get_json() == {"status": "current", "club": "3-wood"} + + def test_control_dispatch_routes_club_command(monkeypatch): monitor = _Monitor() monkeypatch.setattr(server_module, "monitor", monitor) diff --git a/tests/test_phone_catch_up.py b/tests/test_phone_catch_up.py new file mode 100644 index 000000000..e485d7af0 --- /dev/null +++ b/tests/test_phone_catch_up.py @@ -0,0 +1,113 @@ +"""Selecting and caching the session shots a reconnecting phone missed.""" + +import threading + +import pytest + +from openflight.phone_catch_up import ( + CATCH_UP_LIMIT, + PhoneShotCache, + normalize_last_event_id, + select_catch_up, +) + + +def _entries(count): + return [(f"id-{index}", f"payload-{index}".encode()) for index in range(count)] + + +def _ids(entries): + return [event_id for event_id, _payload in entries] + + +class TestSelectCatchUp: + def test_known_id_returns_that_shot_and_later_ones_in_order(self): + """The named shot is resent too: the phone may only have its provisional.""" + assert _ids(select_catch_up(_entries(5), "id-2")) == ["id-2", "id-3", "id-4"] + + def test_latest_id_returns_only_that_shot(self): + assert _ids(select_catch_up(_entries(5), "id-4")) == ["id-4"] + + def test_no_id_returns_whole_session(self): + assert _ids(select_catch_up(_entries(3), None)) == ["id-0", "id-1", "id-2"] + + def test_unknown_id_returns_whole_session(self): + """A cleared session, deleted shot or Pi restart: the phone's anchor is gone.""" + assert _ids(select_catch_up(_entries(3), "not-in-session")) == ["id-0", "id-1", "id-2"] + + def test_empty_session_returns_nothing(self): + assert select_catch_up([], None) == [] + assert select_catch_up([], "id-0") == [] + + def test_caps_to_most_recent_shots(self): + selected = select_catch_up(_entries(CATCH_UP_LIMIT + 5), None) + + assert len(selected) == CATCH_UP_LIMIT + assert selected[0][0] == "id-5" + assert selected[-1][0] == f"id-{CATCH_UP_LIMIT + 4}" + + def test_caps_missed_shots_after_a_known_id(self): + selected = select_catch_up(_entries(30), "id-2", limit=4) + + assert _ids(selected) == ["id-26", "id-27", "id-28", "id-29"] + + def test_payloads_travel_with_their_ids(self): + assert select_catch_up(_entries(2), "id-1") == [("id-1", b"payload-1")] + + def test_rejects_non_positive_limit(self): + with pytest.raises(ValueError): + select_catch_up(_entries(2), None, limit=0) + + +class TestNormalizeLastEventId: + @pytest.mark.parametrize("value", [None, "", " ", 7, True, ["id"], {"id": 1}, "x" * 65]) + def test_invalid_values_mean_no_anchor(self, value): + assert normalize_last_event_id(value) is None + + def test_strips_whitespace(self): + assert normalize_last_event_id(" abc ") == "abc" + + def test_accepts_uuid(self): + event_id = "05dd37ec-49ed-596b-b1a4-953d54e4f239" + assert normalize_last_event_id(event_id) == event_id + + +class TestPhoneShotCache: + def test_remembers_latest_payload_per_event_id(self): + cache = PhoneShotCache() + cache.remember("a", b"provisional") + cache.remember("a", b"final") + + assert cache.get("a") == b"final" + assert cache.get("missing") is None + + def test_evicts_least_recently_written_when_full(self): + cache = PhoneShotCache(max_entries=2) + cache.remember("a", b"1") + cache.remember("b", b"2") + cache.remember("a", b"1-final") # refreshes "a" + cache.remember("c", b"3") + + assert cache.get("b") is None + assert cache.get("a") == b"1-final" + assert cache.get("c") == b"3" + assert len(cache) == 2 + + def test_rejects_non_positive_size(self): + with pytest.raises(ValueError): + PhoneShotCache(max_entries=0) + + def test_concurrent_writers_stay_bounded(self): + cache = PhoneShotCache(max_entries=50) + + def write(prefix): + for index in range(500): + cache.remember(f"{prefix}-{index}", b"x") + + threads = [threading.Thread(target=write, args=(name,)) for name in "abcd"] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert len(cache) == 50 diff --git a/tests/test_phone_transport.py b/tests/test_phone_transport.py index d083e262a..71ca076cc 100644 --- a/tests/test_phone_transport.py +++ b/tests/test_phone_transport.py @@ -1,5 +1,6 @@ -"""Server-level phone behaviour: event fan-out and command routing.""" +"""Server-level phone behaviour: the SSE stream, event fan-out and command routing.""" +import json from datetime import datetime import pytest @@ -9,6 +10,7 @@ from openflight.power import PowerStatus from openflight.power.models import PowerState from openflight.profiles import ProfileStore +from openflight.shot_stream import HEARTBEAT_FRAME, ShotStreamBroker def _shot_data(ball_speed=150.0, **extra): @@ -23,8 +25,19 @@ def _shot_data(ball_speed=150.0, **extra): return data +def _decode(event): + return event.name, json.loads(bytes(event).decode("utf-8")) + + +def _drain(subscriber): + events = [] + while not subscriber.empty(): + events.append(_decode(subscriber.get_nowait())) + return events + + class _PhoneTransport: - """Records what the server hands to the phone ble.""" + """Records what the server hands to one phone transport (BLE or SSE).""" def __init__(self): self.shots = [] @@ -46,9 +59,11 @@ def publish_club(self, club): @pytest.fixture def phones(monkeypatch, tmp_path): - """Capture Socket.IO and BLE output of a monitor-less server.""" + """Capture Socket.IO, SSE and BLE output of a monitor-less server.""" + stream = _PhoneTransport() ble = _PhoneTransport() emitted = [] + monkeypatch.setattr(server_module, "shot_stream", stream) monkeypatch.setattr(server_module, "ble_publisher", ble) monkeypatch.setattr(server_module, "monitor", None) monkeypatch.setattr(server_module, "power_monitor", None) @@ -59,14 +74,95 @@ def phones(monkeypatch, tmp_path): "emit", lambda event, data=None, **_kwargs: emitted.append((event, data)), ) - return ble, emitted + return stream, ble, emitted + + +# -- SSE broker ---------------------------------------------------------------------- + + +def test_every_subscriber_gets_shots_events_and_club_changes(): + broker = ShotStreamBroker() + first = broker.subscribe() + second = broker.subscribe() + + broker.publish_shot(_shot_data(), final=False, enrichment={"status": "pending"}) + broker.publish_event({"schema_version": 2, "type": "shot_processing", "state": "failed"}) + broker.publish_club("pw") + + for subscriber in (first, second): + events = _drain(subscriber) + assert [name for name, _ in events] == ["shot", "shot_processing", "club_changed"] + assert events[0][1]["final"] is False + assert events[2][1] == {"schema_version": 2, "type": "club_changed", "club": "pw"} + + +def test_subscriber_is_seeded_with_state_then_latest_shot(): + broker = ShotStreamBroker() + broker.publish_shot(_shot_data(140.0), final=True) + broker.publish_shot(_shot_data(151.0), final=True) + seed = [{"schema_version": 2, "type": "club_changed", "club": "7-iron"}] + + subscriber = broker.subscribe(initial_events=seed) + + events = _drain(subscriber) + assert [name for name, _ in events] == ["club_changed", "shot"] + assert events[1][1]["ball_speed_mph"] == 151.0 + + +def test_stream_route_seeds_current_state_then_session_shots(monkeypatch, tmp_path): + broker = ShotStreamBroker(heartbeat_interval_s=0.01) + # The session, not the broker's latest shot, decides what a client is + # seeded with (see tests/test_shot_stream_catch_up.py). + monitor = server_module.MockLaunchMonitor() + monitor._shots.append( + Shot( + ball_speed_mph=150.0, + timestamp=datetime(2026, 9, 25, 12, 0, 0, 1), + club=ClubType.DRIVER, + shot_number=1, + ) + ) + monkeypatch.setattr(server_module, "monitor", monitor) + monkeypatch.setattr(server_module, "phone_shot_cache", server_module.PhoneShotCache()) + monkeypatch.setattr(server_module, "shot_stream", broker) + monkeypatch.setattr(server_module, "active_club", ClubType.IRON_7) + monkeypatch.setattr(server_module, "power_monitor", None) + monkeypatch.setattr(server_module, "profile_store", ProfileStore(tmp_path / "profiles.json")) + + response = server_module.app.test_client().get("/api/shots/stream") + try: + assert response.status_code == 200 + frames = response.response + assert next(frames).decode("utf-8") == HEARTBEAT_FRAME + events = [] + for _ in range(3): + frame = next(frames).decode("utf-8") + name = frame.split("\n", 1)[0].removeprefix("event: ") + events.append((name, json.loads(frame.split("data: ", 1)[1]))) + finally: + response.close() + + assert [name for name, _ in events] == ["club_changed", "profiles", "shot"] + assert events[0][1] == {"schema_version": 2, "type": "club_changed", "club": "7-iron"} + assert events[1][1]["profiles"][0]["name"] == "Profile 1" + assert events[2][1]["schema_version"] == 2 and events[2][1]["final"] is True + assert broker.subscriber_count == 0 + + +@pytest.mark.parametrize("schema", ["1", "3"]) +def test_stream_route_rejects_other_schemas(monkeypatch, schema): + monkeypatch.setattr(server_module, "shot_stream", ShotStreamBroker()) + + response = server_module.app.test_client().get(f"/api/shots/stream?schema={schema}") + + assert response.status_code == 400 # -- event fan-out --------------------------------------------------------------------- def test_shot_processing_power_and_profiles_reach_phones(phones): - ble, emitted = phones + stream, ble, emitted = phones status = PowerStatus( available=True, provider="geekworm", @@ -81,23 +177,24 @@ def test_shot_processing_power_and_profiles_reach_phones(phones): server_module._on_power_status(status) server_module.handle_get_profiles() - types = [event["type"] for event in ble.events] - assert types == ["shot_processing", "power_status", "profiles"] - assert ble.events[0]["state"] == "calculating" - assert ble.events[1] == { - "schema_version": 2, - "type": "power_status", - **status.to_dict(), - } + for transport in (stream, ble): + types = [event["type"] for event in transport.events] + assert types == ["shot_processing", "power_status", "profiles"] + assert transport.events[0]["state"] == "calculating" + assert transport.events[1] == { + "schema_version": 2, + "type": "power_status", + **status.to_dict(), + } assert [event for event, _ in emitted] == ["shot_processing", "power_status", "profiles"] def test_club_changes_reach_every_transport(phones): - ble, emitted = phones + stream, ble, emitted = phones server_module.apply_club_selection({"club": "pw"}) - assert ble.clubs == ["pw"] + assert stream.clubs == ble.clubs == ["pw"] assert ("club_changed", {"club": "pw"}) in emitted @@ -105,7 +202,7 @@ def test_club_changes_reach_every_transport(phones): def test_dispatch_uses_the_socketio_operations(phones): - ble, emitted = phones + stream, _ble, emitted = phones store = server_module.get_profile_store() first = store.get_active() second = store.add("Sam") @@ -120,22 +217,25 @@ def test_dispatch_uses_the_socketio_operations(phones): assert (status, response) == (200, {"status": "applied", "active_profile_id": first.id}) assert via_ble == via_socket assert via_ble[0] == "profiles" - assert ble.events[-1]["active_profile_id"] == first.id - assert second.id in {item["id"] for item in ble.events[-1]["profiles"]} + assert stream.events[-1]["active_profile_id"] == first.id + assert second.id in {item["id"] for item in stream.events[-1]["profiles"]} def test_socket_clear_session_also_notifies_phones(phones): - ble, emitted = phones + stream, ble, emitted = phones active = server_module.get_profile_store().get_active().id server_module.handle_clear_session({}) assert emitted == [("session_cleared", {"profile_id": active, "shots": []})] - assert ble.events == [{"schema_version": 2, "type": "session_cleared", "profile_id": active}] + for transport in (stream, ble): + assert transport.events == [ + {"schema_version": 2, "type": "session_cleared", "profile_id": active} + ] def test_socket_delete_shot_notifies_phones_only_on_success(phones, monkeypatch): - ble, emitted = phones + stream, ble, emitted = phones shot = Shot( ball_speed_mph=150.0, timestamp=datetime(2026, 9, 25, 12, 0, 0, 1), @@ -146,23 +246,25 @@ def test_socket_delete_shot_notifies_phones_only_on_success(phones, monkeypatch) monkeypatch.setattr(server_module, "monitor", monitor) server_module.handle_delete_shot({"timestamp": "2026-01-01T00:00:00"}) - assert ble.events == [] + for transport in (stream, ble): + assert transport.events == [] server_module.handle_delete_shot({"timestamp": shot.timestamp.isoformat()}) assert [event for event, _ in emitted] == ["delete_shot_error", "session_state"] assert monitor.get_shots() == [] - assert ble.events == [ - { - "schema_version": 2, - "type": "shot_deleted", - "timestamp": shot.timestamp.isoformat(), - } - ] + for transport in (stream, ble): + assert transport.events == [ + { + "schema_version": 2, + "type": "shot_deleted", + "timestamp": shot.timestamp.isoformat(), + } + ] @pytest.mark.parametrize("command", ["clear_session", "delete_shot"]) def test_destructive_commands_are_not_routed_over_ble(phones, command): - _ble, emitted = phones + _stream, _ble, emitted = phones response, status = server_module.dispatch_phone_control_command( command, {"timestamp": "2026-09-25T12:00:00"} @@ -201,7 +303,7 @@ class _Monitor: def test_fast_path_publishes_one_final_shot_without_enrichment(phones, monkeypatch): - ble, _emitted = phones + stream, ble, _emitted = phones for name in ("kld7_vertical", "kld7_horizontal", "camera_capture_runtime", "iwr6843_runtime"): monkeypatch.setattr(server_module, name, None) for name in ("ball_speed_correction_enabled", "calculated_spin_enabled", "ballistics_enabled"): @@ -225,10 +327,12 @@ def test_fast_path_publishes_one_final_shot_without_enrichment(phones, monkeypat timeout=5, ) - [(shot_data, final, enrichment)] = ble.shots - assert final is True - assert enrichment is None - assert shot_data["ball_speed_mph"] == 150.0 + for transport in (stream, ble): + [(shot_data, final, enrichment)] = transport.shots + assert final is True + assert enrichment is None + assert shot_data["ball_speed_mph"] == 150.0 + assert stream.shots[0][0] is ble.shots[0][0] @pytest.mark.parametrize( diff --git a/tests/test_phone_transport_server.py b/tests/test_phone_transport_server.py index 763d911f5..23cd8a5fb 100644 --- a/tests/test_phone_transport_server.py +++ b/tests/test_phone_transport_server.py @@ -1,4 +1,4 @@ -"""Server-level behaviour of the phone transports (Socket.IO and BLE).""" +"""Server-level behaviour of the phone transports (Socket.IO, SSE and BLE).""" import logging import sys @@ -22,9 +22,11 @@ def publish_club(self, club): @pytest.fixture def no_monitor_transports(monkeypatch): """No launch monitor, with every club transport captured.""" + stream = _ClubPublisher() ble = _ClubPublisher() emitted = [] monkeypatch.setattr(server_module, "monitor", None) + monkeypatch.setattr(server_module, "shot_stream", stream) monkeypatch.setattr(server_module, "ble_publisher", ble) monkeypatch.setattr(server_module, "active_club", ClubType.DRIVER) monkeypatch.setattr( @@ -32,28 +34,57 @@ def no_monitor_transports(monkeypatch): "emit", lambda event, data, **_kwargs: emitted.append((event, data)), ) - return ble, emitted + return stream, ble, emitted def test_socket_set_club_without_monitor_still_broadcasts(no_monitor_transports): """Before the monitor exists, a club change is recorded and broadcast, as before.""" - ble, emitted = no_monitor_transports + stream, ble, emitted = no_monitor_transports server_module.handle_set_club({"club": "7-iron"}) assert server_module.active_club is ClubType.IRON_7 assert emitted == [("club_changed", {"club": "7-iron"})] + assert stream.clubs == ["7-iron"] assert ble.clubs == ["7-iron"] +def test_network_club_post_without_monitor_applies_and_broadcasts(no_monitor_transports): + stream, ble, emitted = no_monitor_transports + + response = server_module.app.test_client().post("/api/club", json={"club": "pw"}) + + assert response.status_code == 200 + assert response.get_json() == {"status": "applied", "club": "pw"} + assert server_module.active_club is ClubType.PW + assert emitted == [("club_changed", {"club": "pw"})] + assert stream.clubs == ["pw"] + assert ble.clubs == ["pw"] + + +@pytest.mark.parametrize("payload", [{"club": "putter"}, {"club": "unknown"}, {}, None, "pw"]) +def test_network_club_post_rejects_invalid_selection(no_monitor_transports, payload): + stream, ble, emitted = no_monitor_transports + + response = server_module.app.test_client().post("/api/club", json=payload) + + assert response.status_code == 400 + assert "error" in response.get_json() + assert server_module.active_club is ClubType.DRIVER + assert emitted == [] + assert stream.clubs == [] + assert ble.clubs == [] + + @pytest.mark.parametrize("payload", [{"club": "putter"}, {"club": "unknown"}, None]) def test_socket_set_club_ignores_invalid_selection(no_monitor_transports, payload): - ble, emitted = no_monitor_transports + stream, ble, emitted = no_monitor_transports server_module.handle_set_club(payload) assert server_module.active_club is ClubType.DRIVER assert emitted == [] + assert stream.clubs == [] assert ble.clubs == [] diff --git a/tests/test_server.py b/tests/test_server.py index 0a8d417df..18550fd5a 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -29,6 +29,7 @@ swing_speed_to_dict, swing_speed_to_shot_dict, ) +from openflight.shot_stream import HEARTBEAT_FRAME, SSE_MIMETYPE, ShotStreamBroker from openflight.swing_speed import SwingSpeedEvent @@ -1236,6 +1237,76 @@ def test_display_route_falls_back_when_dist_missing(self, monkeypatch, tmp_path) assert b'
' in response.data +class TestShotStreamRoute: + """Tests for the Server-Sent Events shot endpoint used by the iOS app.""" + + @staticmethod + def _shot_data(ball_speed=151.4): + return { + "timestamp": "2026-07-29T19:42:10", + "club": "driver", + "ball_speed_mph": ball_speed, + "club_speed_mph": None, + "smash_factor": None, + "estimated_carry_yards": 264, + "launch_angle_vertical": None, + "launch_angle_horizontal": None, + "spin_rpm": None, + "club_path_deg": None, + "spin_axis_deg": None, + } + + def test_stream_replays_session_shots_as_sse(self, monkeypatch): + """A phone that connects mid-session gets the session's shots immediately.""" + broker = ShotStreamBroker(heartbeat_interval_s=0.01) + monitor = MockLaunchMonitor() + monitor._shots.append( + Shot( + ball_speed_mph=151.4, + timestamp=datetime(2026, 7, 29, 19, 42, 10), + club=ClubType.DRIVER, + shot_number=1, + ) + ) + monkeypatch.setattr(server_module, "monitor", monitor) + monkeypatch.setattr(server_module, "phone_shot_cache", server_module.PhoneShotCache()) + monkeypatch.setattr(server_module, "shot_stream", broker) + + response = server_module.app.test_client().get("/api/shots/stream") + try: + assert response.status_code == 200 + assert response.mimetype == SSE_MIMETYPE + assert response.headers["Cache-Control"] == "no-cache" + + frames = response.response + assert next(frames).decode("utf-8") == HEARTBEAT_FRAME + + frame = next(frames).decode("utf-8") + while not frame.startswith("event: shot\n"): # state events come first + frame = next(frames).decode("utf-8") + event = json.loads(frame.split("data: ", 1)[1]) + assert frame.startswith(f"event: shot\nid: {event['event_id']}\ndata: ") + assert event["schema_version"] == 2 + assert event["ball_speed_mph"] == 151.4 + finally: + response.close() + + assert broker.subscriber_count == 0 + + def test_stream_refuses_clients_beyond_the_limit(self, monkeypatch): + """The threaded dev server must not be swamped by stream connections.""" + broker = ShotStreamBroker(heartbeat_interval_s=0.01, max_subscribers=1) + monkeypatch.setattr(server_module, "shot_stream", broker) + client = server_module.app.test_client() + + first = client.get("/api/shots/stream") + try: + second = client.get("/api/shots/stream") + assert second.status_code == 503 + finally: + first.close() + + class TestShotToDict: """Tests for shot_to_dict conversion.""" @@ -4183,6 +4254,69 @@ def test_ble_publish_survives_websocket_failure(self, monkeypatch): assert len(published) == 1 assert published[0]["ball_speed_mph"] == 100.0 + def test_shot_stream_publish_survives_websocket_failure(self, monkeypatch): + """A broken web client must not suppress the independent network transport.""" + broker = ShotStreamBroker() + subscriber = broker.subscribe() + + monkeypatch.setattr(server_module, "monitor", None) + monkeypatch.setattr(server_module, "sim_connectors", []) + monkeypatch.setattr(server_module, "ble_publisher", None) + monkeypatch.setattr(server_module, "shot_stream", broker) + monkeypatch.setattr(server_module, "debug_mode", False) + monkeypatch.setattr(server_module, "get_session_logger", lambda: None) + monkeypatch.setattr( + server_module.socketio, + "emit", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("socket closed")), + ) + + on_shot_detected( + Shot( + ball_speed_mph=100.0, + timestamp=datetime.now(), + club=ClubType.IRON_7, + mode="mock", + ) + ) + _wait_for_shot_finalization_idle() + + streamed = json.loads(subscriber.get_nowait()) + assert streamed["schema_version"] == 2 + assert streamed["ball_speed_mph"] == 100.0 + + def test_shot_stream_failure_does_not_break_shot_processing(self, monkeypatch): + """A raising broker must not escape into the shot pipeline.""" + forwarded = [] + + class BrokenBroker: + def publish(self, _shot_data): + raise RuntimeError("stream is wedged") + + monkeypatch.setattr(server_module, "monitor", None) + monkeypatch.setattr(server_module, "ble_publisher", None) + monkeypatch.setattr(server_module, "shot_stream", BrokenBroker()) + monkeypatch.setattr(server_module, "debug_mode", False) + monkeypatch.setattr(server_module, "get_session_logger", lambda: None) + monkeypatch.setattr(server_module.socketio, "emit", lambda *_a, **_k: None) + monkeypatch.setattr( + server_module, + "_forward_shot_to_simulators", + lambda shot: forwarded.append(shot), + ) + + on_shot_detected( + Shot( + ball_speed_mph=100.0, + timestamp=datetime.now(), + club=ClubType.IRON_7, + mode="mock", + ) + ) + _wait_for_shot_finalization_idle() + + assert len(forwarded) == 1 + def test_implausible_club_aoa_is_rejected(self, monkeypatch): """A +31° club AoA is physically impossible and should be discarded.""" diff --git a/tests/test_shot_stream.py b/tests/test_shot_stream.py new file mode 100644 index 000000000..6de3bb02b --- /dev/null +++ b/tests/test_shot_stream.py @@ -0,0 +1,207 @@ +"""Tests for Server-Sent Events shot delivery, buffering, and isolation.""" + +import json +import logging +import queue + +import pytest + +from openflight.shot_stream import ( + HEARTBEAT_FRAME, + ShotStreamBroker, + ShotStreamFull, + format_event, +) + + +def _shot_data(ball_speed=150.0): + return { + "timestamp": "2026-07-29T19:42:10", + "club": "driver", + "ball_speed_mph": ball_speed, + "club_speed_mph": None, + "smash_factor": None, + "estimated_carry_yards": 250, + "launch_angle_vertical": None, + "launch_angle_horizontal": None, + "spin_rpm": None, + "club_path_deg": None, + "spin_axis_deg": None, + } + + +def _decode_frame(frame): + """Decode one ``shot`` frame: ``event``, ``id`` (the event_id), ``data``, blank line.""" + assert frame.endswith("\n\n") + lines = frame[:-2].split("\n") + assert lines[0] == "event: shot" + assert lines[1].startswith("id: ") + event = json.loads(lines[2].removeprefix("data: ")) + assert lines[1] == f"id: {event['event_id']}" + assert len(lines) == 3 + return event + + +def test_rejects_nonsense_configuration(): + with pytest.raises(ValueError): + ShotStreamBroker(queue_size=0) + with pytest.raises(ValueError): + ShotStreamBroker(heartbeat_interval_s=0) + with pytest.raises(ValueError): + ShotStreamBroker(max_subscribers=0) + + +def test_event_frame_is_one_data_line_terminated_by_blank_line(): + frame = format_event(b'{"schema_version":2}') + + assert frame == 'event: shot\ndata: {"schema_version":2}\n\n' + assert frame.count("\ndata:") == 1 + + +def test_publish_reaches_every_subscriber(): + broker = ShotStreamBroker() + first = broker.subscribe() + second = broker.subscribe() + + assert broker.publish_shot(_shot_data(151.4), final=True) + + assert _decode_frame(format_event(first.get_nowait()))["ball_speed_mph"] == 151.4 + assert _decode_frame(format_event(second.get_nowait()))["ball_speed_mph"] == 151.4 + assert broker.subscriber_count == 2 + + +def test_club_change_reaches_every_subscriber(): + broker = ShotStreamBroker() + first = broker.subscribe() + second = broker.subscribe() + + assert broker.publish_club("3-wood") + + expected = ( + 'event: club_changed\ndata: {"club":"3-wood","schema_version":2,"type":"club_changed"}\n\n' + ) + assert format_event(first.get_nowait()) == expected + assert format_event(second.get_nowait()) == expected + + +def test_subscriber_replays_latest_shot_on_connect(): + broker = ShotStreamBroker() + broker.publish_shot(_shot_data(140.0), final=True) + broker.publish_shot(_shot_data(151.4), final=True) + + subscriber = broker.subscribe() + + replayed = _decode_frame(format_event(subscriber.get_nowait())) + assert replayed["ball_speed_mph"] == 151.4 + assert subscriber.empty() + + +def test_invalid_shot_does_not_raise_into_caller(caplog): + broker = ShotStreamBroker() + subscriber = broker.subscribe() + + with caplog.at_level(logging.WARNING): + assert broker.publish_shot({}, final=True) is False + + assert "Failed to encode shot payload" in caplog.text + assert subscriber.empty() + + +def test_full_queue_drops_oldest_unsent_shot(caplog): + broker = ShotStreamBroker(queue_size=2) + subscriber = broker.subscribe() + + with caplog.at_level(logging.WARNING): + for speed in (100.0, 110.0, 120.0): + broker.publish_shot(_shot_data(speed), final=True) + + assert "dropped oldest unsent shot" in caplog.text + remaining = [ + _decode_frame(format_event(subscriber.get_nowait()))["ball_speed_mph"] for _ in range(2) + ] + assert remaining == [110.0, 120.0] + + +def test_slow_subscriber_does_not_block_a_healthy_one(): + broker = ShotStreamBroker(queue_size=1) + slow = broker.subscribe() + healthy = broker.subscribe() + + broker.publish_shot(_shot_data(100.0), final=True) + healthy.get_nowait() + broker.publish_shot(_shot_data(120.0), final=True) + + assert _decode_frame(format_event(healthy.get_nowait()))["ball_speed_mph"] == 120.0 + assert _decode_frame(format_event(slow.get_nowait()))["ball_speed_mph"] == 120.0 + + +def test_subscriber_limit_is_enforced(): + broker = ShotStreamBroker(max_subscribers=1) + broker.subscribe() + + with pytest.raises(ShotStreamFull): + broker.subscribe() + + +def test_unsubscribe_frees_a_slot_and_tolerates_repeats(): + broker = ShotStreamBroker(max_subscribers=1) + subscriber = broker.subscribe() + + broker.unsubscribe(subscriber) + broker.unsubscribe(subscriber) + + assert broker.subscriber_count == 0 + assert broker.subscribe() is not subscriber + + +def test_stream_opens_with_a_heartbeat_so_response_headers_flush(): + """A WSGI server withholds headers until the first chunk of the body.""" + broker = ShotStreamBroker(heartbeat_interval_s=30.0) + frames = broker.frames(broker.subscribe()) + + assert next(frames) == HEARTBEAT_FRAME + + frames.close() + + +def test_stream_emits_shots_then_heartbeats_while_idle(): + broker = ShotStreamBroker(heartbeat_interval_s=0.01) + broker.publish_shot(_shot_data(151.4), final=True) + frames = broker.frames(broker.subscribe()) + + opening = next(frames) + replayed = next(frames) + idle = next(frames) + + assert opening == HEARTBEAT_FRAME + assert _decode_frame(replayed)["ball_speed_mph"] == 151.4 + assert idle == HEARTBEAT_FRAME + assert broker.subscriber_count == 1 + + frames.close() + assert broker.subscriber_count == 0 + + +def test_stream_unsubscribes_when_client_disconnects(): + broker = ShotStreamBroker(heartbeat_interval_s=0.01) + frames = broker.frames(broker.subscribe()) + next(frames) + + frames.close() + + assert broker.subscriber_count == 0 + broker.publish_shot(_shot_data(), final=True) + assert broker.subscriber_count == 0 + + +def test_publish_after_disconnect_still_replays_to_the_next_client(): + broker = ShotStreamBroker(heartbeat_interval_s=0.01) + frames = broker.frames(broker.subscribe()) + frames.close() + + broker.publish_shot(_shot_data(133.0), final=True) + subscriber = broker.subscribe() + + assert _decode_frame(format_event(subscriber.get_nowait()))["ball_speed_mph"] == 133.0 + with pytest.raises(queue.Empty): + subscriber.get_nowait() diff --git a/tests/test_shot_stream_catch_up.py b/tests/test_shot_stream_catch_up.py new file mode 100644 index 000000000..c52aa102e --- /dev/null +++ b/tests/test_shot_stream_catch_up.py @@ -0,0 +1,215 @@ +"""Network (SSE) schema v2 catch-up: the same rule BLE ``hello`` applies. + +A v2 stream client names the last shot it has with the standard +``Last-Event-ID`` header (or ``?last_event_id=``) and is seeded with the +session shots after it. v2 shot frames carry ``id: `` so an +``EventSource`` sends the header on its own when it reconnects. +""" + +import json +from datetime import datetime + +import pytest + +from openflight import server as server_module +from openflight.ble.protocol import encode_shot_event, stable_shot_event_id +from openflight.launch_monitor import ClubType, Shot +from openflight.phone_catch_up import CATCH_UP_LIMIT, PhoneShotCache +from openflight.profiles import ProfileStore +from openflight.shot_stream import HEARTBEAT_FRAME, ShotStreamBroker, format_event + + +def _shot_data(number=1, ball_speed=150.0): + return { + "timestamp": f"2026-09-28T12:00:{number:02d}.000001", + "club": "driver", + "ball_speed_mph": ball_speed, + "estimated_carry_yards": 250, + "shot_number": number, + } + + +def _catch_up_entry(number): + data = _shot_data(number) + return stable_shot_event_id(data), encode_shot_event(data, final=True) + + +def _drain(subscriber): + events = [] + while not subscriber.empty(): + events.append(subscriber.get_nowait()) + return events + + +def _parse(frame): + """(name, id or None, decoded data) from one SSE frame.""" + fields = dict(line.split(": ", 1) for line in frame.strip().split("\n")) + return fields["event"], fields.get("id"), json.loads(fields["data"]) + + +# -- broker --------------------------------------------------------------------------- + + +def test_catch_up_replaces_latest_seed_and_keeps_order(): + broker = ShotStreamBroker() + broker.publish_shot(_shot_data(9), final=True) + entries = [_catch_up_entry(number) for number in (1, 2, 3)] + state = [{"schema_version": 2, "type": "club_changed", "club": "7-iron"}] + + subscriber = broker.subscribe(initial_events=state, catch_up=entries) + + frames = [_parse(format_event(event)) for event in _drain(subscriber)] + assert [name for name, _id, _data in frames] == ["club_changed", "shot", "shot", "shot"] + assert [event_id for _name, event_id, _data in frames[1:]] == [eid for eid, _ in entries] + assert [data["shot_number"] for _name, _id, data in frames[1:]] == [1, 2, 3] + assert frames[0][1] is None # only shots carry an id + + +def test_empty_catch_up_seeds_no_shot(): + broker = ShotStreamBroker() + broker.publish_shot(_shot_data(9), final=True) + + subscriber = broker.subscribe(catch_up=[]) + + assert _drain(subscriber) == [] + + +def test_without_catch_up_the_latest_shot_is_still_seeded(): + broker = ShotStreamBroker() + broker.publish_shot(_shot_data(9), final=True) + + events = _drain(broker.subscribe()) + + assert [_parse(format_event(event))[2]["shot_number"] for event in events] == [9] + + +def test_full_catch_up_fits_the_seed_queue(): + broker = ShotStreamBroker(queue_size=2) + entries = [_catch_up_entry(number) for number in range(1, CATCH_UP_LIMIT + 1)] + state = [{"schema_version": 2, "type": "club_changed", "club": "driver"}] + + events = _drain(broker.subscribe(initial_events=state, catch_up=entries)) + + assert len(events) == CATCH_UP_LIMIT + 1 + + +def test_live_shot_frames_carry_event_id_and_other_events_do_not(): + broker = ShotStreamBroker() + subscriber = broker.subscribe() + + broker.publish_shot(_shot_data(5), final=False, enrichment={"status": "pending"}) + broker.publish_event({"schema_version": 2, "type": "shot_processing", "state": "failed"}) + + frames = [_parse(format_event(event)) for event in _drain(subscriber)] + assert frames[0][1] == stable_shot_event_id(_shot_data(5)) + assert frames[0][2]["event_id"] == frames[0][1] + assert frames[1][1] is None + + +# -- route ---------------------------------------------------------------------------- + + +@pytest.fixture +def session(monkeypatch, tmp_path): + """A mock monitor with five shots behind the real stream route; returns event ids.""" + monitor = server_module.MockLaunchMonitor() + for index in range(5): + monitor._shots.append( + Shot( + ball_speed_mph=120.0 + index, + timestamp=datetime(2026, 9, 28, 12, 0, index, 1000), + club=ClubType.DRIVER, + shot_number=index + 1, + ) + ) + monkeypatch.setattr(server_module, "monitor", monitor) + monkeypatch.setattr(server_module, "shot_stream", ShotStreamBroker(heartbeat_interval_s=0.01)) + monkeypatch.setattr(server_module, "phone_shot_cache", PhoneShotCache()) + monkeypatch.setattr(server_module, "active_club", ClubType.DRIVER) + monkeypatch.setattr(server_module, "power_monitor", None) + monkeypatch.setattr(server_module, "profile_store", ProfileStore(tmp_path / "profiles.json")) + return [stable_shot_event_id(server_module.shot_to_dict(shot)) for shot in monitor.get_shots()] + + +def _stream_shot_ids(path, headers=None): + """Event ids of the shots seeded on connect (before the first idle heartbeat).""" + response = server_module.app.test_client().get(path, headers=headers or {}) + try: + assert response.status_code == 200 + frames = response.response + assert next(frames).decode("utf-8") == HEARTBEAT_FRAME + ids = [] + for raw in frames: + frame = raw.decode("utf-8") + if frame == HEARTBEAT_FRAME: + break + name, event_id, data = _parse(frame) + if name == "shot": + assert event_id == data.get("event_id") + ids.append(event_id) + return ids + finally: + response.close() + + +def test_route_without_anchor_seeds_whole_session(session): + assert _stream_shot_ids("/api/shots/stream?schema=2") == session + + +def test_route_last_event_id_header_resends_it_and_seeds_missed_shots(session): + ids = _stream_shot_ids("/api/shots/stream?schema=2", {"Last-Event-ID": session[2]}) + + assert ids == session[2:] + + +def test_route_last_event_id_query_parameter_matches_header(session): + assert _stream_shot_ids(f"/api/shots/stream?schema=2&last_event_id={session[2]}") == session[2:] + + +def test_route_header_takes_precedence_over_query(session): + ids = _stream_shot_ids( + f"/api/shots/stream?schema=2&last_event_id={session[0]}", + {"Last-Event-ID": session[3]}, + ) + + assert ids == session[3:] + + +def test_route_unknown_anchor_seeds_whole_session(session): + ids = _stream_shot_ids("/api/shots/stream?schema=2", {"Last-Event-ID": "not-a-shot"}) + + assert ids == session + + +def test_route_without_schema_parameter_is_schema_2_with_catch_up(session): + assert _stream_shot_ids("/api/shots/stream", {"Last-Event-ID": session[2]}) == session[2:] + + +@pytest.mark.parametrize("schema", ["1", "3", "two"]) +def test_route_rejects_other_schemas(session, schema): + response = server_module.app.test_client().get(f"/api/shots/stream?schema={schema}") + + assert response.status_code == 400 + assert server_module.shot_stream.subscriber_count == 0 + + +def test_route_matches_ble_catch_up_for_the_same_anchor(session): + """Both transports apply one rule: same anchor, same shots, same bytes.""" + shared = server_module.phone_catch_up(session[1]) + + assert _stream_shot_ids("/api/shots/stream?schema=2", {"Last-Event-ID": session[1]}) == [ + event_id for event_id, _payload in shared + ] + + +def test_route_survives_catch_up_failure_with_latest_replay(session, monkeypatch): + server_module.shot_stream.publish_shot( + server_module.shot_to_dict(server_module.monitor.get_shots()[-1]), final=True + ) + + def broken(_last_event_id=None): + raise RuntimeError("session unavailable") + + monkeypatch.setattr(server_module, "phone_catch_up", broken) + + assert _stream_shot_ids("/api/shots/stream?schema=2") == [session[-1]] From 42ba71b36669292d2f3c4223760695c06e87bce6 Mon Sep 17 00:00:00 2001 From: btrippcsci Date: Sun, 11 Oct 2026 09:55:49 -0400 Subject: [PATCH 2/2] fix(phone): replay missed deletions and long gaps on reconnect A tester's phone missed a kiosk deletion and 35 of 55 shots taken while it was disconnected: catch-up never told the phone about deletions and sent only the newest 20 shots. - Catch-up now sends the shots deleted while the phone was away first, as shot_deleted events: those made after the named shot was first sent, or every remembered one (the last 100) when the anchor is unknown. PhoneShotCache orders sends and deletions with one counter. - The catch-up limit is 100 shots instead of 20. - A BLE catch-up leaves two delivery-queue slots free, so a live shot taken meanwhile no longer pushes a missed shot out of the queue. - hello advertises deletion_catch_up, and the network stream names the same features in an OpenFlight-Features header, so a phone app can tell a Pi that predates them and suggest an update. Co-Authored-By: Claude Opus 5.5 --- docs/changelog.md | 11 +- docs/ios-ble.md | 40 ++++-- src/openflight/ble/protocol.py | 1 + src/openflight/ble/publisher.py | 70 ++++++++-- src/openflight/phone_catch_up.py | 78 +++++++++-- src/openflight/server.py | 32 ++++- tests/ble_harness.py | 1 + .../ble_goldens/v2_response_hello.json | 53 ++++---- tests/test_ble_catch_up.py | 124 ++++++++++++++++++ tests/test_phone_catch_up.py | 71 ++++++++++ tests/test_shot_stream_catch_up.py | 70 ++++++++++ 11 files changed, 484 insertions(+), 67 deletions(-) diff --git a/docs/changelog.md b/docs/changelog.md index ed9101f02..828feb09b 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -80,9 +80,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 changes are broadcast on the stream too. A reconnecting phone names the newest shot it has (`last_event_id` in BLE `hello`, or `Last-Event-ID` on the stream) and is sent that shot again plus every later current-session shot, - oldest first, up to 20; replays are the exact bytes sent live, and cleared or - deleted shots are never replayed. `hello` advertises the `shot_catch_up` - feature. See [phone app connection](ios-ble.md#catch-up-after-a-reconnect). + oldest first, up to 100; replays are the exact bytes sent live, and cleared + or deleted shots are never replayed. Shots deleted on the kiosk while the + phone was away are sent first, as `shot_deleted` events, and a shot taken + during a BLE catch-up no longer pushes a missed shot out of the delivery + queue. `hello` advertises the `shot_catch_up` and `deletion_catch_up` + features, and the stream names the same list in an `OpenFlight-Features` + header, so phone apps can tell an older Pi apart. See + [phone app connection](ios-ble.md#catch-up-after-a-reconnect). - **Bluetooth LE connection for phone apps.** `--ble` (with the optional `ble` extra, `bless==0.3.0` on Linux) advertises one GATT service with a shot and a control characteristic, speaking schema version 2; without BlueZ or the diff --git a/docs/ios-ble.md b/docs/ios-ble.md index e811dc53c..99e5d519c 100644 --- a/docs/ios-ble.md +++ b/docs/ios-ble.md @@ -206,7 +206,7 @@ timeouts. ``` The result names the schema, the features and the characteristics: ```json - {"ok":true,"request_id":"","result":{"characteristics":{"control":"7BA96E63-12C2-4CE0-BB84-3513C7FD1474","shot":"ED365FE6-3ABF-4FC3-8E44-D9525A22DABD"},"features":["provisional_shots","shot_processing","profiles","power_status","shot_deleted","club","shot_catch_up"],"schema_version":2},"schema_version":2} + {"ok":true,"request_id":"","result":{"characteristics":{"control":"7BA96E63-12C2-4CE0-BB84-3513C7FD1474","shot":"ED365FE6-3ABF-4FC3-8E44-D9525A22DABD"},"features":["provisional_shots","shot_processing","profiles","power_status","shot_deleted","club","shot_catch_up","deletion_catch_up"],"schema_version":2},"schema_version":2} ``` 3. Subscribe to the shot characteristic. The catch-up shots arrive. A client that skipped `hello` gets only the latest shot. @@ -223,7 +223,8 @@ transports: | | BLE | Network | |---|---|---| | Name the newest shot you have | `last_event_id` in the `hello` payload | `Last-Event-ID` request header, or `?last_event_id=` (the header wins) | -| Catch-up arrives | On the shot characteristic, after the `hello` response; held until the phone subscribes to it | Seeded after the state events, before live events | +| Missed deletions arrive | `shot_deleted` events on the control characteristic, right after the `hello` response | `shot_deleted` events seeded after the state events | +| Missed shots arrive | On the shot characteristic, after the `hello` response; held until the phone subscribes to it | Seeded after the missed deletions, before live events | | Without catch-up | A client that skips `hello` gets the latest shot | — (every connection gets catch-up) | - The Pi sends the named shot again, then every current-session shot after @@ -233,8 +234,16 @@ transports: - No `last_event_id`, or one the session no longer holds (the session was cleared, that shot was deleted, or the Pi restarted), means the whole current session. -- At most the 20 most recent shots are sent. Use `shot_number` gaps to tell - that older shots were not synced (deleted shots also leave gaps). +- At most the 100 most recent shots are sent. Use `shot_number` gaps to tell + that older shots were not synced (deleted shots also leave gaps). Over BLE a + long catch-up takes a while: each shot is a few dozen notifications. +- Shots deleted while the phone was away are sent first, as `shot_deleted` + events: every deletion made after the named shot was first sent, or, when + the phone names no shot or one the Pi no longer knows (for example because + that shot was itself deleted), every deletion the Pi remembers (the last + 100, since it started). Deleting a shot the app no longer has must be a + no-op, so an extra deletion is harmless. A cleared session is not replayed: + its shots are simply absent from the catch-up. - Replayed shots are the exact bytes last sent live, including `final` and `enrichment`. A shot no longer cached is rebuilt as a final shot with `enrichment: null`. @@ -248,13 +257,15 @@ transports: `id`, which leaves the last shot id in place. - Over BLE a notification reaches every subscribed phone, so another connected phone receives the catch-up too and upserts it. -- A shot taken while a BLE catch-up is still being sent can push the oldest - queued catch-up shot out of the eight-message delivery queue. The phone then - lacks that one shot until its session is replayed in full (for example after - it reconnects without `last_event_id`). +- A BLE catch-up leaves two slots of the eight-message delivery queue free, + so a shot taken while it is still being sent is delivered alongside it + instead of pushing a missed shot out. -The Pi advertises support with the `shot_catch_up` feature in the `hello` -result. +The Pi advertises support with the `shot_catch_up` (missed shots) and +`deletion_catch_up` (missed deletions) features in the `hello` result. The +network stream has no `hello`, so its response carries the same list in an +`OpenFlight-Features` header (comma-separated). An app can use either to tell +that a Pi predates these features and suggest updating it. ### Shot @@ -317,8 +328,9 @@ unbounded `settings` object could not be guaranteed to fit in one BLE message. accepted for clients that name it; any other `schema` value returns `400`. A stream opens with `: ping`, then the current state as `club_changed`, `profiles` and (with a battery monitor) `power_status`, then the -[catch-up](#catch-up-after-a-reconnect) shots (the whole session unless -`Last-Event-ID` names a shot). Each `shot` frame carries `id: `. +[catch-up](#catch-up-after-a-reconnect) deletions and shots (the whole session +unless `Last-Event-ID` names a shot). The `OpenFlight-Features` response header +lists the Pi's features, as `hello` does over BLE. Each `shot` frame carries `id: `. Event names match the `type` of the payload. Commands stay on `/api/club` and Socket.IO. @@ -338,8 +350,8 @@ message that would not fit instead of sending a truncated one. queued event is dropped if that client cannot keep up. One stalled phone cannot slow down another. - Disconnecting clears that client's queue. On the next connection the client - is [caught up](#catch-up-after-a-reconnect) on the session shots it missed - (up to 20). + is [caught up](#catch-up-after-a-reconnect) on the deletions and session + shots it missed (up to 100 shots). - Clients upsert shots by `event_id`, which ignores replays of shots they already have and merges a provisional shot with its final version. diff --git a/src/openflight/ble/protocol.py b/src/openflight/ble/protocol.py index c1cd80b1a..3ca7467d1 100644 --- a/src/openflight/ble/protocol.py +++ b/src/openflight/ble/protocol.py @@ -68,6 +68,7 @@ "shot_deleted", "club", "shot_catch_up", + "deletion_catch_up", ) EVENT_TYPES = ( diff --git a/src/openflight/ble/publisher.py b/src/openflight/ble/publisher.py index 50e36a145..37dd0304e 100644 --- a/src/openflight/ble/publisher.py +++ b/src/openflight/ble/publisher.py @@ -30,6 +30,13 @@ # Given a client's ``last_event_id`` (or ``None``), the ``(event_id, shot # payload)`` pairs it missed, oldest first. See ``openflight.phone_catch_up``. CatchUpProvider = Callable[[Any], Sequence[tuple[str, bytes]]] +# Given the same ``last_event_id``, the events it missed that are not shots +# (``shot_deleted``), oldest first, sent on the control characteristic. +CatchUpEventsProvider = Callable[[Any], Sequence[Mapping]] +# Delivery-queue slots catch-up leaves free, so a live shot taken while +# catch-up is still being sent never pushes a missed shot out of the queue. +_LIVE_HEADROOM = 2 +_HEADROOM_POLL_S = 0.01 _SHOT = SHOT_CHARACTERISTIC_UUID.lower() _CONTROL = CONTROL_CHARACTERISTIC_UUID.lower() @@ -67,6 +74,7 @@ def __init__( # pylint: disable=too-many-arguments fragment_interval_s: float = 0.01, command_handler: CommandHandler | None = None, catch_up_provider: CatchUpProvider | None = None, + catch_up_events_provider: CatchUpEventsProvider | None = None, ): if queue_size < 1: raise ValueError("BLE queue size must be at least one") @@ -75,6 +83,7 @@ def __init__( # pylint: disable=too-many-arguments self.fragment_interval_s = fragment_interval_s self.command_handler = command_handler self.catch_up_provider = catch_up_provider + self.catch_up_events_provider = catch_up_events_provider self._state_lock = threading.Lock() self._thread: threading.Thread | None = None @@ -361,6 +370,25 @@ async def _load_catch_up(self, last_event_id) -> list[bytes] | None: logger.warning("[BLE] Could not load missed shots for catch-up", exc_info=True) return None + async def _load_catch_up_events(self, last_event_id) -> list[bytes]: + """The encoded deletions a client missed; empty when unavailable.""" + provider = self.catch_up_events_provider + if provider is None: + return [] + try: + events = await asyncio.to_thread(provider, last_event_id) + return [encode_message(event) for event in events] + except Exception: # pylint: disable=broad-exception-caught + # Never fail ``hello`` over catch-up: the shots still arrive. + logger.warning("[BLE] Could not load missed deletions for catch-up", exc_info=True) + return [] + + async def _send_catch_up(self, events: list[bytes], shots: list[bytes] | None) -> None: + """Missed deletions on the control characteristic, then the missed shots.""" + for event_payload in events: + await self._send_control_response(event_payload) + self._schedule_catch_up(shots) + def _schedule_catch_up(self, payloads: list[bytes] | None) -> None: """Send catch-up now if the shot characteristic is subscribed, else on subscribe. @@ -387,14 +415,22 @@ def _start_catch_up(self, payloads: list[bytes]) -> None: asyncio.run_coroutine_threadsafe(self._deliver_catch_up(payloads), loop) async def _deliver_catch_up(self, payloads: list[bytes]) -> None: - """Queue catch-up shots in order, waiting for room instead of dropping any.""" + """Queue catch-up shots in order, waiting for room instead of dropping any. + + Catch-up leaves ``_LIVE_HEADROOM`` slots free, so a live shot finds + room instead of dropping the oldest queued (missed) shot. + """ for payload in payloads: - with self._state_lock: - queue = self._queue - subscribed = self._subscribed - if queue is None or not subscribed: - return - await queue.put(payload) + while True: + with self._state_lock: + queue = self._queue + subscribed = self._subscribed + if queue is None or not subscribed: + return + if queue.qsize() < max(1, queue.maxsize - _LIVE_HEADROOM): + break + await asyncio.sleep(_HEADROOM_POLL_S) + queue.put_nowait(payload) @staticmethod def _offer(queue: asyncio.Queue[bytes] | None, payload: bytes) -> None: @@ -486,6 +522,7 @@ def _on_write_request(self, characteristic, value, **_kwargs) -> None: async def _process_control_payload(self, payload: bytes) -> None: request_id = "unknown" catch_up: list[bytes] | None = None + catch_up_events: list[bytes] = [] try: command = json.loads(payload) if not isinstance(command, dict): @@ -503,7 +540,9 @@ async def _process_control_payload(self, payload: bytes) -> None: raise ValueError("Control command payload must be an object") if command_type == "hello": - response, catch_up = await self._answer_hello(request_id, command_payload) + response, catch_up, catch_up_events = await self._answer_hello( + request_id, command_payload + ) else: if self.command_handler is None: raise ValueError("Phone controls are not configured on this OpenFlight server") @@ -525,22 +564,27 @@ async def _process_control_payload(self, payload: bytes) -> None: ) await self._send_control_response(encode_message(response)) - # After the answer, so a client sees ``hello`` succeed before shots arrive. - self._schedule_catch_up(catch_up) + # After the answer, so a client sees ``hello`` succeed before catch-up arrives. + await self._send_catch_up(catch_up_events, catch_up) async def _answer_hello( self, request_id: str, command_payload: Mapping, - ) -> tuple[dict, list[bytes] | None]: - """Negotiate the transport and load the client's catch-up shots. + ) -> tuple[dict, list[bytes] | None, list[bytes]]: + """Negotiate the transport and load the client's catch-up. ``hello`` is answered by the publisher itself: it is about which characteristics exist, not server state. """ result = build_hello_result(command_payload.get("client_schema_max")) response = build_control_response(request_id, result=result) - return response, await self._load_catch_up(command_payload.get("last_event_id")) + last_event_id = command_payload.get("last_event_id") + return ( + response, + await self._load_catch_up(last_event_id), + await self._load_catch_up_events(last_event_id), + ) async def _send_control_response(self, payload: bytes) -> None: """Send one complete control message without interleaving fragments.""" diff --git a/src/openflight/phone_catch_up.py b/src/openflight/phone_catch_up.py index 60d3a6edf..5d5f2351f 100644 --- a/src/openflight/phone_catch_up.py +++ b/src/openflight/phone_catch_up.py @@ -10,11 +10,17 @@ the most recent ``CATCH_UP_LIMIT`` shots are sent, and clients upsert them by ``event_id`` so a shot they already have is harmless. +Deletions the client missed come first, as ``shot_deleted`` events: every +deletion made after the named shot was first sent, or every remembered one +when the client names no shot or one the Pi no longer knows. Deleting a shot +the phone no longer has is a no-op, so an extra deletion is harmless. + The session itself (``monitor.get_shots()``) decides which shots exist, so -clears, deletes and profile changes need no bookkeeping here. ``PhoneShotCache`` -only remembers the exact bytes last published per ``event_id``, so a replayed -shot carries the same ``final`` and ``enrichment`` state the phone would have -received live. +clears, deletes and profile changes need no bookkeeping for the shots. +``PhoneShotCache`` remembers the exact bytes last published per ``event_id``, +so a replayed shot carries the same ``final`` and ``enrichment`` state the +phone would have received live, and the order in which shots were first sent +and shots were deleted, which places a client's anchor among the deletions. """ from __future__ import annotations @@ -23,9 +29,13 @@ from collections import OrderedDict from typing import Sequence -CATCH_UP_LIMIT = 20 +# Covers a long gap (a tester once missed 55 shots). Over BLE each shot is a +# few dozen 20-byte notifications, so this is about a minute of catch-up. +CATCH_UP_LIMIT = 100 # Far above any realistic session; bounds memory on a Pi left running for days. DEFAULT_CACHE_ENTRIES = 512 +# Deletions replayed to a phone whose anchor is unknown; one kiosk session's worth. +MAX_REMEMBERED_DELETIONS = 100 # Event ids are UUIDs (36 characters); anything much longer is not one of ours. MAX_EVENT_ID_LENGTH = 64 @@ -70,23 +80,47 @@ def select_catch_up( class PhoneShotCache: - """The latest encoded shot per ``event_id``, bounded and thread-safe.""" + """The latest encoded shot per ``event_id`` and the recent deletions. + + Bounded and thread-safe. One counter orders shots (by their first send) + and deletions, so ``deletions_since`` can tell which deletions came after + the shot a reconnecting phone names. + """ - def __init__(self, max_entries: int = DEFAULT_CACHE_ENTRIES): + def __init__( + self, + max_entries: int = DEFAULT_CACHE_ENTRIES, + max_deletions: int = MAX_REMEMBERED_DELETIONS, + ): if max_entries < 1: raise ValueError("Phone shot cache must hold at least one entry") + if max_deletions < 1: + raise ValueError("Phone shot cache must remember at least one deletion") self.max_entries = max_entries + self.max_deletions = max_deletions self._lock = threading.Lock() - self._payloads: OrderedDict[str, bytes] = OrderedDict() + self._counter = 0 + # event_id -> (latest payload, position of its first send) + self._payloads: OrderedDict[str, tuple[bytes, int]] = OrderedDict() + # timestamp -> position of the deletion, oldest first + self._deletions: OrderedDict[str, int] = OrderedDict() def __len__(self) -> int: with self._lock: return len(self._payloads) + def _next_position(self) -> int: + self._counter += 1 + return self._counter + def remember(self, event_id: str, payload: bytes) -> None: """Store the newest payload for a shot (a final replaces its provisional).""" with self._lock: - self._payloads[event_id] = payload + known = self._payloads.get(event_id) + # Keep the first send's position: a phone that got only the + # provisional must still hear of deletions made before the final. + position = known[1] if known is not None else self._next_position() + self._payloads[event_id] = (payload, position) self._payloads.move_to_end(event_id) while len(self._payloads) > self.max_entries: self._payloads.popitem(last=False) @@ -94,4 +128,28 @@ def remember(self, event_id: str, payload: bytes) -> None: def get(self, event_id: str) -> bytes | None: """The last published payload for ``event_id``, if still cached.""" with self._lock: - return self._payloads.get(event_id) + known = self._payloads.get(event_id) + return known[0] if known is not None else None + + def remember_deletion(self, timestamp) -> None: + """Record a deleted shot by its timestamp (the ``shot_deleted`` key).""" + if not isinstance(timestamp, str) or not timestamp: + return + with self._lock: + self._deletions.pop(timestamp, None) + self._deletions[timestamp] = self._next_position() + while len(self._deletions) > self.max_deletions: + self._deletions.popitem(last=False) + + def deletions_since(self, last_event_id: str | None) -> list[str]: + """Timestamps deleted after ``last_event_id`` was first sent, oldest first. + + Every remembered deletion when ``last_event_id`` is ``None`` or no + longer cached (deleted, evicted, or from before a Pi restart). + """ + with self._lock: + known = self._payloads.get(last_event_id) if last_event_id is not None else None + since = known[1] if known is not None else 0 + return [ + timestamp for timestamp, position in self._deletions.items() if position > since + ] diff --git a/src/openflight/server.py b/src/openflight/server.py index 76efd7efa..4a5852ea9 100644 --- a/src/openflight/server.py +++ b/src/openflight/server.py @@ -26,6 +26,7 @@ from .ballistics import resolve_launch, simulate from .ble.protocol import ( + FEATURES as PHONE_FEATURES, build_club_event, build_power_status_event, build_profiles_event, @@ -1613,6 +1614,15 @@ def handle_get_camera_capture_settings(): socketio.emit("camera_capture_settings", _camera_capture_settings_payload()) +def _stream_deletion_catch_up(last_event_id) -> list[dict]: + """Missed deletions for a stream client, or none on failure.""" + try: + return phone_deletion_catch_up(last_event_id) + except Exception: # pylint: disable=broad-exception-caught + logger.warning("[SERVER] Could not load missed deletions for the stream", exc_info=True) + return [] + + def _stream_catch_up(last_event_id) -> list[tuple[str, bytes]] | None: """Catch-up for a stream client, or ``None`` (latest-shot replay) on failure.""" try: @@ -1629,14 +1639,16 @@ def shots_stream(): Speaks schema 2, the only phone schema; ``?schema=2`` is accepted for clients that name it. A client resumes with ``Last-Event-ID`` (or ``?last_event_id=``) and is seeded with the session shots it missed, as - BLE ``hello`` does. + BLE ``hello`` does: missed deletions after the state events, then the + missed shots. ``OpenFlight-Features`` names the same features ``hello`` + returns, so a phone can tell what this Pi supports. """ if request.args.get("schema", "2") != "2": return {"error": "Unsupported schema; use 2"}, 400 last_event_id = request.headers.get("Last-Event-ID") or request.args.get("last_event_id") try: subscriber = shot_stream.subscribe( - initial_events=_phone_state_events(), + initial_events=_phone_state_events() + _stream_deletion_catch_up(last_event_id), catch_up=_stream_catch_up(last_event_id), ) except ShotStreamFull as exc: @@ -1646,6 +1658,7 @@ def shots_stream(): response = Response(shot_stream.frames(subscriber), mimetype=SSE_MIMETYPE) response.headers["Cache-Control"] = "no-cache" response.headers["X-Accel-Buffering"] = "no" + response.headers["OpenFlight-Features"] = ",".join(PHONE_FEATURES) # Covers the case where the response is discarded without ever being # iterated; unsubscribing twice is a no-op. response.call_on_close(lambda: shot_stream.unsubscribe(subscriber)) @@ -2178,6 +2191,8 @@ def apply_delete_shot(payload): return {"error": "Shot not found"}, 404 socketio.emit("session_state", _session_state_payload()) + # Remembered so a phone that is away now hears of it when it reconnects. + phone_shot_cache.remember_deletion(timestamp) _publish_phone_event(build_shot_deleted_event(timestamp)) return {"status": "deleted", "timestamp": timestamp}, 200 @@ -3750,6 +3765,18 @@ def phone_catch_up(last_event_id=None) -> list[tuple[str, bytes]]: return select_catch_up(entries, normalize_last_event_id(last_event_id)) +def phone_deletion_catch_up(last_event_id=None) -> list[dict]: + """``shot_deleted`` events a reconnecting phone missed, oldest first. + + Shared by BLE ``hello`` and the network stream; sent before the catch-up + shots. See ``openflight.phone_catch_up`` for the rule. + """ + return [ + build_shot_deleted_event(timestamp) + for timestamp in phone_shot_cache.deletions_since(normalize_last_event_id(last_event_id)) + ] + + def _publish_phone_event(event: dict) -> None: """Hand one event to the BLE and SSE phone transports; never raises.""" transports = [("network stream", shot_stream)] @@ -5400,6 +5427,7 @@ def main(): ble_publisher = BleShotPublisher( command_handler=dispatch_phone_control_command, catch_up_provider=phone_catch_up, + catch_up_events_provider=phone_deletion_catch_up, ) ble_publisher.start() print("Bluetooth LE enabled (advertising as OpenFlight)") diff --git a/tests/ble_harness.py b/tests/ble_harness.py index f04d6ecb3..bd2b3152c 100644 --- a/tests/ble_harness.py +++ b/tests/ble_harness.py @@ -382,6 +382,7 @@ def emit(event, payload=None, **_kwargs): monkeypatch, command_handler=server_module.dispatch_phone_control_command, catch_up_provider=server_module.phone_catch_up, + catch_up_events_provider=server_module.phone_deletion_catch_up, ) with loopback: monkeypatch.setattr(server_module, "ble_publisher", loopback.publisher) diff --git a/tests/fixtures/ble_goldens/v2_response_hello.json b/tests/fixtures/ble_goldens/v2_response_hello.json index a018b6264..cbf19811e 100644 --- a/tests/fixtures/ble_goldens/v2_response_hello.json +++ b/tests/fixtures/ble_goldens/v2_response_hello.json @@ -18,7 +18,8 @@ "power_status", "shot_deleted", "club", - "shot_catch_up" + "shot_catch_up", + "deletion_catch_up" ], "characteristics": { "shot": "ED365FE6-3ABF-4FC3-8E44-D9525A22DABD", @@ -26,30 +27,32 @@ } } }, - "payload_hex": "7b226f6b223a747275652c22726571756573745f6964223a2235453046324334412d384231442d344333452d394636412d374432423143304539413834222c22726573756c74223a7b22636861726163746572697374696373223a7b22636f6e74726f6c223a2237424139364536332d313243322d344345302d424238342d333531334337464431343734222c2273686f74223a2245443336354645362d334142462d344643332d384534342d443935323541323244414244227d2c226665617475726573223a5b2270726f766973696f6e616c5f73686f7473222c2273686f745f70726f63657373696e67222c2270726f66696c6573222c22706f7765725f737461747573222c2273686f745f64656c65746564222c22636c7562222c2273686f745f63617463685f7570225d2c22736368656d615f76657273696f6e223a327d2c22736368656d615f76657273696f6e223a327d", + "payload_hex": "7b226f6b223a747275652c22726571756573745f6964223a2235453046324334412d384231442d344333452d394636412d374432423143304539413834222c22726573756c74223a7b22636861726163746572697374696373223a7b22636f6e74726f6c223a2237424139364536332d313243322d344345302d424238342d333531334337464431343734222c2273686f74223a2245443336354645362d334142462d344643332d384534342d443935323541323244414244227d2c226665617475726573223a5b2270726f766973696f6e616c5f73686f7473222c2273686f745f70726f63657373696e67222c2270726f66696c6573222c22706f7765725f737461747573222c2273686f745f64656c65746564222c22636c7562222c2273686f745f63617463685f7570222c2264656c6574696f6e5f63617463685f7570225d2c22736368656d615f76657273696f6e223a327d2c22736368656d615f76657273696f6e223a327d", "frames_hex": [ - "01000000177b226f6b223a747275652c22726571", - "0100000117756573745f6964223a223545304632", - "01000002174334412d384231442d344333452d39", - "01000003174636412d3744324231433045394138", - "010000041734222c22726573756c74223a7b2263", - "0100000517686172616374657269737469637322", - "01000006173a7b22636f6e74726f6c223a223742", - "01000007174139364536332d313243322d344345", - "0100000817302d424238342d3335313343374644", - "010000091731343734222c2273686f74223a2245", - "0100000a17443336354645362d334142462d3446", - "0100000b1743332d384534342d44393532354132", - "0100000c173244414244227d2c22666561747572", - "0100000d176573223a5b2270726f766973696f6e", - "0100000e17616c5f73686f7473222c2273686f74", - "0100000f175f70726f63657373696e67222c2270", - "0100001017726f66696c6573222c22706f776572", - "01000011175f737461747573222c2273686f745f", - "010000121764656c65746564222c22636c756222", - "01000013172c2273686f745f63617463685f7570", - "0100001417225d2c22736368656d615f76657273", - "0100001517696f6e223a327d2c22736368656d61", - "01000016175f76657273696f6e223a327d" + "01000000197b226f6b223a747275652c22726571", + "0100000119756573745f6964223a223545304632", + "01000002194334412d384231442d344333452d39", + "01000003194636412d3744324231433045394138", + "010000041934222c22726573756c74223a7b2263", + "0100000519686172616374657269737469637322", + "01000006193a7b22636f6e74726f6c223a223742", + "01000007194139364536332d313243322d344345", + "0100000819302d424238342d3335313343374644", + "010000091931343734222c2273686f74223a2245", + "0100000a19443336354645362d334142462d3446", + "0100000b1943332d384534342d44393532354132", + "0100000c193244414244227d2c22666561747572", + "0100000d196573223a5b2270726f766973696f6e", + "0100000e19616c5f73686f7473222c2273686f74", + "0100000f195f70726f63657373696e67222c2270", + "0100001019726f66696c6573222c22706f776572", + "01000011195f737461747573222c2273686f745f", + "010000121964656c65746564222c22636c756222", + "01000013192c2273686f745f63617463685f7570", + "0100001419222c2264656c6574696f6e5f636174", + "010000151963685f7570225d2c22736368656d61", + "01000016195f76657273696f6e223a327d2c2273", + "01000017196368656d615f76657273696f6e223a", + "0100001819327d" ] } diff --git a/tests/test_ble_catch_up.py b/tests/test_ble_catch_up.py index 4a2b4fe48..77cd79b6d 100644 --- a/tests/test_ble_catch_up.py +++ b/tests/test_ble_catch_up.py @@ -275,3 +275,127 @@ def test_no_monitor_means_no_catch_up(pi): settle(pi, rounds=10) assert _received_ids(phone) == [] + + +def _publish_all_live(): + """Publish every session shot as the live path did when it was taken.""" + for shot in server_module.monitor.get_shots(): + server_module._publish_phone_shot( + server_module.shot_to_dict(shot), final=True, enrichment=None + ) + + +def _delete(index): + """Delete the session's ``index``-th shot on the kiosk; returns its timestamp.""" + timestamp = server_module.monitor.get_shots()[index].timestamp.isoformat() + assert server_module.apply_delete_shot({"timestamp": timestamp})[1] == 200 + return timestamp + + +def _deleted_timestamps(phone): + return [ + message["timestamp"] + for message in phone.decoded(CONTROL_CHARACTERISTIC_UUID) + if message.get("type") == "shot_deleted" + ] + + +def test_hello_advertises_deletion_catch_up(pi): + phone = pi.central() + phone.subscribe(CONTROL_CHARACTERISTIC_UUID) + + assert "deletion_catch_up" in _hello(phone)["result"]["features"] + + +def test_deletion_made_while_the_phone_was_away_is_replayed(pi, monkeypatch): + """Tester report: a shot deleted on the kiosk while the phone was disconnected stayed.""" + ids = _session(monkeypatch, 4) + _publish_all_live() + deleted = _delete(1) + + phone = _connect(pi, last_event_id=ids[-1]) + + phone.wait_for(CONTROL_CHARACTERISTIC_UUID, lambda m: m.get("type") == "shot_deleted") + settle(pi) + assert _deleted_timestamps(phone) == [deleted] + + +def test_deletion_replay_arrives_after_the_hello_answer(pi, monkeypatch): + ids = _session(monkeypatch, 3) + _publish_all_live() + _delete(0) + + phone = _connect(pi, last_event_id=ids[-1]) + phone.wait_for(CONTROL_CHARACTERISTIC_UUID, lambda m: m.get("type") == "shot_deleted") + + kinds = [m.get("type", "response") for m in phone.decoded(CONTROL_CHARACTERISTIC_UUID)] + assert kinds == ["response", "shot_deleted"] + + +def test_deletions_the_phone_already_saw_are_not_replayed(pi, monkeypatch): + ids = _session(monkeypatch, 4) + shots = server_module.monitor.get_shots() + for shot in shots[:2]: + server_module._publish_phone_shot( + server_module.shot_to_dict(shot), final=True, enrichment=None + ) + _delete(0) # before the phone's newest shot was sent: it heard this one live + for shot in server_module.monitor.get_shots()[1:]: + server_module._publish_phone_shot( + server_module.shot_to_dict(shot), final=True, enrichment=None + ) + + phone = _connect(pi, last_event_id=ids[-1]) + phone.wait_for(SHOT_CHARACTERISTIC_UUID) + settle(pi, rounds=10) + + assert _deleted_timestamps(phone) == [] + + +def test_deleted_anchor_replays_every_remembered_deletion(pi, monkeypatch): + """The phone's newest shot was the one deleted: its anchor is gone, so replay them all.""" + ids = _session(monkeypatch, 3) + _publish_all_live() + first = _delete(0) + newest = _delete(1) # now index 1 is the old third shot, the phone's newest + + phone = _connect(pi, last_event_id=ids[-1]) + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, 1) + settle(pi, rounds=10) + + assert _deleted_timestamps(phone) == [first, newest] + + +def test_no_deletion_replay_without_a_monitor_or_deletions(pi): + phone = _connect(pi, last_event_id="anything") + settle(pi, rounds=10) + + assert _deleted_timestamps(phone) == [] + + +def test_a_gap_of_55_shots_is_delivered_in_full(pi, monkeypatch): + """Tester report: 55 shots missed while away, only the newest 20 came back.""" + ids = _session(monkeypatch, 56) + _publish_all_live() + + phone = _connect(pi, last_event_id=ids[0]) + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, 56, timeout=30) + settle(pi) + assert _received_ids(phone) == ids + + +def test_a_live_shot_during_catch_up_does_not_push_a_catch_up_shot_out(pi, monkeypatch): + count = pi.publisher.queue_size * 3 + ids = _session(monkeypatch, count + 1) + live = server_module.monitor._shots.pop() + monkeypatch.setattr(pi.publisher, "fragment_interval_s", 0.002) + + phone = _connect(pi) + phone.wait_for(SHOT_CHARACTERISTIC_UUID) + server_module.monitor._shots.append(live) + _publish_latest_live() + + phone.wait_for_count(SHOT_CHARACTERISTIC_UUID, count + 1, timeout=30) + settle(pi) + assert sorted(_received_ids(phone)) == sorted(ids) diff --git a/tests/test_phone_catch_up.py b/tests/test_phone_catch_up.py index e485d7af0..9daf226d4 100644 --- a/tests/test_phone_catch_up.py +++ b/tests/test_phone_catch_up.py @@ -6,6 +6,7 @@ from openflight.phone_catch_up import ( CATCH_UP_LIMIT, + MAX_REMEMBERED_DELETIONS, PhoneShotCache, normalize_last_event_id, select_catch_up, @@ -54,6 +55,11 @@ def test_caps_missed_shots_after_a_known_id(self): def test_payloads_travel_with_their_ids(self): assert select_catch_up(_entries(2), "id-1") == [("id-1", b"payload-1")] + def test_limit_covers_a_long_gap(self): + """A tester missed 55 shots; the old limit of 20 lost the oldest 35.""" + assert CATCH_UP_LIMIT >= 100 + assert len(select_catch_up(_entries(55), "id-0")) == 55 + def test_rejects_non_positive_limit(self): with pytest.raises(ValueError): select_catch_up(_entries(2), None, limit=0) @@ -111,3 +117,68 @@ def write(prefix): thread.join() assert len(cache) == 50 + + +class TestDeletionsSince: + """Deletions a reconnecting phone missed, in the order the Pi made them.""" + + def test_deletions_after_the_anchor_was_sent(self): + cache = PhoneShotCache() + cache.remember("a", b"1") + cache.remember_deletion("ts-old") + cache.remember("b", b"2") + cache.remember_deletion("ts-1") + cache.remember_deletion("ts-2") + + assert cache.deletions_since("b") == ["ts-1", "ts-2"] + + def test_no_deletions_after_the_anchor(self): + cache = PhoneShotCache() + cache.remember_deletion("ts-old") + cache.remember("a", b"1") + + assert cache.deletions_since("a") == [] + + @pytest.mark.parametrize("anchor", [None, "unknown"]) + def test_no_or_unknown_anchor_returns_every_remembered_deletion(self, anchor): + """The phone's anchor may itself have been deleted: replay all, they are idempotent.""" + cache = PhoneShotCache() + cache.remember_deletion("ts-1") + cache.remember("a", b"1") + cache.remember_deletion("ts-2") + + assert cache.deletions_since(anchor) == ["ts-1", "ts-2"] + + def test_anchor_position_is_its_first_send(self): + """A phone may hold only the provisional: deletions after it still count.""" + cache = PhoneShotCache() + cache.remember("a", b"provisional") + cache.remember_deletion("ts-1") + cache.remember("a", b"final") + + assert cache.deletions_since("a") == ["ts-1"] + + def test_a_repeated_deletion_is_reported_once_at_its_latest_position(self): + cache = PhoneShotCache() + cache.remember_deletion("ts-1") + cache.remember("a", b"1") + cache.remember_deletion("ts-1") + + assert cache.deletions_since(None) == ["ts-1"] + assert cache.deletions_since("a") == ["ts-1"] + + def test_keeps_only_the_most_recent_deletions(self): + cache = PhoneShotCache() + for index in range(MAX_REMEMBERED_DELETIONS + 3): + cache.remember_deletion(f"ts-{index}") + + deletions = cache.deletions_since(None) + assert len(deletions) == MAX_REMEMBERED_DELETIONS + assert deletions[0] == "ts-3" + + @pytest.mark.parametrize("timestamp", [None, "", 7]) + def test_ignores_invalid_timestamps(self, timestamp): + cache = PhoneShotCache() + cache.remember_deletion(timestamp) + + assert cache.deletions_since(None) == [] diff --git a/tests/test_shot_stream_catch_up.py b/tests/test_shot_stream_catch_up.py index c52aa102e..bedea09b0 100644 --- a/tests/test_shot_stream_catch_up.py +++ b/tests/test_shot_stream_catch_up.py @@ -213,3 +213,73 @@ def broken(_last_event_id=None): monkeypatch.setattr(server_module, "phone_catch_up", broken) assert _stream_shot_ids("/api/shots/stream?schema=2") == [session[-1]] + + +def _seeded_events(path, headers=None): + """``(name, data)`` of every event seeded on connect, and the response headers.""" + response = server_module.app.test_client().get(path, headers=headers or {}) + try: + assert response.status_code == 200 + frames = response.response + assert next(frames).decode("utf-8") == HEARTBEAT_FRAME + events = [] + for raw in frames: + frame = raw.decode("utf-8") + if frame == HEARTBEAT_FRAME: + break + name, _event_id, data = _parse(frame) + events.append((name, data)) + return events, response.headers + finally: + response.close() + + +def _publish_session_live(monkeypatch): + """Publish every session shot as the live path did when it was taken.""" + monkeypatch.setattr(server_module, "ble_publisher", None) + for shot in server_module.monitor.get_shots(): + server_module._publish_phone_shot( + server_module.shot_to_dict(shot), final=True, enrichment=None + ) + + +def test_route_replays_deletions_missed_while_away_before_the_shots(session, monkeypatch): + _publish_session_live(monkeypatch) + timestamp = server_module.monitor.get_shots()[1].timestamp.isoformat() + assert server_module.apply_delete_shot({"timestamp": timestamp})[1] == 200 + + events, _headers = _seeded_events("/api/shots/stream?schema=2", {"Last-Event-ID": session[-1]}) + + names = [name for name, _data in events] + assert [data["timestamp"] for name, data in events if name == "shot_deleted"] == [timestamp] + assert names.index("shot_deleted") < names.index("shot") + + +def test_route_matches_ble_deletion_replay_for_the_same_anchor(session, monkeypatch): + _publish_session_live(monkeypatch) + timestamp = server_module.monitor.get_shots()[0].timestamp.isoformat() + server_module.apply_delete_shot({"timestamp": timestamp}) + + events, _headers = _seeded_events("/api/shots/stream?schema=2", {"Last-Event-ID": session[-1]}) + + assert [data for name, data in events if name == "shot_deleted"] == ( + server_module.phone_deletion_catch_up(session[-1]) + ) + + +def test_route_names_its_features_in_a_header(session): + """A stream has no ``hello``; the app reads the same feature list from this header.""" + _events, headers = _seeded_events("/api/shots/stream?schema=2") + + features = headers["OpenFlight-Features"].split(",") + assert "shot_catch_up" in features + assert "deletion_catch_up" in features + + +def test_route_survives_deletion_replay_failure(session, monkeypatch): + def broken(_last_event_id=None): + raise RuntimeError("deletions unavailable") + + monkeypatch.setattr(server_module, "phone_deletion_catch_up", broken) + + assert _stream_shot_ids("/api/shots/stream?schema=2") == session