feat(nextion): add NextionDisplay module with event parsing tests

backend/modules/nextion_display.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Nextion 串口屏协议封装。"""
+
+import logging
+import threading
+import time
+
+import serial
+
+logger = logging.getLogger('nextion_display')
+
+
+def parse_nextion_events(data: bytes):
+    """从字节流中解析 Nextion 触控事件（0x65 事件头，7 字节一包）。"""
+    events = []
+    buf = bytearray(data)
+    while len(buf) >= 7:
+        idx = -1
+        for i in range(len(buf)):
+            if buf[i] == 0x65:
+                idx = i
+                break
+        if idx < 0:
+            break
+        if idx > 0:
+            del buf[:idx]
+        if len(buf) < 7:
+            break
+        if buf[4] == 0xFF and buf[5] == 0xFF and buf[6] == 0xFF:
+            events.append({
+                'type': 'button',
+                'page': buf[1],
+                'comp': buf[2],
+                'event': buf[3],
+            })
+            del buf[:7]
+        else:
+            del buf[:1]
+    return events
+
+
+class NextionDisplay:
+    """Nextion 串口屏管理类。"""
+
+    def __init__(self):
+        self.ser = None
+        self.lock = threading.Lock()
+        self.read_thread = None
+        self.stop_event = threading.Event()
+        self.button_callback = None
+        self.port = None
+        self.baudrate = None
+        self.connected = False
+
+    def connect(self, port: str, baudrate: int = 9600, timeout: float = 0.1):
+        try:
+            with self.lock:
+                if self.connected:
+                    self.disconnect()
+                logger.info(f"连接 Nextion 屏幕串口: {port} @ {baudrate}")
+                self.ser = serial.Serial(port, baudrate, timeout=timeout)
+                self.port = port
+                self.baudrate = baudrate
+                self.connected = True
+                self.stop_event.clear()
+                self.read_thread = threading.Thread(target=self._read_loop, daemon=True)
+                self.read_thread.start()
+                logger.info(f"Nextion 屏幕串口已连接: {port}")
+                return True, f"已连接 {port} @ {baudrate}"
+        except Exception as e:
+            self.connected = False
+            self.ser = None
+            msg = f"连接 Nextion 屏幕串口失败: {e}"
+            logger.error(msg)
+            return False, msg
+
+    def disconnect(self):
+        try:
+            with self.lock:
+                self.stop_event.set()
+                if self.read_thread and self.read_thread.is_alive():
+                    self.read_thread.join(timeout=2.0)
+                if self.ser and self.ser.is_open:
+                    try:
+                        self.ser.close()
+                    except Exception as e:
+                        logger.error(f"关闭 Nextion 串口失败: {e}")
+                self.ser = None
+                self.connected = False
+                self.port = None
+                self.baudrate = None
+            logger.info("Nextion 屏幕串口已断开")
+            return True, "已断开"
+        except Exception as e:
+            logger.error(f"断开 Nextion 屏幕串口失败: {e}")
+            return False, str(e)
+
+    def _read_loop(self):
+        logger.info("启动 Nextion 事件监听线程")
+        buf = bytearray()
+        while not self.stop_event.is_set():
+            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:
+                            logger.debug(f"Nextion 事件: {evt}")
+                            if self.button_callback:
+                                try:
+                                    self.button_callback(evt['page'], evt['comp'], evt['event'])
+                                except Exception as e:
+                                    logger.error(f"按钮回调异常: {e}")
+                else:
+                    time.sleep(0.01)
+            except Exception as e:
+                logger.error(f"Nextion 读取线程异常: {e}")
+                time.sleep(0.5)
+        logger.info("Nextion 事件监听线程结束")
+
+    def _encode_cmd(self, cmd: str) -> bytes:
+        return cmd.encode('gb2312', errors='replace') + b'\xFF\xFF\xFF'
+
+    def send_cmd(self, cmd: str, delay_ms: float = 0) -> bool:
+        try:
+            with self.lock:
+                if not self.connected or not self.ser or not self.ser.is_open:
+                    logger.warning(f"Nextion 未连接，无法发送: {cmd}")
+                    return False
+                data = self._encode_cmd(cmd)
+                self.ser.write(data)
+                self.ser.flush()
+                logger.debug(f"[Nextion TX] {cmd}")
+            if delay_ms > 0:
+                time.sleep(delay_ms / 1000.0)
+            return True
+        except Exception as e:
+            logger.error(f"发送 Nextion 指令失败 '{cmd}': {e}")
+            self.connected = False
+            return False
+
+    def send_raw(self, data: bytes) -> bool:
+        try:
+            with self.lock:
+                if not self.connected or not self.ser or not self.ser.is_open:
+                    logger.warning("Nextion 未连接，无法发送原始数据")
+                    return False
+                self.ser.write(data)
+                self.ser.flush()
+                logger.debug(f"[Nextion TX RAW] {data.hex()}")
+            return True
+        except Exception as e:
+            logger.error(f"发送 Nextion 原始数据失败: {e}")
+            self.connected = False
+            return False
+
+    def set_button_callback(self, callback):
+        self.button_callback = callback
+
+    def get_status(self):
+        return {
+            'connected': self.connected and self.ser is not None and self.ser.is_open,
+            'port': self.port,
+            'baudrate': self.baudrate,
+        }
+
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.DEBUG)
+    disp = NextionDisplay()
+    ok, msg = disp.connect('/dev/ttyS9', 9600)
+    print(ok, msg)
+    if ok:
+        disp.send_cmd('t2.txt="测试"')
+        time.sleep(1)
+        disp.disconnect()

backend/tests/test_nextion_display.py

@@ -0,0 +1,31 @@
+import pytest
+import sys
+import os
+
+# 添加项目根目录到Python路径
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+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)
+    assert events == [{'type': 'button', 'page': 0, 'comp': 5, 'event': 1}]
+
+
+def test_parse_multiple_events():
+    data = (
+        bytes([0x65, 0x00, 0x05, 0x01, 0xFF, 0xFF, 0xFF]) +
+        bytes([0x65, 0x00, 0x06, 0x01, 0xFF, 0xFF, 0xFF])
+    )
+    events = parse_nextion_events(data)
+    assert len(events) == 2
+    assert events[1] == {'type': 'button', 'page': 0, 'comp': 6, 'event': 1}
+
+
+def test_encode_nextion_cmd():
+    display = NextionDisplay()
+    encoded = display._encode_cmd('t2.txt="终端1"')
+    assert encoded.endswith(b'\xFF\xFF\xFF')
+    assert encoded.startswith(b't2.txt=')