Kaynağa Gözat

fix(nextion): preserve incomplete event bytes in read loop

wenhongquan 1 gün önce
ebeveyn
işleme
54a1b0ed5b

+ 22 - 22
backend/modules/nextion_display.py

@@ -12,32 +12,32 @@ logger = logging.getLogger('nextion_display')
 
 
 def parse_nextion_events(data: bytes):
-    """从字节流中解析 Nextion 触控事件(0x65 事件头,7 字节一包)。"""
+    """从字节流中解析 Nextion 触控事件(0x65 事件头,7 字节一包)。
+
+    返回 (events, remaining_bytes),其中 remaining_bytes 是未消费的字节,
+    可能是一个不完整事件包的开头,需要保留给下一次读取继续解析。
+    """
     events = []
-    buf = bytearray(data)
-    while len(buf) >= 7:
-        idx = -1
-        for i in range(len(buf)):
-            if buf[i] == 0x65:
-                idx = i
-                break
+    i = 0
+    remaining = b''
+    while i < len(data):
+        idx = data.find(0x65, i)
         if idx < 0:
             break
-        if idx > 0:
-            del buf[:idx]
-        if len(buf) < 7:
+        if idx + 7 > len(data):
+            remaining = data[idx:]
             break
-        if buf[4] == 0xFF and buf[5] == 0xFF and buf[6] == 0xFF:
+        if data[idx + 4] == 0xFF and data[idx + 5] == 0xFF and data[idx + 6] == 0xFF:
             events.append({
                 'type': 'button',
-                'page': buf[1],
-                'comp': buf[2],
-                'event': buf[3],
+                'page': data[idx + 1],
+                'comp': data[idx + 2],
+                'event': data[idx + 3],
             })
-            del buf[:7]
+            i = idx + 7
         else:
-            del buf[:1]
-    return events
+            i = idx + 1
+    return events, remaining
 
 
 class NextionDisplay:
@@ -103,10 +103,10 @@ class NextionDisplay:
             try:
                 if self.ser and self.ser.is_open and self.ser.in_waiting > 0:
                     buf.extend(self.ser.read(self.ser.in_waiting))
-                    events = parse_nextion_events(bytes(buf))
-                    if events:
-                        buf.clear()
-                        for evt in events:
+                    events, remaining = parse_nextion_events(bytes(buf))
+                    buf.clear()
+                    buf.extend(remaining)
+                    for evt in events:
                             logger.debug(f"Nextion 事件: {evt}")
                             if self.button_callback:
                                 try:

+ 11 - 2
backend/tests/test_nextion_display.py

@@ -10,8 +10,9 @@ from modules.nextion_display import NextionDisplay, parse_nextion_events
 
 def test_parse_button_event():
     data = bytes([0x65, 0x00, 0x05, 0x01, 0xFF, 0xFF, 0xFF])
-    events = parse_nextion_events(data)
+    events, remaining = parse_nextion_events(data)
     assert events == [{'type': 'button', 'page': 0, 'comp': 5, 'event': 1}]
+    assert remaining == b''
 
 
 def test_parse_multiple_events():
@@ -19,9 +20,17 @@ def test_parse_multiple_events():
         bytes([0x65, 0x00, 0x05, 0x01, 0xFF, 0xFF, 0xFF]) +
         bytes([0x65, 0x00, 0x06, 0x01, 0xFF, 0xFF, 0xFF])
     )
-    events = parse_nextion_events(data)
+    events, remaining = parse_nextion_events(data)
     assert len(events) == 2
     assert events[1] == {'type': 'button', 'page': 0, 'comp': 6, 'event': 1}
+    assert remaining == b''
+
+
+def test_parse_event_with_trailing_partial_bytes():
+    data = bytes([0x65, 0x00, 0x05, 0x01, 0xFF, 0xFF, 0xFF, 0x65])
+    events, remaining = parse_nextion_events(data)
+    assert events == [{'type': 'button', 'page': 0, 'comp': 5, 'event': 1}]
+    assert remaining == bytes([0x65])
 
 
 def test_encode_nextion_cmd():