dzxj_dtu/backend/modules/nextion_display.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Nextion 串口屏协议封装。"""

import logging
import threading
import time

import serial

logger = logging.getLogger('nextion_display')

EVENT_HEADER = 0x65
TERMINATOR = b'\xFF\xFF\xFF'


def parse_nextion_events(data: bytes) -> tuple[list[dict], bytes]:
    """从字节流中解析 Nextion 触控事件（0x65 事件头，7 字节一包）。

    返回 (events, remaining_bytes)，其中 remaining_bytes 是未消费的字节，
    可能是一个不完整事件包的开头，需要保留给下一次读取继续解析。
    """
    events = []
    i = 0
    remaining = b''
    while i < len(data):
        idx = data.find(EVENT_HEADER, i)
        if idx < 0:
            break
        if idx + 7 > len(data):
            remaining = data[idx:]
            break
        if data[idx + 4:idx + 7] == TERMINATOR:
            events.append({
                'type': 'button',
                'page': data[idx + 1],
                'comp': data[idx + 2],
                'event': data[idx + 3],
            })
            i = idx + 7
        else:
            i = idx + 1
    return events, remaining


class NextionDisplay:
    """Nextion 串口屏管理类。"""

    def __init__(self):
        self.ser = None
        self.lock = threading.RLock()
        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):
        """连接 Nextion 屏幕串口。

        如果已经连接，会先断开当前连接。

        Args:
            port: 串口设备路径。
            baudrate: 波特率，默认 9600。
            timeout: 串口读取超时时间，默认 0.1 秒。

        Returns:
            tuple[bool, str]: 成功时返回 (True, 描述信息)，失败时返回 (False, 错误信息)。
        """
        with self.lock:
            if self.connected:
                # 释放锁后再调用 disconnect，避免在同一把锁上产生死锁。
                self.lock.release()
                try:
                    self.disconnect()
                finally:
                    self.lock.acquire()

        try:
            logger.info(f"连接 Nextion 屏幕串口: {port} @ {baudrate}")
            ser = serial.Serial(port, baudrate, timeout=timeout)
        except Exception as e:
            with self.lock:
                self.connected = False
                self.ser = None
            msg = f"连接 Nextion 屏幕串口失败: {e}"
            logger.error(msg)
            return False, msg

        with self.lock:
            self.ser = ser
            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}"

    def disconnect(self):
        try:
            with self.lock:
                was_connected = self.connected
                self.stop_event.set()
                ser = self.ser
                self.ser = None
                self.connected = False
                self.port = None
                self.baudrate = None

            if self.read_thread and self.read_thread.is_alive():
                self.read_thread.join(timeout=2.0)

            if ser and ser.is_open:
                try:
                    ser.close()
                except Exception as e:
                    logger.error(f"关闭 Nextion 串口失败: {e}")

            if was_connected:
                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:
                with self.lock:
                    ser = self.ser
                if ser and ser.is_open and ser.in_waiting > 0:
                    buf.extend(ser.read(ser.in_waiting))
                    events, remaining = parse_nextion_events(bytes(buf))
                    buf.clear()
                    buf.extend(remaining)
                    for evt in events:
                        logger.debug(f"Nextion 事件: {evt}")
                        with self.lock:
                            callback = self.button_callback
                        if callback:
                            try:
                                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') + TERMINATOR

    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}")
            with self.lock:
                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}")
            with self.lock:
                self.connected = False
            return False

    def set_button_callback(self, callback):
        with self.lock:
            self.button_callback = callback

    def get_status(self):
        with self.lock:
            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()