From d69d94eaa2fa1fdf1021e3b642c35ffac275f4a1 Mon Sep 17 00:00:00 2001 From: Sean Doran Date: Fri, 27 Feb 2026 18:44:28 -0500 Subject: [PATCH] fix: eliminate LED flicker during knob turns Two targeted fixes: 1. Deduplicate LED writes in handle_knob: compute new colours and only call send_leds() when they differ from last_led_colors. A single physical knob turn produces 20-50 ADC samples in quick succession; without this guard each sample triggers a 47-byte write and the firmware can't keep up, causing visible flicker. 2. Gate reapply_app_volumes on a 200 ms knob-quiet period. Calling playerctl/pulsectl while the user is actively turning a knob stalls the main loop, backing up serial data, missing heartbeats, and causing the device LEDs to time out mid-turn. Both state variables (last_led_colors, last_knob_event) are mutable lists initialised in main() and threaded through handle_knob, keeping module-level globals out of the picture. Adds 6 new unit tests (56 total, all passing). --- src/turnup/turnupd.py | 41 +++++++++++++++++-- tests/test_protocol.py | 89 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 126 insertions(+), 4 deletions(-) diff --git a/src/turnup/turnupd.py b/src/turnup/turnupd.py index c83c2f4..5e6efc4 100755 --- a/src/turnup/turnupd.py +++ b/src/turnup/turnupd.py @@ -251,8 +251,20 @@ def handle_knob( pulse: PulseController, ser: serial.Serial, knob_norms: list[float], + last_led_colors: list[tuple[int, int, int]], + last_knob_event: list[float], ) -> None: - """Dispatch a knob event, update PulseAudio, then refresh the LEDs.""" + """Dispatch a knob event, update PulseAudio, then refresh the LEDs. + + *last_led_colors* is a length-5 list used to suppress duplicate LED + packets — if the computed colours are identical to the last send we skip + the write, eliminating the LED storm that causes visible flicker during a + fast knob turn. + + *last_knob_event* is a single-element list (mutable float box) whose + value is updated to ``time.monotonic()`` on every call so the main loop + can gate ``reapply_app_volumes`` on a quiet period after knob activity. + """ knob_cfg = config.get("knobs", {}).get(str(knob_id)) if not knob_cfg: return @@ -282,7 +294,16 @@ def handle_knob( log.info("Knob %d → group %s = %.2f", knob_id, knob_cfg.get("targets"), vol) knob_norms[knob_id] = norm - send_leds(ser, all_led_colors(config, knob_norms)) + last_knob_event[0] = time.monotonic() + + # Only send an LED packet when the colour actually changes. A single + # physical knob turn generates 20-50 ADC samples in rapid succession; + # without this guard every sample triggers a write and the firmware can't + # keep up, causing visible flicker. + new_colors = all_led_colors(config, knob_norms) + if new_colors != last_led_colors: + send_leds(ser, new_colors) + last_led_colors[:] = new_colors def handle_button( @@ -326,6 +347,14 @@ def main() -> None: knob_norms = init_knob_norms(config, pulse) buf = bytearray() + # Mutable state shared between the main loop and handle_knob: + # last_led_colors — suppress duplicate LED writes during fast knob turns + # last_knob_event — timestamp of most recent knob message; used to gate + # reapply_app_volumes so we don't stall the loop + # mid-turn (200 ms quiet period required) + last_led_colors: list[tuple[int, int, int]] = [(0, 0, 0)] * NUM_KNOBS + last_knob_event: list[float] = [0.0] + # Track config file mtime so we can restart when it changes. try: config_mtime: float | None = os.stat(DEFAULT_CONFIG_PATH).st_mtime @@ -361,6 +390,7 @@ def main() -> None: handle_knob( msg["id"], msg["value"], config, pulse, ser, knob_norms, + last_led_colors, last_knob_event, ) elif msg["type"] == "button": handle_button( @@ -386,7 +416,12 @@ def main() -> None: # streams (e.g. Spotify starting a new song resets to 100 %). # Also trigger immediately on any PA sink-input event so # PA-only apps (e.g. Brave) are corrected within ~0.1 s. - if pulse.drain_events() or now - last_reapply >= 1.0: + # Guard on a 200 ms knob-quiet period: calling playerctl / + # pulsectl while the user is actively turning a knob can + # stall the main loop long enough for serial data to back + # up, which in turn causes heartbeat misses and LED flicker. + knob_quiet = now - last_knob_event[0] >= 0.2 + if (pulse.drain_events() or now - last_reapply >= 1.0) and knob_quiet: last_reapply = now reapply_app_volumes(config, pulse, knob_norms) diff --git a/tests/test_protocol.py b/tests/test_protocol.py index e8e48b8..55274fb 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -4,8 +4,17 @@ No external dependencies — these are pure-function tests. """ import pytest +from unittest.mock import MagicMock, call, patch -from turnup.turnupd import knob_to_norm, knob_to_volume, parse_messages, KNOB_MAX, VOLUME_MAX +from turnup.turnupd import ( + KNOB_MAX, + NUM_KNOBS, + VOLUME_MAX, + handle_knob, + knob_to_norm, + knob_to_volume, + parse_messages, +) # ── knob_to_norm ────────────────────────────────────────────────────────────── @@ -126,3 +135,81 @@ class TestParseMessages: msgs, remainder = parse_messages(bytearray()) assert msgs == [] assert remainder == bytearray() + + +# ── handle_knob ─────────────────────────────────────────────────────────────── + +def _make_knob_fixtures(knob_id: int = 0, action: str = "sink_volume"): + """Return (config, pulse_mock, ser_mock, knob_norms, last_led_colors, last_knob_event).""" + config = { + "knobs": { + str(knob_id): { + "action": action, + "target": "default", + "led": {"low_color": "#000000", "high_color": "#ffffff"}, + } + } + } + pulse = MagicMock() + ser = MagicMock() + knob_norms = [0.0] * NUM_KNOBS + last_led_colors = [(0, 0, 0)] * NUM_KNOBS + last_knob_event = [0.0] + return config, pulse, ser, knob_norms, last_led_colors, last_knob_event + + +class TestHandleKnobLEDDedup: + """Fix 1 — duplicate LED packets must be suppressed.""" + + def test_sends_leds_when_color_changes(self): + config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures() + # Turn knob 0 to max; color should change from (0,0,0). + handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke) + assert ser.write.called + + def test_skips_leds_when_color_unchanged(self): + config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures() + # First call sets the color. + handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke) + write_count_after_first = ser.write.call_count + # Second call with same value — color is identical, no new write. + handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke) + assert ser.write.call_count == write_count_after_first + + def test_last_led_colors_updated_after_send(self): + config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures() + handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke) + # last_led_colors must not remain all-zero after a successful send. + assert llc != [(0, 0, 0)] * NUM_KNOBS + + def test_color_change_after_stable_triggers_send(self): + config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures() + # Reach a stable state at max value. + handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke) + handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke) + count_stable = ser.write.call_count + # Now change to zero — should trigger another send. + handle_knob(0, 0, config, pulse, ser, knob_norms, llc, lke) + assert ser.write.call_count == count_stable + 1 + + +class TestHandleKnobLastKnobEvent: + """Fix 2 — last_knob_event[0] must be updated on every call.""" + + def test_last_knob_event_updated(self): + import time + config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures() + before = time.monotonic() + handle_knob(0, 500, config, pulse, ser, knob_norms, llc, lke) + after = time.monotonic() + assert before <= lke[0] <= after + + def test_last_knob_event_updated_even_when_led_unchanged(self): + import time + config, pulse, ser, knob_norms, llc, lke = _make_knob_fixtures() + handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke) + first_ts = lke[0] + import time as _t; _t.sleep(0.01) + handle_knob(0, KNOB_MAX, config, pulse, ser, knob_norms, llc, lke) + # Timestamp must advance even though color did not change. + assert lke[0] > first_ts