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):