Nextion 串口屏显示接入实现计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 在现有 dzxj_dtu 后端接入 /dev/ttyS9 上的 Nextion 串口屏，实时显示 DTU 管理的 panel 端口状态。

Architecture: 新增独立 NextionDisplay 模块管理屏幕串口与协议；在 app.py 中维护 panel 到“终端”的映射，Modbus 轮询和屏上按钮事件触发屏幕刷新；通过独立串口与现有 RS485 通信隔离。

Tech Stack: Python 3, Flask, Flask-SocketIO, pyserial

文件结构

新建 backend/modules/nextion_display.py：Nextion 协议封装（串口连接、指令发送、按钮事件监听）。
修改 backend/config.py：新增 Nextion 相关配置项。
修改 backend/app.py：初始化屏幕、集成刷新逻辑、新增 API、修正 screen_connected 状态上报。
新建 backend/tests/test_nextion_display.py：NextionDisplay 核心功能单元测试（事件解析、指令编码）。

Task 1: 新增 Nextion 配置项

Files:

Modify: backend/config.py

上下文： 在 backend/config.py 末尾 DHT11 配置之后追加 Nextion 配置。

Step 1: 编辑 backend/config.py 追加配置

在文件末尾追加：

# Nextion 串口屏配置（接 /dev/ttyS9）
NEXTION_ENABLED = os.getenv('NEXTION_ENABLED', 'true').lower() in ('1', 'true', 'yes')
NEXTION_SCREEN_PORT = os.getenv('NEXTION_SCREEN_PORT', '/dev/ttyS9')
NEXTION_SCREEN_BAUD = int(os.getenv('NEXTION_SCREEN_BAUD', '9600'))
NEXTION_PANEL_NAME_PREFIX = os.getenv('NEXTION_PANEL_NAME_PREFIX', '终端')
NEXTION_CMD_DELAY_MS = int(os.getenv('NEXTION_CMD_DELAY_MS', '25'))  # 连续指令间隔，避免缓冲溢出

Step 2: 验证配置文件语法

Run:

cd /Users/wenhongquan/Documents/trae_projects/dzxj_dtu/backend
python -c "import config; print(config.NEXTION_SCREEN_PORT, config.NEXTION_SCREEN_BAUD)"

Expected: 输出 /dev/ttyS9 9600

[ ] Step 3: Commit

git add backend/config.py
git commit -m "config: add Nextion display settings"

Task 2: 实现 NextionDisplay 模块

Files:

Create: backend/modules/nextion_display.py
Test: backend/tests/test_nextion_display.py

Step 1: 写失败测试

[ ] 创建 backend/tests/test_nextion_display.py：

import pytest
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 b't2.txt="\xd6\xd5\xb6\xcb1"' in encoded or encoded.startswith(b't2.txt=')

[ ] Step 2: 运行测试确认失败

Run:

cd /Users/wenhongquan/Documents/trae_projects/dzxj_dtu/backend
pytest tests/test_nextion_display.py -v

Expected: ModuleNotFoundError 或 ImportError

Step 3: 实现模块

[ ] 创建 backend/modules/nextion_display.py：

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

参考 nextion_web_control.py 实现，提供：
- 串口连接/断开
- Nextion 指令发送（GB2312 + 0xFF 0xFF 0xFF 结尾）
- 原始字节发送
- 触控事件监听（0x65 事件头）
"""

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:
    """将字符串指令编码为 Nextion 协议数据包。"""
    return cmd.encode('gb2312', errors='replace') + b'\xFF\xFF\xFF'

def send_cmd(self, cmd: str, delay_ms: float = 0) -> bool:
    """发送一条 Nextion 指令。"""
    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):
    """设置按钮事件回调：callback(page_id, comp_id, event)。"""
    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()

[ ] Step 4: 运行测试确认通过

Run:

cd /Users/wenhongquan/Documents/trae_projects/dzxj_dtu/backend
pytest tests/test_nextion_display.py -v

Expected: 3 个测试全部 PASS

[ ] Step 5: Commit

git add backend/modules/nextion_display.py backend/tests/test_nextion_display.py
git commit -m "feat(nextion): add NextionDisplay module with event parsing tests"

Task 3: 在 `app.py` 中集成 Nextion 显示

Files:

Modify: backend/app.py

Step 1: 导入与初始化

[ ] 在 backend/app.py 中，找到 from modules.dht11_sensor import DHT11Sensor 这一行，在其下方新增：
```
from modules.nextion_display import NextionDisplay
```

[ ] 在 serial_client = SerialPort() 附近新增全局实例：

# Nextion 串口屏实例
screen_display = NextionDisplay()
screen_current_panel_index = 0

Step 2: 新增 panel 排序与屏幕刷新函数

[ ] 在 app.py 中合适位置（建议在 port_state 定义之后）新增以下函数：

def get_sorted_panels():
"""按 position 排序返回 panel 列表。"""
return sorted(panel_config.items(), key=lambda x: x[1].get('position', 0))


def _port_status_to_pic(port_state_entry):
"""将端口状态推导为 Nextion pic 编号。"""
if not port_state_entry:
    return 0
last_uid = port_state_entry.get('last_uid')
expected_uid = port_state_entry.get('expected_uid')
last_polled = port_state_entry.get('last_polled_at')
if not last_polled:
    return 0  # UNKNOWN
if last_uid is None:
    return 1  # DISCONNECTED
if expected_uid and last_uid != expected_uid:
    return 2  # ILLEGAL
return 3  # CONNECTED


def refresh_screen_panel(panel_id):
"""将指定 panel 的数据下发到 Nextion 屏幕。"""
global screen_current_panel_index
panels = get_sorted_panels()
if not panels:
    return False, '没有可用的 panel'
panel_ids = [p[0] for p in panels]
if panel_id not in panel_ids:
    return False, f'panel 不存在: {panel_id}'
if not screen_display.get_status().get('connected'):
    return False, '屏幕串口未连接'

ps = port_state.get(panel_id, {})
now = time.time()
online_count = sum(1 for pid in panel_ids if device_last_seen.get(pid, 0) > now - 60)
idx = panel_ids.index(panel_id)
screen_current_panel_index = idx

# MQTT 状态
mqtt_st = mqtt_client.get_status()
mqtt_connected = mqtt_st.get('connected', False) if isinstance(mqtt_st, dict) else bool(mqtt_st)

delay_ms = config.NEXTION_CMD_DELAY_MS

screen_display.send_cmd(f't2.txt="{config.NEXTION_PANEL_NAME_PREFIX}{idx + 1}"', delay_ms=delay_ms)
screen_display.send_cmd(f't1.txt="{panel_id}"', delay_ms=delay_ms)
screen_display.send_cmd(f't4.txt="{dtu_config.get("firmware_version", "v1.0.0")}"', delay_ms=delay_ms)

mqtt_str = '已连接' if mqtt_connected else '未连接'
screen_display.send_cmd(f'g0.txt="MQTT：{mqtt_str} 当前终端数：{online_count}"', delay_ms=delay_ms)
screen_display.send_cmd(f'g0.pco={0 if mqtt_connected else 63488}', delay_ms=delay_ms)

for port_id in range(1, 25):
    pic = _port_status_to_pic(ps.get(port_id, {}))
    screen_display.send_cmd(f'p{port_id - 1}.pic={pic}', delay_ms=delay_ms)

logger.info(f"屏幕已刷新: {panel_id} (终端{idx + 1})")
return True, '屏幕已刷新'


def switch_screen_panel(direction):
"""切换当前显示的 panel。"""
panels = get_sorted_panels()
if not panels:
    return False, '没有可用的 panel'
global screen_current_panel_index
screen_current_panel_index = (screen_current_panel_index + direction) % len(panels)
panel_id = panels[screen_current_panel_index][0]
logger.info(f"屏幕切换: 终端{screen_current_panel_index + 1} -> {panel_id}")
return refresh_screen_panel(panel_id)


def screen_button_event_handler(page_id, comp_id, event):
"""处理屏幕按钮事件。"""
if event != 0x01:
    return
if page_id == 0x00 and comp_id == 0x05:
    logger.info('串口屏事件: 上一终端')
    switch_screen_panel(-1)
elif page_id == 0x00 and comp_id == 0x06:
    logger.info('串口屏事件: 下一终端')
    switch_screen_panel(1)

注意：如果 config 在 app.py 中是通过 from config import (...) 导入的，需要把 NEXTION_CMD_DELAY_MS、NEXTION_PANEL_NAME_PREFIX 等加入导入列表。

Step 3: 修正 MQTT 状态上报中的 `screen_connected`

[ ] 找到 dtu_publish_status() 函数中 screen_connected = False 的代码块（约第 641-650 行），替换为：

    # 屏幕连接状态：从 Nextion 屏幕实例读取
    screen_connected = screen_display.get_status().get('connected', False)

Step 4: 在启动序列中连接屏幕串口

[ ] 在 app.py 的 __main__ 中，找到 auto_connect_serial() 调用之后，添加：

    # 连接 Nextion 串口屏
    if config.NEXTION_ENABLED:
        try:
            ok, msg = screen_display.connect(config.NEXTION_SCREEN_PORT, config.NEXTION_SCREEN_BAUD)
            if ok:
                screen_display.set_button_callback(screen_button_event_handler)
                logger.info("Nextion 屏幕连接成功")
                # 如果已有 panel，立即显示第一个
                panels = get_sorted_panels()
                if panels:
                    refresh_screen_panel(panels[0][0])
            else:
                logger.warning(f"Nextion 屏幕连接失败: {msg}")
        except Exception as e:
            logger.error(f"Nextion 屏幕初始化异常: {e}")

Step 5: 在端口轮询后触发屏幕刷新

[ ] 在 port_poll_loop 中，找到 dtu_publish_panel_status(panel_id, addr, panel_ports) 调用之后，添加：

                        # 如果该 panel 是当前显示的 panel，刷新屏幕
                        panels = get_sorted_panels()
                        if panels and panels[screen_current_panel_index][0] == panel_id:
                            try:
                                refresh_screen_panel(panel_id)
                            except Exception as e:
                                logger.error(f"刷新屏幕失败: {e}")

Step 6: MQTT 状态变化时刷新屏幕

[ ] 在 mqtt_status_handler(status) 函数末尾（在 if status and dtu_config.get('enabled'): 分支内，订阅主题之后）添加：

        # MQTT 状态变化时刷新当前屏幕显示
        panels = get_sorted_panels()
        if panels and screen_display.get_status().get('connected'):
            try:
                refresh_screen_panel(panels[screen_current_panel_index][0])
            except Exception as e:
                logger.error(f"MQTT 状态变化刷新屏幕失败: {e}")

Step 7: 新增 HTTP API

[ ] 在 app.py 任意 API 路由区域添加：

@app.route('/api/screen/status', methods=['GET'])
def get_screen_status():
"""获取 Nextion 屏幕状态。"""
panels = get_sorted_panels()
current_panel_id = None
if panels and 0 <= screen_current_panel_index < len(panels):
    current_panel_id = panels[screen_current_panel_index][0]
status = screen_display.get_status()
return jsonify({
    'success': True,
    'data': {
        'connected': status.get('connected', False),
        'port': status.get('port'),
        'baudrate': status.get('baudrate'),
        'current_panel_index': screen_current_panel_index,
        'current_panel_id': current_panel_id,
        'panel_count': len(panels)
    }
})


@app.route('/api/screen/refresh', methods=['POST'])
def refresh_screen():
"""手动刷新当前 panel 到屏幕。"""
panels = get_sorted_panels()
if not panels:
    return jsonify({'success': False, 'message': '没有可用的 panel'}), 400
panel_id = panels[screen_current_panel_index][0]
success, message = refresh_screen_panel(panel_id)
if success:
    return jsonify({'success': True, 'message': message})
return jsonify({'success': False, 'message': message}), 400

Step 8: 验证 `app.py` 语法

Run:

cd /Users/wenhongquan/Documents/trae_projects/dzxj_dtu/backend
python -m py_compile app.py

Expected: 无输出（编译成功）

Step 9: Commit

git add backend/app.py
git commit -m "feat(nextion): integrate Nextion display with app, add screen APIs"

Task 4: 测试与验证

Files:

Test: backend/tests/test_nextion_display.py
Manual: backend/app.py
[ ] Step 1: 运行单元测试

Run:

cd /Users/wenhongquan/Documents/trae_projects/dzxj_dtu/backend
pytest tests/test_nextion_display.py -v

Expected: 3 个测试全部 PASS

Step 2: 启动后端并检查屏幕状态 API

在 149 设备上启动后端（确保 /dev/ttyS9 存在）：

cd /Users/wenhongquan/Documents/trae_projects/dzxj_dtu/backend
python app.py

在另一个终端调用 API：

curl http://127.0.0.1:5001/api/screen/status

Expected:

{
  "success": true,
  "data": {
    "connected": true,
    "port": "/dev/ttyS9",
    "baudrate": 9600,
    "current_panel_index": 0,
    "current_panel_id": "PANEL_dtu_001_1",
    "panel_count": 1
  }
}

[ ] Step 3: 手动刷新屏幕

curl -X POST http://127.0.0.1:5001/api/screen/refresh

Expected:

{"success": true, "message": "屏幕已刷新"}

同时观察 149 设备上的屏幕是否显示：终端名、版本、MQTT 状态、24 端口图标。

Step 4: 测试屏上按钮切换

按屏幕上的“上一终端 / 下一终端”按钮，观察日志是否输出：

串口屏事件: 上一终端
屏幕切换: 终端X -> PANEL_...

Step 5: 测试端口状态实时刷新

插拔跳线或触发 Modbus 轮询，观察当前显示 panel 的端口图标是否实时变化。

Step 6: Commit 测试结果/日志（可选）

如果测试通过，可标记版本：

git tag -a v1.0.1-nextion -m "Nextion display integration verified"

Self-Review Checklist

Spec coverage:
- 独立 Nextion 模块 → Task 2
- 配置项 → Task 1
- app.py 集成（初始化、刷新、按钮事件、状态上报） → Task 3
- HTTP API → Task 3 Step 7
- 测试验证 → Task 4
- 无遗漏。
Placeholder scan:
- 无 TBD/TODO。
- 所有代码均给出完整实现或调用示例。
Type consistency：
- NextionDisplay.connect 返回 (bool, str)，与使用处一致。
- screen_display.get_status() 返回 {'connected': bool, 'port': str|None, 'baudrate': int|None}，与使用处一致。
- screen_current_panel_index 为全局 int，跨函数一致。
已知限制：
- 屏幕串口打开失败不会阻塞 DTU 启动；后续手动调用 /api/screen/refresh 会返回失败。
- 当前未实现屏幕串口定时自动重连；如需此功能，可在 refresh_screen_panel 检测到未连接时触发一次重连。

2026-07-06-nextion-display.md 21 KB Verlauf Originalformat

Nextion 串口屏显示接入实现计划

文件结构

Task 1: 新增 Nextion 配置项

Task 2: 实现 NextionDisplay 模块

Step 1: 写失败测试

Step 3: 实现模块

Task 3: 在 app.py 中集成 Nextion 显示

Step 1: 导入与初始化

Step 2: 新增 panel 排序与屏幕刷新函数

Step 3: 修正 MQTT 状态上报中的 screen_connected

Step 4: 在启动序列中连接屏幕串口

Step 5: 在端口轮询后触发屏幕刷新

Step 6: MQTT 状态变化时刷新屏幕

Step 7: 新增 HTTP API

Step 8: 验证 app.py 语法

Step 9: Commit

Task 4: 测试与验证

Self-Review Checklist

2026-07-06-nextion-display.md 21 KB

Verlauf Originalformat

Task 3: 在 `app.py` 中集成 Nextion 显示

Step 3: 修正 MQTT 状态上报中的 `screen_connected`

Step 8: 验证 `app.py` 语法