fix(vegas): lock the scroll to the panel refresh; encode the preview off the render thread

Vegas advanced by elapsed time, blended neighbouring columns every frame,
and paced itself with a sleep to target_fps. On hdpi (4x128x64 on one
chain, a 120Hz cap the chain cannot reach, ~95-100Hz real) that ran at
73fps with target 90 and ~89fps with target 125: the sleep drifted
against the refresh and missed a vsync every few frames, and the blend
read as shimmer on the panel (and as "anti-aliased" text in the preview).

smooth_scroll now means the crisp pacing the plugin tickers already use:
a whole number of pixels per presented frame, each held for frame_hold
refreshes, with SwapOnVSync as the clock. The speed is solved against the
panel's measured refresh, timed from our own swaps once scrolling starts,
because the configured limit is only a cap -- at "120Hz" 90px/s solves to
3px every 4 refreshes, at the real ~97Hz to 1px every refresh. The old
blend stays available as sub_pixel_blend (default off).

With the web preview open, the render thread also PNG-encoded the whole
512x64 frame five times a second, 12-14ms each -- longer than a refresh.
Mid-scroll that encode now runs on a single-slot writer thread (Pillow
releases the GIL while compressing); static frames still write inline.

Measured on hdpi, 3-minute soak with the preview open: 3 of 17,280
frames held an extra refresh (0.02%), down from ~6-20%.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

test/test_display_dirty_tracking.py

@@ -357,3 +357,57 @@ class TestFrameHoldLifetime:
             assert dm._frame_hold == 1
         finally:
             dm.set_scrolling_state(False)
+
+
+class TestSnapshotOffRenderThread:
+    """Mid-scroll, the preview PNG is encoded off the render thread.
+
+    At 512x64 the encode takes 12-14ms on a Pi 4 -- longer than a refresh --
+    so doing it inline made the next swap miss its vsync five times a second
+    whenever the web preview was open.
+    """
+
+    def _record_saves(self, dm, monkeypatch):
+        import threading
+        threads = []
+        done = threading.Event()
+        real = dm._save_snapshot
+
+        def recording(image):
+            threads.append(threading.current_thread().name)
+            real(image)
+            done.set()
+
+        monkeypatch.setattr(dm, "_save_snapshot", recording)
+        return threads, done
+
+    def _due(self, dm, tmp_path, colour):
+        dm._snapshot_path = str(tmp_path / "snap.png")
+        dm._last_snapshot_ts = 0.0
+        dm._last_snapshot_touch_ts = 0.0
+        dm._last_snapshot_digest = None
+        dm.draw.rectangle([0, 0, 10, 4], fill=colour)
+
+    def test_scrolling_frames_are_encoded_on_the_writer_thread(
+            self, dm, tmp_path, monkeypatch):
+        import threading
+        threads, done = self._record_saves(dm, monkeypatch)
+        self._due(dm, tmp_path, (0, 255, 255))
+        dm.set_scrolling_state(True)
+        try:
+            dm.update_display()
+            assert done.wait(5), "the snapshot writer never wrote the frame"
+        finally:
+            dm.set_scrolling_state(False)
+        assert threads == ["snapshot-writer"]
+        assert threads[0] != threading.current_thread().name
+        assert os.path.exists(dm._snapshot_path)
+
+    def test_static_frames_are_still_written_inline(
+            self, dm, tmp_path, monkeypatch):
+        import threading
+        threads, _ = self._record_saves(dm, monkeypatch)
+        self._due(dm, tmp_path, (255, 0, 255))
+        dm.set_scrolling_state(False)
+        dm.update_display()
+        assert threads == [threading.current_thread().name]

test/test_display_pending_changes.py

@@ -140,6 +140,9 @@ def vegas_coordinator(controller):
         'enabled': True, 'max_cycle_duration': VEGAS_ITERATION_SECONDS}}})
     assert coord.vegas_config.continuous_scroll
     coord.render_pipeline = MagicMock()
+    # Real numbers: the loop sleeps and reports against these.
+    coord.render_pipeline.frame_interval = coord.vegas_config.get_frame_interval()
+    coord.render_pipeline.target_fps = float(coord.vegas_config.target_fps)
     coord.stream_manager = MagicMock()
     coord.display_manager = controller.display_manager
     coord.stats = {'cycles_completed': 0, 'interruptions': 0}

test/test_vegas_crisp_pacing.py

@@ -0,0 +1,130 @@
+"""Vegas scrolls in whole pixels locked to the panel refresh.
+
+It used to advance by elapsed time, blend neighbouring columns, and pace itself
+with a sleep to target_fps. On a 512x64 chain refreshing at 95Hz that ran at
+73-89fps with p99 frames of 20-28ms: the sleep drifted against the refresh and
+missed a vsync every few frames, and the blend shimmered on the panel.
+"""
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.common.scroll_config import solve_crisp  # noqa: E402
+from src.vegas_mode import render_pipeline as rp_module  # noqa: E402
+from src.vegas_mode.config import VegasModeConfig  # noqa: E402
+from src.vegas_mode.render_pipeline import RenderPipeline  # noqa: E402
+
+W, H = 128, 32
+
+
+class FakeStream:
+    def get_grouped_content_for_composition(self):
+        return [('a', [Image.new('RGB', (4000, H), (255, 255, 255))])]
+
+    def get_active_plugin_ids(self):
+        return ['a']
+
+
+class FakeDM:
+    width = W
+    height = H
+
+    def __init__(self, refresh_hz=100.0, hardware=True):
+        self.refresh_hz = refresh_hz
+        self.matrix = object() if hardware else None
+        self.image = Image.new('RGB', (W, H))
+        self.holds = []
+
+    def set_scrolling_state(self, is_scrolling, frame_hold=1):
+        self.holds.append(frame_hold)
+
+    def update_display(self):
+        pass
+
+
+def _pipeline(dm=None, **cfg):
+    p = RenderPipeline(VegasModeConfig(lead_in_width=0, **cfg), dm or FakeDM(),
+                       FakeStream())
+    assert p.compose_scroll_content()
+    return p
+
+
+def test_default_steps_whole_pixels_and_holds_frames():
+    dm = FakeDM(refresh_hz=100.0)
+    p = _pipeline(dm, scroll_speed=50)
+    want = solve_crisp(50, 100.0)
+    assert p.scroll_helper.fixed_pixels_per_frame == want.pixels_per_frame
+    assert not p.scroll_helper.sub_pixel_scrolling
+
+    before = p.scroll_helper.scroll_position
+    p.render_frame()
+    assert p.scroll_helper.scroll_position - before == want.pixels_per_frame
+    # The hold is what makes 1px every 2 refreshes 50px/s rather than 100.
+    assert dm.holds[-1] == want.frame_hold == 2
+    assert p.target_fps == want.frames_per_second
+
+
+def test_sleep_floor_stays_below_the_refresh_period():
+    # A floor at or above the real period accumulates until a frame misses.
+    p = _pipeline(FakeDM(refresh_hz=100.0), scroll_speed=50)
+    assert p.frame_interval < p._frame_hold / 100.0
+
+
+def test_sub_pixel_blend_keeps_the_old_time_based_blend():
+    dm = FakeDM()
+    p = _pipeline(dm, sub_pixel_blend=True, target_fps=90)
+    assert p.scroll_helper.fixed_pixels_per_frame is None
+    assert p.scroll_helper.sub_pixel_scrolling
+    p.render_frame()
+    assert dm.holds[-1] == 1
+    assert p.frame_interval == 1.0 / 90
+    assert p.target_fps == 90
+
+
+def _run_swaps(p, period, frames):
+    """Render `frames` frames whose swaps are `period` seconds apart."""
+    clock = [1000.0]
+
+    def monotonic():
+        return clock[0]
+
+    with patch.object(rp_module.time, 'monotonic', monotonic):
+        for _ in range(frames):
+            p.render_frame()
+            clock[0] += period
+
+
+def test_a_panel_below_its_cap_is_measured_and_the_speed_re_solved():
+    # 4x128x64 on one chain: capped at 120Hz, really 95Hz. Against the cap
+    # 90px/s solves to 3px every 4 refreshes; against 95Hz, 1px every one.
+    p = _pipeline(FakeDM(refresh_hz=120.0), scroll_speed=90)
+    assert p._crisp.pixels_per_frame == 3
+
+    frames = RenderPipeline.REFRESH_WARMUP_FRAMES + RenderPipeline.REFRESH_SAMPLES + 2
+    _run_swaps(p, p._frame_hold / 95.0, frames)
+
+    assert p._measured_hz == 95.0
+    assert (p._crisp.pixels_per_frame, p._crisp.frame_hold) == (1, 1)
+    # The floor still comes from the cap, not the measurement.
+    assert p.frame_interval < 1 / 95.0
+
+
+def test_a_panel_that_keeps_up_with_its_cap_is_left_alone():
+    p = _pipeline(FakeDM(refresh_hz=100.0), scroll_speed=50)
+    crisp = p._crisp
+    frames = RenderPipeline.REFRESH_WARMUP_FRAMES + RenderPipeline.REFRESH_SAMPLES + 2
+    _run_swaps(p, p._frame_hold / 99.5, frames)
+    assert p._measured_hz == 100.0
+    assert p._crisp == crisp
+
+
+def test_no_measurement_without_hardware():
+    # Nothing blocks in the emulator, so swap gaps say nothing about a panel.
+    p = _pipeline(FakeDM(refresh_hz=120.0, hardware=False), scroll_speed=90)
+    frames = RenderPipeline.REFRESH_WARMUP_FRAMES + RenderPipeline.REFRESH_SAMPLES + 2
+    _run_swaps(p, 1 / 50.0, frames)
+    assert p._measured_hz is None

test/test_vegas_density.py

@@ -841,6 +841,10 @@ class TestCycleEndsBeforeWrap:
         return p
 
     def _advance_to(self, pipeline, distance):
+        # render_frame() steps before it checks, and a whole-pixel pace steps
+        # a fixed amount; start one step short so the checked frame is at
+        # `distance`.
+        distance -= pipeline.scroll_helper.fixed_pixels_per_frame or 0
         pipeline.scroll_helper.total_distance_scrolled = distance
         pipeline.scroll_helper.scroll_position = float(distance)