dzxj_dtu/backend/app.py

from flask import Flask, jsonify, request, abort
from flask_cors import CORS
from flask_socketio import SocketIO, emit
import threading
import queue
import time
import json
import os
import logging
import uuid

# 配置文件路径
SERIAL_CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'serial_config.json')

# 导入配置
from config import (
    MAX_BUFFER_SIZE,
    FLASK_SECRET_KEY,
    FLASK_DEBUG,
    FLASK_HOST,
    FLASK_PORT,
    LOG_LEVEL,
    LOG_FORMAT,
    LOG_FILE,
    SOCKETIO_ASYNC_MODE,
    SOCKETIO_ALLOWED_ORIGINS,
    DEFAULT_FORWARD_SERIAL_TO_MQTT,
    DEFAULT_FORWARD_MQTT_TO_SERIAL,
    DEFAULT_MQTT_PUBLISH_TOPIC,
    SOCKETIO_NAMESPACE_DATA,
    SOCKETIO_NAMESPACE_STATUS,
    SOCKETIO_NAMESPACE_CONTROL,
    ERROR_CODES,
    ERROR_MESSAGES,
    DEFAULT_CUSTOMER_ID,
    DEFAULT_DTU_ID,
    DTU_HEARTBEAT_INTERVAL,
    DTU_OFFLINE_TIMEOUT,
    DEFAULT_MQTT_TOPIC_PREFIX,
    DTU_REGISTRATION_TOPIC,
    DTU_STATUS_TOPIC,
    DTU_CONTROL_TOPIC,
    DTU_RESPONSE_TOPIC,
    DTU_EVENT_TOPIC,
    DTU_ALARM_TOPIC,
    DTU_BROADCAST_TOPIC,
    MAX_PANELS,
    MAX_PORTS_PER_PANEL,
    DHT11_ENABLED,
    DHT11_GPIO_PIN,
    DHT11_GPIO_CHIP,
    DHT11_POLL_INTERVAL,
    DHT11_SIMULATE,
    NEXTION_ENABLED,
    NEXTION_SCREEN_PORT,
    NEXTION_SCREEN_BAUD,
    NEXTION_PANEL_NAME_PREFIX,
    NEXTION_CMD_DELAY_MS
)

# 配置日志
logging_config = {
    'level': getattr(logging, LOG_LEVEL),
    'format': LOG_FORMAT
}
if LOG_FILE:
    logging_config['filename'] = LOG_FILE

logging.basicConfig(**logging_config)
logger = logging.getLogger('serial_mqtt_gateway')

from modules.serial_port import SerialPort
from modules.mqtt_client import MQTTClient
from modules.network_config import network_manager
from modules.modbus_rtu import ModbusRTUClient, ANTENNA_ADDRESSES, AddressConfigProtocol, build_broadcast_query, build_confirm_address, build_assign_address
from modules.dht11_sensor import DHT11Sensor
from modules.nextion_display import NextionDisplay

app = Flask(__name__)

app.config['SECRET_KEY'] = FLASK_SECRET_KEY
# 配置CORS以允许nginx代理的前端访问
CORS(app, resources={r"/api/*": {"origins": "*"}, r"/socket.io/*": {"origins": "*"}})

# 初始化SocketIO
socketio = SocketIO(
    app, 
    cors_allowed_origins=SOCKETIO_ALLOWED_ORIGINS, 
    async_mode=SOCKETIO_ASYNC_MODE,
    manage_session=False,  # 禁用会话管理以提高性能
    ping_timeout=30,  # 心跳超时时间
    ping_interval=25,  # 心跳间隔
    logger=FLASK_DEBUG,  # 根据Flask调试模式决定是否记录SocketIO日志
    engineio_logger=FLASK_DEBUG
)

# 初始化串口和MQTT客户端
serial_client = SerialPort()
mqtt_client = MQTTClient()
modbus_client = ModbusRTUClient(serial_client)
address_config = AddressConfigProtocol(serial_client)

# Nextion 串口屏实例
screen_display = NextionDisplay()
screen_current_panel_index = 0
screen_lock = threading.Lock()
screen_refresh_queue = queue.Queue()
screen_refresh_thread = None

# 初始化 DHT11 传感器（后续在 __main__ 中设置回调并启动）
dht11_sensor = None
if DHT11_ENABLED and (DHT11_GPIO_PIN is not None or DHT11_SIMULATE):
    dht11_sensor = DHT11Sensor(
        gpio_pin=DHT11_GPIO_PIN,
        gpio_chip=DHT11_GPIO_CHIP,
        poll_interval=DHT11_POLL_INTERVAL,
        simulate=DHT11_SIMULATE
    )
elif DHT11_ENABLED and DHT11_GPIO_PIN is None and not DHT11_SIMULATE:
    logger.warning("DHT11 已启用但未配置 GPIO pin (DHT11_GPIO_PIN)，跳过本地传感器启动")

# 转发标志
forward_serial_to_mqtt = DEFAULT_FORWARD_SERIAL_TO_MQTT
forward_mqtt_to_serial = DEFAULT_FORWARD_MQTT_TO_SERIAL
mqtt_publish_topic = DEFAULT_MQTT_PUBLISH_TOPIC

# DTU MQTT协议配置（运行时可修改）
dtu_config = {
    'topic_prefix': DEFAULT_MQTT_TOPIC_PREFIX,
    'customer_id': DEFAULT_CUSTOMER_ID,
    'dtu_id': DEFAULT_DTU_ID,
    'firmware_version': 'v1.0.0',
    'hardware_version': 'v1.0',
    'heartbeat_interval': DTU_HEARTBEAT_INTERVAL,
    'enabled': True
}

DTU_CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'dtu_config.json')

def load_dtu_config():
    global dtu_config
    try:
        if os.path.exists(DTU_CONFIG_FILE):
            with open(DTU_CONFIG_FILE, 'r') as f:
                saved = json.load(f)
                dtu_config.update(saved)
    except Exception as e:
        logger.warning(f"加载DTU配置失败: {e}")

def save_dtu_config():
    try:
        with open(DTU_CONFIG_FILE, 'w') as f:
            json.dump(dtu_config, f, indent=2)
    except Exception as e:
        logger.warning(f"保存DTU配置失败: {e}")

# 端口状态追踪（用于事件检测）
port_state = {}  # {panel_id: {port_id: {'last_uid': str, 'expected_uid': str, 'alarm_count': int}}}


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 _screen_refresh_worker():
    """在后台线程中异步刷新 Nextion 屏幕。"""
    while True:
        try:
            item = screen_refresh_queue.get()
            if item is None:
                break
            panel_id = item[0]
            refresh_screen_panel(panel_id)
        except Exception as e:
            logger.error(f"屏幕刷新工作线程异常: {e}")
        finally:
            try:
                screen_refresh_queue.task_done()
            except Exception:
                pass


def refresh_screen_panel(panel_id):
    """将指定 panel 的数据下发到 Nextion 屏幕。"""
    global screen_current_panel_index
    with screen_lock:
        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_st = mqtt_client.get_status()
        mqtt_connected = mqtt_st.get('connected', False) if isinstance(mqtt_st, dict) else bool(mqtt_st)

        delay_ms = NEXTION_CMD_DELAY_MS

        screen_display.send_cmd(f't2.txt="{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。"""
    global screen_current_panel_index
    with screen_lock:
        panels = get_sorted_panels()
        if not panels:
            return False, '没有可用的 panel'
        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)


# 面板配置（从地址配置模块加载）
panel_config = {}  # {panel_id: {'address': int, 'position': int, 'panel_uid': str}}

# 数据存储缓冲区
serial_data_buffer = []
mqtt_data_buffer = []

# 状态标志
serial_status = False
mqtt_status = False

# 串口配置保存和加载函数
def save_serial_config(port, baudrate=9600, timeout=1.0, bytesize=8, parity='N', stopbits=1):
    """保存串口配置"""
    config = {
        'port': port,
        'baudrate': baudrate,
        'timeout': timeout,
        'bytesize': bytesize,
        'parity': parity,
        'stopbits': stopbits
    }
    try:
        with open(SERIAL_CONFIG_FILE, 'w') as f:
            json.dump(config, f)
        logger.info(f"串口配置已保存: {port} @ {baudrate}")
    except Exception as e:
        logger.error(f"保存串口配置失败: {e}")

def load_serial_config():
    """加载串口配置"""
    if os.path.exists(SERIAL_CONFIG_FILE):
        try:
            with open(SERIAL_CONFIG_FILE, 'r') as f:
                config = json.load(f)
            logger.info(f"已加载串口配置: {config.get('port')} @ {config.get('baudrate')}")
            return config
        except Exception as e:
            logger.error(f"加载串口配置失败: {e}")
    return None

def auto_connect_serial():
    """自动连接上次使用的串口"""
    config = load_serial_config()
    if config and config.get('port'):
        port = config.get('port')
        baudrate = config.get('baudrate', 9600)
        timeout = config.get('timeout', 1.0)
        bytesize = config.get('bytesize', 8)
        parity = config.get('parity', 'N')
        stopbits = config.get('stopbits', 1)
        logger.info(f"尝试自动连接串口: {port} @ {baudrate}")
        success, message = serial_client.connect(port, baudrate=baudrate, timeout=timeout, bytesize=bytesize, parity=parity, stopbits=stopbits)
        if success:
            logger.info(f"自动连接串口成功: {port}")
            return True
        else:
            logger.warning(f"自动连接串口失败: {message}")
    return False

# 客户端连接管理
connected_clients = {
    'data': set(),
    'status': set(),
    'control': set()
}

# 设置回调函数
def serial_data_handler(data):
    """处理串口接收的数据"""
    try:
        timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
        serial_data_buffer.append({
            'timestamp': timestamp,
            'data': data,
            'direction': 'in'
        })
        if len(serial_data_buffer) > MAX_BUFFER_SIZE:
            serial_data_buffer.pop(0)
        
        socketio.emit('serial_data', {
            'timestamp': timestamp,
            'data': data,
            'direction': 'in'
        }, namespace=SOCKETIO_NAMESPACE_DATA)
        
        if forward_serial_to_mqtt and mqtt_client.get_status():
            success, msg = mqtt_client.publish(mqtt_publish_topic, data)
            if not success:
                logger.warning(f"串口数据转发到MQTT失败: {msg}")
    except Exception as e:
        logger.error(f"处理串口数据时出错: {str(e)}")

def serial_send_handler(data):
    """处理串口发送的数据"""
    try:
        timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
        serial_data_buffer.append({
            'timestamp': timestamp,
            'data': data,
            'direction': 'out'
        })
        if len(serial_data_buffer) > MAX_BUFFER_SIZE:
            serial_data_buffer.pop(0)
        
        socketio.emit('serial_data', {
            'timestamp': timestamp,
            'data': data,
            'direction': 'out'
        }, namespace=SOCKETIO_NAMESPACE_DATA)
    except Exception as e:
        logger.error(f"处理串口发送数据时出错: {str(e)}")

def serial_status_handler(status):
    """处理串口状态变化"""
    try:
        global serial_status
        serial_status = status
        
        # 通过WebSocket广播状态变化
        socketio.emit('serial_status', {
            'connected': status
        }, namespace=SOCKETIO_NAMESPACE_STATUS)
        
        logger.info(f"串口状态更新: {'已连接' if status else '已断开'}")
    except Exception as e:
        logger.error(f"处理串口状态时出错: {str(e)}")

def mqtt_data_handler(data):
    """处理MQTT接收的数据"""
    try:
        # 添加到缓冲区
        timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
        mqtt_data_buffer.append({
            'timestamp': timestamp,
            'topic': data['topic'],
            'payload': data['payload']
        })
        # 保持缓冲区大小
        if len(mqtt_data_buffer) > MAX_BUFFER_SIZE:
            mqtt_data_buffer.pop(0)
        
        # 通过WebSocket广播数据
        socketio.emit('mqtt_data', {
            'timestamp': timestamp,
            'topic': data['topic'],
            'payload': data['payload']
        }, namespace=SOCKETIO_NAMESPACE_DATA)
        
        # 如果启用了转发且串口已连接，转发数据到串口
        if forward_mqtt_to_serial and serial_client.get_status():
            success, msg = serial_client.send_data(data['payload'])
            if not success:
                logger.warning(f"MQTT数据转发到串口失败: {msg}")
    except Exception as e:
        logger.error(f"处理MQTT数据时出错: {str(e)}")

def mqtt_status_handler(status):
    """处理MQTT状态变化"""
    try:
        global mqtt_status
        mqtt_status = status

        # 通过WebSocket广播状态变化
        socketio.emit('mqtt_status', {
            'connected': status
        }, namespace=SOCKETIO_NAMESPACE_STATUS)

        logger.info(f"MQTT状态更新: {'已连接' if status else '已断开'}")

        # 如果MQTT连接成功且启用了DTU协议，发送注册消息和订阅控制主题
        if status and dtu_config.get('enabled'):
            socketio.sleep(1)  # 等待连接稳定

            # 发送DTU注册消息
            dtu_register()

            # 订阅控制主题
            control_topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'control')
            mqtt_client.subscribe(control_topic, qos=1)
            logger.info(f"已订阅控制主题: {control_topic}")

            # 订阅广播主题 (DTU发现 7.14 和批量配置 7.15)
            discover_topic = build_dtu_topic('broadcast', 'dtu', 'discover')
            config_topic = build_dtu_topic('broadcast', 'dtu', 'config')
            mqtt_client.subscribe(discover_topic, qos=1)
            mqtt_client.subscribe(config_topic, qos=1)
            logger.info(f"已订阅广播主题: {discover_topic}, {config_topic}")

            # MQTT 状态变化时刷新当前屏幕显示
            panels = get_sorted_panels()
            if panels and screen_display.get_status().get('connected'):
                try:
                    with screen_lock:
                        current_idx = screen_current_panel_index
                    if 0 <= current_idx < len(panels):
                        screen_refresh_queue.put((panels[current_idx][0],))
                except Exception as e:
                    logger.error(f"MQTT 状态变化刷新屏幕失败: {e}")
    except Exception as e:
        logger.error(f"处理MQTT状态时出错: {str(e)}")


# ========== DTU MQTT协议处理函数 ==========

def build_dtu_topic(*parts):
    """构建DTU MQTT主题"""
    prefix = dtu_config.get('topic_prefix', '线架系统')
    return '/'.join([prefix] + list(parts))


def dtu_register(discovery_request_id=None):
    """发送DTU注册消息"""
    if not mqtt_client.get_status():
        logger.warning("MQTT未连接，无法发送注册消息")
        return False

    try:
        # 加载面板配置
        devices = address_config.get_stored_devices()
        # 优先使用 dtu_config 中预设的 panel_id 映射 (uid_hex -> panel_id)
        panel_id_map = dtu_config.get('panel_id_mapping', {}) or {}
        panels = []
        panel_id = 1
        for uid_hex, addr in devices.items():
            custom_id = panel_id_map.get(uid_hex) or panel_id_map.get(addr)
            if custom_id:
                pid = str(custom_id)
            else:
                pid = f"PANEL_{dtu_config['dtu_id']}_{addr}"
            panels.append({
                'panel_id': pid,
                'address': addr,
                'position': panel_id
            })
            panel_id += 1

        # 检测网络类型
        network_type = 'ethernet'
        signal_strength = None
        try:
            interfaces = network_manager.get_status().get('interfaces', {})
            for name, info in interfaces.items():
                if name in ('lo', 'docker0') or name.startswith(('br-', 'veth')):
                    continue
                if info.get('status') != 'UP':
                    continue
                if name.startswith(('wlan', 'wlp')):
                    network_type = 'wifi'
                    # 读取信号强度 (dBm, 越接近 0 越强)
                    try:
                        import re
                        with open('/proc/net/wireless') as f:
                            for line in f.readlines()[2:]:
                                m = re.match(r'\s*(\S+):.*\s(\d+)\.\s+(-?\d+)\.', line)
                                if m:
                                    signal_strength = int(m.group(3))
                                    break
                    except Exception:
                        pass
                elif name.startswith(('wwan', 'ppp', 'usb', 'eth')):
                    # 移动网卡
                    if name.startswith(('wwan', 'ppp', 'usb')):
                        network_type = 'cellular'
                break
        except Exception:
            pass

        payload = {
            'msg_id': f"reg_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'REGISTER',
            'payload': {
                'firmware_version': dtu_config.get('firmware_version', 'v1.0.0'),
                'hardware_version': dtu_config.get('hardware_version', 'v1.0'),
                'panel_count': len(panels),
                'network_type': network_type,
                'signal_strength': signal_strength,
                'uptime': int(time.time() * 1000),
                'panels': panels,
                'discovery_request_id': discovery_request_id
            }
        }

        if discovery_request_id:
            payload['payload']['discovery_request_id'] = discovery_request_id

        topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'register')
        success, msg = mqtt_client.publish(topic, json.dumps(payload), qos=1)

        if success:
            logger.info(f"DTU注册消息已发送: {topic}")
        else:
            logger.error(f"DTU注册消息发送失败: {msg}")

        return success
    except Exception as e:
        logger.error(f"发送DTU注册消息失败: {str(e)}")
        return False


_discover_dedup = {}

def handle_broadcast_discover(payload):
    """处理广播发现消息 (7.14)"""
    try:
        if not isinstance(payload, dict) or payload.get('type') != 'DISCOVER':
            return

        p = payload.get('payload', {})
        cust_id = p.get('customer_id', '')
        if cust_id and cust_id != dtu_config.get('customer_id'):
            return

        request_id = p.get('request_id', '')
        now = time.time()
        if request_id and request_id in _discover_dedup:
            if now - _discover_dedup[request_id] < 60:
                return
        _discover_dedup[request_id] = now
        # 清理超过 60s 的旧条目，防止内存泄漏
        if len(_discover_dedup) > 256:
            for k in list(_discover_dedup.keys()):
                if now - _discover_dedup[k] > 60:
                    _discover_dedup.pop(k, None)

        logger.info(f"收到DISCOVER广播，响应注册消息 (request_id={request_id})")
        import random
        delay = random.uniform(0, 3.0)
        time.sleep(delay)

        dtu_register(discovery_request_id=request_id)
        dtu_publish_status()
    except Exception as e:
        logger.error(f"处理广播发现消息失败: {str(e)}")


def handle_broadcast_config(payload):
    """处理广播批量配置消息 (7.15)"""
    try:
        if isinstance(payload, dict) and payload.get('type') == 'CONFIG':
            cfg = payload.get('payload', {})
            cust_id = cfg.get('customer_id', '')
            if cust_id and cust_id != dtu_config.get('customer_id'):
                return

            config_version = cfg.get('config_version', '')
            force_apply = cfg.get('force_apply', False)
            items = cfg.get('items', {})

            local_version = dtu_config.get('config_version', '')
            incoming_msg_id = payload.get('msg_id', '')
            if local_version == config_version and not force_apply:
                logger.info(f"配置版本 {config_version} 已存在，跳过")
                _send_config_response(config_version, [], True, 1011, incoming_msg_id)
                return

            valid_ranges = {
                'heartbeat_interval': (10, 3600),
                'poll_interval_ms': (100, 10000),
                'modbus_timeout_ms': (200, 5000),
                'mqtt_keepalive': (10, 300),
                'event_debounce_ms': (0, 5000),
            }
            applied = []
            for key, val in items.items():
                if key in valid_ranges:
                    lo, hi = valid_ranges[key]
                    if not (lo <= val <= hi):
                        _send_config_response(config_version, applied, False, 1012, incoming_msg_id)
                        return
                    if key == 'heartbeat_interval':
                        dtu_config['heartbeat_interval'] = val
                    elif key == 'mqtt_keepalive':
                        dtu_config['mqtt_keepalive'] = val
                    elif key == 'event_debounce_ms':
                        dtu_config['event_debounce_ms'] = val
                    elif key == 'poll_interval_ms':
                        dtu_config['poll_interval_ms'] = val
                    elif key == 'modbus_timeout_ms':
                        dtu_config['modbus_timeout_ms'] = val
                    applied.append(key)
                elif key in ('alarm_enabled',):
                    dtu_config['alarm_enabled'] = bool(val)
                    applied.append(key)
                else:
                    _send_config_response(config_version, applied, False, 1013, incoming_msg_id)
                    return

            if applied:
                dtu_config['config_version'] = config_version
                try:
                    save_dtu_config()
                except Exception as e:
                    logger.error(f"配置写入持久化失败: {e}")
                    _send_config_response(config_version, applied, False, 1014, incoming_msg_id)
                    return

            _send_config_response(config_version, applied, True, 0, incoming_msg_id)

    except Exception as e:
        logger.error(f"处理广播配置消息失败: {str(e)}")


def _send_config_response(config_version, applied_items, success, error_code, original_msg_id=''):
    """发送批量配置响应 (随机 0-2000ms 延迟避免广播风暴)"""
    import time as _t, json as _j, random as _r
    delay = _r.uniform(0, 2.0)

    def _delayed_publish():
        _t.sleep(delay)
        topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'response')
        payload = {
            'msg_id': f"rsp_cfg_{int(_t.time() * 1000)}",
            'timestamp': int(_t.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'RESPONSE',
            'payload': {
                'original_msg_id': original_msg_id or f"cfg_{config_version}",
                'command': 'BROADCAST_CONFIG',
                'success': success,
                'result': {
                    'applied_version': config_version,
                    'applied_items': applied_items,
                    'restart_required': False
                } if success else None,
                'error_code': error_code,
                'error_message': get_ota_error_message(error_code) if error_code and error_code >= 1000 else None
            }
        }
        mqtt_client.publish(topic, _j.dumps(payload), qos=1)

    # 异步执行避免阻塞 MQTT 主循环
    threading.Thread(target=_delayed_publish, daemon=True).start()


def dtu_publish_status(force=False):
    """发送DTU状态/心跳消息

    Args:
        force: 保留参数。事件/告警/查询触发时传入 True，用于立即上报（区别于定时心跳）
    """
    if not mqtt_client.get_status() or not dtu_config.get('enabled'):
        return False
    # 调试日志: 区分定时 vs 触发上报
    if force:
        logger.debug("dtu_publish_status(force=True) - 事件触发立即上报")

    try:
        # panel_online 计数: 在最近 60s 内所有 24 个端口都被成功轮询过
        # 严格语义: 全部 24 端口在最近一轮 poll 周期内被 poll 成功 → online
        now_ms = int(time.time() * 1000)
        threshold_ms = 60_000  # 60 秒未轮询视为 offline
        poll_ok = 0
        for pid, cfg in panel_config.items():
            ps = port_state.get(pid, {})
            expected_ports = cfg.get('port_count', 24)
            polled_recently = sum(
                1 for p in ps.values()
                if p.get('last_polled_at') and (now_ms - p.get('last_polled_at', 0)) < threshold_ms
            )
            if polled_recently >= expected_ports:
                poll_ok += 1
        panel_online = poll_ok
        panel_offline = max(0, len(panel_config) - panel_online)

        # 检查串口状态
        serial_st = serial_client.get_status()
        rs485_status = 'normal' if (isinstance(serial_st, dict) and serial_st.get('connected', False)) else 'error'

        # 网络状态: 检查实际接口 UP 状态
        # 规则: 有线 (eth*/wlan*) UP → online; 仅 wifi 信号弱 (无 link/低质量) → weak; 否则 offline
        network_status = 'offline'
        try:
            interfaces = network_manager.get_status().get('interfaces', {})
            real_ifs = {i: s for i, s in interfaces.items()
                        if s.get('status') == 'UP'
                        and i not in ('lo', 'docker0')
                        and not i.startswith(('br-', 'veth'))}
            if real_ifs:
                # 优先有线连接
                if any(i.startswith(('eth', 'en')) for i in real_ifs):
                    network_status = 'online'
                else:
                    # 仅无线: 检查信号强度 (Linux: /proc/net/wireless)
                    try:
                        import re
                        with open('/proc/net/wireless') as f:
                            lines = f.readlines()
                        quality = None
                        for line in lines[2:]:  # 跳过表头
                            m = re.match(r'\s*(\S+):.*\s(\d+)\.\s+(\d+)\.\s+(\d+)', line)
                            if m:
                                # link quality (字段 2, 0-70); signal level (字段 3, dBm, 越低越差)
                                link = int(m.group(2))
                                sig = int(m.group(3))
                                # 估算: link < 30 或 sig < -75 dBm 视为弱
                                if quality is None or link < quality:
                                    quality = link
                        network_status = 'weak' if (quality is not None and quality < 30) else 'online'
                    except Exception:
                        network_status = 'online'
        except Exception:
            network_status = 'offline'

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

        # CPU/内存使用率 (Linux 才有意义)
        cpu_usage = None
        memory_usage = None
        dtu_temperature = None
        try:
            import psutil
            cpu_usage = psutil.cpu_percent(interval=None)
            memory_usage = psutil.virtual_memory().percent
            # 主板温度: 优先 psutil.sensors_temperatures
            try:
                temps = psutil.sensors_temperatures(fahrenheit=False) or {}
                for chip_name, entries in temps.items():
                    for entry in entries:
                        if entry.current and entry.current > 0:
                            dtu_temperature = round(entry.current, 1)
                            break
                    if dtu_temperature is not None:
                        break
            except Exception:
                pass
            # 回退: 直接读 /sys/class/thermal/thermal_zone*/temp
            if dtu_temperature is None:
                import glob as _glob
                for tz in sorted(_glob.glob('/sys/class/thermal/thermal_zone*/temp')):
                    try:
                        with open(tz) as f:
                            v = int(f.read().strip())
                            if v > 0:
                                dtu_temperature = round(v / 1000.0, 1)
                                break
                    except Exception:
                        continue
        except Exception:
            pass

        # 优先使用外部上报的 dtu_temperature (来自 MQTT/RS485 传感器)
        dtu_temperature = env_sensor_data.get('dtu_temperature') or dtu_temperature

        payload = {
            'msg_id': f"hbt_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'STATUS',
            'payload': {
                'cpu_usage': cpu_usage,
                'memory_usage': memory_usage,
                'temperature': dtu_temperature,
                'network_status': network_status,
                'panel_online': panel_online,
                'panel_offline': panel_offline,
                'screen_connected': screen_connected,
                'mqtt_connected': mqtt_client.get_status().get('connected', False) if isinstance(mqtt_client.get_status(), dict) else bool(mqtt_client.get_status()),
                'rs485_status': rs485_status
            }
        }

        topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'status')
        success, _ = mqtt_client.publish(topic, json.dumps(payload), qos=1)
        return success
    except Exception as e:
        logger.error(f"发送DTU状态消息失败: {str(e)}")
        return False


def dtu_publish_event(panel_id, port_id, event_type, jumper_uid, previous_jumper_uid=None):
    """发送端口事件消息"""
    if not mqtt_client.get_status() or not dtu_config.get('enabled'):
        return False

    try:
        payload = {
            'msg_id': f"evt_{int(time.time() * 1000)}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'EVENT',
            'payload': {
                'panel_id': panel_id,
                'port_id': port_id,
                'event_type': event_type,
                'event_id': f"evt_{uuid.uuid4().hex[:8]}",
                'jumper_uid': jumper_uid,
                'previous_jumper_uid': previous_jumper_uid
            }
        }

        topic = build_dtu_topic(dtu_config['customer_id'], 'patchpanel', dtu_config['dtu_id'], panel_id, 'event')
        success, _ = mqtt_client.publish(topic, json.dumps(payload), qos=1)

        if success:
            logger.info(f"端口事件已发送: {panel_id}:{port_id} - {event_type}")
            # 记录到事件历史
            port_event_history.append({
                'timestamp': int(time.time() * 1000),
                'panel_id': panel_id,
                'port_id': port_id,
                'event_type': event_type,
                'jumper_uid': jumper_uid,
                'previous_jumper_uid': previous_jumper_uid
            })
            # 限制历史记录数量
            if len(port_event_history) > 1000:
                port_event_history[:] = port_event_history[-1000:]

        # 事件驱动: 立即上报一次 status (符合 7.4 "状态变化时立即上报")
        try:
            dtu_publish_status(force=True)
        except Exception:
            pass
        return success
    except Exception as e:
        logger.error(f"发送端口事件失败: {str(e)}")
        return False


def dtu_publish_alarm(panel_id, port_id, alarm_type, expected_jumper_uid, actual_jumper_uid, severity='WARNING'):
    """发送非法告警消息"""
    if not mqtt_client.get_status() or not dtu_config.get('enabled'):
        return False

    try:
        description = f"端口{port_id}期望跳线{expected_jumper_uid}，实际{'未读到' if not actual_jumper_uid else actual_jumper_uid}"

        payload = {
            'msg_id': f"alm_{int(time.time() * 1000)}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'ALARM',
            'payload': {
                'panel_id': panel_id,
                'port_id': port_id,
                'alarm_type': alarm_type,
                'severity': severity,
                'expected_jumper_uid': expected_jumper_uid,
                'actual_jumper_uid': actual_jumper_uid,
                'description': description
            }
        }

        topic = build_dtu_topic(dtu_config['customer_id'], 'patchpanel', dtu_config['dtu_id'], panel_id, 'alarm')
        success, _ = mqtt_client.publish(topic, json.dumps(payload), qos=1)

        if success:
            logger.warning(f"非法告警已发送: {panel_id}:{port_id} - {alarm_type}")
            # 记录到事件历史
            port_event_history.append({
                'timestamp': int(time.time() * 1000),
                'panel_id': panel_id,
                'port_id': port_id,
                'event_type': 'ALARM',
                'alarm_type': alarm_type,
                'expected_jumper_uid': expected_jumper_uid,
                'actual_jumper_uid': actual_jumper_uid
            })
            # 限制历史记录数量
            if len(port_event_history) > 1000:
                port_event_history[:] = port_event_history[-1000:]

        # 事件驱动: 告警后立即上报一次 status
        try:
            dtu_publish_status(force=True)
        except Exception:
            pass
        return success
    except Exception as e:
        logger.error(f"发送非法告警失败: {str(e)}")
        return False


def dtu_publish_panel_status(panel_id, address, ports_data):
    """发送配线架状态消息 (7.7)"""
    if not mqtt_client.get_status() or not dtu_config.get('enabled'):
        return False
    try:
        topic = build_dtu_topic(dtu_config['customer_id'], 'patchpanel', dtu_config['dtu_id'], panel_id, 'status')
        payload = {
            'msg_id': f"pst_{int(time.time() * 1000)}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'STATUS',
            'payload': {
                'panel_id': panel_id,
                'position': panel_config.get(panel_id, {}).get('position', 0),
                'address': address,
                'online': True,
                'last_poll_time': int(time.time() * 1000),
                'ports': ports_data
            }
        }
        mqtt_client.publish(topic, json.dumps(payload), qos=0)
        return True
    except Exception as e:
        logger.error(f"发送面板状态失败: {str(e)}")
        return False


def dtu_publish_jumper_status():
    """发送跳线状态汇总 (7.8) - STATUS envelope with panels[] array"""
    if not mqtt_client.get_status() or not dtu_config.get('enabled'):
        return False
    try:
        panels_arr = []
        for panel_id, cfg in panel_config.items():
            ports_arr = []
            ps = port_state.get(panel_id, {})
            for port_id in range(1, 25):
                p = ps.get(port_id, {})
                last_uid = p.get('last_uid')
                expected_uid = p.get('expected_uid')
                last_polled = p.get('last_polled_at')
                if not last_polled:
                    status = 'UNKNOWN'
                elif last_uid is None:
                    status = 'DISCONNECTED'
                    if expected_uid:
                        status = 'ILLEGAL'
                elif expected_uid and last_uid != expected_uid:
                    status = 'ILLEGAL'
                else:
                    status = 'CONNECTED'
                ports_arr.append({
                    'port_id': port_id,
                    'status': status,
                    'jumper_uid': last_uid
                })
            panels_arr.append({
                'panel_id': panel_id,
                'position': cfg.get('position', 0),
                'ports': ports_arr
            })

        topic = build_dtu_topic(dtu_config['customer_id'], 'jumper', dtu_config['dtu_id'], 'status')
        payload = {
            'msg_id': f"jmp_{int(time.time() * 1000)}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'STATUS',
            'payload': {'panels': panels_arr}
        }
        mqtt_client.publish(topic, json.dumps(payload), qos=0)
        return True
    except Exception as e:
        logger.error(f"发送跳线状态汇总失败: {str(e)}")
        return False


_env_sensor_data = {
    'temperature': None,
    'humidity': None,
    'sensor_update_time': None
}


def update_env_sensor_data(temperature, humidity, dtu_temperature=None, sensor_update_time=None):
    """更新环境传感器数据 - 委托给完整实现（带历史和告警）"""
    update_env_sensor_data_full(temperature, humidity, dtu_temperature, sensor_update_time)


def dtu_publish_env_sensor():
    """发送环境传感器数据 (7.9) - STATUS envelope"""
    if not mqtt_client.get_status() or not dtu_config.get('enabled'):
        return False
    try:
        topic = build_dtu_topic(dtu_config['customer_id'], 'env', dtu_config['dtu_id'], 'sensor')
        payload = {
            'msg_id': f"sen_{int(time.time() * 1000)}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'STATUS',
            'payload': {
                'temperature': _env_sensor_data.get('temperature'),
                'humidity': _env_sensor_data.get('humidity'),
                'sensor_update_time': _env_sensor_data.get('sensor_update_time')
            }
        }
        mqtt_client.publish(topic, json.dumps(payload), qos=0)
        return True
    except Exception as e:
        logger.error(f"发送环境传感器数据失败: {str(e)}")
        return False


def dtu_handle_control(topic, payload):
    """处理下行控制指令"""
    try:
        if not dtu_config.get('enabled'):
            return

        command = payload.get('payload', {}).get('command')
        target = payload.get('payload', {}).get('target')
        params = payload.get('payload', {}).get('params', {})

        logger.info(f"收到控制指令: command={command}, target={target}")

        response_payload = {
            'msg_id': f"rsp_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'RESPONSE',
            'payload': {
                'original_msg_id': payload.get('msg_id'),
                'command': command,
                'success': True,
                'result': {},
                'error_code': 0,
                'error_message': None
            }
        }
        topic_response_dtu = build_dtu_topic(
            dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'response')

        # 处理各命令
        if command == 'SET_PORT_LED':
            # 设置端口LED
            port_id = params.get('port_id')
            led_mode = params.get('led_mode', 'OFF')

            # 参数校验
            if port_id is None or not (1 <= int(port_id) <= 24):
                response_payload['payload']['success'] = False
                response_payload['payload']['error_code'] = 1003
                response_payload['payload']['error_message'] = f"端口号非法: {port_id}"
                mqtt_client.publish(topic_response_dtu, json.dumps(response_payload), qos=1)
                return

            if led_mode not in ('OFF', 'BLINK_RED', 'BLINK_GREEN', 'BLINK_BLUE'):
                response_payload['payload']['success'] = False
                response_payload['payload']['error_code'] = 1004
                response_payload['payload']['error_message'] = f"LED模式非法: {led_mode}"
                mqtt_client.publish(topic_response_dtu, json.dumps(response_payload), qos=1)
                return

            # LED模式映射
            led_mode_map = {'OFF': 0, 'BLINK_RED': 1, 'BLINK_GREEN': 2, 'BLINK_BLUE': 3}
            color = led_mode_map.get(led_mode, 0)

            # 查找设备地址
            device_address = 1  # 默认
            for panel_id, cfg in panel_config.items():
                if panel_id == target:
                    device_address = cfg.get('address', 1)
                    break

            result = modbus_client.set_rgb_led(device_address, port_id, color)
            response_payload['payload']['success'] = 'error' not in result
            response_payload['payload']['result'] = result
            if 'error' in result:
                response_payload['payload']['error_code'] = 1005
                response_payload['payload']['error_message'] = result['error']

        elif command == 'QUERY_DTU_STATUS':
            # 查询DTU状态
            dtu_publish_status(force=True)

        elif command == 'SYNC_PORT_MAPPING':
            # 同步单端口期望映射
            port_id = params.get('port_id')
            jumper_uid = params.get('jumper_uid')

            if port_id is None or target is None or target not in panel_config:
                response_payload['payload']['success'] = False
                response_payload['payload']['error_code'] = 1006 if port_id is None else 1002
                response_payload['payload']['error_message'] = "参数缺失" if port_id is None else f"目标面板不存在: {target}"
                mqtt_client.publish(topic_response_dtu, json.dumps(response_payload), qos=1)
                return

            if target not in port_state:
                port_state[target] = {}
            if port_id not in port_state[target]:
                port_state[target][port_id] = {'last_uid': None, 'expected_uid': None, 'alarm_count': 0}

            port_state[target][port_id]['expected_uid'] = jumper_uid
            save_dtu_config()

        elif command == 'SYNC_ALL_MAPPING':
            # 批量同步期望映射
            mappings = params.get('mappings', [])
            if not isinstance(mappings, list):
                response_payload['payload']['success'] = False
                response_payload['payload']['error_code'] = 1006
                response_payload['payload']['error_message'] = "mappings 字段格式错误"
                mqtt_client.publish(topic_response_dtu, json.dumps(response_payload), qos=1)
                return
            try:
                for mapping in mappings:
                    panel_id = mapping.get('panel_id')
                    port_id = mapping.get('port_id')
                    jumper_uid = mapping.get('jumper_uid')

                    if panel_id not in panel_config:
                        raise ValueError(f"目标面板不存在: {panel_id}")
                    if panel_id not in port_state:
                        port_state[panel_id] = {}
                    if port_id not in port_state[panel_id]:
                        port_state[panel_id][port_id] = {'last_uid': None, 'expected_uid': None, 'alarm_count': 0}
                    port_state[panel_id][port_id]['expected_uid'] = jumper_uid
                save_dtu_config()
            except Exception as e:
                response_payload['payload']['success'] = False
                response_payload['payload']['error_code'] = 1007
                response_payload['payload']['error_message'] = f"同步映射失败: {e}"
                mqtt_client.publish(topic_response_dtu, json.dumps(response_payload), qos=1)
                return

        elif command == 'REBOOT':
            # 重启DTU: 1s 延迟后执行 reboot
            import subprocess
            logger.warning(f"MQTT REBOOT 命令收到，1 秒后执行系统重启")
            threading.Thread(target=lambda: (
                time.sleep(1),
                subprocess.run(['reboot'], capture_output=True)
            ), daemon=True).start()
            response_payload['payload']['result'] = {'message': 'Reboot scheduled in 1s'}

        elif command == 'READ_PANEL_STATUS':
            # 读取面板状态
            # 读取面板的多个寄存器获取状态信息
            # 寄存器地址定义: 0x0000=运行状态, 0x0001=LED控制, 0x0002-0x0009=天线卡状态
            device_address = 1
            for panel_id, cfg in panel_config.items():
                if panel_id == target:
                    device_address = cfg.get('address', 1)
                    break

            # 读取面板状态寄存器 (地址0x0000开始，读取10个寄存器)
            status_result = modbus_client.read_holding_registers(device_address, 0x0000, 10)

            # 解析状态
            registers = status_result.get('registers', [])
            panel_status = {
                'device_address': device_address,
                'run_status': registers[0] if len(registers) > 0 else None,
                'led_control': registers[1] if len(registers) > 1 else None,
                'antenna_status': registers[2:10] if len(registers) >= 10 else [],
                'raw_registers': registers
            }

            response_payload['payload']['success'] = 'error' not in status_result
            response_payload['payload']['result'] = panel_status

        elif command == 'QUERY_JUMPER_STATUS':
            # 查询跳线器状态汇总 (7.8) - 触发 STATUS 上行至 jumper/{dtu_id}/status
            dtu_publish_jumper_status()
            return  # dtu_publish_jumper_status 已发送响应

        elif command == 'QUERY_ENV_SENSOR':
            # 查询环境传感器 (7.9) - 触发 STATUS 上行至 env/{dtu_id}/sensor
            dtu_publish_env_sensor()
            return  # dtu_publish_env_sensor 已发送响应

        elif command == 'OTA_UPGRADE':
            params = payload.get('payload', {}).get('params', {})
            target_version = params.get('firmware_version')
            force_upgrade = params.get('force_upgrade', False)
            current_version = dtu_config.get('firmware_version', 'v1.0.0')
            incoming_msg_id = payload.get('msg_id', '')
            if target_version == current_version and not force_upgrade:
                response_payload['payload']['success'] = False
                response_payload['payload']['error_code'] = 1021
                response_payload['payload']['error_message'] = f"已是目标版本: {target_version}"
                mqtt_client.publish(topic_response_dtu, json.dumps(response_payload), qos=1)
                return
            if params.get('firmware_url'):
                import threading
                threading.Thread(target=lambda: _run_ota({**params, '_incoming_msg_id': incoming_msg_id}), daemon=True).start()
            response_payload['payload']['success'] = True
            response_payload['payload']['result'] = {
                'firmware_version': target_version,
                'ota_status': 'DOWNLOADING',
                'restart_required': False
            }

        elif command == 'OTA_CANCEL':
            # 取消OTA升级
            # 向OTA控制寄存器写入取消命令 (0x0000=取消)
            device_address = 1
            for panel_id, cfg in panel_config.items():
                if panel_id == target:
                    device_address = cfg.get('address', 1)
                    break

            # OTA控制寄存器地址: 0x0300
            # 值: 0x0000=取消升级
            cancel_result = modbus_client.write_single_register(device_address, 0x0300, 0x0000)

            response_payload['payload']['success'] = 'error' not in cancel_result
            response_payload['payload']['result'] = {
                'message': 'OTA upgrade cancelled' if 'error' not in cancel_result else 'Failed to cancel OTA',
                'raw_result': cancel_result
            }

        else:
            # 未知命令
            response_payload['payload']['success'] = False
            response_payload['payload']['error_code'] = 1001
            response_payload['payload']['error_message'] = f"未知命令: {command}"
            mqtt_client.publish(topic_response_dtu, json.dumps(response_payload), qos=1)
            return

        # 发送响应（QUERY_DTU_STATUS 已由 dtu_publish_status 发送；QUERY_JUMPER_STATUS/QUERY_ENV_SENSOR 已在分支中发布）
        if command == 'QUERY_DTU_STATUS':
            pass
        else:
            cmd = command
            if cmd == 'READ_PANEL_STATUS':
                if target and target != 'all' and target in panel_config:
                    # 单个面板: 使用 panel_id 作为 topic segment
                    topic = build_dtu_topic(dtu_config['customer_id'], 'patchpanel',
                                            dtu_config['dtu_id'], target, 'status')
                    mqtt_client.publish(topic, json.dumps(response_payload), qos=0)
                else:
                    # target=all: 遍历所有面板发布到各自的 topic
                    for pid in panel_config.keys():
                        topic = build_dtu_topic(dtu_config['customer_id'], 'patchpanel',
                                                dtu_config['dtu_id'], pid, 'status')
                        mqtt_client.publish(topic, json.dumps(response_payload), qos=0)
            else:
                topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'response')
                p = response_payload['payload']
                if not p.get('success', True) and p.get('error_code', 0) == 0:
                    p['error_code'] = 9999
                    p['error_message'] = p.get('error_message') or '未知错误'
                mqtt_client.publish(topic, json.dumps(response_payload), qos=1)

    except Exception as e:
        logger.error(f"处理控制指令失败: {str(e)}")


# 启动DTU心跳定时器
def start_dtu_heartbeat():
    """启动DTU心跳定时任务"""
    def heartbeat_task():
        env_tick = 0
        while True:
            socketio.sleep(dtu_config.get('heartbeat_interval', DTU_HEARTBEAT_INTERVAL))
            if mqtt_client.get_status() and dtu_config.get('enabled'):
                dtu_publish_status()
                # 每 2 个心跳周期 (120s) 主动推送一次环境传感器数据 (7.9)
                env_tick += 1
                if env_tick >= 2:
                    env_tick = 0
                    dtu_publish_env_sensor()

    socketio.start_background_task(target=heartbeat_task)


# 修改mqtt_data_handler以处理控制指令
def mqtt_data_handler_extended(data):
    """处理MQTT接收的数据（扩展版，含DTU协议）"""
    try:
        topic = data.get('topic', '')
        payload_str = data.get('payload', '')

        # 尝试解析JSON
        try:
            payload = json.loads(payload_str) if isinstance(payload_str, str) else payload_str
        except:
            payload = payload_str

        # 添加到缓冲区
        timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
        mqtt_data_buffer.append({
            'timestamp': timestamp,
            'topic': topic,
            'payload': payload_str
        })
        if len(mqtt_data_buffer) > MAX_BUFFER_SIZE:
            mqtt_data_buffer.pop(0)

        # 通过WebSocket广播
        socketio.emit('mqtt_data', {
            'timestamp': timestamp,
            'topic': topic,
            'payload': payload_str
        }, namespace=SOCKETIO_NAMESPACE_DATA)

        # 检查是否是控制指令主题
        expected_control_topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'control')
        if topic == expected_control_topic:
            dtu_handle_control(topic, payload)

        # 检查是否是状态主题，用于处理OTA状态上报
        expected_status_topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'status')
        if topic == expected_status_topic and isinstance(payload, dict):
            if 'ota_status' in payload or 'firmware_version' in payload:
                handle_ota_status(dtu_config['dtu_id'], payload)

        # 检查是否是环境传感器数据主题 (兼容旧 sensor/.../data 路径，spec 7.9 为 env/.../sensor)
        sensor_topic = build_dtu_topic(dtu_config['customer_id'], 'sensor', dtu_config['dtu_id'], 'data')
        env_topic = build_dtu_topic(dtu_config['customer_id'], 'env', dtu_config['dtu_id'], 'sensor')
        if topic in (sensor_topic, env_topic) and isinstance(payload, dict):
            if 'temperature' in payload or 'humidity' in payload:
                update_env_sensor_data(
                    payload.get('temperature'),
                    payload.get('humidity'),
                    payload.get('dtu_temperature'),
                    payload.get('sensor_update_time')
                )

        # 检查是否是广播主题 - DTU发现 (7.14)
        discover_topic = build_dtu_topic('broadcast', 'dtu', 'discover')
        if topic == discover_topic:
            handle_broadcast_discover(payload)

        # 检查是否是广播主题 - 批量配置 (7.15)
        config_topic = build_dtu_topic('broadcast', 'dtu', 'config')
        if topic == config_topic:
            handle_broadcast_config(payload)

        # 转发到串口（如果启用）
        if forward_mqtt_to_serial and serial_client.get_status():
            success, msg = serial_client.send_data(payload_str)
            if not success:
                logger.warning(f"MQTT数据转发到串口失败: {msg}")
    except Exception as e:
        logger.error(f"处理MQTT数据时出错: {str(e)}")

# WebSocket事件处理
@socketio.on('connect', namespace=SOCKETIO_NAMESPACE_DATA)
def handle_data_connect():
    """处理数据命名空间的连接"""
    try:
        client_id = str(uuid.uuid4())
        connected_clients['data'].add(client_id)
        logger.info(f'客户端已连接到数据命名空间，当前连接数: {len(connected_clients["data"])}')
        
        # 发送当前的缓冲区数据，限制发送的历史记录数量
        max_history = 100  # 限制发送的历史记录数量以提高性能
        emit('serial_data_history', {'data': serial_data_buffer[-max_history:]})
        emit('mqtt_data_history', {'data': mqtt_data_buffer[-max_history:]})
        
        # 存储客户端ID以便断开连接时使用
        socketio.start_background_task(target=lambda: None)  # 确保上下文可用
    except Exception as e:
        logger.error(f"处理数据命名空间连接时出错: {str(e)}")

@socketio.on('connect', namespace=SOCKETIO_NAMESPACE_STATUS)
def handle_status_connect():
    """处理状态命名空间的连接"""
    try:
        client_id = str(uuid.uuid4())
        connected_clients['status'].add(client_id)
        logger.info(f'客户端已连接到状态命名空间，当前连接数: {len(connected_clients["status"])}')
        
        # 发送当前状态
        emit('serial_status', {'connected': serial_status})
        emit('mqtt_status', {'connected': mqtt_status})
        emit('forward_status', {
            'serial_to_mqtt': forward_serial_to_mqtt,
            'mqtt_to_serial': forward_mqtt_to_serial,
            'publish_topic': mqtt_publish_topic
        })
    except Exception as e:
        logger.error(f"处理状态命名空间连接时出错: {str(e)}")

@socketio.on('connect', namespace=SOCKETIO_NAMESPACE_CONTROL)
def handle_control_connect():
    """处理控制命名空间的连接"""
    try:
        client_id = str(uuid.uuid4())
        connected_clients['control'].add(client_id)
        logger.info(f'客户端已连接到控制命名空间，当前连接数: {len(connected_clients["control"])}')
    except Exception as e:
        logger.error(f"处理控制命名空间连接时出错: {str(e)}")

@socketio.on('disconnect', namespace=SOCKETIO_NAMESPACE_DATA)
def handle_data_disconnect():
    """处理数据命名空间的断开连接"""
    try:
        # 清理客户端连接记录
        # 在实际应用中，可能需要更复杂的逻辑来追踪具体哪个客户端断开了连接
        if len(connected_clients['data']) > 0:
            # 这里简化处理，实际应该维护session到client_id的映射
            connected_clients['data'].pop()  # 注意：这是一个简化的实现
        logger.info(f'客户端已断开数据命名空间的连接，当前连接数: {len(connected_clients["data"])}')
    except Exception as e:
        logger.error(f"处理数据命名空间断开连接时出错: {str(e)}")

@socketio.on('disconnect', namespace=SOCKETIO_NAMESPACE_STATUS)
def handle_status_disconnect():
    """处理状态命名空间的断开连接"""
    try:
        # 清理客户端连接记录
        if len(connected_clients['status']) > 0:
            connected_clients['status'].pop()
        logger.info(f'客户端已断开状态命名空间的连接，当前连接数: {len(connected_clients["status"])}')
    except Exception as e:
        logger.error(f"处理状态命名空间断开连接时出错: {str(e)}")

@socketio.on('disconnect', namespace=SOCKETIO_NAMESPACE_CONTROL)
def handle_control_disconnect():
    """处理控制命名空间的断开连接"""
    try:
        # 清理客户端连接记录
        if len(connected_clients['control']) > 0:
            connected_clients['control'].pop()
        logger.info(f'客户端已断开控制命名空间的连接，当前连接数: {len(connected_clients["control"])}')
    except Exception as e:
        logger.error(f"处理控制命名空间断开连接时出错: {str(e)}")

@socketio.on('serial_send', namespace=SOCKETIO_NAMESPACE_CONTROL)
def handle_serial_send(data):
    """通过WebSocket处理串口发送数据请求"""
    try:
        message = data.get('message', '')
        if not message:
            emit('serial_send_response', {
                'success': False,
                'message': '消息内容不能为空',
                'error_code': ERROR_CODES['CONFIG_ERROR']
            })
            return
        
        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            emit('serial_send_response', {
                'success': False,
                'message': '串口未连接',
                'error_code': ERROR_CODES['SERIAL_CONNECTION_ERROR']
            })
            return
            
        success, msg = serial_client.send_data(message)
        emit('serial_send_response', {
            'success': success,
            'message': msg,
            'error_code': ERROR_CODES['SUCCESS'] if success else ERROR_CODES['SERIAL_SEND_ERROR']
        })
        
        if success:
            logger.info(f"通过WebSocket发送串口数据成功: {message[:50]}..." if len(message) > 50 else message)
    except Exception as e:
        error_msg = f"处理串口发送请求时出错: {str(e)}"
        logger.error(error_msg)
        emit('serial_send_response', {
            'success': False,
            'message': error_msg,
            'error_code': ERROR_CODES['UNKNOWN_ERROR']
        })

@socketio.on('mqtt_publish', namespace=SOCKETIO_NAMESPACE_CONTROL)
def handle_mqtt_publish(data):
    """通过WebSocket处理MQTT发布数据请求"""
    try:
        topic = data.get('topic', '')
        message = data.get('message', '')
        
        if not topic or not message:
            emit('mqtt_publish_response', {
                'success': False,
                'message': '主题或消息内容不能为空',
                'error_code': ERROR_CODES['CONFIG_ERROR']
            })
            return
        
        if not mqtt_client.get_status():
            emit('mqtt_publish_response', {
                'success': False,
                'message': 'MQTT未连接',
                'error_code': ERROR_CODES['MQTT_CONNECTION_ERROR']
            })
            return
            
        success, msg = mqtt_client.publish(topic, message)
        emit('mqtt_publish_response', {
            'success': success,
            'message': msg,
            'error_code': ERROR_CODES['SUCCESS'] if success else ERROR_CODES['MQTT_PUBLISH_ERROR']
        })
        
        if success:
            logger.info(f"通过WebSocket发布MQTT消息成功: 主题={topic}, 消息={message[:50]}..." if len(message) > 50 else message)
    except Exception as e:
        error_msg = f"处理MQTT发布请求时出错: {str(e)}"
        logger.error(error_msg)
        emit('mqtt_publish_response', {
            'success': False,
            'message': error_msg,
            'error_code': ERROR_CODES['UNKNOWN_ERROR']
        })

@socketio.on('update_forward_config', namespace=SOCKETIO_NAMESPACE_CONTROL)
def handle_update_forward_config(data):
    """通过WebSocket更新转发配置"""
    global forward_serial_to_mqtt, forward_mqtt_to_serial, mqtt_publish_topic
    
    try:
        # 更新转发标志
        if 'serial_to_mqtt' in data:
            forward_serial_to_mqtt = bool(data['serial_to_mqtt'])
        if 'mqtt_to_serial' in data:
            forward_mqtt_to_serial = bool(data['mqtt_to_serial'])
        if 'publish_topic' in data:
            new_topic = str(data['publish_topic'])
            if not new_topic.strip():
                raise ValueError("发布主题不能为空")
            mqtt_publish_topic = new_topic
        
        # 广播配置更新
        socketio.emit('forward_status', {
            'serial_to_mqtt': forward_serial_to_mqtt,
            'mqtt_to_serial': forward_mqtt_to_serial,
            'publish_topic': mqtt_publish_topic
        }, namespace=SOCKETIO_NAMESPACE_STATUS)
        
        emit('update_forward_config_response', {
            'success': True,
            'message': '转发配置已更新',
            'error_code': ERROR_CODES['SUCCESS']
        })
        
        logger.info(f"转发配置已更新 - 串口到MQTT: {forward_serial_to_mqtt}, MQTT到串口: {forward_mqtt_to_serial}, 发布主题: {mqtt_publish_topic}")
    except ValueError as e:
        error_msg = str(e)
        logger.warning(f"转发配置更新失败: {error_msg}")
        emit('update_forward_config_response', {
            'success': False,
            'message': error_msg,
            'error_code': ERROR_CODES['CONFIG_ERROR']
        })
    except Exception as e:
        error_msg = f'更新转发配置失败: {str(e)}'
        logger.error(error_msg)
        emit('update_forward_config_response', {
            'success': False,
            'message': error_msg,
            'error_code': ERROR_CODES['UNKNOWN_ERROR']
        })

@socketio.on('clear_data_buffer', namespace=SOCKETIO_NAMESPACE_CONTROL)
def handle_clear_data_buffer(data):
    """通过WebSocket清空数据缓冲区"""
    try:
        buffer_type = data.get('type', '')
        
        if buffer_type == 'serial':
            serial_data_buffer.clear()
            emit('clear_data_buffer_response', {
                'success': True,
                'message': '串口数据缓冲区已清空',
                'error_code': ERROR_CODES['SUCCESS']
            })
            logger.info("串口数据缓冲区已清空")
        elif buffer_type == 'mqtt':
            mqtt_data_buffer.clear()
            emit('clear_data_buffer_response', {
                'success': True,
                'message': 'MQTT数据缓冲区已清空',
                'error_code': ERROR_CODES['SUCCESS']
            })
            logger.info("MQTT数据缓冲区已清空")
        elif buffer_type == 'all':
            serial_data_buffer.clear()
            mqtt_data_buffer.clear()
            emit('clear_data_buffer_response', {
                'success': True,
                'message': '所有数据缓冲区已清空',
                'error_code': ERROR_CODES['SUCCESS']
            })
            logger.info("所有数据缓冲区已清空")
        else:
            emit('clear_data_buffer_response', {
                'success': False,
                'message': '无效的缓冲区类型',
                'error_code': ERROR_CODES['CONFIG_ERROR']
            })
            logger.warning(f"清空缓冲区失败: 无效的缓冲区类型 '{buffer_type}'")
    except Exception as e:
        error_msg = f'清空缓冲区失败: {str(e)}'
        logger.error(error_msg)
        emit('clear_data_buffer_response', {
            'success': False,
            'message': error_msg,
            'error_code': ERROR_CODES['UNKNOWN_ERROR']
        })

# 设置回调
serial_client.set_data_callback(serial_data_handler)
serial_client.set_send_callback(serial_send_handler)
serial_client.set_status_callback(serial_status_handler)
mqtt_client.set_data_callback(mqtt_data_handler_extended)
mqtt_client.set_status_callback(mqtt_status_handler)

# 设备配置文件路径
DEVICE_CONFIG_FILE = '/root/dzxj_dtu/devices.json'
confirm_loop_running = False


def save_device_config():
    """保存设备配置到文件"""
    try:
        filepath = DEVICE_CONFIG_FILE
        success = address_config.save_config(filepath)
        if success:
            logger.info(f"设备配置已保存到: {filepath}")
    except Exception as e:
        logger.error(f"保存设备配置失败: {str(e)}")


def load_device_config():
    """加载设备配置"""
    try:
        filepath = DEVICE_CONFIG_FILE
        if os.path.exists(filepath):
            success = address_config.load_config(filepath)
            if success:
                devices = address_config.get_stored_devices()
                logger.info(f"已加载 {len(devices)} 个设备配置")
                return devices
    except Exception as e:
        logger.error(f"加载设备配置失败: {str(e)}")
    return {}


def confirm_loop():
    """后台确认线程：每10秒对所有已存储设备发送 confirm_address"""
    global confirm_loop_running
    confirm_loop_running = True
    logger.info("启动设备确认线程 (间隔10秒)")

    while confirm_loop_running:
        try:
            devices = address_config.get_stored_devices()
            if not devices:
                time.sleep(10)
                continue

            if not serial_client.get_status():
                time.sleep(10)
                continue

            _st = serial_client.get_status()
            if not (isinstance(_st, dict) and _st.get('connected', False)):
                time.sleep(10)
                continue

            for uid_hex, addr in devices.items():
                try:
                    uid_bytes = bytes.fromhex(uid_hex)
                    cmd = build_confirm_address(addr, uid_bytes)
                    success, msg = serial_client.send_raw(cmd)
                    if success:
                        logger.debug(f"确认设备: 地址={addr}, UID={uid_hex[:16]}...")
                    else:
                        logger.warning(f"确认失败 地址={addr}: {msg}")
                except Exception as e:
                    logger.error(f"确认异常 地址={addr}: {str(e)}")
                time.sleep(0.1)

        except Exception as e:
            logger.error(f"确认线程异常: {str(e)}")

        time.sleep(10)


# API路由
# 移除静态文件服务，前端由nginx提供服务

# 根路径路由
@app.route('/')
def index():
    return send_from_directory(STATIC_FOLDER, 'index.html')

# 404错误处理器 - 解决SPA路由刷新问题
@app.errorhandler(404)
def page_not_found(e):
    """处理所有404错误，对于非API路径返回index.html"""
    path = request.path
    # 检查是否是API请求
    if path.startswith('/api/'):
        # 对于API请求，返回404错误
        return jsonify({
            'success': False,
            'message': 'API endpoint not found'
        }), 404
    # 对于所有非API路径，返回index.html让前端路由处理
    return send_from_directory(STATIC_FOLDER, 'index.html'), 200

@app.route('/api/serial/ports', methods=['GET'])
def get_serial_ports():
    """获取可用串口列表"""
    ports = serial_client.list_ports()
    return jsonify({
        'success': True,
        'ports': ports
    })

@app.route('/api/serial/connect', methods=['POST'])
def serial_connect():
    """连接串口"""
    try:
        data = request.json
        port = data.get('port')
        baudrate = data.get('baudrate', 9600)
        bytesize = data.get('bytesize', 8)
        parity = data.get('parity', 'N')
        stopbits = data.get('stopbits', 1)
        timeout = data.get('timeout', 0.1)
        
        if not port:
            logger.warning("连接串口请求缺少串口名称")
            return jsonify({
                'success': False, 
                'message': '串口名称不能为空',
                'error_code': ERROR_CODES['CONFIG_ERROR']
            }), 400
        
        # 先断开之前的连接
        _status = serial_client.get_status()
        if isinstance(_status, dict) and _status.get('connected', False):
            _port = serial_client.current_config.port if serial_client.current_config else 'unknown'
            logger.info(f"断开现有串口连接: {_port}")
            serial_client.disconnect()
        
        # 连接新的串口
        logger.info(f"尝试连接串口: {port}, 波特率: {baudrate}")
        success, message = serial_client.connect(port, baudrate=baudrate, timeout=timeout, bytesize=bytesize, parity=parity, stopbits=stopbits)
        
        status_code = 200 if success else 400
        error_code = ERROR_CODES['SUCCESS'] if success else ERROR_CODES['SERIAL_CONNECTION_ERROR']
        
        response = {
            'success': success, 
            'message': message,
            'error_code': error_code
        }

        if success:
            logger.info(f"串口连接成功: {port}")
            # 保存串口配置
            save_serial_config(port, baudrate, timeout, bytesize, parity, stopbits)
        else:
            logger.error(f"串口连接失败: {message}")

        return jsonify(response), status_code
    except Exception as e:
        error_msg = f'连接串口时出错: {str(e)}'
        logger.exception(error_msg)  # 使用exception记录完整堆栈
        return jsonify({
            'success': False, 
            'message': error_msg,
            'error_code': ERROR_CODES['UNKNOWN_ERROR']
        }), 500

@app.route('/api/serial/disconnect', methods=['POST'])
def serial_disconnect():
    """断开串口连接"""
    success, message = serial_client.disconnect()
    return jsonify({
        'success': success,
        'message': message
    })

@app.route('/api/serial/status', methods=['GET'])
def serial_get_status():
    """获取串口状态"""
    _st = serial_client.get_status()
    saved_config = {}
    try:
        with open(SERIAL_CONFIG_FILE, 'r') as f:
            saved_config = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        pass
    result = {'saved_config': saved_config}
    if isinstance(_st, dict):
        result['connected'] = _st.get('connected', False)
        if result['connected']:
            cfg = _st.get('config')
            result['port'] = cfg.port if cfg else None
        else:
            result['port'] = None
    else:
        result['connected'] = bool(_st)
        result['port'] = None
    return jsonify(result)

@app.route('/api/serial/send', methods=['POST'])
def serial_send():
    """发送数据到串口"""
    data = request.json
    message = data.get('message')
    
    if not message:
        return jsonify({
            'success': False,
            'message': '消息内容不能为空'
        })
    
    success, message = serial_client.send_data(message)
    return jsonify({
        'success': success,
        'message': message
    })

@app.route('/api/mqtt/connect', methods=['POST'])
def mqtt_connect():
    """连接MQTT服务器"""
    try:
        data = request.json
        host = data.get('broker') or data.get('host', 'localhost')
        port = data.get('port', 1883)
        client_id = data.get('client_id', f'serial_gateway_{int(time.time())}')
        username = data.get('username')
        password = data.get('password')
        keepalive = data.get('keepalive', 60)
        
        # 先断开之前的连接
        if mqtt_client.get_status():
            logger.info(f"断开现有MQTT连接: {mqtt_client.host}:{mqtt_client.port}")
            mqtt_client.disconnect()
        
        # 连接新的MQTT服务器
        logger.info(f"尝试连接MQTT服务器: {host}:{port}, 客户端ID: {client_id}")
        will_topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'status')
        will_payload = json.dumps({
            'dtu_id': dtu_config['dtu_id'], 'type': 'STATUS',
            'payload': {'online': False, 'reason': 'CONNECTION_LOST'}
        })
        success, message = mqtt_client.connect(
            host=host, port=port, client_id=client_id,
            username=username, password=password,
            keepalive=keepalive,
            will_topic=will_topic, will_payload=will_payload, will_qos=1, will_retain=False
        )
        
        # 如果连接成功，订阅主题
        if success and 'topics' in data:
            mqtt_client.subscribe(data['topics'])
        
        status_code = 200 if success else 400
        error_code = ERROR_CODES['SUCCESS'] if success else ERROR_CODES['MQTT_CONNECTION_ERROR']
        
        response = {
            'success': success, 
            'message': message,
            'error_code': error_code
        }
        
        if success:
            logger.info(f"MQTT服务器连接成功: {host}:{port}")
        else:
            logger.error(f"MQTT服务器连接失败: {message}")
        
        return jsonify(response), status_code
    except Exception as e:
        error_msg = f'连接MQTT服务器时出错: {str(e)}'
        logger.exception(error_msg)
        return jsonify({
            'success': False, 
            'message': error_msg,
            'error_code': ERROR_CODES['UNKNOWN_ERROR']
        }), 500

@app.route('/api/mqtt/disconnect', methods=['POST'])
def mqtt_disconnect():
    """断开MQTT连接"""
    success, message = mqtt_client.disconnect()
    return jsonify({
        'success': success,
        'message': message
    })

@app.route('/api/mqtt/status', methods=['GET'])
def mqtt_get_status():
    """获取MQTT状态"""
    _st = mqtt_client.get_status()
    return jsonify({
        'connected': _st.get('connected', False) if isinstance(_st, dict) else bool(_st)
    })

@app.route('/api/mqtt/publish', methods=['POST'])
def mqtt_publish():
    """发布MQTT消息"""
    data = request.json
    topic = data.get('topic')
    message = data.get('message')
    
    if not topic or not message:
        return jsonify({
            'success': False,
            'message': '主题和消息内容不能为空'
        })
    
    success, message = mqtt_client.publish(topic, message)
    return jsonify({
        'success': success,
        'message': message
    })

@app.route('/api/mqtt/subscribe', methods=['POST'])
def mqtt_subscribe():
    """订阅MQTT主题"""
    data = request.json
    topics = data.get('topics', [])

    if not topics:
        return jsonify({
            'success': False,
            'message': '请至少订阅一个主题'
        })

    success, message = mqtt_client.subscribe(topics)
    return jsonify({
        'success': success,
        'message': message
    })


@app.route('/api/mqtt/broadcast_discover', methods=['POST'])
def mqtt_broadcast_discover():
    """发送DTU发现广播 (7.14)

    服务器通过广播主题发送DISCOVER消息，
    触发所有在线DTU进行注册响应
    """
    data = request.json or {}
    customer_id = data.get('customer_id', dtu_config.get('customer_id', 'default'))

    discover_topic = build_dtu_topic('broadcast', 'dtu', 'discover')
    request_id = f"dsc_{int(time.time() * 1000)}"
    payload = {
        'msg_id': request_id,
        'timestamp': int(time.time() * 1000),
        'dtu_id': 'PLATFORM',
        'type': 'DISCOVER',
        'payload': {
            'request_id': request_id,
            'customer_id': customer_id
        }
    }

    success, message = mqtt_client.publish(discover_topic, json.dumps(payload), qos=1)
    logger.info(f"发送DTU发现广播: topic={discover_topic}, payload={payload}")

    return jsonify({
        'success': success,
        'message': 'DTU发现广播已发送' if success else message,
        'topic': discover_topic
    })


@app.route('/api/mqtt/broadcast_config', methods=['POST'])
def mqtt_broadcast_config():
    """发送DTU批量配置广播 (7.15)

    服务器通过广播主题发送CONFIG消息，
    用于批量配置所有在线DTU
    """
    data = request.json or {}
    customer_id = data.get('customer_id', dtu_config.get('customer_id', 'default'))
    config_items = data.get('items', {})
    config_version = data.get('config_version', f"cfg_{int(time.time())}")
    force_apply = data.get('force_apply', False)

    config_topic = build_dtu_topic('broadcast', 'dtu', 'config')
    payload = {
        'msg_id': f"cfg_{int(time.time() * 1000)}",
        'timestamp': int(time.time() * 1000),
        'dtu_id': 'PLATFORM',
        'type': 'CONFIG',
        'payload': {
            'customer_id': customer_id,
            'config_version': config_version,
            'force_apply': force_apply,
            'items': config_items
        }
    }

    success, message = mqtt_client.publish(config_topic, json.dumps(payload), qos=1)
    logger.info(f"发送DTU批量配置广播: topic={config_topic}, payload={payload}")

    return jsonify({
        'success': success,
        'message': 'DTU批量配置广播已发送' if success else message,
        'topic': config_topic,
        'config': {
            'config_version': config_version,
            'items': config_items,
            'force_apply': force_apply
        }
    })


@app.route('/api/data/serial', methods=['GET'])
def get_serial_data():
    """获取串口数据"""
    return jsonify({
        'data': serial_data_buffer
    })

@app.route('/api/data/mqtt', methods=['GET'])
def get_mqtt_data():
    """获取MQTT数据"""
    return jsonify({
        'data': mqtt_data_buffer
    })

@app.route('/api/forward/config', methods=['POST'])
def set_forward_config():
    """设置转发配置"""
    global forward_serial_to_mqtt, forward_mqtt_to_serial, mqtt_publish_topic
    
    data = request.json
    forward_serial_to_mqtt = data.get('serial_to_mqtt', False)
    forward_mqtt_to_serial = data.get('mqtt_to_serial', False)
    if 'publish_topic' in data:
        mqtt_publish_topic = data['publish_topic']
    
    return jsonify({
        'success': True,
        'message': '转发配置已更新',
        'config': {
            'serial_to_mqtt': forward_serial_to_mqtt,
            'mqtt_to_serial': forward_mqtt_to_serial,
            'publish_topic': mqtt_publish_topic
        }
    })

@app.route('/api/forward/status', methods=['GET'])
def get_forward_status():
    """获取转发状态"""
    return jsonify({
        'serial_to_mqtt': forward_serial_to_mqtt,
        'mqtt_to_serial': forward_mqtt_to_serial,
        'publish_topic': mqtt_publish_topic
    })

# 健康检查端点
@app.route('/api/health', methods=['GET'])
def health_check():
    """健康检查端点"""
    try:
        # 获取客户端连接数量
        client_counts = {
            'data': len(connected_clients['data']),
            'status': len(connected_clients['status']),
            'control': len(connected_clients['control'])
        }
        
        # 获取缓冲区大小
        buffer_sizes = {
            'serial': len(serial_data_buffer),
            'mqtt': len(mqtt_data_buffer)
        }
        
        # 执行系统负载检查
        # 注意：这只是一个简化的负载检查，实际应用中可能需要更复杂的监控
        is_healthy = True
        load_warnings = []
        
        # 检查缓冲区是否过大
        if buffer_sizes['serial'] > MAX_BUFFER_SIZE * 0.8:
            is_healthy = False
            load_warnings.append(f"串口缓冲区接近最大容量: {buffer_sizes['serial']}/{MAX_BUFFER_SIZE}")
            
        if buffer_sizes['mqtt'] > MAX_BUFFER_SIZE * 0.8:
            is_healthy = False
            load_warnings.append(f"MQTT缓冲区接近最大容量: {buffer_sizes['mqtt']}/{MAX_BUFFER_SIZE}")
        
        # 检查WebSocket连接数是否过多
        total_clients = sum(client_counts.values())
        if total_clients > 100:  # 设置合理的阈值
            is_healthy = False
            load_warnings.append(f"WebSocket连接数过多: {total_clients}")
        
        # 尝试获取网络状态信息（使用try-except包装，防止网络模块出错导致健康检查失败）
        network_info = None
        try:
            network_info = network_manager.get_network_status()
        except Exception as e:
            logger.warning(f"获取网络状态时出错: {str(e)}")
            # 不影响整体健康检查，只添加警告
            load_warnings.append(f"网络状态获取失败: {str(e)}")
        
        response = {
            'status': 'healthy' if is_healthy else 'warning',
            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
            'services': {
                'serial': serial_status,
                'mqtt': mqtt_status,
                'websocket': True
            },
            'client_counts': client_counts,
            'buffer_sizes': buffer_sizes,
            'warnings': load_warnings
        }
        
        # 仅当获取到网络信息时添加
        if network_info:
            response['network'] = network_info
        
        return jsonify(response), 200
    except Exception as e:
        # 捕获所有异常，确保健康检查不会返回500错误
        logger.error(f"健康检查端点出错: {str(e)}")
        # 返回一个基础的健康状态，至少显示服务在运行
        return jsonify({
            'status': 'error',
            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
            'error': str(e),
            'services': {
                'api': True,  # API服务本身是运行的
                'serial': None,
                'mqtt': None,
                'websocket': None
            }
        }), 200  # 仍然返回200，避免健康检查导致的连锁反应


@app.route('/api/screen/status', methods=['GET'])
def get_screen_status():
    """获取 Nextion 屏幕状态。"""
    with screen_lock:
        panels = get_sorted_panels()
        current_index = screen_current_panel_index
        current_panel_id = None
        if panels and 0 <= current_index < len(panels):
            current_panel_id = panels[current_index][0]
        panel_count = len(panels)
    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': current_index,
            'current_panel_id': current_panel_id,
            'panel_count': panel_count
        }
    })


@app.route('/api/screen/refresh', methods=['POST'])
def refresh_screen():
    """手动刷新当前 panel 到屏幕。"""
    with screen_lock:
        panels = get_sorted_panels()
        if not panels:
            return jsonify({'success': False, 'message': '没有可用的 panel'}), 400
        if not (0 <= screen_current_panel_index < len(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


# 网络配置相关API
@app.route('/api/network/config', methods=['GET'])
def get_network_config():
    """获取当前网络配置"""
    try:
        config = network_manager.get_network_config()
        return jsonify(config)
    except Exception as e:
        logger.error(f'获取网络配置失败: {str(e)}')
        return jsonify({'success': False, 'message': f'获取网络配置失败: {str(e)}'}), 500

@app.route('/api/network/config', methods=['POST'])
def update_network_config():
    """更新网络配置"""
    try:
        config_data = request.json
        result = network_manager.update_network_config(config_data)
        if result['success']:
            return jsonify(result)
        else:
            return jsonify(result), 400
    except Exception as e:
        logger.error(f'更新网络配置失败: {str(e)}')
        return jsonify({'success': False, 'message': f'更新网络配置失败: {str(e)}'}), 500

@app.route('/api/network/status', methods=['GET'])
def get_network_status():
    """获取网络状态信息"""
    try:
        status = network_manager.get_network_status()
        return jsonify(status)
    except Exception as e:
        logger.error(f'获取网络状态失败: {str(e)}')
        return jsonify({'success': False, 'message': f'获取网络状态失败: {str(e)}'}), 500

@app.route('/api/network/restart', methods=['POST'])
def restart_network_service():
    """重启网络服务以应用新配置"""
    try:
        result = network_manager.restart_network_service()
        if result['success']:
            return jsonify(result)
        else:
            return jsonify(result), 500
    except Exception as e:
        logger.error(f'重启网络服务失败: {str(e)}')
        return jsonify({'success': False, 'message': f'重启网络服务失败: {str(e)}'}), 500

# Modbus RTU API
@app.route('/api/modbus/antenna_addresses', methods=['GET'])
def get_antenna_addresses():
    """获取天线地址映射表"""
    return jsonify({
        'success': True,
        'antennas': {str(k): f"0x{v:04x}" for k, v in ANTENNA_ADDRESSES.items()}
    })


@app.route('/api/modbus/read_antenna', methods=['POST'])
def modbus_read_antenna():
    """读取指定天线的卡号

    请求参数:
    {
        "device_address": 1,      // 设备地址 (1-247)
        "antenna": 1,             // 天线编号 (1-24)
        "timeout": 1.0            // 可选，超时时间(秒)
    }
    """
    try:
        data = request.json
        device_address = data.get('device_address', 1)
        antenna = data.get('antenna', 1)
        timeout = data.get('timeout')

        if antenna < 1 or antenna > 24:
            return jsonify({
                'success': False,
                'message': f'无效的天线编号: {antenna}, 必须是1-24'
            }), 400

        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            return jsonify({
                'success': False,
                'message': '串口未连接'
            }), 400

        result = modbus_client.read_antenna_card(
            device_address=device_address,
            antenna_num=antenna,
            timeout=timeout
        )

        if 'error' in result:
            return jsonify({
                'success': False,
                'message': result['error'],
                'raw_data': result.get('raw_data', '')
            }), 400

        return jsonify({
            'success': True,
            'data': result
        })

    except Exception as e:
        logger.error(f'读取天线数据失败: {str(e)}')
        return jsonify({
            'success': False,
            'message': str(e)
        }), 500


@app.route('/api/modbus/read_registers', methods=['POST'])
def modbus_read_registers():
    """读取保持寄存器

    请求参数:
    {
        "device_address": 1,
        "start_address": 2,       // 起始地址 (十六进制如0x0002或十进制如2)
        "quantity": 4,            // 寄存器数量
        "timeout": 1.0
    }
    """
    try:
        data = request.json
        device_address = data.get('device_address', 1)
        start_address = data.get('start_address', 0)
        quantity = data.get('quantity', 1)
        timeout = data.get('timeout')

        # 支持十六进制字符串
        if isinstance(start_address, str):
            start_address = int(start_address, 16)
        if isinstance(device_address, str):
            device_address = int(device_address, 16)

        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            return jsonify({
                'success': False,
                'message': '串口未连接'
            }), 400

        result = modbus_client.read_holding_registers(
            device_address=device_address,
            start_address=start_address,
            quantity=quantity,
            timeout=timeout
        )

        if 'error' in result:
            return jsonify({
                'success': False,
                'message': result['error'],
                'raw_data': result.get('raw_data', '')
            }), 400

        return jsonify({
            'success': True,
            'data': result
        })

    except Exception as e:
        logger.error(f'读取寄存器失败: {str(e)}')
        return jsonify({
            'success': False,
            'message': str(e)
        }), 500


@app.route('/api/modbus/write_register', methods=['POST'])
def modbus_write_register():
    """写单个寄存器

    请求参数:
    {
        "device_address": 1,
        "register_address": 1,    // 寄存器地址
        "value": 256,             // 写入的值
        "timeout": 1.0
    }
    """
    try:
        data = request.json
        device_address = data.get('device_address', 1)
        register_address = data.get('register_address', 1)
        value = data.get('value', 0)
        timeout = data.get('timeout')

        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            return jsonify({
                'success': False,
                'message':'串口未连接'
            }), 400

        result = modbus_client.write_single_register(
            device_address=device_address,
            register_address=register_address,
            value=value,
            timeout=timeout
        )

        if 'error' in result:
            return jsonify({
                'success': False,
                'message': result['error'],
                'raw_data': result.get('raw_data', '')
            }), 400

        return jsonify({
            'success': True,
            'data': result
        })

    except Exception as e:
        logger.error(f'写寄存器失败: {str(e)}')
        return jsonify({
            'success': False,
            'message': str(e)
        }), 500


@app.route('/api/modbus/set_rgb_led', methods=['POST'])
def modbus_set_rgb_led():
    """设置RGB灯状态

    请求参数:
    {
        "device_address": 1,
        "led_number": 1,          // 灯编号 (1-24)
        "color": 1,               // 颜色: 0=灭, 1=红灯, 2=绿灯, 3=蓝灯
        "timeout": 1.0
    }
    """
    try:
        data = request.json
        device_address = data.get('device_address', 1)
        led_number = data.get('led_number', 1)
        color = data.get('color', 0)
        timeout = data.get('timeout')

        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            return jsonify({
                'success': False,
                'message': '串口未连接'
            }), 400

        result = modbus_client.set_rgb_led(
            device_address=device_address,
            led_number=led_number,
            color=color,
            timeout=timeout
        )

        if 'error' in result:
            return jsonify({
                'success': False,
                'message': result['error'],
                'raw_data': result.get('raw_data', '')
            }), 400

        # 更新 LED 状态追踪
        led_states[led_number] = color

        return jsonify({
            'success': True,
            'data': result
        })

    except Exception as e:
        logger.error(f'设置RGB灯失败: {str(e)}')
        return jsonify({
            'success': False,
            'message': str(e)
        }), 500


@app.route('/api/modbus/scan', methods=['POST'])
def modbus_scan_devices():
    """扫描在线设备

    请求参数:
    {
        "max_address": 247,       // 最大设备地址
        "timeout": 0.2            // 单个设备超时时间
    }
    """
    try:
        data = request.json or {}
        max_address = data.get('max_address', 247)
        timeout = data.get('timeout', 0.2)

        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            return jsonify({
                'success': False,
                'message': '串口未连接'
            }), 400

        # 异步扫描可能更好，但这里先同步实现
        result = modbus_client.scan_devices(max_address)

        return jsonify({
            'success': True,
            'devices': result
        })

    except Exception as e:
        logger.error(f'扫描设备失败: {str(e)}')
        return jsonify({
            'success': False,
            'message': str(e)
        }), 500


# ========== 地址配置协议 API ==========

@app.route('/api/modbus/broadcast_query', methods=['POST'])
def modbus_broadcast_query():
    """发送广播查询指令"""
    try:
        data = request.json or {}
        timeout = data.get('timeout')

        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            return jsonify({'success': False, 'message': '串口未连接'}), 400

        responses = address_config.broadcast_query(timeout)
        assign_results = []
        if responses:
            assign_results = address_config.process_responses(responses)
            for r in assign_results:
                uid_bytes = r.get('uid', r.get('request', ''))
            save_device_config()
        return jsonify({
            'success': True,
            'responses': responses,
            'assign_results': assign_results,
            'count': len(responses)
        })

    except Exception as e:
        logger.error(f'广播查询失败: {str(e)}')
        return jsonify({'success': False, 'message': str(e)}), 500


@app.route('/api/modbus/auto_configure', methods=['POST'])
def modbus_auto_configure():
    """自动配置设备地址"""
    try:
        data = request.json or {}
        timeout = data.get('timeout')

        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            return jsonify({'success': False, 'message': '串口未连接'}), 400

        result = address_config.auto_configure(timeout)
        if result.get('success') and (result.get('discovered', 0) > 0 or result.get('confirmed', 0) > 0 or result.get('assigned', 0) > 0):
            save_device_config()
        return jsonify(result)

    except Exception as e:
        logger.error(f'自动配置失败: {str(e)}')
        return jsonify({'success': False, 'message': str(e)}), 500


@app.route('/api/modbus/stored_devices', methods=['GET'])
def get_stored_devices():
    """获取已存储的设备列表"""
    return jsonify({'success': True, 'devices': address_config.get_stored_devices()})


@app.route('/api/modbus/stored_devices', methods=['POST'])
def add_stored_device():
    """添加已存储的设备"""
    try:
        data = request.json
        uid = data.get('uid', '').lower()
        address = data.get('address', 1)

        if len(uid) != 24:
            return jsonify({'success': False, 'message': 'UID长度必须是24个十六进制字符'}), 400

        address_config.add_stored_device(uid, address)
        save_device_config()
        return jsonify({'success': True, 'message': f'已添加设备: UID={uid}, 地址={address}'})

    except Exception as e:
        return jsonify({'success': False, 'message': str(e)}), 500


@app.route('/api/modbus/load_config', methods=['POST'])
def modbus_load_config():
    """从文件加载设备配置"""
    try:
        data = request.json
        filepath = data.get('filepath', '/tmp/modbus_devices.json')
        success = address_config.load_config(filepath)
        return jsonify({'success': success, 'devices': address_config.get_stored_devices()})
    except Exception as e:
        return jsonify({'success': False, 'message': str(e)}), 500


@app.route('/api/modbus/save_config', methods=['POST'])
def modbus_save_config():
    """保存设备配置到文件"""
    try:
        data = request.json
        filepath = data.get('filepath', '/tmp/modbus_devices.json')
        success = address_config.save_config(filepath)
        return jsonify({'success': success, 'message': f'配置已保存到: {filepath}' if success else '保存失败'})
    except Exception as e:
        return jsonify({'success': False, 'message': str(e)}), 500


@app.route('/api/modbus/confirm_devices', methods=['POST'])
def confirm_devices():
    """手动触发对所有已存储设备的确认"""
    try:
        devices = address_config.get_stored_devices()
        if not devices:
            return jsonify({'success': False, 'message': '没有已存储的设备'}), 400

        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            return jsonify({'success': False, 'message': '串口未连接'}), 400

        results = []
        for uid_hex, addr in devices.items():
            try:
                uid_bytes = bytes.fromhex(uid_hex)
                cmd = build_confirm_address(addr, uid_bytes)
                success, msg = serial_client.send_raw(cmd)
                logger.info(f"确认设备 地址={addr}: {'成功' if success else '失败 ' + msg}")
                results.append({'address': addr, 'uid': uid_hex, 'success': success, 'message': msg})
            except Exception as e:
                results.append({'address': addr, 'uid': uid_hex, 'success': False, 'message': str(e)})
            time.sleep(0.1)

        return jsonify({'success': True, 'results': results, 'count': len(results)})
    except Exception as e:
        return jsonify({'success': False, 'message': str(e)}), 500


# 端口状态追踪 - 用于存储每个端口的事件状态
port_event_history = []  # 存储最近的端口事件


@app.route('/api/port/status', methods=['GET'])
def get_port_status():
    """获取所有端口状态"""
    return jsonify({'success': True, 'port_state': port_state})


@app.route('/api/port/events', methods=['GET'])
def get_port_events():
    """获取端口事件历史"""
    limit = request.args.get('limit', 100, type=int)
    return jsonify({
        'success': True,
        'events': port_event_history[-limit:]
    })


@app.route('/api/port/clear_events', methods=['POST'])
def clear_port_events():
    """清除端口事件历史"""
    global port_event_history
    port_event_history = []
    return jsonify({'success': True, 'message': '事件历史已清除'})


# 设备最后响应时间跟踪
device_last_seen = {}

@app.route('/api/panel/status', methods=['GET'])
def get_panel_status():
    """获取所有面板状态"""
    now = time.time()
    PANEL_OFFLINE_TIMEOUT = 60

    panel_status = {}
    for panel_id, ports in port_state.items():
        port_count = len(ports)
        alarm_count = sum(1 for p in ports.values() if p.get('alarm_count', 0) > 0)
        connected_count = sum(1 for p in ports.values() if p.get('last_uid'))
        panel_status[panel_id] = {
            'panel_id': panel_id,
            'port_count': port_count,
            'connected_count': connected_count,
            'alarm_count': alarm_count,
            'status': 'online' if connected_count > 0 else 'offline'
        }
    for panel_id, cfg in panel_config.items():
        last_seen = device_last_seen.get(panel_id, 0)
        is_online = (now - last_seen) < PANEL_OFFLINE_TIMEOUT
        if panel_id in panel_status:
            panel_status[panel_id].update({
                'address': cfg.get('address'),
                'position': cfg.get('position'),
                'panel_uid': cfg.get('panel_uid'),
                'status': 'online' if is_online else 'offline'
            })
        else:
            panel_status[panel_id] = {
                'panel_id': panel_id,
                'address': cfg.get('address'),
                'position': cfg.get('position'),
                'panel_uid': cfg.get('panel_uid'),
                'port_count': 0,
                'connected_count': 0,
                'alarm_count': 0,
                'status': 'online' if is_online else 'offline'
            }
    return jsonify({'success': True, 'panels': panel_status})


# LED 状态追踪 (内存中跟踪每个灯的最后设置状态)
led_states = {i: 0 for i in range(1, 25)}  # 0=off, 1=red, 2=green, 3=blue


@app.route('/api/modbus/led_status', methods=['GET'])
def get_led_status():
    """获取所有 LED 状态"""
    return jsonify({'success': True, 'leds': led_states})


@app.route('/api/modbus/set_all_leds', methods=['POST'])
def set_all_leds():
    """批量设置 LED
    请求: {"device_address": 1, "color": 1}
    设置所有 24 个 LED 到指定颜色
    """
    try:
        data = request.json
        device_address = data.get('device_address', 1)
        color = data.get('color', 0)
        timeout = data.get('timeout')

        if not (isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False)):
            return jsonify({'success': False, 'message': '串口未连接'}), 400

        results = []
        for led_num in range(1, 25):
            result = modbus_client.set_rgb_led(
                device_address=device_address,
                led_number=led_num,
                color=color,
                timeout=timeout
            )
            if 'error' in result:
                results.append({'led': led_num, 'success': False, 'error': result['error']})
            else:
                results.append({'led': led_num, 'success': True})
                led_states[led_num] = color

        return jsonify({'success': True, 'results': results})

    except Exception as e:
        logger.error(f'批量设置LED失败: {str(e)}')
        return jsonify({'success': False, 'message': str(e)}), 500


# 在 set_rgb_led 中更新 LED 状态追踪
# 不再需要静态文件目录，前端由nginx提供服务


# ========== DTU MQTT协议配置API ==========

@app.route('/api/dtu/config', methods=['GET'])
def get_dtu_config():
    """获取DTU配置"""
    return jsonify({
        'success': True,
        'data': dtu_config
    })


@app.route('/api/dtu/config', methods=['POST'])
def update_dtu_config():
    """更新DTU配置"""
    try:
        data = request.json
        old_config = dtu_config.copy()

        # 更新配置项
        if 'topic_prefix' in data:
            dtu_config['topic_prefix'] = data['topic_prefix']
        if 'customer_id' in data:
            dtu_config['customer_id'] = data['customer_id']
        if 'dtu_id' in data:
            dtu_config['dtu_id'] = data['dtu_id']
        if 'firmware_version' in data:
            dtu_config['firmware_version'] = data['firmware_version']
        if 'hardware_version' in data:
            dtu_config['hardware_version'] = data['hardware_version']
        if 'heartbeat_interval' in data:
            dtu_config['heartbeat_interval'] = data['heartbeat_interval']
        if 'enabled' in data:
            dtu_config['enabled'] = data['enabled']

        # 如果MQTT已连接且主题配置发生变化，重新订阅
        if mqtt_status and dtu_config.get('enabled'):
            old_control_topic = build_dtu_topic(
                old_config.get('customer_id', DEFAULT_CUSTOMER_ID),
                'dtu',
                old_config.get('dtu_id', DEFAULT_DTU_ID),
                'control'
            )
            new_control_topic = build_dtu_topic(
                dtu_config['customer_id'],
                'dtu',
                dtu_config['dtu_id'],
                'control'
            )

            if old_control_topic != new_control_topic:
                # 取消旧订阅，订阅新主题
                mqtt_client.unsubscribe(old_control_topic)
                mqtt_client.subscribe(new_control_topic)
                logger.info(f"控制主题已更新: {old_control_topic} -> {new_control_topic}")

                # 重新发送注册消息
                dtu_register()

        return jsonify({
            'success': True,
            'message': 'DTU配置已更新',
            'data': dtu_config
        })
    except Exception as e:
        logger.error(f"更新DTU配置失败: {str(e)}")
        return jsonify({'success': False, 'message': str(e)}), 500


@app.route('/api/dtu/register', methods=['POST'])
def manual_dtu_register():
    """手动触发DTU注册"""
    try:
        if not mqtt_status:
            return jsonify({'success': False, 'message': 'MQTT未连接'}), 400

        success = dtu_register()
        if success:
            return jsonify({'success': True, 'message': '注册消息已发送'})
        else:
            return jsonify({'success': False, 'message': '注册消息发送失败'}), 500
    except Exception as e:
        logger.error(f"手动触发DTU注册失败: {str(e)}")
        return jsonify({'success': False, 'message': str(e)}), 500


def _get_cpu_usage():
    try:
        import psutil
        return psutil.cpu_percent(interval=None)
    except Exception:
        return None


def _get_memory_usage():
    try:
        import psutil
        return psutil.virtual_memory().percent
    except Exception:
        return None


@app.route('/api/dtu/status', methods=['GET'])
def get_dtu_status():
    """获取DTU状态"""
    try:
        # 获取串口状态
        serial_st = serial_client.get_status()

        # 获取面板状态
        devices = address_config.get_stored_devices()

        # 主板温度: 优先外部上报，否则本地读取
        dtu_temperature = env_sensor_data.get('dtu_temperature')
        if dtu_temperature is None:
            try:
                import psutil
                temps = psutil.sensors_temperatures(fahrenheit=False) or {}
                for chip_name, entries in temps.items():
                    for entry in entries:
                        if entry.current and entry.current > 0:
                            dtu_temperature = round(entry.current, 1)
                            break
                    if dtu_temperature is not None:
                        break
            except Exception:
                pass
            if dtu_temperature is None:
                import glob as _glob
                for tz in sorted(_glob.glob('/sys/class/thermal/thermal_zone*/temp')):
                    try:
                        with open(tz) as f:
                            v = int(f.read().strip())
                            if v > 0:
                                dtu_temperature = round(v / 1000.0, 1)
                                break
                    except Exception:
                        continue

        status = {
            'dtu_id': dtu_config.get('dtu_id'),
            'mqtt_connected': mqtt_status,
            'serial_connected': isinstance(serial_st, dict) and serial_st.get('connected', False),
            'dtu_enabled': dtu_config.get('enabled', True),
            'topic_prefix': dtu_config.get('topic_prefix'),
            'customer_id': dtu_config.get('customer_id'),
            'panel_count': len(devices),
            'cpu_usage': _get_cpu_usage(),
            'memory_usage': _get_memory_usage(),
            'temperature': dtu_temperature,
            'dtu_temperature': dtu_temperature,
            'firmware_version': dtu_config.get('firmware_version', 'v1.0.0'),
            'topics': {
                'register': build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'register'),
                'status': build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'status'),
                'control': build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'control'),
                'response': build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'response'),
                'event': build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'event'),
                'alarm': build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'alarm')
            }
        }

        return jsonify({'success': True, 'data': status})
    except Exception as e:
        logger.error(f"获取DTU状态失败: {str(e)}")
        return jsonify({'success': False, 'message': str(e)}), 500


@app.route('/api/dtu/control', methods=['POST'])
def dtu_control():
    """发送DTU控制命令（HTTP 入口，复用 dtu_handle_control 逻辑）"""
    try:
        data = request.json or {}
        # 兼容两种入参格式:
        #   1) {"command": "...", "target": "...", "params": {...}}  简洁格式
        #   2) {"payload": {"command": "...", "target": "...", "params": {...}}}  MQTT envelope 格式
        inner = data.get('payload') if isinstance(data.get('payload'), dict) else data
        command = inner.get('command')

        # REBOOT 走快速通道立即执行
        if command == 'REBOOT':
            import subprocess
            logger.warning("执行系统重启命令")
            threading.Thread(target=lambda: (
                time.sleep(1),
                subprocess.run(['reboot'], capture_output=True)
            ), daemon=True).start()
            return jsonify({'success': True, 'message': '系统正在重启...'})

        # 其他命令委托给 MQTT 处理器（构造 MQTT envelope 后调用）
        mqtt_envelope = {
            'msg_id': data.get('msg_id', f"http_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}"),
            'timestamp': data.get('timestamp', int(time.time() * 1000)),
            'dtu_id': data.get('dtu_id', dtu_config.get('dtu_id')),
            'type': 'CONTROL',
            'payload': {
                'command': command,
                'target': inner.get('target'),
                'params': inner.get('params', {})
            }
        }
        # 调用 MQTT 处理器（其内部会发送响应到 dtu/.../response 主题）
        dtu_handle_control('http/control', mqtt_envelope)
        # 如果命令是 QUERY_JUMPER_STATUS / QUERY_ENV_SENSOR / QUERY_DTU_STATUS，
        # 实际响应已发布到对应主题，HTTP 端点仅返回成功标记
        if command in ('QUERY_JUMPER_STATUS', 'QUERY_ENV_SENSOR', 'QUERY_DTU_STATUS'):
            return jsonify({
                'success': True,
                'message': f'{command} 已发布到 MQTT 主题'
            })
        # 其他命令: 从 MQTT 主题同步拉取最新响应（简化处理：直接返回已发布标记）
        return jsonify({
            'success': True,
            'message': f'{command} 命令已下发，等待 MQTT 响应'
        })
    except Exception as e:
        logger.error(f"DTU控制命令失败: {str(e)}")
        return jsonify({'success': False, 'message': str(e)}), 500


# OTA状态存储
ota_status = {
    'status': 'IDLE',  # IDLE, DOWNLOADING, VERIFYING, FLASHING, SUCCESS, FAILED
    'progress': 0,
    'firmware_version': None,
    'target_version': None,
    'error_code': None,
    'error_message': None,
    'last_update': None
}


@app.route('/api/dtu/ota_detect', methods=['POST'])
def ota_detect():
    """检测固件包信息（自动下载并解析 manifest）"""
    try:
        url = request.json.get('url', '')
        if not url:
            return jsonify({'success': False, 'message': '请提供固件 URL'}), 400

        import urllib.request, tempfile, tarfile, json, hashlib
        tmp = tempfile.mktemp(suffix='.tar.gz')
        try:
            urllib.request.urlretrieve(url, tmp)
        except Exception as e:
            return jsonify({'success': False, 'message': f'下载失败: {str(e)}'}), 400

        file_size = os.path.getsize(tmp)
        h = hashlib.md5()
        with open(tmp, 'rb') as f:
            for chunk in iter(lambda: f.read(65536), b''):
                h.update(chunk)
        md5sum = h.hexdigest()

        version = ''
        try:
            with tarfile.open(tmp, 'r:gz') as tar:
                m = tar.extractfile('firmware/firmware.json')
                if m:
                    manifest = json.loads(m.read())
                    version = manifest.get('version', '')
        except Exception as e:
            logger.warning(f"解析 firmware.json 失败: {e}")

        os.remove(tmp)
        return jsonify({
            'success': True,
            'data': {
                'file_size': file_size,
                'checksum': md5sum,
                'checksum_type': 'MD5',
                'firmware_version': version
            }
        })
    except Exception as e:
        logger.error(f"检测固件失败: {str(e)}")
        return jsonify({'success': False, 'message': str(e)}), 500


@app.route('/api/dtu/ota_status', methods=['GET'])
def get_ota_status():
    """获取OTA升级状态"""
    try:
        return jsonify({
            'success': True,
            'data': {
                'current_firmware': dtu_config.get('firmware_version', 'v1.0.0'),
                'ota_status': ota_status.get('status', 'IDLE'),
                'ota_progress': ota_status.get('progress', 0),
                'target_version': ota_status.get('target_version'),
                'error_code': ota_status.get('error_code'),
                'error_message': ota_status.get('error_message'),
                'last_update': ota_status.get('last_update')
            }
        })
    except Exception as e:
        logger.error(f"获取OTA状态失败: {str(e)}")
        return jsonify({'success': False, 'message': str(e)}), 500


def handle_ota_status(dtu_id, payload):
    """处理MQTT上报的OTA状态"""
    ota_status['status'] = payload.get('ota_status', 'IDLE')
    ota_status['progress'] = payload.get('ota_progress', 0)
    ota_status['firmware_version'] = payload.get('firmware_version')
    ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
    error_code = payload.get('error_code')
    if error_code and error_code != 0:
        ota_status['error_code'] = error_code
        ota_status['error_message'] = payload.get('error_message', get_ota_error_message(error_code))
        ota_status['status'] = 'FAILED'
    elif ota_status['status'] == 'SUCCESS':
        ota_status['error_code'] = None
        ota_status['error_message'] = None
        if ota_status.get('firmware_version'):
            dtu_config['firmware_version'] = ota_status['firmware_version']
            save_dtu_config()
    logger.info(f"MQTT OTA状态更新: {ota_status['status']}, 进度: {ota_status['progress']}%")


def get_ota_error_message(error_code):
    error_messages = {1021: '已是目标版本', 1022: '校验失败', 1023: '下载失败', 1024: '写入失败', 1025: '存储空间不足'}
    return error_messages.get(error_code, f'未知错误码: {error_code}')


def _publish_ota_progress():
    """通过MQTT发布OTA进度"""
    try:
        if not mqtt_client.get_status():
            return
        topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'status')
        payload = {
            'msg_id': f"ota_status_{int(time.time() * 1000)}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'STATUS',
            'payload': {
                'ota_status': ota_status.get('status', 'IDLE'),
                'ota_progress': ota_status.get('progress', 0),
                'firmware_version': ota_status.get('target_version', ''),
                'error_code': ota_status.get('error_code', 0),
                'error_message': ota_status.get('error_message')
            }
        }
        mqtt_client.publish(topic, json.dumps(payload), qos=1)
    except Exception:
        pass


def _send_ota_error_response(error_code, original_msg_id=''):
    """通过MQTT发送OTA错误响应（1021-1025）"""
    try:
        if not mqtt_client.get_status():
            return
        topic = build_dtu_topic(dtu_config['customer_id'], 'dtu', dtu_config['dtu_id'], 'response')
        payload = {
            'msg_id': f"rsp_ota_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}",
            'timestamp': int(time.time() * 1000),
            'dtu_id': dtu_config['dtu_id'],
            'type': 'RESPONSE',
            'payload': {
                'original_msg_id': original_msg_id,
                'command': 'OTA_UPGRADE',
                'success': False,
                'result': None,
                'error_code': error_code,
                'error_message': get_ota_error_message(error_code)
            }
        }
        mqtt_client.publish(topic, json.dumps(payload), qos=1)
    except Exception:
        pass


def _run_ota(params):
    """执行OTA升级（后台线程，供HTTP和MQTT共用）"""
    import subprocess, os, hashlib, shutil, tarfile, tempfile, urllib.request
    url = params['firmware_url']
    version = params['firmware_version']
    file_size = params['file_size']
    checksum = params['checksum']
    checksum_type = params.get('checksum_type', 'MD5').upper()
    force = params.get('force_upgrade', False)
    ota_status['status'] = 'DOWNLOADING'
    ota_status['progress'] = 0
    ota_status['target_version'] = version
    ota_status['error_code'] = None
    ota_status['error_message'] = None
    ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
    _publish_ota_progress()
    try:
        tmp_dir = tempfile.mkdtemp(prefix='ota_')
        fw_path = os.path.join(tmp_dir, 'firmware.tar.gz')
        ota_status['progress'] = 5
        ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
        logger.info(f"OTA: 开始下载固件 {url}")
        req = urllib.request.Request(url, headers={'User-Agent': 'OTA-Updater'})
        # 进度上报节流: 每 10% 或每 5 秒
        last_published_pct = 0
        last_publish_time = time.time()
        with urllib.request.urlopen(req, timeout=120) as resp:
            with open(fw_path, 'wb') as f:
                total = int(resp.headers.get('Content-Length', 0))
                downloaded = 0
                while True:
                    chunk = resp.read(65536)
                    if not chunk: break
                    f.write(chunk)
                    downloaded += len(chunk)
                    if total:
                        pct = 5 + int(downloaded / total * 30)
                        ota_status['progress'] = min(pct, 35)
                        now = time.time()
                        if pct - last_published_pct >= 10 or (now - last_publish_time) >= 5:
                            last_published_pct = pct
                            last_publish_time = now
                            ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
                            _publish_ota_progress()
        dl_size = os.path.getsize(fw_path)
        ota_status['progress'] = 40
        ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
        _publish_ota_progress()
        if abs(dl_size - file_size) > 1024:
            raise Exception(f"文件大小不匹配: 预期{file_size}, 实际{dl_size}")
        ota_status['status'] = 'VERIFYING'
        ota_status['progress'] = 50
        ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
        _publish_ota_progress()
        h = hashlib.new(checksum_type)
        with open(fw_path, 'rb') as f:
            for chunk in iter(lambda: f.read(65536), b''): h.update(chunk)
        actual_checksum = h.hexdigest().lower()
        if actual_checksum != checksum.lower():
            raise Exception(f"校验和不匹配: 预期{checksum}, 实际{actual_checksum}")
        ota_status['progress'] = 60
        ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
        _publish_ota_progress()
        extract_dir = os.path.join(tmp_dir, 'firmware')
        os.makedirs(extract_dir, exist_ok=True)
        with tarfile.open(fw_path, 'r:gz') as tar: tar.extractall(extract_dir)
        manifest_path = os.path.join(extract_dir, 'firmware', 'firmware.json')
        if not os.path.exists(manifest_path):
            raise Exception("固件包缺少 firmware.json")
        with open(manifest_path, 'r') as f: manifest = json.load(f)
        fw_version = manifest.get('version', version)
        if not force:
            current_ver = dtu_config.get('firmware_version', 'v0.0.0')
            if fw_version == current_ver:
                raise Exception(f'已是目标版本 {current_ver}')
        ota_status['status'] = 'FLASHING'
        ota_status['progress'] = 70
        ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
        _publish_ota_progress()
        fw_root = os.path.join(extract_dir, 'firmware')
        project_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        excludes = {'__pycache__', '.git', 'log', 'venv'}
        for root, dirs, files in os.walk(fw_root):
            rel = os.path.relpath(root, fw_root)
            if rel == '.': rel = ''
            parts = rel.split(os.sep) if rel else []
            if parts and parts[0] in excludes: continue
            for fname in files:
                if fname == 'firmware.json': continue
                src = os.path.join(root, fname)
                dst = os.path.join(project_dir, rel, fname)
                os.makedirs(os.path.dirname(dst), exist_ok=True)
                shutil.copy2(src, dst)
        ota_status['progress'] = 90
        ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
        dtu_config['firmware_version'] = fw_version
        save_dtu_config()
        ota_status['status'] = 'SUCCESS'
        ota_status['progress'] = 100
        ota_status['firmware_version'] = fw_version
        ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
        _publish_ota_progress()
        shutil.rmtree(tmp_dir, ignore_errors=True)
        for i in range(10, 0, -1):
            ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
            time.sleep(1)
        logger.info("OTA: 重启服务...")
        subprocess.Popen(
            [subprocess.sys.executable, '-m', 'flask', 'run', '--host=0.0.0.0', '--port=5001'],
            cwd=os.path.dirname(os.path.abspath(__file__)),
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
        )
        os._exit(0)
    except Exception as e:
        ota_status['status'] = 'FAILED'
        ota_status['error_message'] = str(e)
        ota_status['last_update'] = time.strftime('%Y-%m-%d %H:%M:%S')
        # 映射错误到标准错误码 (1021-1025)
        msg = str(e)
        if '校验和' in msg or 'firmware.json' in msg:
            ota_status['error_code'] = 1022
        elif '已是目标版本' in msg:
            ota_status['error_code'] = 1021
        elif '下载' in msg or 'urlopen' in msg or 'Connection' in msg or 'timeout' in msg:
            ota_status['error_code'] = 1023
        elif '磁盘' in msg or '空间' in msg or 'No space' in msg:
            ota_status['error_code'] = 1025
        elif '写入' in msg or 'flash' in msg.lower() or 'shutil' in msg:
            ota_status['error_code'] = 1024
        else:
            ota_status['error_code'] = 9999
        _publish_ota_progress()
        # 通过MQTT发送错误响应（同时适配HTTP和MQTT触发）
        _send_ota_error_response(ota_status['error_code'], params.get('_incoming_msg_id', ''))
        logger.error(f"OTA: 升级失败 - {str(e)}")
        if 'tmp_dir' in dir() and tmp_dir and os.path.exists(tmp_dir):
            shutil.rmtree(tmp_dir, ignore_errors=True)


@app.route('/api/dtu/ota_upgrade', methods=['POST'])
def trigger_ota_upgrade():
    """触发OTA升级"""
    try:
        data = request.json
        if not data:
            return jsonify({'success': False, 'message': '请求体不能为空'}), 400

        required_fields = ['firmware_url', 'firmware_version', 'file_size', 'checksum', 'checksum_type']
        missing = [f for f in required_fields if not data.get(f)]
        if missing:
            return jsonify({'success': False, 'message': f'缺少必填字段: {", ".join(missing)}'}), 400

        # 已是目标版本且未强制: 1021 (同时通过 MQTT 上报响应)
        force_upgrade = data.get('force_upgrade', False)
        current_version = dtu_config.get('firmware_version', 'v1.0.0')
        if data['firmware_version'] == current_version and not force_upgrade:
            # 通过 MQTT 上报 1021 响应 (与其他错误码一致)
            try:
                _send_ota_error_response(1021, data.get('msg_id', ''))
            except Exception:
                pass
            return jsonify({
                'success': False,
                'error_code': 1021,
                'message': f'已是目标版本: {current_version}'
            }), 400

        # 注入 _incoming_msg_id 使 MQTT 错误响应可回填
        ota_params = {**data, '_incoming_msg_id': data.get('msg_id', '')}
        t = threading.Thread(target=_run_ota, args=(ota_params,), daemon=True)
        t.start()

        return jsonify({
            'success': True,
            'message': 'OTA升级已启动',
            'data': {
                'target_version': data['firmware_version'],
                'ota_status': 'DOWNLOADING'
            }
        })
    except Exception as e:
        logger.error(f"触发OTA升级失败: {str(e)}")
        return jsonify({'success': False, 'message': str(e)}), 500


# ==================== 环境传感器 API ====================

# 环境传感器数据存储
env_sensor_data = {
    'temperature': None,          # 环境温度 (℃)
    'humidity': None,             # 环境湿度 (%)
    'dtu_temperature': None,      # DTU主板温度 (℃)
    'update_time': None,          # 数据更新时间
    'sensor_update_time': None,   # 传感器更新时间
    'connected': False            # 传感器连接状态
}

# 环境传感器告警阈值
env_sensor_threshold = {
    'temp_high': 45.0,            # 环境温度上限 (℃)
    'temp_low': -10.0,            # 环境温度下限 (℃)
    'humidity_high': 80.0,        # 环境湿度上限 (%)
    'humidity_low': 20.0,         # 环境湿度下限 (%)
    'dtu_temp_high': 70.0         # DTU主板温度上限 (℃)
}

# 环境传感器历史数据 (保留最近1000条)
env_sensor_history = []


@app.route('/api/sensor/env', methods=['GET'])
def get_env_sensor_data():
    """获取环境传感器当前数据"""
    data = dict(env_sensor_data)
    # DTU 主板温度回退: 若外部未上报，从 sysfs 本地读取
    if data.get('dtu_temperature') is None:
        try:
            import psutil
            temps = psutil.sensors_temperatures(fahrenheit=False) or {}
            for chip_name, entries in temps.items():
                for entry in entries:
                    if entry.current and entry.current > 0:
                        data['dtu_temperature'] = round(entry.current, 1)
                        break
                if data['dtu_temperature'] is not None:
                    break
        except Exception:
            pass
        if data.get('dtu_temperature') is None:
            import glob as _glob
            for tz in sorted(_glob.glob('/sys/class/thermal/thermal_zone*/temp')):
                try:
                    with open(tz) as f:
                        v = int(f.read().strip())
                        if v > 0:
                            data['dtu_temperature'] = round(v / 1000.0, 1)
                            data['update_time'] = data.get('update_time') or __import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                            break
                except Exception:
                    continue
    # 附加 CPU/内存使用率（始终本地读取）
    data['cpu_usage'] = _get_cpu_usage()
    data['memory_usage'] = _get_memory_usage()
    return jsonify({
        'success': True,
        'data': data
    })


@app.route('/api/sensor/threshold', methods=['GET', 'POST'])
def handle_env_sensor_threshold():
    """获取或设置环境传感器告警阈值"""
    if request.method == 'GET':
        return jsonify({
            'success': True,
            'data': env_sensor_threshold
        })
    else:
        try:
            data = request.get_json()
            for key in env_sensor_threshold.keys():
                if key in data:
                    env_sensor_threshold[key] = float(data[key])
            logger.info(f"更新环境传感器告警阈值: {env_sensor_threshold}")
            return jsonify({
                'success': True,
                'message': '阈值设置成功',
                'data': env_sensor_threshold
            })
        except Exception as e:
            logger.error(f"设置告警阈值失败: {str(e)}")
            return jsonify({'success': False, 'message': str(e)}), 400


@app.route('/api/sensor/history', methods=['GET'])
def get_env_sensor_history():
    """获取环境传感器历史数据"""
    try:
        import datetime
        # 获取查询参数
        range_param = request.args.get('range', '24h')
        start_time = request.args.get('start_time')
        end_time = request.args.get('end_time')
        limit = request.args.get('limit', 100, type=int)

        filtered_history = env_sensor_history

        # 解析range参数
        if not start_time and not end_time:
            now = datetime.datetime.now()
            if range_param == '1h':
                start_time = (now - datetime.timedelta(hours=1)).strftime('%Y-%m-%d %H:%M:%S')
            elif range_param == '6h':
                start_time = (now - datetime.timedelta(hours=6)).strftime('%Y-%m-%d %H:%M:%S')
            elif range_param == '24h':
                start_time = (now - datetime.timedelta(hours=24)).strftime('%Y-%m-%d %H:%M:%S')
            elif range_param == '7d':
                start_time = (now - datetime.timedelta(days=7)).strftime('%Y-%m-%d %H:%M:%S')
            elif range_param == '30d':
                start_time = (now - datetime.timedelta(days=30)).strftime('%Y-%m-%d %H:%M:%S')

        # 按时间过滤
        if start_time:
            filtered_history = [h for h in filtered_history if h.get('update_time') >= start_time]
        if end_time:
            filtered_history = [h for h in filtered_history if h.get('update_time') <= end_time]

        # 按时间排序 (新到旧)
        filtered_history = sorted(filtered_history, key=lambda x: x.get('update_time', ''), reverse=True)

        # 限制返回数量
        filtered_history = filtered_history[:limit]

        return jsonify({
            'success': True,
            'data': filtered_history,
            'total': len(filtered_history)
        })
    except Exception as e:
        logger.error(f"获取历史数据失败: {str(e)}")
        return jsonify({'success': False, 'message': str(e)}), 500


def update_env_sensor_data_full(temperature, humidity, dtu_temperature, sensor_update_time):
    """更新环境传感器数据 (由MQTT消息触发) - 完整版，含历史和告警"""
    import datetime
    now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    env_sensor_data['temperature'] = temperature
    env_sensor_data['humidity'] = humidity
    env_sensor_data['dtu_temperature'] = dtu_temperature
    env_sensor_data['sensor_update_time'] = sensor_update_time
    env_sensor_data['update_time'] = now
    env_sensor_data['connected'] = True

    # 同步到 MQTT 上行所用的 _env_sensor_data
    _env_sensor_data['temperature'] = temperature
    _env_sensor_data['humidity'] = humidity
    _env_sensor_data['sensor_update_time'] = sensor_update_time

    # 添加到历史记录
    history_record = {
        'temperature': temperature,
        'humidity': humidity,
        'dtu_temperature': dtu_temperature,
        'sensor_update_time': sensor_update_time,
        'update_time': now
    }
    env_sensor_history.append(history_record)

    # 保留最近1000条
    if len(env_sensor_history) > 1000:
        env_sensor_history[:] = env_sensor_history[-1000:]

    # 检查告警
    check_env_sensor_alarms(temperature, humidity, dtu_temperature)

    logger.info(f"环境传感器数据更新: 温度={temperature}℃, 湿度={humidity}%, DTU温度={dtu_temperature}℃")


def dht11_data_callback(temperature, humidity):
    """DHT11 本地 GPIO 传感器数据回调。

    将读取到的温湿度更新到环境传感器数据区，并通过 MQTT 上报。
    保留已有的 dtu_temperature（主板温度），仅更新环境温湿度字段。
    """
    try:
        sensor_update_time = int(time.time() * 1000)
        dtu_temperature = env_sensor_data.get('dtu_temperature')

        update_env_sensor_data_full(temperature, humidity, dtu_temperature, sensor_update_time)

        # 立即通过 MQTT 上报环境传感器数据
        try:
            dtu_publish_env_sensor()
        except Exception as pub_err:
            logger.warning(f"DHT11 数据 MQTT 上报失败: {pub_err}")

        logger.info(f"DHT11 本地传感器数据已处理: 温度={temperature}°C, 湿度={humidity}%")
    except Exception as e:
        logger.error(f"处理 DHT11 本地传感器数据失败: {e}")


def check_env_sensor_alarms(temperature, humidity, dtu_temperature):
    """检查环境传感器告警"""
    alarms = []

    if temperature is not None:
        if temperature > env_sensor_threshold['temp_high']:
            alarms.append({
                'type': 'temperature_high',
                'message': f'环境温度过高: {temperature}℃ (阈值: {env_sensor_threshold["temp_high"]}℃)',
                'level': 'warning'
            })
        elif temperature < env_sensor_threshold['temp_low']:
            alarms.append({
                'type': 'temperature_low',
                'message': f'环境温度过低: {temperature}℃ (阈值: {env_sensor_threshold["temp_low"]}℃)',
                'level': 'warning'
            })

    if humidity is not None:
        if humidity > env_sensor_threshold['humidity_high']:
            alarms.append({
                'type': 'humidity_high',
                'message': f'环境湿度过高: {humidity}% (阈值: {env_sensor_threshold["humidity_high"]}%)',
                'level': 'warning'
            })
        elif humidity < env_sensor_threshold['humidity_low']:
            alarms.append({
                'type': 'humidity_low',
                'message': f'环境湿度过低: {humidity}% (阈值: {env_sensor_threshold["humidity_low"]}%)',
                'level': 'warning'
            })

    if dtu_temperature is not None:
        if dtu_temperature > env_sensor_threshold['dtu_temp_high']:
            alarms.append({
                'type': 'dtu_temp_high',
                'message': f'DTU主板温度过高: {dtu_temperature}℃ (阈值: {env_sensor_threshold["dtu_temp_high"]}℃)',
                'level': 'critical'
            })

    # 发送告警通知
    for alarm in alarms:
        logger.warning(f"环境传感器告警: {alarm['message']}")
        # 可以通过WebSocket发送告警
        socketio.emit('env_sensor_alarm', alarm)


if __name__ == '__main__':
    try:
        # 启动前的初始化工作
        logger.info('启动串口-MQTT网关服务...')
        logger.info(f"配置信息: 主机={FLASK_HOST}, 端口={FLASK_PORT}, 调试模式={FLASK_DEBUG}")

        # 加载DTU配置
        load_dtu_config()

        # 加载设备配置
        loaded = load_device_config()
        logger.info(f"已加载 {len(loaded)} 个设备配置")
        panel_config.clear()
        for uid_hex, addr in loaded.items():
            panel_id = f"PANEL_{dtu_config.get('dtu_id', 'DTU001')}_{addr}"
            panel_config[panel_id] = {'address': addr, 'position': addr, 'panel_uid': uid_hex}
        logger.info(f"已加载 {len(panel_config)} 个面板配置")
        # 启动自动发现循环
        def auto_discover_loop():
            while True:
                time.sleep(30)
                try:
                    _st = serial_client.get_status()
                    if not (isinstance(_st, dict) and _st.get('connected', False)):
                        continue
                    if not dtu_config.get('enabled'):
                        continue
                    result = address_config.auto_configure(timeout=2.0)
                    if result.get('discovered', 0) > 0:
                        logger.info(f"自动发现: {result.get('discovered')} 个设备")
                        save_device_config()
                    now = time.time()
                    for uid in address_config.get_stored_devices():
                        device_last_seen[uid] = now
                    for pid in panel_config:
                        device_last_seen[pid] = now
                except Exception as e:
                    logger.error(f"自动发现异常: {str(e)}")
        import threading
        t = threading.Thread(target=auto_discover_loop, daemon=True)
        t.start()
        logger.info("启动自动发现线程 (间隔30秒)")

        # 启动DTU心跳定时器
        if dtu_config.get('enabled'):
            start_dtu_heartbeat()
            logger.info("启动DTU心跳定时器")

        # 启动端口轮询循环
        def port_poll_loop():
            while True:
                interval = dtu_config.get('poll_interval_ms', 5000) / 1000.0
                time.sleep(max(1, interval))
                try:
                    _st = serial_client.get_status()
                    if not (isinstance(_st, dict) and _st.get('connected', False)):
                        continue
                    if not dtu_config.get('enabled'):
                        continue
                    if not panel_config:
                        continue
                    online = 0
                    offline = 0
                    for panel_id, cfg in panel_config.items():
                        addr = cfg.get('address', 1)
                        panel_ok = False
                        panel_ports = []
                        for port_id in range(1, 25):
                            try:
                                result = modbus_client.read_antenna_card(addr, port_id, timeout=1.0)
                            except Exception:
                                result = {'error': 'exception'}
                            if 'error' in result:
                                panel_ports.append({'port_id': port_id, 'status': 'UNKNOWN', 'jumper_uid': None})
                                continue
                            panel_ok = True
                            card_str = result.get('card_number_str', '')
                            uid = card_str.upper() if card_str and card_str != '0000000000000000' else ''
                            if panel_id not in port_state:
                                port_state[panel_id] = {}
                            if port_id not in port_state[panel_id]:
                                port_state[panel_id][port_id] = {'last_uid': None, 'expected_uid': None, 'alarm_count': 0, 'last_polled_at': int(time.time() * 1000)}
                            ps = port_state[panel_id][port_id]
                            ps['last_polled_at'] = int(time.time() * 1000)
                            last_uid = ps.get('last_uid')
                            if uid and uid != last_uid:
                                dtu_publish_event(panel_id, port_id, 'CONNECT' if not last_uid else 'MOVE', uid, last_uid)
                                ps['last_uid'] = uid
                            elif not uid and last_uid:
                                dtu_publish_event(panel_id, port_id, 'DISCONNECT', None, last_uid)
                                ps['last_uid'] = None

                            ps_exp = ps.get('expected_uid')
                            if dtu_config.get('alarm_enabled', True):
                                if ps_exp and uid and uid != ps_exp:
                                    ps['alarm_count'] = ps.get('alarm_count', 0) + 1
                                    sev = 'CRITICAL' if ps['alarm_count'] >= 3 else 'WARNING'
                                    dtu_publish_alarm(panel_id, port_id, 'ILLEGAL_CONNECT', ps_exp, uid, sev)
                                elif ps_exp and not uid:
                                    ps['alarm_count'] = ps.get('alarm_count', 0) + 1
                                    sev = 'CRITICAL' if ps['alarm_count'] >= 3 else 'WARNING'
                                    dtu_publish_alarm(panel_id, port_id, 'ILLEGAL_DISCONNECT', ps_exp, None, sev)
                                elif ps_exp and uid == ps_exp:
                                    ps['alarm_count'] = 0

                            if uid:
                                port_status = 'ILLEGAL' if (ps_exp and uid != ps_exp) else 'CONNECTED'
                            else:
                                port_status = 'DISCONNECTED'
                            panel_ports.append({'port_id': port_id, 'status': port_status, 'jumper_uid': uid or None})

                        if panel_ok:
                            online += 1
                            device_last_seen[panel_id] = time.time()
                            device_last_seen[cfg.get('panel_uid', '')] = time.time()
                            dtu_publish_panel_status(panel_id, addr, panel_ports)
                            # 如果该 panel 是当前显示的 panel，刷新屏幕
                            panels = get_sorted_panels()
                            trigger_panel_id = None
                            with screen_lock:
                                if panels and 0 <= screen_current_panel_index < len(panels) and panels[screen_current_panel_index][0] == panel_id:
                                    trigger_panel_id = panel_id
                            if trigger_panel_id:
                                try:
                                    screen_refresh_queue.put((trigger_panel_id,))
                                except Exception as e:
                                    logger.error(f"刷新屏幕失败: {e}")
                        else:
                            offline += 1

                    dtu_publish_status(force=True)
                except Exception as e:
                    logger.error(f"端口轮询异常: {str(e)}")
        t2 = threading.Thread(target=port_poll_loop, daemon=True)
        t2.start()
        logger.info("启动端口轮询线程 (间隔5秒)")

        # 启动 DHT11 本地传感器读取线程
        if dht11_sensor is not None:
            dht11_sensor.on_data = dht11_data_callback
            dht11_sensor.start()

        # 自动连接上次使用的串口
        logger.info("尝试自动连接串口...")
        auto_connect_serial()

        # 连接 Nextion 串口屏
        if NEXTION_ENABLED:
            try:
                ok, msg = screen_display.connect(NEXTION_SCREEN_PORT, NEXTION_SCREEN_BAUD)
                if ok:
                    screen_display.set_button_callback(screen_button_event_handler)
                    logger.info("Nextion 屏幕连接成功")
                    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}")

        # 启动 Nextion 屏幕异步刷新工作线程
        screen_refresh_thread = threading.Thread(target=_screen_refresh_worker, daemon=True)
        screen_refresh_thread.start()
        logger.info("启动 Nextion 屏幕异步刷新线程")

        # 启动服务
        socketio.run(
            app, 
            host=FLASK_HOST, 
            port=FLASK_PORT, 
            debug=FLASK_DEBUG,
            use_reloader=False,  # 禁用重载器以避免重复初始化问题
            log_output=False,  # 禁用Flask的日志输出，使用我们自己的日志配置
            allow_unsafe_werkzeug=True
        )
    except KeyboardInterrupt:
        # 优雅退出
        logger.info('正在关闭应用...')
        try:
            if isinstance(serial_client.get_status(), dict) and serial_client.get_status().get("connected", False):
                serial_client.disconnect()
                logger.info('串口连接已断开')
            if mqtt_client.get_status():
                mqtt_client.disconnect()
                logger.info('MQTT连接已断开')
            if dht11_sensor is not None:
                dht11_sensor.stop()
                logger.info('DHT11 传感器线程已停止')
        except Exception as e:
            logger.error(f'关闭连接时出错: {str(e)}')

        # 清理WebSocket连接
        for client_type in connected_clients:
            connected_clients[client_type].clear()
        
        logger.info('应用已安全关闭')
    except Exception as e:
        logger.exception(f'应用启动失败')  # 使用exception记录完整堆栈
        # 确保资源被释放
        try:
            serial_client.disconnect()
            mqtt_client.disconnect()
        except:
            pass