Nextion 串口屏显示接入设计

背景

在 149 设备 /dev/ttyS9 上已接入并烧录好 HMI 的 Nextion 串口屏。参考 /Users/wenhongquan/Downloads/gwc/nextion_web_control.py 的控制方式，完善当前 dzxj_dtu 系统，使 DTU 显示屏能实时显示当前面板端口状态。

参考脚本的核心假设：

串口屏组件：t1.txt（IP/标识）、t2.txt（终端名）、t4.txt（版本号）、g0.txt/g0.pco（MQTT 状态）、p0.pic ~ p23.pic（24 端口状态图）。
通信协议：指令使用 GB2312 编码，以 0xFF 0xFF 0xFF 结尾；触控事件头为 0x65，7 字节一组。
串口参数：COM3 @ 9600；Linux 环境下对应 /dev/ttyS9。

目标

在现有 DTU 系统中接入 Nextion 串口屏，使用独立串口 /dev/ttyS9 @ 9600。
将当前 DTU 管理的 panel（配线架）映射为串口屏上的“终端”，每个 panel 的 24 个端口映射为 p0.pic ~ p23.pic。
屏上“上一终端 / 下一终端”按钮可切换当前显示 panel。
Modbus 轮询更新端口状态时，自动刷新当前显示 panel。
MQTT 状态上报中的 screen_connected 真实反映屏幕串口连接状态。

方案选择

选择 方案 A：新增独立 Nextion 显示模块 + 独立串口。理由：

现有 SerialPort 类为 RS485/Modbus 设计（默认 115200、send_data 自动追加 \n），不适合 Nextion 协议。
Nextion 协议需要 GB2312 编码、0xFF 结尾、按钮事件解析，独立封装更清晰。
独立串口不干扰现有 RS485 通信。
便于后续扩展屏幕显示内容（温湿度、告警等）。

架构

+-----------------+        /dev/ttyS9 @ 9600         +-----------------+
|   app.py        | <--------------------------------> | Nextion 串口屏  |
|                 |                                    | (HMI 已烧录)    |
|  +-----------+  |                                    +-----------------+
|  | Nextion   |  |
|  | Display   |  |
|  +-----------+  |
|        ^        |
|        | reads panel_config / port_state
|        v        |
|  panel_config   |
|  port_state     |
|  mqtt_status    |
+-----------------+

新增模块

`backend/modules/nextion_display.py`

封装 NextionDisplay 类，职责：

打开/关闭屏幕串口。
发送 Nextion 指令（字符串 + 0xFF 0xFF 0xFF）。
发送原始字节。
后台线程监听屏幕返回的按钮事件，解析 0x65 事件包。
提供 on_button_event 回调注册接口。
提供 get_status() 返回连接状态。

关键方法：

connect(port, baudrate=9600, timeout=0.1) -> (bool, str)
disconnect() -> (bool, str)
send_cmd(cmd: str) -> bool
send_raw(data: bytes) -> bool
set_button_callback(callback)
get_status() -> dict

`backend/config.py` 新增配置

# Nextion 串口屏配置
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', '终端')

`app.py` 集成点

1. 初始化

在 app.py 中导入并创建全局实例：

from modules.nextion_display import NextionDisplay

screen_display = NextionDisplay()
screen_current_panel_index = 0

2. 启动连接

在 __main__ 启动序列中，RS485 串口自动连接完成后，增加屏幕串口连接：

if NEXTION_ENABLED:
    try:
        screen_display.connect(NEXTION_SCREEN_PORT, NEXTION_SCREEN_BAUD)
        logger.info(f"Nextion 屏幕串口已连接: {NEXTION_SCREEN_PORT} @ {NEXTION_SCREEN_BAUD}")
    except Exception as e:
        logger.warning(f"Nextion 屏幕串口连接失败: {e}")

3. Panel 排序与映射

维护一个按 position 排序的 panel 列表：

def get_sorted_panels():
    return sorted(panel_config.items(), key=lambda x: x[1].get('position', 0))

切换显示 panel 时：

def switch_screen_panel(direction):
    panels = get_sorted_panels()
    if not panels:
        return
    global screen_current_panel_index
    screen_current_panel_index = (screen_current_panel_index + direction) % len(panels)
    refresh_screen_panel(panels[screen_current_panel_index][0])

4. 屏幕刷新函数

def refresh_screen_panel(panel_id):
    panels = get_sorted_panels()
    if not panels:
        return
    if panel_id not in [p[0] for p in panels]:
        return
    if not screen_display.get_status().get('connected'):
        return

    cfg = panel_config.get(panel_id, {})
    ps = port_state.get(panel_id, {})
    online_count = sum(1 for p in panels if device_last_seen.get(p[0], 0) > time.time() - 60)

    # 端口状态 -> pic 编号
    # 注意：port_state 中只存储 last_uid / expected_uid / last_polled_at，
    # 需要按与 port_poll_loop 相同的规则推导状态
    def _port_status(p):
        last_uid = p.get('last_uid')
        expected_uid = p.get('expected_uid')
        last_polled = p.get('last_polled_at')
        if not last_polled:
            return 'UNKNOWN'
        if last_uid is None:
            return 'DISCONNECTED'
        if expected_uid and last_uid != expected_uid:
            return 'ILLEGAL'
        return 'CONNECTED'

    status_to_pic = {
        'UNKNOWN': 0,
        'DISCONNECTED': 1,
        'ILLEGAL': 2,
        'CONNECTED': 3,
    }
    port_pics = []
    for port_id in range(1, 25):
        p = ps.get(port_id, {})
        status = _port_status(p)
        port_pics.append(status_to_pic.get(status, 0))

    # 找到当前 panel 在排序列表中的索引
    idx = [p[0] for p in panels].index(panel_id)

    # 下发到屏幕
    screen_display.send_cmd(f't2.txt="{NEXTION_PANEL_NAME_PREFIX}{idx + 1}"')
    screen_display.send_cmd(f't1.txt="{panel_id}"')
    screen_display.send_cmd(f't4.txt="{dtu_config.get("firmware_version", "v1.0.0")}"')

    mqtt_connected = mqtt_client.get_status().get('connected', False) if isinstance(mqtt_client.get_status(), dict) else bool(mqtt_client.get_status())
    mqtt_str = '已连接' if mqtt_connected else '未连接'
    screen_display.send_cmd(f'g0.txt="MQTT：{mqtt_str} 当前终端数：{online_count}"')
    screen_display.send_cmd(f'g0.pco={0 if mqtt_connected else 63488}')

    for i, pic in enumerate(port_pics):
        screen_display.send_cmd(f'p{i}.pic={pic}')

5. 按钮事件回调

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)

screen_display.set_button_callback(screen_button_event_handler)

6. 自动刷新触发点

在 port_poll_loop 每次轮询完一个 panel 后：

# 如果该 panel 是当前显示的 panel，刷新屏幕
panels = get_sorted_panels()
if panels and panels[screen_current_panel_index][0] == panel_id:
    refresh_screen_panel(panel_id)

MQTT 状态变化时（mqtt_status_handler）也刷新一次当前 panel。

7. 状态上报修正

将 dtu_publish_status() 中的 screen_connected 改为：

screen_connected = screen_display.get_status().get('connected', False)

8. 新增 HTTP API

GET /api/screen/status：返回屏幕串口连接状态、当前 panel 索引。
POST /api/screen/refresh：手动刷新当前 panel 到屏幕。

请求/响应示例：

// GET /api/screen/status
{
  "success": true,
  "data": {
    "connected": true,
    "port": "/dev/ttyS9",
    "baudrate": 9600,
    "current_panel_index": 0,
    "current_panel_id": "PANEL_dtu_001_1"
  }
}

// POST /api/screen/refresh
{
  "success": true,
  "message": "屏幕已刷新"
}

端口状态映射

DTU 系统状态	含义	Nextion `pic`
`UNKNOWN`	Modbus 读取失败	0
`DISCONNECTED`	未读到跳线	1
`ILLEGAL`	读到跳线但与期望不符	2
`CONNECTED`	读到跳线且与期望一致	3

事件映射

屏幕返回字节序列	含义	动作
`65 00 05 01 FF FF FF`	page 0, comp 5, 按下	上一终端
`65 00 06 01 FF FF FF`	page 0, comp 6, 按下	下一终端

错误处理

串口打开失败：记录 warning，不阻塞 DTU 启动；后续在 panel 刷新时如检测到未连接可尝试重连。
发送失败：单次失败只记录 error；连续失败则标记断开。
事件解析异常：读取线程捕获异常并继续。
panel 为空：屏幕显示等待状态，设备上线后自动刷新。

启动顺序

加载 DTU 配置、设备配置。
启动自动发现线程、DTU 心跳、端口轮询线程。
自动连接 RS485 串口。
连接 Nextion 屏幕串口 /dev/ttyS9 @ 9600。
如果 panel_config 已加载且屏幕已连接，下发第一个 panel。

依赖

pyserial（已依赖）。
无新增第三方包。

测试建议

单独运行 python -m modules.nextion_display 测试串口打开、指令发送、按钮事件解析。
启动 app.py，确认 /api/screen/status 返回 connected: true。
触发一次 /api/screen/refresh，确认屏幕显示终端名、版本、MQTT 状态、24 端口图标。
手动插拔跳线，观察屏幕当前 panel 的端口图标是否实时变化。
按屏上“上一终端 / 下一终端”按钮，观察 panel 切换。

2026-07-06-nextion-display-design.md 9.5 KB Vēsture Neapstrādāts