feat(schedule): show which asset is on screen right now

.claude/skills/anthias-viewer/SKILL.md

@@ -28,6 +28,8 @@ known; OPEN items are flagged.
 ## Viewer ↔ Redis pub/sub messaging
 
 - **Command channel is `anthias.viewer`** (Redis pub/sub); server↔viewer request-reply uses `anthias.viewer` publish + BLPOP on `anthias.reply.<correlation-id>`.
+- **Now-playing is a published fact, not a round trip (#3177).** The viewer writes `viewer:now_playing_asset_id` on every rotation; `anthias_common.now_playing` announces changes on `anthias.now_playing` (deduped, plus a 1 s floor: `duration` may be 0, so rotation is not self-limiting). `page_context.assets()` reads the key to flag `Asset.is_now_playing`, a transient attribute rather than a column. **The TTL is liveness, not content**: 180 s, refreshed by the viewer's `now-playing-refresher` thread every 60 s. Deriving it from the asset's duration is the obvious-looking mistake, because durations run to a year. `blank` calls `clear()`; `stop` deliberately does not, because it leaves a frame on screen. Absent key = no highlight, never a guess.
+- **One process-wide WS subscriber, never one per socket.** `app/consumers.py` bridges `anthias.now_playing` onto `WS_GROUP` with `group_send`, refcounted by open socket. `/ws` is unauthenticated and `vendor.ts` opens it on every page, so per-socket would let any reachable origin pin a Redis connection per socket. The bridge **drops the payload** and sends the generic `'*'` sentinel: the browser ignores the body anyway, and asset ids on an unauthenticated socket leak what is on screen. It reads with `get_message(timeout=...)` rather than `listen()`, because `PubSub.check_health` only runs between reads.
 - **Viewer subscriber can die silently and keep playing (OPEN).** Seen on a 1 GB arm64 board (~16 h uptime): `redis-cli pubsub channels` listed only `hostcmd`, **no `anthias.viewer`**, so `publish anthias.viewer "viewer blank"` returned `subscribers=0` and went nowhere — board deaf to next/previous/stop/blank, yet still cycling assets. `docker restart` restores it. **Diagnose with `redis-cli pubsub channels`, not the ready flag.** `viewer-subscriber-ready` lies: set once on subscribe, only cleared on `redis.ConnectionError`, so it stays `1` when the thread dies any other way. Suspected: `_consume()` raising anything that isn't `redis.ConnectionError` escapes `run()`'s except and kills `ViewerSubscriber.run()` for the process lifetime. No issue filed.
 - **Subscriber topic-prefix trap:** the viewer subscriber splits commands via `data.partition(' ')` and drops anything without the `viewer ` topic prefix. `processing.py` celery publishes raw `'reload'` (no prefix) which would be dropped — unverified suspect.
 - **stop/blank pause path — NOT a bug (issue #3136).** QA reported `stop`/`blank` not halting rotation on headless pi5, but it **did not reproduce** on the real pi5 RC (2026-07-08). The `global loop_is_stopped` fix (#3065) is an ancestor of the RC. The QA symptom was a measurement artifact (4 s durations → the natural `Showing asset...` line fires within ~20 ms of the publish). Recommend closing #3136 as not-reproducible.

CLAUDE.md

@@ -16,7 +16,7 @@ Anthias runs as a set of Docker containers:
 - **redis** (port 6379) — Celery broker + result backend, Channels channel layer, and the viewer signalling bus (pub/sub channel + per-correlation-ID reply lists).
 - **webview** — Qt-based browser for rendering content on the display; fetches `/anthias_assets/` from anthias-server.
 
-Inter-service messaging is all Redis: WebSocket fan-out from Celery to browsers goes via Channels/Redis, and server↔viewer commands/replies use Redis pub/sub on `anthias.viewer` with BLPOP on `anthias.reply.<correlation-id>` for the few request-reply paths. The primary database is SQLite stored at `~/.anthias/anthias.db`, with configuration in `~/.anthias/anthias.conf`. (Pre-rebrand installations have these at `~/.screenly/screenly.db` and `~/.screenly/screenly.conf`; `bin/migrate_legacy_paths.sh` migrates them on upgrade and leaves back-compat symlinks.)
+Inter-service messaging is all Redis: WebSocket fan-out from Celery to browsers goes via Channels/Redis, and server↔viewer commands/replies use Redis pub/sub on `anthias.viewer` with BLPOP on `anthias.reply.<correlation-id>` for the few request-reply paths. The viewer also publishes plain facts the server reads directly (CEC availability, SMART, and the now-playing asset at `viewer:now_playing_asset_id`, whose changes are also announced on `anthias.now_playing`) rather than answering a round trip per render. The primary database is SQLite stored at `~/.anthias/anthias.db`, with configuration in `~/.anthias/anthias.conf`. (Pre-rebrand installations have these at `~/.screenly/screenly.db` and `~/.screenly/screenly.conf`; `bin/migrate_legacy_paths.sh` migrates them on upgrade and leaves back-compat symlinks.)
 
 ### Key Directories
 

src/anthias_common/now_playing.py

@@ -0,0 +1,174 @@
+"""The asset the viewer currently has on screen.
+
+anthias-viewer writes this on every rotation; anthias-server reads it
+when it renders the schedule table, so the operator can see which row
+is live. Play order can't answer that once shuffle is on (#3177).
+
+A published fact rather than a request-reply round trip: the table
+re-renders every 5s per open browser, so asking the viewer per render
+would put a blocking BLPOP on the render path and wake the display
+loop for it. Same shape as ``cec:available`` and the SMART fact.
+
+The TTL is a liveness signal, not a content one: a short window kept
+alive by :func:`refresh` on a timer, like the display-resolution fact.
+Deriving it from the asset's duration instead was tempting and wrong.
+Durations run to a year, so a viewer that died mid-rotation could have
+gone on claiming a row for months, and a paused viewer would have
+dropped the highlight off a picture that was still on screen.
+
+What :func:`refresh` re-asserts is this process's own memory of what it
+last put on screen, never whatever happens to be in Redis. Blind-
+extending the key with EXPIRE looks equivalent and is not: a restarted
+viewer would inherit its dead predecessor's claim and renew it while
+its own screen is still on the splash, and one that crash-loops faster
+than the TTL would renew it forever, which is the exact failure the
+TTL exists to end. Re-asserting also restores the fact by itself if
+Redis is flushed under us, which EXPIRE cannot do.
+
+Every call is best-effort: the write sits in the display loop, where
+an exception would take the screen down, and the read gates a page
+render that must still work with the viewer down.
+"""
+
+import logging
+from time import monotonic
+from typing import Any
+
+from anthias_common.warn_once import WarnOnce
+
+logger = logging.getLogger(__name__)
+
+#: Redis key holding the asset_id the viewer is displaying.
+NOW_PLAYING_KEY = 'viewer:now_playing_asset_id'
+
+#: Pub/sub channel announcing that the value moved, for anyone who
+#: wants it pushed rather than polled. Its own channel, not the viewer
+#: command bus: the audience is browsers. The id rides along for
+#: ``redis-cli subscribe``, but nothing is entitled to it -- the
+#: consumer drops the payload, because /ws is unauthenticated (see
+#: :mod:`anthias_server.app.consumers`).
+NOW_PLAYING_CHANNEL = 'anthias.now_playing'
+
+#: How long the fact outlives the last thing the viewer said. Same
+#: 3-minute window as the display-resolution fact, and for the same
+#: reason: it must survive an ordinary slow tick but not a dead viewer.
+TTL_S = 180
+
+#: Refresh cadence. Comfortably inside TTL_S so two missed ticks in a
+#: row still don't expire a fact that is merely late.
+REFRESH_INTERVAL_S = 60
+
+#: Floor on how often a change is announced; the key write is never
+#: skipped. ``duration`` may be 0 (v2 serializer: ``min_value=0``), so
+#: rotation is not self-limiting, and every announcement costs each
+#: open tab a full table render. The 5s poll used to be the ceiling on
+#: that; this keeps one. A dropped nudge costs one poll of staleness.
+MIN_ANNOUNCE_INTERVAL_S = 1.0
+
+#: What THIS process last put on screen, or None if it hasn't put
+#: anything there yet. Only :func:`refresh` reads it; the module
+#: docstring says why it exists rather than trusting the key.
+_believed: str | None = None
+
+#: When the last announcement went out, for MIN_ANNOUNCE_INTERVAL_S.
+_last_announced_at: float | None = None
+
+#: Warn-once latch for this module's Redis calls.
+_latch = WarnOnce(logger)
+
+
+def _announce(client: Any, payload: str) -> None:
+    """Nudge the browsers, at most once per MIN_ANNOUNCE_INTERVAL_S.
+
+    The second gate after the callers' dedup, for a value that moves
+    faster than a browser can usefully redraw. No catch-up: the 5s
+    poll already carries whatever a dropped nudge would have.
+    """
+    global _last_announced_at
+    now = monotonic()
+    if (
+        _last_announced_at is not None
+        and now - _last_announced_at < MIN_ANNOUNCE_INTERVAL_S
+    ):
+        return
+    client.publish(NOW_PLAYING_CHANNEL, payload)
+    # After the publish, not before: a failed one is caught upstream,
+    # and advancing the floor anyway would drop the next genuine
+    # change too.
+    _last_announced_at = now
+
+
+def publish(client: Any, asset_id: str | None) -> None:
+    """Record — and announce — the asset now on screen."""
+    global _believed
+    if not asset_id:
+        clear(client)
+        return
+    try:
+        # SET always runs: it is what refreshes the TTL. Only the
+        # announcement is deduped, because a single-asset playlist
+        # would otherwise re-render every open table on every loop for
+        # no news. ``get=True`` returns the old value (Redis >= 6.2).
+        previous = client.set(NOW_PLAYING_KEY, asset_id, ex=TTL_S, get=True)
+        _believed = asset_id
+        if previous != asset_id:
+            _announce(client, asset_id)
+        _latch.worked('publish')
+    except Exception as exc:
+        _latch.warn('publish', 'Could not publish the now-playing asset', exc)
+
+
+def refresh(client: Any) -> None:
+    """Re-assert what this process last put on screen.
+
+    On a timer rather than per rotation, because a rotation can be an
+    hour long. A no-op until this process has displayed something, so
+    a restart cannot keep its predecessor's claim alive and
+    :func:`clear` retires the fact for good rather than for one tick.
+    Silent: it re-states a value the browsers already have.
+    """
+    if _believed is None:
+        return
+    try:
+        client.set(NOW_PLAYING_KEY, _believed, ex=TTL_S)
+        _latch.worked('refresh')
+    except Exception as exc:
+        _latch.warn('refresh', 'Could not refresh the now-playing asset', exc)
+
+
+def clear(client: Any) -> None:
+    """Record that nothing is on screen.
+
+    Announced only when something actually stopped: the viewer clears
+    on every tick of an empty playlist, and announcing each one would
+    re-render every open table several times a minute on an idle
+    device.
+    """
+    global _believed
+    _believed = None
+    try:
+        if client.delete(NOW_PLAYING_KEY):
+            _announce(client, '')
+        _latch.worked('clear')
+    except Exception as exc:
+        _latch.warn('clear', 'Could not clear the now-playing asset', exc)
+
+
+def read(client: Any) -> str | None:
+    """The asset_id the viewer last reported, or ``None``.
+
+    ``None`` covers every unknown: the viewer hasn't reported yet, the
+    fact expired because the viewer stopped saying it, or Redis is
+    unreachable.
+    """
+    try:
+        raw = client.get(NOW_PLAYING_KEY)
+        _latch.worked('read')
+    except Exception as exc:
+        _latch.warn('read', 'Could not read the now-playing asset', exc)
+        return None
+    if not raw:
+        return None
+    if isinstance(raw, bytes):
+        raw = raw.decode('utf-8', errors='replace')
+    return str(raw)

src/anthias_common/storage_health.py

@@ -84,6 +84,7 @@
 from typing import Any
 
 from anthias_common import smart
+from anthias_common.warn_once import WarnOnce
 
 logger = logging.getLogger(__name__)
 
@@ -219,29 +220,9 @@
 }
 
 
-_warned: set[str] = set()
-
-
-def _warn_once(key: str, message: str) -> None:
-    """Log ``message`` at WARNING the first time, DEBUG thereafter.
-
-    Deliberately a small local copy of the helper in
-    :mod:`anthias_common.undervoltage` rather than a shared import:
-    the two modules are siblings with no dependency between them, and
-    the alternative was reaching into another module's private
-    helper.
-
-    The conditions this guards (an unreadable boot id) are properties
-    of the device, not of an individual reading, so they are worth
-    stating once. The watcher and every page render call in here, so
-    without the throttle one persistent fault would bury the device
-    log.
-    """
-    if key in _warned:
-        logger.debug(message)
-        return
-    _warned.add(key)
-    logger.warning(message)
+#: As in :mod:`anthias_common.undervoltage` — its own instance so the
+#: shared ``no_boot_id`` key can't silence that module's warning.
+_warn = WarnOnce(logger)
 
 
 def _read_text(path: str) -> str | None:
@@ -866,7 +847,7 @@ def _save_latch(
     check whether the card is wearing out would be self-defeating.
     """
     if boot_id is None:
-        _warn_once(
+        _warn.warn(
             'no_boot_id',
             'No kernel boot id available; reporting storage health from '
             'live readings only and not persisting history.',

src/anthias_common/undervoltage.py

@@ -54,6 +54,8 @@
 from datetime import UTC, datetime
 from typing import Any
 
+from anthias_common.warn_once import WarnOnce
+
 # The hwmon class dir is kernel-global and readable from inside an
 # unprivileged container, which is the whole point of using it.
 HWMON_ROOT = '/sys/class/hwmon'
@@ -188,29 +190,10 @@ def _empty_state() -> dict[str, Any]:
     }
 
 
-_warned: set[str] = set()
-
-
-def _warn_once(key: str, message: str) -> None:
-    """Log ``message`` at WARNING the first time, DEBUG thereafter.
-
-    These conditions (an unreadable boot id, a Redis that will not
-    answer) are properties of the device, not of the individual
-    reading, so they are worth stating once and not repeating. The
-    watcher and every page render call into this module, so without
-    the throttle a single persistent fault would bury everything else
-    in the device log.
-
-    Scoped to the process, which for the celery worker means once per
-    boot in practice. It deliberately is not keyed on the kernel boot
-    id: the main caller is the branch that fires precisely *because*
-    the boot id could not be read.
-    """
-    if key in _warned:
-        logger.debug(message)
-        return
-    _warned.add(key)
-    logger.warning(message)
+#: Device-level faults here (an unreadable boot id, a Redis that will
+#: not answer) are worth stating once, not on every reading: the
+#: watcher and every page render call in here.
+_warn = WarnOnce(logger)
 
 
 def _coerce_count(value: Any) -> int:
@@ -261,7 +244,7 @@ def _load_latch(
     try:
         raw = redis_client.get(REDIS_KEY)
     except Exception:
-        _warn_once(
+        _warn.warn(
             'latch_unreadable',
             'Could not read the under-voltage latch from Redis; '
             'treating history as unknown rather than empty.',
@@ -344,7 +327,7 @@ def record_observation(
     # state; the live reading is still accurate, and the paired
     # discard in ``_load_latch`` keeps the two halves consistent.
     if boot_id is None:
-        _warn_once(
+        _warn.warn(
             'no_boot_id',
             'No kernel boot id available; reporting under-voltage from '
             'the live sensor only and not persisting history.',

src/anthias_common/utils.py

@@ -16,6 +16,7 @@
 import certifi
 import pytz
 import redis
+import redis.asyncio
 import requests
 import sh
 import urllib3
@@ -729,6 +730,38 @@ def connect_to_redis() -> 'redis.Redis':
     return redis.Redis(host='redis', decode_responses=True, port=6379, db=0)
 
 
+def connect_to_redis_async() -> 'redis.asyncio.Redis':
+    """Async twin of :func:`connect_to_redis`, for the ASGI paths.
+
+    A separate client rather than a shared one: redis-py's sync and
+    async clients can't share a connection pool.
+
+    Both timeouts are safe on a pub/sub client under redis-py 8.x:
+    ``PubSub.parse_response`` hands ``read_response`` ``math.inf`` for
+    a blocking read, which is its documented opt-out from
+    ``socket_timeout``. They therefore bound the dial, the SUBSCRIBE
+    and the retry layer's reconnect/AUTH/HELLO, but never a read that
+    is legitimately waiting for the next rotation.
+
+    The same opt-out means ``socket_timeout`` cannot notice a half-open
+    socket under a blocking ``listen()``, and neither can
+    ``health_check_interval``: it only fires from
+    ``PubSub.check_health``, which runs when ``parse_response`` is
+    re-entered. Detecting that is the caller's job -- see
+    ``_watch_now_playing`` in :mod:`anthias_server.app.consumers`,
+    which reads with an explicit timeout in a loop.
+    """
+    return redis.asyncio.Redis(
+        host='redis',
+        decode_responses=True,
+        port=6379,
+        db=0,
+        socket_connect_timeout=5,
+        socket_timeout=5,
+        health_check_interval=30,
+    )
+
+
 def is_docker() -> bool:
     return os.path.isfile('/.dockerenv')