Merge pull request #6 from sean351/fix/led-flicker-partial-frame

fix: prevent LED flicker from dropped knob messages at read-buffer boundary

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)

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
+        # A partial knob frame at the end of the buffer must be returned in
-        # incomplete knob frame (missing terminator 0xFF) is consumed as garbage.
+        # 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):