# 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` 追加配置**

在文件末尾追加：

```python
# 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:
```bash
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**

```bash
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`：

```python
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:
```bash
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`：

```python
#!/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:
```bash
cd /Users/wenhongquan/Documents/trae_projects/dzxj_dtu/backend
pytest tests/test_nextion_display.py -v
```

Expected: 3 个测试全部 PASS

- [ ] **Step 5: Commit**

```bash
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` 这一行，在其下方新增：

```python
from modules.nextion_display import NextionDisplay
```

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

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

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

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

```python
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 行），替换为：

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

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

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

```python
        # 连接 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)` 调用之后，添加：

```python
                            # 如果该 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'):` 分支内，订阅主题之后）添加：

```python
            # 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 路由区域添加：

```python
@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:
```bash
cd /Users/wenhongquan/Documents/trae_projects/dzxj_dtu/backend
python -m py_compile app.py
```

Expected: 无输出（编译成功）

### Step 9: Commit

```bash
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:
```bash
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` 存在）：

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

在另一个终端调用 API：

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

Expected:
```json
{
  "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: 手动刷新屏幕**

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

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

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

- [ ] **Step 4: 测试屏上按钮切换**

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

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

- [ ] **Step 5: 测试端口状态实时刷新**

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

- [ ] **Step 6: Commit 测试结果/日志（可选）**

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

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

---

## Self-Review Checklist

1. **Spec coverage:**
   - 独立 Nextion 模块 → Task 2
   - 配置项 → Task 1
   - `app.py` 集成（初始化、刷新、按钮事件、状态上报） → Task 3
   - HTTP API → Task 3 Step 7
   - 测试验证 → Task 4
   - 无遗漏。

2. **Placeholder scan:**
   - 无 TBD/TODO。
   - 所有代码均给出完整实现或调用示例。

3. **Type consistency：**
   - `NextionDisplay.connect` 返回 `(bool, str)`，与使用处一致。
   - `screen_display.get_status()` 返回 `{'connected': bool, 'port': str|None, 'baudrate': int|None}`，与使用处一致。
   - `screen_current_panel_index` 为全局 `int`，跨函数一致。

4. **已知限制：**
   - 屏幕串口打开失败不会阻塞 DTU 启动；后续手动调用 `/api/screen/refresh` 会返回失败。
   - 当前未实现屏幕串口定时自动重连；如需此功能，可在 `refresh_screen_panel` 检测到未连接时触发一次重连。
