From d9f7bae867a575b24dc83e2617cfe0e8475d3a57 Mon Sep 17 00:00:00 2001 From: trekkateer Date: Sat, 17 Oct 2026 02:16:33 -0400 Subject: [PATCH] Fixed the issue where switching off title slides led to an offset. Also fixed a race condition where if an edit where made on air, the API would indicate a signature change 10 seconds before the SQLite DB got updated, causing new changes to be ignored --- docs/PROCLAIM_DATA_FORMAT.md | 2 + proclaim_feed.py | 48 +++++++++++++++-------- proclaim_lib.py | 9 ++++- proclaim_service.py | 11 +++++- tests/test_proclaim_feed.py | 75 ++++++++++++++++++++++++++++++++++-- 5 files changed, 124 insertions(+), 21 deletions(-) diff --git a/docs/PROCLAIM_DATA_FORMAT.md b/docs/PROCLAIM_DATA_FORMAT.md index 1bc5063..adeb299 100644 --- a/docs/PROCLAIM_DATA_FORMAT.md +++ b/docs/PROCLAIM_DATA_FORMAT.md @@ -100,6 +100,8 @@ The `Content` JSON structure varies by `ServiceItemKind`. All text content is st - **`UseCustomTransition`** — `"true"` if using custom transitions. - **`CustomTransitionKind`** — e.g., `"LyricScrolling"`. - **`CustomTransitionDuration`** — e.g., `"0"`. +- **`ShowTitle`** - `"true"` if the title of a song is displayed +- **`TitleSlide`** - `"Title"`, `"Final"`, `"FirstLyric"`, or `"Every"`. Title and Final will add a title slide to the deck. #### `Content` diff --git a/proclaim_feed.py b/proclaim_feed.py index d1cd773..db5f093 100644 --- a/proclaim_feed.py +++ b/proclaim_feed.py @@ -33,6 +33,13 @@ DEFAULT_PROCLAIM_BASE_URL = 'http://127.0.0.1:52195' +# How long after an item's revision changes we keep re-reading it from the DB. Proclaim's +# API reports a new localRevision as soon as the operator edits an item (e.g. its title +# slide setting), but the SQLite row can be written later; a single read at the moment the +# revision moves can see the old Content and would then be cached under the new revision +# for the rest of the run. +DEFAULT_REVISION_SETTLE_WINDOW = 10.0 + class ProclaimClient: """Client for Proclaim's local HTTP API.""" @@ -83,8 +90,9 @@ class ProclaimFeed: """A ``SlideFeed`` backed by Proclaim's HTTP API + SQLite DB. ``poll`` never raises on a source problem (an unreachable/off-air Proclaim yields an - off-air snapshot). Items are re-parsed only when their Proclaim ``localRevision`` - signature changes; ``existing_translation`` grounding is stamped onto each item at + off-air snapshot). Items are re-parsed when their Proclaim ``localRevision`` + signature changes, and on every order sync for ``revision_settle_window`` seconds + after that (the DB can lag the API); ``existing_translation`` grounding is stamped onto each item at parse time so downstream consumers (and, later, a replay feed) need no DB access. """ @@ -94,11 +102,13 @@ def __init__( proclaim_base_url: str = DEFAULT_PROCLAIM_BASE_URL, db: Optional[ProclaimDB] = None, order_sync_interval: float = 2.0, + revision_settle_window: float = DEFAULT_REVISION_SETTLE_WINDOW, report_exception: Optional[Callable[[Exception], None]] = None, ): self.client = ProclaimClient(proclaim_base_url) self.db = db if db is not None else ProclaimDB() self.order_sync_interval = order_sync_interval + self.revision_settle_window = revision_settle_window self._report_exception = report_exception or (lambda _e: None) self._seq = 0 self.reset() @@ -111,6 +121,7 @@ def reset(self) -> None: """Drop all source-side caches (called on doc rollover).""" self._items: Dict[str, FeedItem] = {} self._item_revisions: Dict[str, str] = {} + self._settling_until: Dict[str, float] = {} self._order: List[str] = [] self._last_order_sync = float('-inf') self._presentation_id: Optional[str] = None @@ -154,7 +165,7 @@ async def poll(self) -> FeedSnapshot: ): presentation = await self._fetch_onair_presentation() if presentation: - self._sync_order(presentation) + self._sync_order(presentation, now) self._last_order_sync = now return FeedSnapshot( @@ -194,12 +205,14 @@ async def _fetch_onair_presentation(self) -> Optional[Dict[str, Any]]: # -- order sync / parsing -------------------------------------------------- - def _sync_order(self, presentation: Dict[str, Any]) -> None: + def _sync_order(self, presentation: Dict[str, Any], now: float) -> None: """Update the cached order + items from the on-air presentation. - Only items whose Proclaim ``localRevision`` signature changed (or that we haven't - seen) are re-read from the DB; a DB/parse error for one item is logged and skipped - (its revision is left uncached so we retry next sync) rather than dropping the order. + Items whose Proclaim ``localRevision`` signature changed (or that we haven't seen) + are re-read from the DB, and keep being re-read until ``revision_settle_window`` + has passed since the change, so a DB write that lands after the API's revision bump + is still picked up. A DB/parse error for one item is logged and skipped (its + revision is left uncached so we retry next sync) rather than dropping the order. """ signatures = service_item_signatures(presentation) self._order = [sig['id'] for sig in signatures if sig['id']] @@ -208,11 +221,14 @@ def _sync_order(self, presentation: Dict[str, Any]) -> None: item_id = sig['id'] if not item_id: continue - unchanged = ( - self._item_revisions.get(item_id) == sig['revision'] - and item_id in self._items + cached = self._items.get(item_id) + revision_changed = ( + self._item_revisions.get(item_id) != sig['revision'] or cached is None ) - if unchanged: + if revision_changed: + self._settling_until[item_id] = now + self.revision_settle_window + elif now >= self._settling_until.get(item_id, float('-inf')): + self._settling_until.pop(item_id, None) continue try: @@ -223,11 +239,13 @@ def _sync_order(self, presentation: Dict[str, Any]) -> None: if not parsed: continue - self._items[item_id] = self._to_feed_item(parsed) + item = self._to_feed_item(parsed) + self._items[item_id] = item self._item_revisions[item_id] = sig['revision'] - logger.info( - f"Synced item {item_id} ({parsed.title}) with {len(parsed.slides)} original slides" - ) + if cached is None or cached.slides_hash != item.slides_hash: + logger.info( + f"Synced item {item_id} ({parsed.title}) with {len(parsed.slides)} original slides" + ) def _to_feed_item(self, parsed) -> FeedItem: """Convert a parsed ServiceItemWithSlides into a FeedItem with hash + grounding.""" diff --git a/proclaim_lib.py b/proclaim_lib.py index 63e4f53..60fc0c7 100644 --- a/proclaim_lib.py +++ b/proclaim_lib.py @@ -302,8 +302,13 @@ def _slides_from_source_xml(item_kind: str, content: dict, source_xml: str) -> L sections = split_into_song_sections(source_text) order_str = content.get('CustomOrderSequence') or '' slides = get_slides_in_order(sections, order_str) - if title := content.get('SongDisplayTitle'): - slides.insert(0, title) + + """If the title is displayed, add title slide. Otherwise don't""" + if (title := content.get('SongDisplayTitle')) and content.get('ShowTitle') == "true": + if content.get('TitleSlide') == 'Title': + slides.insert(0, title) + elif content.get('TitleSlide') == 'Final': + slides.append(title) return slides return split_into_slides(source_text) diff --git a/proclaim_service.py b/proclaim_service.py index 3ce4b83..5f228b2 100644 --- a/proclaim_service.py +++ b/proclaim_service.py @@ -19,6 +19,7 @@ import signal import socket import subprocess +import sys import time from pathlib import Path from typing import Any, Dict, Optional @@ -209,6 +210,11 @@ def build_pusher( async def signal_handler(cancel_scope: anyio.CancelScope): + # asyncio on Windows has no add_signal_handler, so open_signal_receiver raises + # NotImplementedError and takes the whole task group down with it. There, Ctrl+C + # arrives as KeyboardInterrupt instead, which the __main__ block turns into a clean exit. + if sys.platform == 'win32': + return with anyio.open_signal_receiver(signal.SIGINT, signal.SIGTERM) as signals: async for signum in signals: signal_name = signal.strsignal(signum) or str(signum) @@ -284,4 +290,7 @@ async def main(): if __name__ == '__main__': - anyio.run(main) + try: + anyio.run(main) + except KeyboardInterrupt: + logger.info("Received Ctrl+C, shutting down...") diff --git a/tests/test_proclaim_feed.py b/tests/test_proclaim_feed.py index 16fb63a..d8f22b6 100644 --- a/tests/test_proclaim_feed.py +++ b/tests/test_proclaim_feed.py @@ -13,11 +13,19 @@ pytestmark = pytest.mark.anyio -def make_feed(order_sync_interval: float = 0.0) -> ProclaimFeed: - """Build a feed with the DB faked and the order always eligible to sync.""" +def make_feed(order_sync_interval: float = 0.0, revision_settle_window: float = 0.0) -> ProclaimFeed: + """Build a feed with the DB faked and the order always eligible to sync. + + The settle window defaults to 0 here so tests that aren't about it see exactly one + re-parse per revision change. + """ db = mock.MagicMock() db.get_presentation.return_value = None - return ProclaimFeed(db=db, order_sync_interval=order_sync_interval) + return ProclaimFeed( + db=db, + order_sync_interval=order_sync_interval, + revision_settle_window=revision_settle_window, + ) def onair_status(item_id="i1", slide_index=0, presentation_id="pres-1"): @@ -79,6 +87,65 @@ def fake_parse(_db, item_id): assert calls == ["i1", "i2", "i1"] +def test_lagging_db_write_is_picked_up_within_settle_window(): + """The API bumps the revision before Proclaim writes the DB row: the first re-read sees + the old Content, so the item keeps being re-read until the window closes.""" + feed = make_feed(revision_settle_window=10.0) + presentation = presentation_with([ + {"id": "i1", "title": "Song", "kind": "SongLyrics", "slides": [{"localRevision": 1}]}, + ]) + db_slides = ["Verse"] + calls = [] + + def fake_parse(_db, item_id): + calls.append(item_id) + return ServiceItemWithSlides(item_id, "Song", list(db_slides), "SongLyrics") + + with mock.patch.object(pf, "parse_item_original", fake_parse): + feed._sync_order(presentation, now=0.0) + # Past the first sighting's window: settled, no re-read. + feed._sync_order(presentation, now=20.0) + assert len(calls) == 1 + + # Operator adds a title slide: revision moves, DB still has the old row. + presentation["serviceItems"][0]["slides"].append({"localRevision": 2}) + feed._sync_order(presentation, now=30.0) + assert feed._items["i1"].slides == ["Verse"] + + # DB catches up; the same revision is re-read inside the window. + db_slides[:] = ["Song", "Verse"] + feed._sync_order(presentation, now=32.0) + assert feed._items["i1"].slides == ["Song", "Verse"] + assert len(calls) == 3 + + # Window closed: back to trusting the revision. + feed._sync_order(presentation, now=41.0) + feed._sync_order(presentation, now=43.0) + assert len(calls) == 3 + + +def test_settle_window_restarts_on_each_revision_change(): + """A second edit inside the window extends re-reading from that edit.""" + feed = make_feed(revision_settle_window=10.0) + presentation = presentation_with([ + {"id": "i1", "title": "X", "kind": "Content", "slides": [{"localRevision": 1}]}, + ]) + calls = [] + + def fake_parse(_db, item_id): + calls.append(item_id) + return ServiceItemWithSlides(item_id, "X", ["A"], "Content") + + with mock.patch.object(pf, "parse_item_original", fake_parse): + feed._sync_order(presentation, now=0.0) + presentation["serviceItems"][0]["slides"][0]["localRevision"] = 2 + feed._sync_order(presentation, now=8.0) + feed._sync_order(presentation, now=15.0) # inside the window opened at 8 + assert len(calls) == 3 + feed._sync_order(presentation, now=18.0) + assert len(calls) == 3 + + async def test_poll_force_syncs_when_active_item_uncached(): """Even before the throttle elapses, an unseen active item forces an order fetch.""" feed = make_feed(order_sync_interval=999.0) # throttle would otherwise block a sync @@ -168,6 +235,7 @@ def test_reset_clears_caches(): feed = make_feed() feed._items = {"i1": object()} feed._item_revisions = {"i1": "r"} + feed._settling_until = {"i1": 5.0} feed._order = ["i1"] feed._session_pres_id = "pres-1" feed._translation_screen_idx = 3 @@ -176,6 +244,7 @@ def test_reset_clears_caches(): assert feed._items == {} assert feed._item_revisions == {} + assert feed._settling_until == {} assert feed._order == [] assert feed._session_pres_id is None assert feed._translation_screen_idx is None