feat(audio): warm-up gate on the adaptive NMI-rate loop to kill the start/seek pitch glide

c64cast/audio.py

@@ -391,6 +391,18 @@ def encode_floats_to_dac(
 # bounds each move so even fast acquisition glides rather than jumps.
 NMI_RATE_LOOP_ACQUIRE_ALPHA = 0.4  # responsive EMA during acquisition
 NMI_RATE_LOOP_ACQUIRE_DECIDE_CHUNKS = 2  # decide every ~2 chunks while acquiring
+# Warm-up gate: hold the latch at the (near-converged) seed and SUPPRESS decisions
+# for this long after the consumer starts or a large playback disturbance, while
+# still feeding R into the EMA. The first R samples after a start/seek are
+# unrepresentative — the video pipeline's bus load hasn't reached steady state yet
+# (post-seek decode catch-up + the playlist's frame-drop snap), so R reads high
+# until the steady-state VIC/DMA tick-loss arrives (~3 s on HW: R≈11.5k→10.1k). The
+# old loop seeded its EMA off that transient and chased the latch *away* from the
+# seed and back — an audible start-of-playback pitch glide of wasted motion, since
+# the bitmap seed (= ceiling) is already the converged latch. Holding the seed
+# through the transient, then deciding once from a warm EMA, lands on the converged
+# latch with no glide. Re-armed by note_playback_disturbance() on a big frame-drop.
+NMI_RATE_LOOP_WARMUP_S = 3.0
 # Per-mode-class seed for the loop's starting latch, so playback begins near the
 # converged rate (minimal/zero start glide) instead of ramping up from nominal.
 # Bitmap modes lose ~10% of NMI ticks to the REU bank-swap + badline DMA →
@@ -1184,6 +1196,11 @@ def __init__(
         self._last_r_time = 0.0
         self._nmi_loop_chunk_count = 0
         self._nmi_loop_acquiring = True
+        # Warm-up gate deadline (monotonic). While now < this, the loop measures R
+        # into the EMA but holds the latch (see NMI_RATE_LOOP_WARMUP_S). 0.0 = open
+        # (no warm-up pending), so direct _update_nmi_rate_loop calls act at once.
+        # Armed at consumer start and re-armed by note_playback_disturbance().
+        self._nmi_warmup_until = 0.0
         # Current display mode (set by set_nmi_latch_for_mode) + an in-session
         # cache of each mode's converged latch. The loop SEEDS the starting latch
         # from these so playback begins at ~the right rate (no start-of-playback
@@ -1414,6 +1431,9 @@ def set_nmi_latch_for_mode(
                     self._nmi_latch = seed
                     self.api.write_regs(f"{CIA2.TIMER_A_LO:04X}", seed & 0xFF, (seed >> 8) & 0xFF)
                 self._nmi_loop_acquiring = True
+                # A mode change shifts the bus-halt profile (hence R); hold the
+                # latch at the new seed until the new mode's load settles.
+                self._nmi_warmup_until = time.monotonic() + NMI_RATE_LOOP_WARMUP_S
             return
 
         # `hires_edges` scenes report display_mode.name == "hires" (same VIC
@@ -1567,6 +1587,12 @@ def _worker(self) -> None:
                         self._last_r_time = 0.0
                         self._nmi_loop_chunk_count = 0
                         self._nmi_loop_acquiring = True
+                        # Hold the rate loop at the seed until the start/seek
+                        # transient settles (post-seek decode catch-up + the
+                        # playlist's frame-drop snap), so it acquires from a steady
+                        # R instead of chasing the spin-up reading. See
+                        # NMI_RATE_LOOP_WARMUP_S.
+                        self._nmi_warmup_until = time.monotonic() + NMI_RATE_LOOP_WARMUP_S
                         # Pace the next write one chunk_period out so the
                         # PREBUFFER_CHUNKS slack stays steady instead of
                         # getting eaten up immediately.
@@ -1646,6 +1672,14 @@ def _update_nmi_rate_loop(self, r_addr: int) -> None:
         self._last_r_addr = r_addr
         self._last_r_time = now
 
+        # Warm-up gate: during the post-start / post-disturbance settle window the
+        # EMA keeps warming (above) but the latch is held at the seed — the seed is
+        # already near-converged, so this plays at ~the right rate instead of
+        # chasing the unrepresentative spin-up R and gliding back. The chunk counter
+        # is not advanced, so the normal decide cadence resumes cleanly on release.
+        if now < self._nmi_warmup_until:
+            return
+
         self._nmi_loop_chunk_count += 1
         decide_every = (
             NMI_RATE_LOOP_ACQUIRE_DECIDE_CHUNKS
@@ -1684,6 +1718,18 @@ def _update_nmi_rate_loop(self, r_addr: int) -> None:
         self._nmi_latch = new_latch
         self.api.write_regs(f"{CIA2.TIMER_A_LO:04X}", new_latch & 0xFF, (new_latch >> 8) & 0xFF)
 
+    def note_playback_disturbance(self) -> None:
+        """Re-arm the adaptive NMI-rate loop's warm-up gate after a large playback
+        disturbance (the playlist calls this when it snaps the deadline forward and
+        drops a big batch of frames — a seek catch-up or a stream rebuffer).
+
+        Holds the latch at its current value while R rides through the disturbance
+        and re-settles, so the loop doesn't chase the abnormal bus load and glitch
+        the pitch. The EMA is left intact (not re-seeded) so it keeps tracking
+        across the gap. Cheap + thread-safe: a single monotonic write. A no-op in
+        effect when the rate loop isn't running (open-loop / REU pump / static)."""
+        self._nmi_warmup_until = time.monotonic() + NMI_RATE_LOOP_WARMUP_S
+
     # ---- sample tap ----------------------------------------------------------
     def _push_to_tap(self, mono_floats: np.ndarray) -> None:
         """Append float samples in [-1, 1] to the FFT tap ring buffer."""

c64cast/playlist.py

@@ -28,6 +28,12 @@
 InterstitialFactory = Callable[[str], Scene]
 FollowerSceneFactory = Callable[["SceneCfg"], Scene]
 
+# A deadline snap-forward dropping at least this many seconds of frames counts as
+# a "large" playback disturbance (seek catch-up, stream rebuffer) worth telling the
+# audio streamer about, so its adaptive NMI-rate loop re-arms its warm-up gate
+# instead of chasing the abnormal bus load. Routine 1-3 frame drops stay below it.
+_AUDIO_DISTURBANCE_DROP_S = 0.5
+
 
 class Playlist:
     def __init__(
@@ -618,6 +624,11 @@ def _run_one_frame(self, scene: Scene, next_deadline: float) -> float:
                     dropped,
                     (now - next_deadline + frame_time) * 1000,
                 )
+                # A large snap (seek catch-up / stream rebuffer) abnormally loads
+                # the bus; tell the audio loop to hold its NMI rate steady through
+                # it instead of chasing the transient and gliding the pitch.
+                if self.audio is not None and dropped * frame_time >= _AUDIO_DISTURBANCE_DROP_S:
+                    self.audio.note_playback_disturbance()
         return next_deadline
 
     def _handle_broadcast_interrupt(self) -> None:

config/c64cast.example.toml

@@ -268,7 +268,12 @@ nmi_rate_adaptive = true            # Adaptive NMI-rate compensation. The
                                     # handler cycle budget so it never overruns.
                                     # Supersedes pitch_mult_* below (those apply
                                     # only when this is false). Host-DMA path
-                                    # only; on by default.
+                                    # only; on by default. A brief warm-up gate
+                                    # holds the rate steady right after a start /
+                                    # seek / big frame-drop (the spin-up R reads
+                                    # are unrepresentative) so it doesn't audibly
+                                    # glide the pitch chasing them, then trims
+                                    # from a settled estimate.
 source_alignment_marker = false     # **DEBUG/CAPTURE ONLY** — prepend a
                                     # 100 ms 200→3500 Hz linear chirp to the
                                     # pre-encoded REU audio so Cam Link

tests/test_audio_lifecycle.py

@@ -490,6 +490,60 @@ def test_loop_seeds_rate_on_valid_read(self):
         s._update_nmi_rate_loop(audio_mod.RING_BUFFER_ADDR + 1000)  # ~1000 B in ~0.1 s
         self.assertGreater(s._r_rate_ema, 0.0)  # seeded to ~10 kB/s (timing-slop)
 
+    # ---- warm-up gate ----
+    def _slow_r_primed(self) -> AudioStreamer:
+        """A streamer primed so the next _update_nmi_rate_loop call WOULD step the
+        latch (slow R, past the decide cadence) absent any warm-up hold."""
+        s = _make(sample_rate=10500, nmi_rate_adaptive=True)
+        s._nmi_timer_started = True
+        s._nmi_latch = s._nmi_latch_value()  # nominal
+        s._r_rate_ema = 9456.0  # ~9.9% slow → coarse step
+        s._last_r_addr = -1  # skip the EMA update this call (use the pre-seed)
+        s._nmi_loop_chunk_count = max(1, round(s.sample_rate / s.chunk_size)) - 1
+        return s
+
+    def test_warmup_holds_latch(self):
+        # Within the warm-up window the loop must NOT move the latch, even with a
+        # slow R that would otherwise step it (the start/seek transient hold).
+        s = self._slow_r_primed()
+        s._nmi_warmup_until = time.monotonic() + 5.0  # warm-up in effect
+        s._update_nmi_rate_loop(audio_mod.RING_BUFFER_ADDR)
+        self.assertEqual(s._nmi_latch, s._nmi_latch_value())  # unchanged
+
+    def test_warmup_still_updates_ema(self):
+        # The EMA keeps warming during warm-up so the first post-warm-up decision
+        # acts on a settled estimate rather than re-seeding off one sample.
+        s = _make(sample_rate=10500, nmi_rate_adaptive=True)
+        s._nmi_timer_started = True
+        s._nmi_warmup_until = time.monotonic() + 5.0
+        s._last_r_addr = audio_mod.RING_BUFFER_ADDR
+        s._last_r_time = time.monotonic() - 0.1
+        s._r_rate_ema = -1.0
+        s._update_nmi_rate_loop(audio_mod.RING_BUFFER_ADDR + 1000)
+        self.assertGreater(s._r_rate_ema, 0.0)  # measured + seeded despite the hold
+
+    def test_acts_after_warmup(self):
+        # Past the warm-up deadline the same slow R steps the latch (gate released).
+        s = self._slow_r_primed()
+        s._nmi_warmup_until = time.monotonic() - 0.01  # warm-up elapsed
+        s._update_nmi_rate_loop(audio_mod.RING_BUFFER_ADDR)
+        self.assertEqual(s._nmi_latch, 92)  # 96 - capped coarse step 4
+
+    def test_note_playback_disturbance_rearms_warmup(self):
+        s = _make(sample_rate=10500, nmi_rate_adaptive=True)
+        before = time.monotonic()
+        s.note_playback_disturbance()
+        self.assertGreaterEqual(
+            s._nmi_warmup_until, before + audio_mod.NMI_RATE_LOOP_WARMUP_S - 0.05
+        )
+
+    def test_disturbance_then_held(self):
+        # End-to-end: a disturbance arms warm-up, which then holds a would-be step.
+        s = self._slow_r_primed()
+        s.note_playback_disturbance()
+        s._update_nmi_rate_loop(audio_mod.RING_BUFFER_ADDR)
+        self.assertEqual(s._nmi_latch, s._nmi_latch_value())  # held by the re-arm
+
 
 class DigiBoostTest(unittest.TestCase):
     def test_enable_writes_all_voices(self):

tests/test_playlist.py

@@ -913,5 +913,49 @@ def test_skip_event_is_cleared_after_use(self):
         self.assertFalse(pl.skip_event.is_set())
 
 
+class _DisturbanceAudio:
+    """Records note_playback_disturbance() calls (the only audio method the
+    frame-drop path touches)."""
+
+    def __init__(self):
+        self.disturbances = 0
+
+    def note_playback_disturbance(self):
+        self.disturbances += 1
+
+
+class FrameDropDisturbanceTest(unittest.TestCase):
+    """A large deadline snap-forward signals the audio loop; a small one doesn't."""
+
+    def _playlist(self, audio):
+        return Playlist(
+            [FakeScene("A", frames_until_done=10_000)],
+            FakeApi(),
+            target_fps=10000.0,  # frame_time = 1e-4 s
+            heartbeat_interval=0.0,
+            stop_event=threading.Event(),
+            interstitial_factory=_transition_factory()[0],
+            audio=audio,
+        )
+
+    def test_large_drop_signals_audio(self):
+        audio = _DisturbanceAudio()
+        pl = self._playlist(audio)
+        # Deadline ~1 s in the past → drops ≫ _AUDIO_DISTURBANCE_DROP_S of frames.
+        pl._run_one_frame(pl.scenes[0], time.time() - 1.0)
+        self.assertEqual(audio.disturbances, 1)
+
+    def test_small_drop_does_not_signal(self):
+        audio = _DisturbanceAudio()
+        pl = self._playlist(audio)
+        # ~10 ms behind: a real (>2 frame) drop, but well under the 0.5 s bar.
+        pl._run_one_frame(pl.scenes[0], time.time() - 0.01)
+        self.assertEqual(audio.disturbances, 0)
+
+    def test_no_audio_is_safe(self):
+        pl = self._playlist(None)
+        pl._run_one_frame(pl.scenes[0], time.time() - 1.0)  # must not raise
+
+
 if __name__ == "__main__":
     unittest.main()