perf(display): snapshot mirror — viewer gating, digest skip, keepalive (#404)

The display service PNG-encoded its frame to /tmp/led_matrix_preview.png
at 5 fps, 24/7 — identical frames, no viewers, per-call imports and a
chmod every write. On the devpi baseline the display service idles at
~92% CPU; this was one of its biggest fixed costs.

- New pure policy (src/common/snapshot_policy.py, unit-tested off-Pi):
  WRITE changed frames at full rate only while a viewer is watching,
  at a 30s idle cadence otherwise; NEVER re-encode unchanged frames —
  bump mtime (os.utime) every 20s instead, keeping the health check's
  snapshot-age liveness proxy (60s threshold in api_v3) green. Cross-
  referencing comments guard the two constants.
- Viewer detection: the web SSE display broadcaster (which only runs
  while browsers are subscribed) touches /tmp/led_matrix_preview_viewer
  each loop; the display service stats it at most 1/s. On viewer
  arrival the write clock resets so the first frame lands within ~1s.
- Hoisted the per-call pathlib/permission_utils imports; directory
  permissions ensured once instead of every frame.


Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

Co-authored-by: Chuck <chuck@example.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/common/snapshot_policy.py

@@ -0,0 +1,68 @@
+"""Snapshot write policy for the display preview mirror.
+
+The display service mirrors frames to /tmp/led_matrix_preview.png, which
+serves two consumers with different needs:
+
+- The web UI's live preview (SSE reader in web_interface/app.py) wants
+  fresh frames — but only while a browser is actually watching.
+- The health check (web_interface/blueprints/api_v3.py, hardware status)
+  uses the file's AGE as a liveness proxy: age >= 60s reads as degraded.
+
+PNG-encoding every frame at 5 fps forever — identical frames, no viewers —
+was one of the biggest fixed CPU costs on the Pi. This module is the pure
+decision logic (extracted so it's unit-testable off-Pi; display_manager
+imports rgbmatrix unconditionally and can't be):
+
+    WRITE  — encode + atomically replace the snapshot file
+    TOUCH  — os.utime only: keeps the health-check mtime fresh and lets
+             the SSE reader (mtime-gated) resend at a low rate, without
+             paying for a PNG encode of an unchanged frame
+    SKIP   — do nothing
+
+Policy:
+- With a fresh viewer marker: changed frames write at up to 1/VIEWER_INTERVAL.
+- Without viewers: changed frames still write at 1/IDLE_INTERVAL so the
+  preview page shows something recent on open.
+- Unchanged frames are never re-encoded; the mtime is touched every
+  TOUCH_INTERVAL so the health check (60s threshold) never degrades.
+
+If any constant here changes, re-check the health threshold in
+api_v3.py (get_hardware_status) — TOUCH_INTERVAL must stay well under it.
+"""
+
+from enum import Enum
+
+# Snapshot cadence with a browser preview open (seconds).
+VIEWER_INTERVAL = 0.2
+# Snapshot cadence with no viewers — cheap freshness for page-open (seconds).
+IDLE_INTERVAL = 30.0
+# Max age of the last write/touch before bumping mtime for the health
+# check. MUST stay well under api_v3's 60s degraded threshold.
+TOUCH_INTERVAL = 20.0
+# A viewer marker older than this no longer counts as a live viewer.
+VIEWER_MARKER_FRESH_SEC = 5.0
+
+
+class SnapshotAction(Enum):
+    WRITE = "write"
+    TOUCH = "touch"
+    SKIP = "skip"
+
+
+def decide(now: float, last_write_ts: float, last_touch_ts: float,
+           viewer_fresh: bool, frame_changed: bool) -> SnapshotAction:
+    """Decide what to do with the current frame.
+
+    Args:
+        now: current monotonic-ish timestamp (same clock as the ts args)
+        last_write_ts: when a frame was last actually encoded+written
+        last_touch_ts: when the file mtime was last bumped (write or touch)
+        viewer_fresh: a browser preview is currently watching
+        frame_changed: the frame differs from the last WRITTEN frame
+    """
+    interval = VIEWER_INTERVAL if viewer_fresh else IDLE_INTERVAL
+    if frame_changed and (now - last_write_ts) >= interval:
+        return SnapshotAction.WRITE
+    if (now - max(last_write_ts, last_touch_ts)) >= TOUCH_INTERVAL:
+        return SnapshotAction.TOUCH
+    return SnapshotAction.SKIP

src/display_manager.py

@@ -31,14 +31,24 @@ if os.getenv("EMULATOR", "false") == "true":
 else:
     from rgbmatrix import RGBMatrix, RGBMatrixOptions
 from contextlib import contextmanager
+from pathlib import Path
 from PIL import Image, ImageDraw, ImageFont
 import time
 from collections import OrderedDict
 from typing import Dict, Any, List, Optional, Tuple
 import logging
 import math
+import zlib
 import freetype
 
+from src.common import snapshot_policy
+from src.common.permission_utils import (
+    ensure_directory_permissions,
+    ensure_file_permissions,
+    get_assets_dir_mode,
+    get_assets_file_mode,
+)
+
 # Get logger without configuring
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)  # Set to INFO level
@@ -191,10 +201,19 @@ class DisplayManager:
         # Cleared on _load_fonts() so stale entries don't survive a font reload.
         self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
         self._TEXT_WIDTH_CACHE_MAX = 1024
-        # Snapshot settings for web preview integration (service writes, web reads)
+        # Snapshot mirror for web preview + health check (service writes, web
+        # reads). Cadence/skip decisions live in src/common/snapshot_policy.py:
+        # full rate only while the web SSE broadcaster keeps the viewer marker
+        # fresh; unchanged frames are never re-encoded, only mtime-touched.
         self._snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path intentional; web UI reads same path
-        self._snapshot_min_interval_sec = 0.2  # max ~5 fps
+        self._viewer_marker_path = "/tmp/led_matrix_preview_viewer"  # nosec B108 - touched by web SSE broadcaster
         self._last_snapshot_ts = 0.0
+        self._last_snapshot_touch_ts = 0.0
+        self._last_snapshot_digest: Optional[int] = None
+        self._snapshot_dir_prepared = False
+        self._viewer_check_ts = 0.0
+        self._viewer_fresh = False
+        self._viewer_was_fresh = False
         # Snapshot failures are logged as warnings, rate-limited so a
         # persistent failure (e.g. an unwritable file) can't spam the log —
         # but is never silent: the snapshot's mtime doubles as the web UI's
@@ -1145,27 +1164,56 @@ class DisplayManager:
             'deferred_update_ttl': self._scrolling_state['deferred_update_ttl']
         }
 
+    def _viewer_is_fresh(self, now: float) -> bool:
+        """True when a browser preview is watching (marker file touched by
+        the web SSE broadcaster). The marker is stat'd at most once per
+        second — at 125 fps loops a per-call stat would be pure overhead."""
+        if (now - self._viewer_check_ts) >= 1.0:
+            self._viewer_check_ts = now
+            try:
+                marker_age = now - os.stat(self._viewer_marker_path).st_mtime
+                self._viewer_fresh = marker_age < snapshot_policy.VIEWER_MARKER_FRESH_SEC
+            except OSError:
+                self._viewer_fresh = False
+        return self._viewer_fresh
+
     def _write_snapshot_if_due(self) -> None:
-        """Write the current image to a PNG snapshot file at a limited frequency."""
+        """Mirror the current frame to the preview snapshot when the policy
+        says it's worth it — see src/common/snapshot_policy.py. Unchanged
+        frames are never re-encoded; without viewers the cadence drops to
+        the idle keepalive."""
         try:
             now = time.time()
-            if (now - self._last_snapshot_ts) < self._snapshot_min_interval_sec:
+            viewer_fresh = self._viewer_is_fresh(now)
+            if viewer_fresh and not self._viewer_was_fresh:
+                # A preview just opened: let the next changed frame through
+                # immediately instead of waiting out the idle interval.
+                self._last_snapshot_ts = 0.0
+            self._viewer_was_fresh = viewer_fresh
+
+            digest = zlib.adler32(self.image.tobytes())
+            action = snapshot_policy.decide(
+                now, self._last_snapshot_ts, self._last_snapshot_touch_ts,
+                viewer_fresh, digest != self._last_snapshot_digest)
+            if action is snapshot_policy.SnapshotAction.SKIP:
                 return
-            # Ensure directory exists with proper permissions
+            if action is snapshot_policy.SnapshotAction.TOUCH:
-            from pathlib import Path
+                # mtime bump only: keeps the health check (snapshot age)
-            from src.common.permission_utils import (
+                # green without paying for a PNG encode of an unchanged frame
-                ensure_directory_permissions,
+                os.utime(self._snapshot_path, None)
-                ensure_file_permissions,
+                self._last_snapshot_touch_ts = now
-                get_assets_dir_mode,
+                return
-                get_assets_file_mode
+
-            )
+            # WRITE: ensure directory permissions once, not per frame
             snapshot_path_obj = Path(self._snapshot_path)
-            # Only ensure permissions on non-system directories
+            if not self._snapshot_dir_prepared:
-            # Never modify /tmp permissions - it has special system permissions (1777)
+                # Never modify /tmp permissions - it has special system
-            # that must not be changed or it breaks apt and other system tools
+                # permissions (1777) that must not be changed or it breaks
+                # apt and other system tools
                 parent_dir = snapshot_path_obj.parent
                 if parent_dir and str(parent_dir) != '/tmp':  # nosec B108 - guard to skip /tmp for permission ops
                     ensure_directory_permissions(parent_dir, get_assets_dir_mode())
+                self._snapshot_dir_prepared = True
             # Write atomically: temp then replace
             tmp_path = f"{self._snapshot_path}.tmp"
             self.image.save(tmp_path, format='PNG')
@@ -1180,6 +1228,8 @@ class DisplayManager:
             except Exception:
                 pass
             self._last_snapshot_ts = now
+            self._last_snapshot_touch_ts = now
+            self._last_snapshot_digest = digest
         except Exception as e:
             # Snapshot failures must never break display — but they must not
             # be silent either: the snapshot's mtime is the web UI's display

test/test_snapshot_policy.py

@@ -0,0 +1,93 @@
+"""Tests for the snapshot write policy (src/common/snapshot_policy.py).
+
+The invariants that matter:
+- unchanged frames are NEVER re-encoded (the old code PNG-encoded identical
+  frames at 5 fps, 24/7)
+- the file mtime never goes stale enough to trip the health check's 60s
+  degraded threshold (api_v3 get_hardware_status)
+- a viewer gets full cadence; no viewer drops to the idle keepalive
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.common.snapshot_policy import (  # noqa: E402
+    IDLE_INTERVAL,
+    TOUCH_INTERVAL,
+    VIEWER_INTERVAL,
+    SnapshotAction,
+    decide,
+)
+
+
+class TestViewerCadence:
+    def test_changed_frame_with_viewer_writes_at_full_rate(self):
+        assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL,
+                      last_touch_ts=0, viewer_fresh=True,
+                      frame_changed=True) is SnapshotAction.WRITE
+
+    def test_changed_frame_with_viewer_respects_min_interval(self):
+        assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL / 2,
+                      last_touch_ts=100.0, viewer_fresh=True,
+                      frame_changed=True) is SnapshotAction.SKIP
+
+    def test_unchanged_frame_with_viewer_never_writes(self):
+        """A static screen with a viewer must not burn PNG encodes."""
+        assert decide(now=100.0, last_write_ts=90.0, last_touch_ts=90.0,
+                      viewer_fresh=True,
+                      frame_changed=False) is SnapshotAction.SKIP
+
+
+class TestIdleCadence:
+    def test_changed_frame_without_viewer_waits_for_idle_interval(self):
+        assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL / 2,
+                      last_touch_ts=100.0, viewer_fresh=False,
+                      frame_changed=True) is SnapshotAction.SKIP
+
+    def test_changed_frame_without_viewer_writes_at_idle_rate(self):
+        assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL,
+                      last_touch_ts=0, viewer_fresh=False,
+                      frame_changed=True) is SnapshotAction.WRITE
+
+
+class TestHealthKeepalive:
+    def test_stale_mtime_gets_touched(self):
+        """Whatever else happens, mtime must be bumped within TOUCH_INTERVAL
+        so the health check (60s threshold) never reads the display as dead."""
+        assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL,
+                      last_touch_ts=100.0 - TOUCH_INTERVAL, viewer_fresh=False,
+                      frame_changed=False) is SnapshotAction.TOUCH
+
+    def test_touch_applies_with_viewer_too(self):
+        """Viewer watching a static screen: no writes, but health stays green."""
+        assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL - 1,
+                      last_touch_ts=100.0 - TOUCH_INTERVAL - 1, viewer_fresh=True,
+                      frame_changed=False) is SnapshotAction.TOUCH
+
+    def test_recent_touch_suppresses_another(self):
+        assert decide(now=100.0, last_write_ts=0.0,
+                      last_touch_ts=100.0 - TOUCH_INTERVAL / 2, viewer_fresh=False,
+                      frame_changed=False) is SnapshotAction.SKIP
+
+    def test_touch_interval_stays_under_health_threshold(self):
+        """api_v3's hardware status treats snapshot age >= 60s as degraded.
+        Keep a 2x margin so scheduling jitter can't trip it."""
+        assert TOUCH_INTERVAL <= 30
+
+    def test_worst_case_mtime_age_is_bounded(self):
+        """Simulate any interleaving: from any state, within one policy call
+        after TOUCH_INTERVAL elapses, mtime gets refreshed (WRITE or TOUCH)."""
+        for viewer in (True, False):
+            for changed in (True, False):
+                action = decide(now=1000.0, last_write_ts=900.0,
+                                last_touch_ts=900.0, viewer_fresh=viewer,
+                                frame_changed=changed)
+                assert action in (SnapshotAction.WRITE, SnapshotAction.TOUCH)
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

web_interface/app.py

@@ -610,8 +610,22 @@ def display_preview_generator():
     import io
 
     snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path matches display_manager; only read here
+    # Viewer marker: this generator only runs while the broadcaster has
+    # subscribers (it exits with no clients), so touching the marker each
+    # loop tells the DISPLAY service a browser is actually watching — it
+    # only pays for full-rate PNG snapshot encodes while this stays fresh
+    # (see src/common/snapshot_policy.py).
+    viewer_marker_path = "/tmp/led_matrix_preview_viewer"  # nosec B108 - fixed path matches display_manager
     last_modified = None
 
+    def _touch_viewer_marker():
+        try:
+            with open(viewer_marker_path, 'a'):
+                pass
+            os.utime(viewer_marker_path, None)
+        except OSError:
+            pass  # display side treats a missing marker as "no viewer"
+    
     # Get display dimensions from config
     try:
         main_config = config_manager.load_config()
@@ -627,6 +641,7 @@ def display_preview_generator():
     
     while True:
         try:
+            _touch_viewer_marker()
             # Check if snapshot file exists and has been modified
             if os.path.exists(snapshot_path):
                 current_modified = os.path.getmtime(snapshot_path)