From 390e290b570bfa4f8b30f222c5ee549b5d44b3dd Mon Sep 17 00:00:00 2001 From: Sean Doran Date: Sat, 28 Feb 2026 14:45:46 -0500 Subject: [PATCH] fix: prevent LED flicker from dropped knob messages at read-buffer boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit parse_messages() previously advanced past a 0xFE start byte whenever no complete frame could be matched, silently consuming partial frames at the end of each ser.read(64) call as garbage. When the device sends 20-50 rapid knob messages during a fast turn and one message straddles the 64-byte read boundary, the split message was discarded: the daemon skipped a knob-position update, the LED color lagged, and when the next complete message arrived the color jumped — producing a visible flicker. Fix: break out of the parse loop when a 0xFE is followed by a known type byte (0x02/0x03/0x06/0x07) but the buffer does not yet hold a complete frame. The partial bytes are returned in the remainder and prepended to the next ser.read() chunk, so no message is ever lost. Also sync last_led_colors after the initial send_leds() on connect so the dedup guard is accurate from the first knob event instead of from the first heartbeat. Adds 5 tests covering partial heartbeat, button, and knob frames, the split-read round-trip, and unknown frame type skipping (61 pass total). --- src/turnup/turnupd.py | 20 +++++++++++++++++- tests/test_protocol.py | 47 ++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/src/turnup/turnupd.py b/src/turnup/turnupd.py index 112aed1..9520c93 100755 --- a/src/turnup/turnupd.py +++ b/src/turnup/turnupd.py @@ -81,6 +81,22 @@ def parse_messages(buf: bytearray) -> tuple[list[dict], bytearray]: i += 6 else: + # buf[i] == 0xFE but no complete frame matched. If the following + # byte identifies a known frame type, this is a partial (split) + # frame at the end of the read buffer — break and return it in the + # remainder so the next serial read can complete it. Only skip + # past this 0xFE when we can rule out a valid frame start (unknown + # type byte with enough bytes already present to decide). + if remaining < 2: + break # Lone 0xFE — could be the start of any frame type. + type_byte = buf[i + 1] + if type_byte == 0x02 and remaining < 3: + break # Partial heartbeat: waiting for terminator 0xFF. + if type_byte in (0x06, 0x07) and remaining < 4: + break # Partial button: waiting for id + 0xFF. + if type_byte == 0x03 and remaining < 6: + break # Partial knob: waiting for id, hi, lo, 0xFF. + # Unknown / corrupted frame — skip this 0xFE byte. i += 1 return messages, bytearray(buf[i:]) @@ -378,7 +394,9 @@ def main() -> None: with serial.Serial(port, baud, timeout=0.1) as ser: log.info("Connected to %s", port) buf.clear() - send_leds(ser, all_led_colors(config, knob_norms)) + initial_colors = all_led_colors(config, knob_norms) + send_leds(ser, initial_colors) + last_led_colors[:] = initial_colors while True: data = ser.read(64) diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 55274fb..a5321bf 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -124,11 +124,54 @@ class TestParseMessages: assert msgs == [{"type": "heartbeat"}] def test_incomplete_frame_stays_in_remainder(self): - # The parser advances byte-by-byte when a frame is unrecognised, so an - # incomplete knob frame (missing terminator 0xFF) is consumed as garbage. + # A partial knob frame at the end of the buffer must be returned in + # the remainder so the next serial read can complete it. Previously + # the parser consumed the leading 0xFE as garbage, silently dropping + # the message and causing LED colors to miss updates (flicker). buf = bytearray([0xFE, 0x03, 0x01, 0x03]) msgs, remainder = parse_messages(buf) assert msgs == [] + assert remainder == bytearray([0xFE, 0x03, 0x01, 0x03]) + + def test_partial_frame_completed_on_next_read(self): + # Simulate a knob message split across two ser.read() calls. + # First read: 4 of 6 bytes. + partial = bytearray([0xFE, 0x03, 0x01, 0x03]) + msgs1, remainder1 = parse_messages(partial) + assert msgs1 == [] + assert remainder1 == partial # preserved, not discarded + # Second read supplies the remaining 2 bytes. + msgs2, remainder2 = parse_messages(remainder1 + bytearray([0xF4, 0xFF])) + assert msgs2 == [{"type": "knob", "id": 1, "value": 1012}] + assert remainder2 == bytearray() + + def test_partial_heartbeat_stays_in_remainder(self): + # Lone 0xFE 0x02 (missing 0xFF terminator) must be preserved. + buf = bytearray([0xFE, 0x02]) + msgs, remainder = parse_messages(buf) + assert msgs == [] + assert remainder == bytearray([0xFE, 0x02]) + + def test_partial_button_stays_in_remainder(self): + # FE 06 id — missing 0xFF terminator. + buf = bytearray([0xFE, 0x06, 0x02]) + msgs, remainder = parse_messages(buf) + assert msgs == [] + assert remainder == bytearray([0xFE, 0x06, 0x02]) + + def test_lone_start_byte_stays_in_remainder(self): + # A single 0xFE with no following bytes must be preserved. + buf = bytearray([0xFE]) + msgs, remainder = parse_messages(buf) + assert msgs == [] + assert remainder == bytearray([0xFE]) + + def test_unknown_frame_type_skipped(self): + # 0xFE followed by an unknown type byte (and enough bytes to decide) + # should be skipped, not stall the parser. + buf = bytearray([0xFE, 0x99, 0x00, 0x00, 0x00, 0x00, 0xFE, 0x02, 0xFF]) + msgs, remainder = parse_messages(buf) + assert msgs == [{"type": "heartbeat"}] assert remainder == bytearray() def test_empty_buffer(self):