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


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