#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
设备处理服务模块
基于VDA5050协议的设备管理框架，支持MQTT通信
消息直接透传，不进行编解码处理
"""

import asyncio
import json
import time
import socket
import threading
from datetime import datetime
from typing import Dict, Any, Optional, Callable, List, Union
from dataclasses import dataclass, asdict
from enum import Enum
from utils.logger import get_logger
from .mqtt_service import MQTTService
from .async_mqtt_service import AsyncMQTTService
from config.tf_api_config import MQTT_TOPIC_CONFIG

logger = get_logger("services.device_handler")


class DeviceType(Enum):
    """设备类型枚举"""
    VEHICLE = "vehicle"      # 小车
    DOOR = "door"           # 门
    CALLER = "caller"       # 呼叫器
    LIFT = "lift"           # 电梯
    CONVEYOR = "conveyor"   # 输送带
    SENSOR = "sensor"       # 传感器
    ROBOT = "robot"         # 机器人
    CAMERA = "camera"       # 摄像头
    SCANNER = "scanner"     # 扫描器
    CUSTOM = "custom"       # 自定义设备


class VDA5050CommandType(Enum):
    """VDA5050协议指令类型"""
    ORDER = "order"                    # 订单指令（仅小车）
    STATE = "state"                    # 状态查询（仅小车）
    FACTSHEET = "factsheet"           # 设备信息（仅小车）
    INSTANT_ACTIONS = "instantActions" # 即时动作（所有设备）
    CONNECTION = "connection"          # 连接管理
    VISUALIZATION = "visualization"    # 可视化


class ProtocolType(Enum):
    """协议类型"""
    VDA5050 = "vda5050"    # VDA5050协议（默认）
    TCP = "tcp"            # TCP协议


class DeviceBrand(Enum):
    """设备品牌枚举"""
    HUARUI = "huarui"       # 华睿
    SEER = "seer"           # 仙工
    QUICKTRON = "quicktron" # 快仓
    GEEK = "geek"           # 极智嘉
    MUSHINY = "mushiny"     # 木蚁
    FLASHHOLD = "flashhold" # 闪电匣
    HIKROBOT = "hikrobot"   # 海康机器人
    STANDARD = "standard"   # 标准协议
    CUSTOM = "custom"       # 自定义品牌


class DeviceProtocol:
    """设备协议定义基类"""
    
    def __init__(self, brand: DeviceBrand, device_type: DeviceType, 
                 protocol_type: ProtocolType = ProtocolType.VDA5050):
        self.brand = brand
        self.device_type = device_type
        self.protocol_type = protocol_type
        
    def encode_command(self, command_type: str, command: Dict[str, Any]) -> Dict[str, Any]:
        """将VDA5050标准指令编码为设备特定格式"""
        raise NotImplementedError("子类必须实现encode_command方法")
    
    def decode_response(self, message_type: str, response: Dict[str, Any]) -> Dict[str, Any]:
        """将设备响应解码为VDA5050标准格式"""
        raise NotImplementedError("子类必须实现decode_response方法")
    
    def validate_command(self, command_type: str, command: Dict[str, Any]) -> bool:
        """验证指令格式是否正确"""
        return True
    
    def get_supported_command_types(self) -> List[str]:
        """获取支持的VDA5050指令类型列表"""
        if self.device_type == DeviceType.VEHICLE:
            return [
                VDA5050CommandType.ORDER.value,
                VDA5050CommandType.STATE.value,
                VDA5050CommandType.FACTSHEET.value,
                VDA5050CommandType.INSTANT_ACTIONS.value
            ]
        else:
            # 其他设备只支持即时动作
            return [VDA5050CommandType.INSTANT_ACTIONS.value]


class VDA5050Protocol(DeviceProtocol):
    """VDA5050协议实现（标准协议）"""
    
    def __init__(self, brand: DeviceBrand = DeviceBrand.STANDARD, device_type: DeviceType = DeviceType.VEHICLE):
        super().__init__(brand, device_type, ProtocolType.VDA5050)
    
    def encode_command(self, command_type: str, command: Dict[str, Any]) -> Dict[str, Any]:
        """VDA5050标准指令编码"""
        if command_type == VDA5050CommandType.ORDER.value:
            return self._encode_order(command)
        elif command_type == VDA5050CommandType.INSTANT_ACTIONS.value:
            return self._encode_instant_actions(command)
        elif command_type == VDA5050CommandType.STATE.value:
            return self._encode_state_request(command)
        elif command_type == VDA5050CommandType.FACTSHEET.value:
            return self._encode_factsheet_request(command)
        else:
            return command
    
    def decode_response(self, message_type: str, response: Dict[str, Any]) -> Dict[str, Any]:
        """VDA5050标准响应解码"""
        if message_type == VDA5050CommandType.STATE.value:
            return self._decode_state_response(response)
        elif message_type == VDA5050CommandType.FACTSHEET.value:
            return self._decode_factsheet_response(response)
        else:
            return response
    
    def _encode_order(self, command: Dict[str, Any]) -> Dict[str, Any]:
        """编码VDA5050订单指令"""
        return {
            "headerId": command.get("header_id", f"order_{int(time.time())}"),
            "timestamp": command.get("timestamp", datetime.now().isoformat()),
            "version": command.get("version", "2.0.0"),
            "manufacturer": self.brand.value,
            "serialNumber": command.get("serial_number", ""),
            "orderId": command.get("order_id", f"order_{int(time.time())}"),
            "orderUpdateId": command.get("order_update_id", 0),
            "zoneSetId": command.get("zone_set_id", ""),
            "nodes": command.get("nodes", []),
            "edges": command.get("edges", [])
        }
    
    def _encode_instant_actions(self, command: Dict[str, Any]) -> Dict[str, Any]:
        """编码VDA5050即时动作指令"""
        actions = command.get("actions", [])
        
        # 标准化动作格式
        formatted_actions = []
        for action in actions:
            if isinstance(action, str):
                # 简单字符串转换为标准VDA5050动作格式
                formatted_actions.append({
                    "actionType": action,
                    "actionId": f"{action}_{int(time.time())}",
                    "actionDescription": f"{self.device_type.value} {action} action",
                    "blockingType": "SOFT"
                })
            else:
                formatted_actions.append(action)
        
        return {
            "headerId": command.get("header_id", f"instant_{int(time.time())}"),
            "timestamp": command.get("timestamp", datetime.now().isoformat()),
            "version": command.get("version", "2.0.0"),
            "manufacturer": self.brand.value,
            "serialNumber": command.get("serial_number", ""),
            "instantActions": formatted_actions
        }
    
    def _encode_state_request(self, command: Dict[str, Any]) -> Dict[str, Any]:
        """编码VDA5050状态请求"""
        return {
            "requestId": command.get("request_id", f"state_{int(time.time())}"),
            "timestamp": command.get("timestamp", datetime.now().isoformat())
        }
    
    def _encode_factsheet_request(self, command: Dict[str, Any]) -> Dict[str, Any]:
        """编码VDA5050设备信息请求"""
        return {
            "requestId": command.get("request_id", f"factsheet_{int(time.time())}"),
            "timestamp": command.get("timestamp", datetime.now().isoformat())
        }
    
    def _decode_state_response(self, response: Dict[str, Any]) -> Dict[str, Any]:
        """解码VDA5050状态响应"""
        return {
            "command_type": "state_response",
            "order_id": response.get("orderId", ""),
            "order_update_id": response.get("orderUpdateId", 0),
            "last_node_id": response.get("lastNodeId", ""),
            "last_node_sequence_id": response.get("lastNodeSequenceId", 0),
            "driving": response.get("driving", False),
            "paused": response.get("paused", False),
            "new_base_request": response.get("newBaseRequest", False),
            "distance_since_last_node": response.get("distanceSinceLastNode", 0.0),
            "operating_mode": response.get("operatingMode", ""),
            "node_states": response.get("nodeStates", []),
            "edge_states": response.get("edgeStates", []),
            "agv_position": response.get("agvPosition", {}),
            "velocity": response.get("velocity", {}),
            "loads": response.get("loads", []),
            "action_states": response.get("actionStates", []),
            "battery_state": response.get("batteryState", {}),
            "safety_state": response.get("safetyState", {}),
            "information": response.get("information", []),
            "errors": response.get("errors", []),
            "timestamp": response.get("timestamp", "")
        }
    
    def _decode_factsheet_response(self, response: Dict[str, Any]) -> Dict[str, Any]:
        """解码VDA5050设备信息响应"""
        return {
            "command_type": "factsheet_response",
            "manufacturer": response.get("manufacturer", ""),
            "serial_number": response.get("serialNumber", ""),
            "type_specification": response.get("typeSpecification", {}),
            "physical_parameters": response.get("physicalParameters", {}),
            "protocol_limits": response.get("protocolLimits", {}),
            "protocol_features": response.get("protocolFeatures", {}),
            "agv_geometry": response.get("agvGeometry", {}),
            "load_specification": response.get("loadSpecification", {}),
            "timestamp": response.get("timestamp", "")
        }


class TCPProtocol(DeviceProtocol):
    """TCP协议实现（VDA5050通过TCP传输）"""
    
    def __init__(self, brand: DeviceBrand = DeviceBrand.STANDARD, device_type: DeviceType = DeviceType.VEHICLE):
        super().__init__(brand, device_type, ProtocolType.TCP)
        self.vda_protocol = VDA5050Protocol(brand, device_type)
        
    def encode_command(self, command_type: str, command: Dict[str, Any]) -> Dict[str, Any]:
        """TCP协议编码（基于VDA5050）"""
        # 先通过VDA5050协议编码
        vda_data = self.vda_protocol.encode_command(command_type, command)
        
        # 添加TCP传输包装
        tcp_packet = {
            "protocol": "vda5050",
            "transport": "tcp",
            "timestamp": datetime.now().isoformat(),
            "data": vda_data
        }
        
        return tcp_packet
    
    def decode_response(self, message_type: str, response: Dict[str, Any]) -> Dict[str, Any]:
        """TCP协议解码（基于VDA5050）"""
        # 如果是TCP包装格式，提取VDA5050数据
        if "data" in response and response.get("protocol") == "vda5050":
            vda_data = response["data"]
            return self.vda_protocol.decode_response(message_type, vda_data)
        else:
            # 直接作为VDA5050数据处理
            return self.vda_protocol.decode_response(message_type, response)


class DeviceProtocolRegistry:
    """设备协议注册中心"""
    
    def __init__(self):
        self.protocols: Dict[str, DeviceProtocol] = {}
        self._register_builtin_protocols()
    
    def _register_builtin_protocols(self):
        """注册内置协议"""
        # 默认VDA5050协议（适用于所有设备）
        default_vda5050 = VDA5050Protocol()
        self.register_protocol("vda5050", default_vda5050)
        
        # 各品牌VDA5050协议
        for brand in [DeviceBrand.HUARUI, DeviceBrand.SEER, DeviceBrand.QUICKTRON, 
                     DeviceBrand.GEEK, DeviceBrand.MUSHINY, DeviceBrand.FLASHHOLD, 
                     DeviceBrand.HIKROBOT]:
            # 小车VDA5050协议
            vehicle_protocol = VDA5050Protocol(brand, DeviceType.VEHICLE)
            self.register_protocol(f"{brand.value}_vda5050", vehicle_protocol)
            
            # TCP协议（基于VDA5050）
            tcp_protocol = TCPProtocol(brand, DeviceType.VEHICLE)
            self.register_protocol(f"{brand.value}_tcp", tcp_protocol)
        
        # 非小车设备的VDA5050协议（即时动作）
        for device_type in [DeviceType.DOOR, DeviceType.CALLER, DeviceType.LIFT, 
                           DeviceType.CONVEYOR, DeviceType.SENSOR, DeviceType.CAMERA, 
                           DeviceType.SCANNER, DeviceType.ROBOT, DeviceType.CUSTOM]:
            vda_protocol = VDA5050Protocol(DeviceBrand.STANDARD, device_type)
            self.register_protocol(f"{device_type.value}_vda5050", vda_protocol)
            
            tcp_protocol = TCPProtocol(DeviceBrand.STANDARD, device_type)
            self.register_protocol(f"{device_type.value}_tcp", tcp_protocol)
    
    def register_protocol(self, protocol_key: str, protocol: DeviceProtocol):
        """注册设备协议"""
        self.protocols[protocol_key] = protocol
        logger.info(f"注册设备协议: {protocol_key} ({protocol.brand.value}_{protocol.device_type.value})")
    
    def get_protocol(self, protocol_key: str) -> Optional[DeviceProtocol]:
        """获取设备协议"""
        return self.protocols.get(protocol_key)
    
    def get_protocol_by_brand_type(self, brand: Union[str, DeviceBrand], 
                                  device_type: Union[str, DeviceType]) -> Optional[DeviceProtocol]:
        """根据品牌和设备类型获取协议"""
        if isinstance(brand, str):
            brand = DeviceBrand(brand)
        if isinstance(device_type, str):
            device_type = DeviceType(device_type)
        
        for protocol in self.protocols.values():
            if protocol.brand == brand and protocol.device_type == device_type:
                return protocol
        return None
    
    def list_protocols(self) -> Dict[str, Dict[str, str]]:
        """列出所有协议"""
        return {
            key: {
                "brand": protocol.brand.value,
                "device_type": protocol.device_type.value,
                "supported_commands": protocol.get_supported_commands()
            }
            for key, protocol in self.protocols.items()
        }
    
    def register_custom_protocol(self, protocol_key: str, brand: str, device_type: str,
                                encode_func: Callable = None, decode_func: Callable = None):
        """注册自定义协议（基于VDA5050）"""
        
        # 解析设备类型
        try:
            device_enum = DeviceType(device_type)
        except ValueError:
            device_enum = DeviceType.CUSTOM
            
        # 解析品牌
        try:
            brand_enum = DeviceBrand(brand)
        except ValueError:
            brand_enum = DeviceBrand.CUSTOM
        
        # 创建基于VDA5050的自定义协议
        custom_protocol = VDA5050Protocol(brand_enum, device_enum)
        
        # 如果提供了自定义编解码函数，包装协议
        if encode_func or decode_func:
            original_encode = custom_protocol.encode_command
            original_decode = custom_protocol.decode_response
            
            def custom_encode(command_type: str, command: Dict[str, Any]) -> Dict[str, Any]:
                vda_result = original_encode(command_type, command)
                if encode_func:
                    return encode_func(command_type, vda_result)
                return vda_result
            
            def custom_decode(message_type: str, response: Dict[str, Any]) -> Dict[str, Any]:
                if decode_func:
                    processed_response = decode_func(message_type, response)
                    return original_decode(message_type, processed_response)
                return original_decode(message_type, response)
            
            custom_protocol.encode_command = custom_encode
            custom_protocol.decode_response = custom_decode
        
        self.register_protocol(protocol_key, custom_protocol)


# 全局协议注册中心
_global_protocol_registry = DeviceProtocolRegistry()


def get_protocol_registry() -> DeviceProtocolRegistry:
    """获取全局协议注册中心"""
    return _global_protocol_registry


class TopicManager:
    """MQTT Topic管理器"""

    def __init__(self):
        self.topic_config = MQTT_TOPIC_CONFIG
        self.instance_id = self.topic_config["instance_id"]
        self.supported_brands = self.topic_config["supported_brands"]
        self.command_directions = self.topic_config["command_directions"]

        # 扁平化品牌配置便于查找
        self._flat_brands = {}
        for category in ["vehicle", "other"]:
            if category in self.supported_brands:
                self._flat_brands.update(self.supported_brands[category])

    def validate_brand(self, brand_name: str) -> bool:
        """验证品牌是否支持"""
        return brand_name.lower() in self._flat_brands

    def get_supported_brands(self, category: str = None) -> List[str]:
        """获取支持的品牌列表

        Args:
            category: 设备类型分类 ("vehicle", "other", 或 None 返回所有)

        Returns:
            品牌名称列表
        """
        if category is None:
            # 返回所有品牌
            return list(self._flat_brands.keys())
        elif category in self.supported_brands:
            # 返回指定分类的品牌
            return list(self.supported_brands[category].keys())
        else:
            logger.warning(f"未知的设备类型分类: {category}")
            return []

    def get_brand_config(self, brand_name: str) -> Optional[Dict[str, Any]]:
        """获取品牌配置

        Args:
            brand_name: 品牌名称

        Returns:
            品牌配置字典，如果不存在返回None
        """
        return self._flat_brands.get(brand_name.lower())
    
    def generate_listen_topics(self, device_ids: List[str], brand_name: str,
                             command_types: List[str] = None, device_type: DeviceType = DeviceType.VEHICLE) -> List[str]:
        """生成监听topics

        Args:
            device_ids: 设备ID列表
            brand_name: 品牌名称
            command_types: 指令类型列表，如果为None则使用默认指令
            device_type: 设备类型

        Returns:
            监听topic列表
        """
        if not self.validate_brand(brand_name):
            raise ValueError(f"不支持的品牌: {brand_name}，支持的品牌: {self.get_supported_brands()}")

        brand_config = self.get_brand_config(brand_name)
        if not brand_config:
            raise ValueError(f"无法获取品牌配置: {brand_name}")

        brand_suffix = brand_config["brand_suffix"]
        oagv_version = brand_config["oagv_version"]

        if command_types is None:
            if device_type == DeviceType.VEHICLE:
                command_types = self.topic_config["default_command_types"]["vehicle"]
            else:
                command_types = self.topic_config["default_command_types"]["other"]

        listen_topics = []
        for command_type in command_types:
            if command_type not in self.command_directions:
                logger.warning(f"未知的指令类型: {command_type}")
                continue

            direction_config = self.command_directions[command_type]
            listen_queue = direction_config["listen"]

            if listen_queue == "oagv":
                # oagv格式: oagv/v2/{instanceId}_BRAND/{deviceId}/order
                for device_id in device_ids:
                    topic = f"oagv/{oagv_version}/{self.instance_id}{brand_suffix}/{device_id}/{command_type}"
                    listen_topics.append(topic)
            elif listen_queue == "uagv":
                # uagv格式: uagv/v2.0.0/BRAND/{deviceId}/state
                uagv_version = brand_config["uagv_version"]
                brand_name_only = brand_suffix[1:]  # 去掉前缀下划线
                for device_id in device_ids:
                    topic = f"uagv/{uagv_version}/{brand_name_only}/{device_id}/{command_type}"
                    listen_topics.append(topic)

        return listen_topics
    
    def generate_forward_topics(self, device_ids: List[str], brand_name: str,
                              command_types: List[str] = None, device_type: DeviceType = DeviceType.VEHICLE) -> List[str]:
        """生成转发topics

        Args:
            device_ids: 设备ID列表
            brand_name: 品牌名称
            command_types: 指令类型列表，如果为None则使用默认指令
            device_type: 设备类型

        Returns:
            转发topic列表
        """
        if not self.validate_brand(brand_name):
            raise ValueError(f"不支持的品牌: {brand_name}，支持的品牌: {self.get_supported_brands()}")

        brand_config = self.get_brand_config(brand_name)
        if not brand_config:
            raise ValueError(f"无法获取品牌配置: {brand_name}")

        brand_suffix = brand_config["brand_suffix"]
        uagv_version = brand_config["uagv_version"]
        oagv_version = brand_config["oagv_version"]

        if command_types is None:
            if device_type == DeviceType.VEHICLE:
                command_types = self.topic_config["default_command_types"]["vehicle"]
            else:
                command_types = self.topic_config["default_command_types"]["other"]

        forward_topics = []

        for command_type in command_types:
            if command_type not in self.command_directions:
                logger.warning(f"未知的指令类型: {command_type}")
                continue

            direction_config = self.command_directions[command_type]
            forward_queue = direction_config["forward"]

            if forward_queue == "uagv":
                # uagv格式: uagv/v2.0.0/BRAND/{deviceId}/order
                brand_name_only = brand_suffix[1:]  # 去掉前缀下划线
                for device_id in device_ids:
                    topic = f"uagv/{uagv_version}/{brand_name_only}/{device_id}/{command_type}"
                    forward_topics.append(topic)
            elif forward_queue == "oagv":
                # oagv格式: oagv/v2/{instanceId}_BRAND/{deviceId}/state
                for device_id in device_ids:
                    topic = f"oagv/{oagv_version}/{self.instance_id}{brand_suffix}/{device_id}/{command_type}"
                    forward_topics.append(topic)

        return forward_topics
    
    def extract_vehicle_id_from_topic(self, topic: str) -> Optional[str]:
        """从topic中提取车辆ID"""
        parts = topic.split('/')
        if len(parts) >= 4:
            if parts[0] == "oagv":
                # oagv格式: oagv/v2/{instanceId}_BRAND/{vehicleId}/command
                return parts[3] if parts[3] != "+" else None
            elif parts[0] == "uagv":
                # uagv格式: uagv/v2.0.0/BRAND/{vehicleId}/command
                return parts[3]
        return None


class MessagePriority(Enum):
    """消息优先级"""
    LOW = 1
    NORMAL = 2
    HIGH = 3
    CRITICAL = 4


@dataclass
class DeviceMessage:
    """设备消息数据结构"""
    device_id: str
    device_type: DeviceType
    topic: str
    payload: Dict[str, Any]
    timestamp: float
    priority: MessagePriority = MessagePriority.NORMAL
    source_topic: Optional[str] = None
    target_topics: Optional[List[str]] = None
    processed: bool = False
    retry_count: int = 0
    max_retries: int = 3
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典格式"""
        data = asdict(self)
        data['device_type'] = self.device_type.value
        data['priority'] = self.priority.value
        return data
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'DeviceMessage':
        """从字典创建消息对象"""
        data['device_type'] = DeviceType(data['device_type'])
        data['priority'] = MessagePriority(data['priority'])
        return cls(**data)


@dataclass
class DeviceHandlerConfig:
    """设备处理器配置"""
    device_id: str
    device_type: DeviceType
    listen_topics: List[str]
    forward_topics: List[str]
    handler_function: str
    script_id: str
    device_brand: Optional[DeviceBrand] = DeviceBrand.STANDARD  # 设备品牌（默认标准）
    protocol_type: ProtocolType = ProtocolType.VDA5050  # 协议类型（默认VDA5050）
    # auto_encode: bool = True  # 编解码功能已移除，消息直接透传
    enabled: bool = True
    auto_retry: bool = True
    timeout_seconds: int = 30
    batch_size: int = 1
    description: str = ""
    metadata: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}
            
    # 已移除: protocol_key 方法，不再需要协议标识



class DeviceHandlerRegistry:
    """设备处理器注册中心"""
    
    def __init__(self, use_async_mqtt: bool =True):
        self.device_handlers: Dict[str, DeviceHandlerConfig] = {}
        self.message_processors: Dict[str, Callable] = {}
        self.topic_mappings: Dict[str, List[str]] = {}  # topic -> device_ids
        self.active_devices: Dict[str, Dict[str, Any]] = {}
        self.message_queue: asyncio.Queue = asyncio.Queue()
        self.processing_stats: Dict[str, Dict[str, Any]] = {}
        
        # 选择MQTT服务类型
        self.use_async_mqtt = use_async_mqtt
        if use_async_mqtt:
            self.mqtt_service = AsyncMQTTService()
            logger.info("使用异步MQTT服务")
        else:
            self.mqtt_service = MQTTService()
            logger.info("使用同步MQTT服务")
        # self.protocol_registry = get_protocol_registry()  # 已移除协议注册中心
        
        # Topic管理器
        self.topic_manager = TopicManager()
        
        # 系统级监听状态管理
        self.system_listen_topics: Dict[str, str] = {}  # topic -> handler_id
        self.custom_device_handlers: Dict[str, str] = {}  # device_id -> handler_id
        self.is_system_listening: bool = False

        # 设备消息队列和处理协程 - 按设备ID分流
        self.device_queues: Dict[str, asyncio.Queue] = {}  # device_id -> queue
        self.device_workers: Dict[str, asyncio.Task] = {}  # device_id -> worker_task
        self.max_queue_size_per_device = 1000  # 每个设备的队列容量（可配置）
        self.device_queue_stats: Dict[str, Dict[str, Any]] = {}  # 队列统计信息

    def set_max_queue_size(self, max_size: int):
        """设置每个设备的最大队列容量

        Args:
            max_size: 最大队列容量，建议500-2000
        """
        if max_size < 10:
            logger.warning("队列大小太小，建议至少设置为10")
            max_size = 10
        elif max_size > 10000:
            logger.warning("队列大小太大，建议不超过10000")
            max_size = 10000

        self.max_queue_size_per_device = max_size
        logger.info(f"设备队列最大容量已设置为: {max_size}")

    def register_device_handler(self, config: DeviceHandlerConfig, processor: Callable):
        """注册并运行设备处理器"""
        device_id = config.device_id
        
        # 标记为自定义设备处理器（从系统监听中排除）
        self.custom_device_handlers[device_id] = config.script_id
        
        # 注册设备配置
        self.device_handlers[device_id] = config
        self.message_processors[device_id] = processor
        
        # 建立topic映射
        for topic in config.listen_topics:
            if topic not in self.topic_mappings:
                self.topic_mappings[topic] = []
            if device_id not in self.topic_mappings[topic]:
                self.topic_mappings[topic].append(device_id)
        
        # 初始化统计信息
        self.processing_stats[device_id] = {
            "total_messages": 0,
            "success_count": 0,
            "error_count": 0,
            "last_processed": None,
            "average_processing_time_ms": None,
            "registered_at": datetime.now().isoformat(),
            "status": "running"
        }
        
        # 标记设备为活跃
        self.active_devices[device_id] = {
            "config": config,
            "status": "running", 
            "last_heartbeat": time.time(),
            "message_count": 0
        }
        
        brand_info = f" 品牌: {config.device_brand.value}" if config.device_brand else ""
        # protocol_info = f" 协议: {config.protocol_key}" if config.protocol_key else ""  # 已移除协议信息
        
        logger.info(f"注册并运行设备处理器: {device_id} ({config.device_type.value}{brand_info})")
        logger.info(f"监听topics: {config.listen_topics}")
        logger.info(f"转发topics: {config.forward_topics}")
        logger.info("消息透传模式: 已启用")
        logger.info(f"设备 {device_id} 已从系统级监听中移除，由自定义处理器接管")
        
        # 如果MQTT服务已连接，立即订阅topics
        if self.mqtt_service.connected:
            # 过滤掉已经被系统级监听覆盖的topics，避免重复订阅
            topics_to_subscribe = self._filter_duplicate_topics(config.listen_topics, config.device_brand.value)
            
            if topics_to_subscribe:
                if self.use_async_mqtt:
                    # 异步MQTT需要在异步上下文中订阅
                    asyncio.create_task(self._subscribe_topics_async(topics_to_subscribe))
                else:
                    self._subscribe_topics(topics_to_subscribe)
                logger.info(f"设备 {device_id} 订阅了 {len(topics_to_subscribe)} 个新topics")
            else:
                logger.info(f"设备 {device_id} 的所有topics已被系统级监听覆盖，无需重复订阅")
        
    def unregister_device_handler(self, device_id: str):
        """停止并注销设备处理器"""
        if device_id in self.device_handlers:
            config = self.device_handlers[device_id]
            
            # 只在MQTT连接时才尝试取消订阅topics（_unsubscribe_topics已内置连接检查）
            self._unsubscribe_topics(config.listen_topics, device_id)
            
            # 清理topic映射
            for topic in config.listen_topics:
                if topic in self.topic_mappings:
                    if device_id in self.topic_mappings[topic]:
                        self.topic_mappings[topic].remove(device_id)
                    if not self.topic_mappings[topic]:
                        del self.topic_mappings[topic]
            
            # 从自定义设备列表中移除（恢复系统级监听）
            if device_id in self.custom_device_handlers:
                del self.custom_device_handlers[device_id]
                logger.info(f"设备 {device_id} 已恢复系统级监听")
            
            # 清理设备队列和worker协程
            if device_id in self.device_workers:
                worker_task = self.device_workers[device_id]
                if not worker_task.done():
                    worker_task.cancel()
                del self.device_workers[device_id]
                logger.info(f"已取消设备 {device_id} 的worker协程")

            if device_id in self.device_queues:
                del self.device_queues[device_id]
                logger.info(f"已清理设备 {device_id} 的消息队列")

            if device_id in self.device_queue_stats:
                del self.device_queue_stats[device_id]

            # 清理注册信息
            del self.device_handlers[device_id]
            del self.message_processors[device_id]
            if device_id in self.active_devices:
                del self.active_devices[device_id]
            if device_id in self.processing_stats:
                del self.processing_stats[device_id]

            logger.info(f"停止并注销设备处理器: {device_id}")
    
    def _subscribe_topics(self, topics: List[str]):
        """订阅MQTT topics（同步版本）"""
        for topic in topics:
            # 检查是否已经为该topic添加了我们的消息处理器
            topic_handlers = getattr(self.mqtt_service, 'message_handlers', {}).get(topic, [])
            sync_handler = self._handle_mqtt_message
            
            # 只有当处理器不在列表中时才添加
            if sync_handler not in topic_handlers:
                self.mqtt_service.add_message_handler(topic, sync_handler)
            self.mqtt_service.subscribe(topic)
    
    async def _subscribe_topics_async(self, topics: List[str]):
        """订阅MQTT topics（异步版本）"""
        for topic in topics:
            # 检查是否已经为该topic添加了我们的消息处理器
            topic_handlers = getattr(self.mqtt_service, 'message_handlers', {}).get(topic, [])
            async_handler = self._handle_mqtt_message_async
            
            # 只有当处理器不在列表中时才添加
            if async_handler not in topic_handlers:
                self.mqtt_service.add_message_handler(topic, async_handler)
            await self.mqtt_service.subscribe(topic)
    
    def _filter_duplicate_topics(self, topics: List[str], brand_name: str) -> List[str]:
        """过滤掉已经被系统级监听覆盖的topics
        
        Args:
            topics: 要检查的topic列表
            brand_name: 设备品牌名称
            
        Returns:
            需要订阅的topic列表（排除已被系统级监听覆盖的）
        """
        if not self.is_system_listening:
            # 如果系统级监听未启动，则所有topics都需要订阅
            return topics
        
        filtered_topics = []
        
        for topic in topics:
            # 检查是否已被系统级监听覆盖
            is_covered_by_system = False
            
            # 检查当前topic是否被任何系统级通配符topic覆盖
            for system_topic in self.system_listen_topics.keys():
                if self._is_topic_covered_by_wildcard(topic, system_topic, brand_name):
                    is_covered_by_system = True
                    logger.debug(f"Topic {topic} 已被系统级监听 {system_topic} 覆盖")
                    break
            
            if not is_covered_by_system:
                filtered_topics.append(topic)
                logger.debug(f"Topic {topic} 需要单独订阅")
        
        return filtered_topics
    
    def _is_topic_covered_by_wildcard(self, specific_topic: str, wildcard_topic: str, brand_name: str) -> bool:
        """检查特定topic是否被通配符topic覆盖
        
        Args:
            specific_topic: 具体的topic，如 "oagv/v2/asbm2_IRAYPLE/AGV001/order"
            wildcard_topic: 通配符topic，如 "oagv/v2/asbm2_IRAYPLE/+/order"
            brand_name: 品牌名称
            
        Returns:
            是否被覆盖
        """
        specific_parts = specific_topic.split('/')
        wildcard_parts = wildcard_topic.split('/')
        
        # 长度必须相同
        if len(specific_parts) != len(wildcard_parts):
            return False
        
        # 检查每个部分是否匹配
        for i, (specific_part, wildcard_part) in enumerate(zip(specific_parts, wildcard_parts)):
            if wildcard_part == '+':
                # 通配符匹配任意值
                continue
            elif specific_part != wildcard_part:
                # 非通配符部分必须完全匹配
                return False
        
        # 额外检查：确保品牌匹配
        # 从topic中提取品牌信息进行验证
        if len(specific_parts) >= 3:
            topic_brand_instance = specific_parts[2]

            # 检查品牌是否匹配（使用新的品牌配置获取方法）
            brand_config = self.topic_manager.get_brand_config(brand_name)
            if brand_config:
                brand_suffix = brand_config["brand_suffix"]

                # 对于oagv格式，检查实例+品牌后缀是否匹配
                if specific_parts[0] == "oagv":
                    expected_instance_brand = f"{self.topic_manager.instance_id}{brand_suffix}"
                    if topic_brand_instance == expected_instance_brand:
                        return True
                # 对于uagv格式，检查品牌名是否匹配
                elif specific_parts[0] == "uagv":
                    brand_name_only = brand_suffix[1:] if brand_suffix.startswith('_') else brand_suffix
                    if topic_brand_instance == brand_name_only:
                        return True

        return False

    def _unsubscribe_topics(self, topics: List[str], device_id: str):
        """取消订阅MQTT topics"""
        # 检查MQTT连接状态，避免在断开连接时产生警告
        if not self.mqtt_service.connected:
            logger.debug(f"MQTT未连接，跳过取消订阅: {topics}")
            return
            
        for topic in topics:
            # 检查是否还有其他设备在监听这个topic
            other_devices = [d for d in self.topic_mappings.get(topic, []) if d != device_id]
            if not other_devices:
                if self.use_async_mqtt:
                    # 异步MQTT需要创建任务来取消订阅
                    asyncio.create_task(self.mqtt_service.unsubscribe(topic))
                else:
                    self.mqtt_service.unsubscribe(topic)
    
    def get_handlers_for_topic(self, topic: str) -> List[str]:
        """获取处理指定topic的设备ID列表（支持通配符匹配）"""
        handlers = []
        
        # 首先尝试精确匹配
        exact_handlers = self.topic_mappings.get(topic, [])
        handlers.extend(exact_handlers)
        
        # 然后尝试通配符匹配
        topic_parts = topic.split('/')
        for registered_topic, topic_handlers in self.topic_mappings.items():
            # 跳过已经精确匹配的topic
            if registered_topic == topic:
                continue
                
            registered_parts = registered_topic.split('/')
            
            # 检查路径长度是否匹配
            if len(topic_parts) == len(registered_parts):
                # 检查每个部分是否匹配（支持+通配符）
                match = True
                for i, (topic_part, registered_part) in enumerate(zip(topic_parts, registered_parts)):
                    if registered_part != '+' and registered_part != topic_part:
                        match = False
                        break
                
                if match:
                    handlers.extend(topic_handlers)
        
        return handlers
    
    def get_device_config(self, device_id: str) -> Optional[DeviceHandlerConfig]:
        """获取设备配置"""
        return self.device_handlers.get(device_id)
    
    def get_device_processor(self, device_id: str) -> Optional[Callable]:
        """获取设备处理器"""
        return self.message_processors.get(device_id)

    def _extract_device_id_from_topic(self, topic: str) -> Optional[str]:
        """快速从topic中提取设备ID（优化版本）"""
        # oagv/v2/instance_BRAND/AGV001/state
        # uagv/v2.0.0/BRAND/AGV001/state
        parts = topic.split('/')

        if len(parts) >= 5:
            # 倒数第二个部分通常是设备ID
            device_id = parts[-2]
            # 过滤通配符
            if device_id != "+":
                return device_id

        return None

    async def _create_device_worker(self, device_id: str):
        """为设备创建独立的处理协程"""
        if device_id in self.device_workers and not self.device_workers[device_id].done():
            return  # 已存在

        # 创建队列
        self.device_queues[device_id] = asyncio.Queue(maxsize=self.max_queue_size_per_device)

        # 初始化统计信息
        self.device_queue_stats[device_id] = {
            "total_enqueued": 0,
            "total_processed": 0,
            "dropped_messages": 0,
            "current_queue_size": 0,
            "worker_started_at": datetime.now().isoformat()
        }

        # 创建处理协程
        self.device_workers[device_id] = asyncio.create_task(
            self._device_message_worker(device_id)
        )

        logger.info(f"为设备 {device_id} 创建了独立消息处理协程")

    async def _device_message_worker(self, device_id: str):
        """设备消息处理协程 - 独立处理该设备的所有消息"""
        logger.info(f"设备 {device_id} 消息处理协程已启动")
        queue = self.device_queues[device_id]

        while True:
            try:
                # 从队列取出消息
                message_data = await queue.get()

                try:
                    # 处理单条消息
                    await self._process_single_device_message(
                        device_id=device_id,
                        topic=message_data["topic"],
                        payload=message_data["payload"]
                    )

                    # 更新统计
                    if device_id in self.device_queue_stats:
                        self.device_queue_stats[device_id]["total_processed"] += 1
                        self.device_queue_stats[device_id]["current_queue_size"] = queue.qsize()

                except Exception as e:
                    logger.error(f"处理设备 {device_id} 消息异常: {e}", exc_info=True)
                finally:
                    queue.task_done()

            except asyncio.CancelledError:
                logger.info(f"设备 {device_id} 消息处理协程被取消")
                break
            except Exception as e:
                logger.error(f"设备 {device_id} 处理协程异常: {e}", exc_info=True)
                await asyncio.sleep(0.1)  # 避免异常风暴

    async def _process_single_device_message(self, device_id: str, topic: str,
                                            payload: Union[str, bytes, Dict[str, Any]]):
        """处理单个设备的单条消息（原来的handle_mqtt_message逻辑）"""
        try:
            # 解析payload
            if isinstance(payload, (str, bytes)):
                payload_data = json.loads(payload)
            else:
                payload_data = payload

            # 获取处理该topic的设备列表
            handler_ids = self.get_handlers_for_topic(topic)
            logger.debug(f"处理topic {topic} 的handler_ids: {handler_ids}")

            # 分离自定义处理器和系统级处理器
            custom_handlers = [hid for hid in handler_ids if not hid.startswith("system_")]
            system_handlers = [hid for hid in handler_ids if hid.startswith("system_")]

            # 优先处理自定义处理器
            if custom_handlers:
                logger.debug(f"使用自定义处理器: {custom_handlers}")
                for handler_id in custom_handlers:
                    config = self.get_device_config(handler_id)
                    if config and config.enabled:
                        # 创建设备消息
                        message = DeviceMessage(
                            device_id=handler_id,
                            device_type=config.device_type,
                            topic=topic,
                            payload=payload_data,
                            timestamp=time.time(),
                            source_topic=topic,
                            target_topics=config.forward_topics
                        )

                        # 处理消息
                        await self._process_device_message(message)

            # 如果没有自定义处理器，使用系统级处理器
            elif system_handlers:
                logger.debug(f"使用系统级处理器: {system_handlers}")
                for handler_id in system_handlers:
                    # 系统级处理器：从topic中提取设备ID
                    extracted_device_id = self.topic_manager.extract_vehicle_id_from_topic(topic)
                    if extracted_device_id:
                        # 根据收到的topic自动生成转发topic
                        forward_topic = self._generate_forward_topic_from_listen_topic(topic)
                        target_topics = [forward_topic] if forward_topic else []

                        # 创建系统级设备消息
                        message = DeviceMessage(
                            device_id=extracted_device_id,
                            device_type=DeviceType.VEHICLE,  # 系统级默认为vehicle
                            topic=topic,
                            payload=payload_data,
                            timestamp=time.time(),
                            source_topic=topic,
                            target_topics=target_topics
                        )

                        # 使用系统级处理器处理
                        processor = self.message_processors.get(handler_id)
                        if processor:
                            await self._process_system_message(handler_id, message, processor)

        except json.JSONDecodeError as e:
            logger.error(f"MQTT消息JSON解析失败: {e}, topic: {topic}, payload: {payload}")
        except Exception as e:
            logger.error(f"处理设备消息异常: device_id={device_id}, topic={topic}, error={e}",
                       exc_info=True)
    
    def _handle_mqtt_message(self, topic: str, payload: str):
        """处理MQTT消息"""
        try:
            # 获取当前事件循环，如果没有则创建新的
            try:
                loop = asyncio.get_running_loop()
                # 在当前循环中创建任务
                asyncio.create_task(self.handle_mqtt_message(topic, payload))
            except RuntimeError:
                # 没有运行中的事件循环，使用线程安全的方式
                def run_async():
                    # 在新线程中运行异步处理
                    asyncio.run(self.handle_mqtt_message(topic, payload))
                
                # 创建并启动线程
                thread = threading.Thread(target=run_async, daemon=True)
                thread.start()
        except Exception as e:
            logger.error(f"处理MQTT消息异常: {e}", exc_info=True)
    
    async def _handle_mqtt_message_async(self, topic: str, payload: str):
        """处理MQTT消息（异步版本）"""
        await self.handle_mqtt_message(topic, payload)
    
    async def handle_mqtt_message(self, topic: str, payload: Union[str, bytes, Dict[str, Any]]):
        """处理MQTT消息 - 快速分发到设备队列"""
        logger.debug(f"收到MQTT消息: topic={topic}")

        try:
            # 快速提取设备ID（不做复杂处理）
            device_id = self._extract_device_id_from_topic(topic)
            if not device_id:
                logger.warning(f"无法从topic中提取设备ID: {topic}")
                return

            # 确保该设备有处理队列
            if device_id not in self.device_queues:
                await self._create_device_worker(device_id)

            # 快速入队，不阻塞（使用put_nowait避免await）
            try:
                message_data = {
                    "topic": topic,
                    "payload": payload,
                    "timestamp": time.time()
                }
                self.device_queues[device_id].put_nowait(message_data)

                # 更新队列统计
                if device_id in self.device_queue_stats:
                    self.device_queue_stats[device_id]["total_enqueued"] += 1
                    self.device_queue_stats[device_id]["current_queue_size"] = self.device_queues[device_id].qsize()

            except asyncio.QueueFull:
                logger.warning(f"设备 {device_id} 消息队列已满({self.max_queue_size_per_device})，丢弃消息")
                # 记录丢弃统计
                if device_id in self.device_queue_stats:
                    self.device_queue_stats[device_id]["dropped_messages"] += 1

        except Exception as e:
            logger.error(f"分发消息异常: topic={topic}, error={e}", exc_info=True)

    async def _process_system_message(self, handler_id: str, message: DeviceMessage, processor: Callable):
        """处理系统级消息"""
        try:
            # 执行系统级处理器
            if asyncio.iscoroutinefunction(processor):
                result = await processor(message)
            else:
                result = processor(message)
            
            # 处理转发逻辑
            if result and isinstance(result, dict):
                await self._handle_system_message_result(message, result)
                
        except Exception as e:
            logger.error(f"系统级消息处理失败: handler_id={handler_id}, device_id={message.device_id}, 错误: {e}", exc_info=True)
    
    async def _handle_system_message_result(self, original_message: DeviceMessage, result: Dict[str, Any]):
        """处理系统级消息处理结果"""
        try:
            # 检查是否需要转发
            if result.get("forward", True):
                # 确定转发的payload
                forward_payload = result.get("payload", original_message.payload)
                
                # 确定转发的topic
                response_topic = result.get("response_topic")
                
                if response_topic:
                    self._publish_mqtt_message(response_topic, forward_payload)
                    logger.debug(f"系统级消息已转发: {original_message.device_id} -> {response_topic}")
                    
        except Exception as e:
            logger.error(f"处理系统级消息结果失败: {e}", exc_info=True)

    async def _process_device_message(self, message: DeviceMessage):
        """处理单个设备消息"""
        device_id = message.device_id
        config = self.device_handlers.get(device_id)
        processor = self.message_processors.get(device_id)
        
        if not config or not processor:
            logger.warning(f"未找到设备处理器: {device_id}")
            return
        
        if not config.enabled:
            logger.debug(f"设备处理器已禁用: {device_id}")
            return
        
        start_time = time.time()
        try:
            # 更新统计信息
            stats = self.processing_stats[device_id]
            stats["total_messages"] += 1
            
            # 执行处理器
            if asyncio.iscoroutinefunction(processor):
                result = await processor(message)
            else:
                result = processor(message)
            # 处理成功
            stats["success_count"] += 1
            stats["last_processed"] = datetime.now().isoformat()
            message.processed = True
            
            # 更新设备心跳
            if device_id in self.active_devices:
                self.active_devices[device_id]["last_heartbeat"] = time.time()
                self.active_devices[device_id]["message_count"] += 1
            
            # 计算平均处理时间
            processing_time_ms = int((time.time() - start_time) * 1000)
            if stats["average_processing_time_ms"] is None:
                stats["average_processing_time_ms"] = processing_time_ms
            else:
                total_time = (stats["average_processing_time_ms"] * 
                             (stats["success_count"] - 1) + processing_time_ms)
                stats["average_processing_time_ms"] = int(total_time / stats["success_count"])
            
            # 处理转发逻辑
            if result and isinstance(result, dict):
                await self._handle_message_result(message, result)
            
            logger.debug(f"设备消息处理成功: {device_id}, 耗时: {processing_time_ms}ms")
            
        except Exception as e:
            # 处理失败
            stats["error_count"] += 1
            logger.error(f"设备消息处理失败: {device_id}, 错误: {e}", exc_info=True)
    
    async def _handle_message_result(self, original_message: DeviceMessage, result: Dict[str, Any]):
        """处理消息处理结果

        用户自定义处理函数返回值格式约束：
        {
            "forward": bool,              # 是否转发消息（可选，默认True）
            "payload": dict,              # 要转发的消息载荷（可选，默认使用原始消息）
            "response_topic": str,        # 自定义响应topic（可选）
            "response": dict              # 自定义响应消息载荷（可选，需配合response_topic）
        }

        示例1 - 直接透传：
        return {"forward": True}  # 使用原始消息转发到默认topics

        示例2 - 修改后转发：
        return {
            "forward": True,
            "payload": {"status": "processed", "data": modified_data}
        }

        示例3 - 不转发：
        return {"forward": False}

        示例4 - 发送自定义响应：
        return {
            "forward": False,
            "response_topic": "custom/response/topic",
            "response": {"result": "success"}
        }
        """
        try:
            config = self.device_handlers.get(original_message.device_id)
            if not config:
                logger.warning(f"未找到设备配置: {original_message.device_id}")
                return

            # 验证返回值格式
            if not isinstance(result, dict):
                logger.error(f"处理器返回值格式错误，必须是字典类型，实际类型: {type(result)}")
                return

            # 检查是否需要转发
            should_forward = result.get("forward", True)
            if should_forward:
                # 确定转发的payload
                forward_payload = result.get("payload", original_message.payload)

                # 验证payload格式
                if not isinstance(forward_payload, dict):
                    logger.error(f"转发payload格式错误，必须是字典类型，实际类型: {type(forward_payload)}")
                    return

                # 确定转发的topics
                forward_topics = config.forward_topics

                if isinstance(forward_topics, str):
                    forward_topics = [forward_topics]

                # 转发到目标topics
                if forward_topics:
                    for topic in forward_topics:
                        self._publish_mqtt_message(topic, forward_payload)
                        logger.debug(f"消息已转发: {original_message.device_id} -> {topic}")
                        logger.debug(f"转发载荷: {forward_payload}")
                else:
                    logger.warning(f"未配置转发topics，设备: {original_message.device_id}")

            # 处理自定义响应
            if "response" in result:
                response_topic = result.get("response_topic")
                response_payload = result["response"]

                # 验证响应格式
                if not response_topic:
                    logger.error("提供了response但未指定response_topic")
                    return

                if not isinstance(response_payload, dict):
                    logger.error(f"响应payload格式错误，必须是字典类型，实际类型: {type(response_payload)}")
                    return

                self._publish_mqtt_message(response_topic, response_payload)
                logger.debug(f"响应已发送: {original_message.device_id} -> {response_topic}")
                logger.debug(f"响应载荷: {response_payload}")

        except Exception as e:
            logger.error(f"处理消息结果失败: {e}", exc_info=True)
    
    
    def _publish_mqtt_message(self, topic: str, payload: Any):
        """发布MQTT消息"""
        if self.use_async_mqtt:
            # 对于异步MQTT，创建异步任务
            asyncio.create_task(self.mqtt_service.publish(topic, payload))
        else:
            self.mqtt_service.publish(topic, payload)
    
    async def start_mqtt_service(self):
        """启动内部MQTT服务"""
        try:
            if self.use_async_mqtt:
                success = await self.mqtt_service.connect()
            else:
                success = self.mqtt_service.connect()
                
            if success:
                logger.info("内部MQTT服务已启动")
                
                # 启动系统级监听所有设备
                await self.start_system_listening()
                
                # 为已注册的设备订阅topics（只订阅，不重复添加消息处理器）
                all_topics = set()
                for config in self.device_handlers.values():
                    all_topics.update(config.listen_topics)
                
                if all_topics:
                    # 直接订阅topic，不添加消息处理器（消息处理器在注册时已添加）
                    for topic in all_topics:
                        if self.use_async_mqtt:
                            await self.mqtt_service.subscribe(topic)
                        else:
                            self.mqtt_service.subscribe(topic)
                return True
            else:
                logger.error("启动内部MQTT服务失败")
                return False
        except Exception as e:
            logger.error(f"启动MQTT服务异常: {e}", exc_info=True)
            return False
    
    async def stop_mqtt_service(self):
        """停止内部MQTT服务"""
        try:
            # 先停止系统级监听
            await self.stop_system_listening()
            
            if self.use_async_mqtt:
                await self.mqtt_service.disconnect()
            else:
                self.mqtt_service.disconnect()
            logger.info("内部MQTT服务已停止")
        except Exception as e:
            logger.error(f"停止MQTT服务异常: {e}", exc_info=True)
    
    def get_device_statistics(self, device_id: str = None) -> Dict[str, Any]:
        """获取设备统计信息"""
        if device_id:
            stats = self.processing_stats.get(device_id, {}).copy()
            if device_id in self.active_devices:
                stats.update({
                    "device_status": self.active_devices[device_id]["status"],
                    "last_heartbeat": self.active_devices[device_id]["last_heartbeat"],
                    "message_count": self.active_devices[device_id]["message_count"]
                })

            # 添加队列统计信息
            if device_id in self.device_queue_stats:
                stats["queue_stats"] = self.device_queue_stats[device_id].copy()
                # 添加实时队列大小
                if device_id in self.device_queues:
                    stats["queue_stats"]["current_queue_size"] = self.device_queues[device_id].qsize()

            return stats
        else:
            # 计算总队列统计
            total_enqueued = sum(s.get("total_enqueued", 0) for s in self.device_queue_stats.values())
            total_processed = sum(s.get("total_processed", 0) for s in self.device_queue_stats.values())
            total_dropped = sum(s.get("dropped_messages", 0) for s in self.device_queue_stats.values())
            total_queue_size = sum(q.qsize() for q in self.device_queues.values())

            return {
                "device_count": len(self.device_handlers),
                "running_devices": len([d for d in self.active_devices.values()
                                       if d["status"] == "running"]),
                "total_topics": len(self.topic_mappings),
                "queue_summary": {
                    "total_workers": len(self.device_workers),
                    "total_queues": len(self.device_queues),
                    "total_enqueued": total_enqueued,
                    "total_processed": total_processed,
                    "total_dropped": total_dropped,
                    "current_total_queue_size": total_queue_size,
                    "max_queue_size_per_device": self.max_queue_size_per_device
                },
                "devices": {device_id: self.get_device_statistics(device_id)
                           for device_id in self.device_handlers.keys()}
            }
    
    def clear_script_device_registrations(self, script_id: str):
        """清理指定脚本的所有设备注册"""
        devices_to_remove = []
        for device_id, config in self.device_handlers.items():
            if config.script_id == script_id:
                devices_to_remove.append(device_id)
        
        for device_id in devices_to_remove:
            self.unregister_device_handler(device_id)
            
        logger.info(f"已清理脚本 {script_id} 的 {len(devices_to_remove)} 个设备注册")
    
    async def start_system_listening(self):
        """启动系统级监听所有设备"""
        if self.is_system_listening:
            logger.info("系统监听已经在运行中")
            return
        
        logger.info("启动系统级设备监听...")
        
        # 生成监听所有设备的topics（使用通配符+）
        system_topics = self._generate_system_listen_topics()
        # 创建系统级透传处理器
        system_handler = self._create_system_forward_handler()
        
        # 为每个topic创建系统级处理器
        for topic in system_topics:
            handler_id = f"system_{topic.replace('/', '_').replace('+', 'all')}"
            # 注册系统级处理器
            self.system_listen_topics[topic] = handler_id
            self.message_processors[handler_id] = system_handler
            # 添加topic映射
            if topic not in self.topic_mappings:
                self.topic_mappings[topic] = []
            self.topic_mappings[topic].append(handler_id)
            
            # 订阅topic（只有在第一次添加时才添加消息处理器）
            if self.mqtt_service.connected:
                # 检查是否已经为该topic添加了我们的消息处理器
                topic_handlers = getattr(self.mqtt_service, 'message_handlers', {}).get(topic, [])
                async_handler = self._handle_mqtt_message_async
                sync_handler = self._handle_mqtt_message
                
                if self.use_async_mqtt:
                    await self.mqtt_service.subscribe(topic)
                    # 只有当处理器不在列表中时才添加
                    if async_handler not in topic_handlers:
                        self.mqtt_service.add_message_handler(topic, async_handler)
                else:
                    self.mqtt_service.subscribe(topic)
                    # 只有当处理器不在列表中时才添加
                    if sync_handler not in topic_handlers:
                        self.mqtt_service.add_message_handler(topic, sync_handler)
        
        self.is_system_listening = True
        logger.info(f"系统级监听已启动，监听topics: {system_topics}")
    
    def _generate_system_listen_topics(self) -> List[str]:
        """生成系统级监听所有设备的topics（包括vehicle和other设备）"""
        topics = []

        # 获取默认监听的指令类型
        default_vehicle_commands = self.topic_manager.topic_config["default_command_types"]["vehicle"]
        default_other_commands = self.topic_manager.topic_config["default_command_types"]["other"]

        # 1. 为vehicle类型设备生成监听topics
        logger.info("生成vehicle设备系统监听topics...")
        vehicle_brands = self.topic_manager.get_supported_brands(category="vehicle")
        for brand_name in vehicle_brands:
            brand_config = self.topic_manager.get_brand_config(brand_name)
            if not brand_config:
                continue

            brand_suffix = brand_config["brand_suffix"]
            oagv_version = brand_config["oagv_version"]
            uagv_version = brand_config["uagv_version"]

            for command_type, direction_config in self.topic_manager.command_directions.items():
                # 只生成vehicle默认监听列表中的指令类型
                if command_type not in default_vehicle_commands:
                    logger.debug(f"[Vehicle] 跳过非默认监听指令类型: {command_type}")
                    continue

                listen_queue = direction_config["listen"]

                if listen_queue == "oagv":
                    # oagv格式: oagv/v2/{instanceId}_BRAND/+/command
                    topic = f"oagv/{oagv_version}/{self.topic_manager.instance_id}{brand_suffix}/+/{command_type}"
                    topics.append(topic)
                    logger.debug(f"[Vehicle] 添加监听topic: {topic}")
                elif listen_queue == "uagv":
                    # uagv格式: uagv/v2.0.0/BRAND/+/command
                    brand_name_only = brand_suffix[1:]  # 去掉前缀下划线
                    topic = f"uagv/{uagv_version}/{brand_name_only}/+/{command_type}"
                    topics.append(topic)
                    logger.debug(f"[Vehicle] 添加监听topic: {topic}")

        # 2. 为other类型设备生成监听topics
        logger.info("生成other设备系统监听topics...")
        other_brands = self.topic_manager.get_supported_brands(category="other")
        for brand_name in other_brands:
            brand_config = self.topic_manager.get_brand_config(brand_name)
            if not brand_config:
                continue

            brand_suffix = brand_config["brand_suffix"]
            oagv_version = brand_config["oagv_version"]
            uagv_version = brand_config["uagv_version"]

            for command_type, direction_config in self.topic_manager.command_directions.items():
                # 只生成other默认监听列表中的指令类型
                if command_type not in default_other_commands:
                    logger.debug(f"[Other] 跳过非默认监听指令类型: {command_type}")
                    continue

                listen_queue = direction_config["listen"]

                if listen_queue == "oagv":
                    # oagv格式: oagv/v2/{instanceId}_BRAND/+/command
                    topic = f"oagv/{oagv_version}/{self.topic_manager.instance_id}{brand_suffix}/+/{command_type}"
                    topics.append(topic)
                    logger.debug(f"[Other] 添加监听topic: {topic}")
                elif listen_queue == "uagv":
                    # uagv格式: uagv/v2.0.0/BRAND/+/command
                    brand_name_only = brand_suffix[1:]  # 去掉前缀下划线
                    topic = f"uagv/{uagv_version}/{brand_name_only}/+/{command_type}"
                    topics.append(topic)
                    logger.debug(f"[Other] 添加监听topic: {topic}")

        logger.info(f"系统监听topics生成完成，共 {len(topics)} 个")
        return topics
    
    def _create_system_forward_handler(self) -> Callable:
        """创建系统级透传处理器"""
        async def system_forward_handler(message: DeviceMessage) -> Dict[str, Any]:
            """系统级透传处理器：直接转发消息，不做任何处理"""
            # 提取设备ID
            device_id = self.topic_manager.extract_vehicle_id_from_topic(message.topic)
            if not device_id:
                logger.warning(f"无法从topic中提取设备ID: {message.topic}")
                return {"forward": False}
            
            # 检查是否为自定义处理的设备 - 关键修复：排除自定义设备
            if device_id in self.custom_device_handlers:
                logger.debug(f"设备 {device_id} 由自定义处理器处理，跳过系统转发")
                return {"forward": False}
            
            # 解析topic信息以确定转发目标
            forward_topic = self._generate_forward_topic_from_listen_topic(message.topic)
            if not forward_topic:
                logger.warning(f"无法生成转发topic: {message.topic}")
                return {"forward": False}
            
            logger.info(f"系统透传转发: {message.topic} -> {forward_topic}, 设备: {device_id}")
            
            # 直接透传消息
            return {
                "forward": True,
                "payload": message.payload,
                "response_topic": forward_topic
            }
        
        return system_forward_handler
    
    def _generate_forward_topic_from_listen_topic(self, listen_topic: str) -> Optional[str]:
        """根据监听topic生成对应的转发topic"""
        # logger.debug(f"开始解析转发topic: {listen_topic}")
        
        parts = listen_topic.split('/')
        if len(parts) < 5:
            logger.warning(f"Topic格式不正确，部分数量: {len(parts)}, topic: {listen_topic}")
            return None
        
        queue_type = parts[0]  # oagv 或 uagv
        version = parts[1]
        brand_instance = parts[2]
        device_id = parts[3]
        command_type = parts[4]
        
        # logger.debug(f"解析结果: queue_type={queue_type}, version={version}, brand_instance={brand_instance}, device_id={device_id}, command_type={command_type}")
        
        # 如果device_id是通配符，无法转发
        if device_id == "+":
            logger.debug("设备ID为通配符，无法转发")
            return None
        
        # 查找指令的转发方向
        if command_type not in self.topic_manager.command_directions:
            logger.warning(f"未知的指令类型: {command_type}")
            return None
        
        direction_config = self.topic_manager.command_directions[command_type]
        forward_queue = direction_config["forward"]
        
        # logger.debug(f"指令 {command_type} 的转发方向: {forward_queue}")
        
        # 解析品牌信息 - 改进品牌匹配逻辑（支持新的分类配置）
        brand_name = None
        # logger.debug(f"开始匹配品牌，当前brand_instance: {brand_instance}")

        # 遍历所有品牌（使用扁平化后的字典）
        for brand, config in self.topic_manager._flat_brands.items():
            brand_suffix = config["brand_suffix"]
            # logger.debug(f"检查品牌 {brand}: suffix={brand_suffix}")

            if queue_type == "oagv":
                # oagv格式: oagv/v2/{instanceId}_BRAND/{deviceId}/command
                # brand_instance 应该是 "asbm2_IRAYPLE" 这样的格式
                if brand_instance.endswith(brand_suffix):
                    brand_name = brand
                    # logger.debug(f"成功匹配oagv品牌: {brand}")
                    break
            elif queue_type == "uagv":
                # uagv格式: uagv/v2/BRAND/{deviceId}/command
                # brand_instance 应该是 "IRAYPLE" 这样的格式（不含下划线前缀）
                brand_name_only = brand_suffix[1:] if brand_suffix.startswith('_') else brand_suffix
                if brand_instance == brand_name_only:
                    brand_name = brand
                    # logger.debug(f"成功匹配uagv品牌: {brand}")
                    break

        if not brand_name:
            logger.warning(f"无法识别品牌信息: queue_type={queue_type}, brand_instance={brand_instance}")
            logger.debug(f"可用的品牌配置: {list(self.topic_manager._flat_brands.keys())}")
            return None

        brand_config = self.topic_manager.get_brand_config(brand_name)
        if not brand_config:
            logger.error(f"无法获取品牌配置: {brand_name}")
            return None
        # logger.debug(f"使用品牌配置: {brand_name} -> {brand_config}")
        
        # 生成转发topic
        forward_topic = None
        if forward_queue == "uagv":
            # 转发到uagv
            uagv_version = brand_config["uagv_version"]
            brand_name_only = brand_config["brand_suffix"][1:] if brand_config["brand_suffix"].startswith('_') else brand_config["brand_suffix"]
            forward_topic = f"uagv/{uagv_version}/{brand_name_only}/{device_id}/{command_type}"
            # logger.debug(f"生成uagv转发topic: {forward_topic}")
        elif forward_queue == "oagv":
            # 转发到oagv
            oagv_version = brand_config["oagv_version"]
            brand_suffix = brand_config["brand_suffix"]
            forward_topic = f"oagv/{oagv_version}/{self.topic_manager.instance_id}{brand_suffix}/{device_id}/{command_type}"
            # logger.debug(f"生成oagv转发topic: {forward_topic}")
        else:
            logger.warning(f"未知的转发队列类型: {forward_queue}")
        
        # logger.info(f"转发topic生成完成: {listen_topic} -> {forward_topic}")
        return forward_topic
    
    async def stop_system_listening(self):
        """停止系统级监听"""
        if not self.is_system_listening:
            return
        
        logger.info("停止系统级设备监听...")
        
        # 取消订阅所有系统级topics
        for topic, handler_id in self.system_listen_topics.items():
            if self.mqtt_service.connected:
                if self.use_async_mqtt:
                    await self.mqtt_service.unsubscribe(topic)
                else:
                    self.mqtt_service.unsubscribe(topic)
            
            # 清理映射
            if topic in self.topic_mappings:
                if handler_id in self.topic_mappings[topic]:
                    self.topic_mappings[topic].remove(handler_id)
                if not self.topic_mappings[topic]:
                    del self.topic_mappings[topic]
            
            # 清理处理器
            if handler_id in self.message_processors:
                del self.message_processors[handler_id]
        
        self.system_listen_topics.clear()
        self.is_system_listening = False
        logger.info("系统级监听已停止")
    
    def get_system_listening_status(self) -> Dict[str, Any]:
        """获取系统级监听状态"""
        return {
            "is_listening": self.is_system_listening,
            "system_topics": list(self.system_listen_topics.keys()),
            "custom_devices": list(self.custom_device_handlers.keys()),
            "excluded_devices_count": len(self.custom_device_handlers)
        }


# 全局设备处理器注册中心
_global_device_registry = DeviceHandlerRegistry()


def get_device_registry() -> DeviceHandlerRegistry:
    """获取全局设备注册中心实例"""
    return _global_device_registry


class DeviceHandlerService:
    """设备处理器服务"""
    
    def __init__(self):
        self.registry = get_device_registry()
        
    async def start_service(self):
        """启动设备处理服务"""
        if await self.registry.start_mqtt_service():
            logger.info("设备处理服务已启动，MQTT服务已连接")
        else:
            logger.warning("设备处理服务已启动，但MQTT连接失败")
    
    async def stop_service(self):
        """停止设备处理服务"""
        # 先清理所有设备注册（这会取消订阅topics）
        device_ids = list(self.registry.device_handlers.keys())
        for device_id in device_ids:
            self.registry.unregister_device_handler(device_id)
        
        # 最后停止MQTT服务
        await self.registry.stop_mqtt_service()
        
        logger.info("设备处理服务已停止")

    
    def stop_handler(self, device_id: str):
        """停止指定设备处理器"""
        self.registry.unregister_device_handler(device_id)

    def set_device_queue_size(self, max_size: int):
        """设置设备消息队列的最大容量

        Args:
            max_size: 最大队列容量，建议500-2000
                     - 如果消息处理速度快，可以设置小一些(如500)
                     - 如果消息处理速度慢或消息频率高，可以设置大一些(如2000)
                     - 队列满了会自动丢弃新消息

        Example:
            device_service = get_device_service()
            device_service.set_device_queue_size(1500)  # 设置为1500
        """
        self.registry.set_max_queue_size(max_size)
    
    def get_running_handlers(self) -> Dict[str, Any]:
        """获取所有正在运行的设备处理器"""
        return self.registry.get_device_statistics()
    
    def get_handler_status(self, device_id: str) -> Dict[str, Any]:
        """获取设备处理器状态"""
        return self.registry.get_device_statistics(device_id)
    
    def get_system_listening_status(self) -> Dict[str, Any]:
        """获取系统级监听状态"""
        return self.registry.get_system_listening_status()
    
    async def publish_message(self, topic: str, payload: Any):
        """主动发布MQTT消息
        
        Args:
            topic: MQTT主题
            payload: 消息载荷
        """
        self.registry._publish_mqtt_message(topic, payload)
    
    def send_order(self, device_id: str, order_data: Dict[str, Any], topic: str = None) -> None:
        """发送订单消息
        
        Args:
            device_id: 设备ID
            order_data: 订单数据
            topic: 发布主题（可选，默认使用vda5050格式）
        """
        import asyncio
        
        if not topic:
            topic = f"vda5050/{device_id}/order"
        
        asyncio.create_task(
            self.publish_message(topic, order_data)
        )
    
    def send_instant_action(self, device_id: str, actions: Union[List[str], List[Dict[str, Any]]], 
                           topic: str = None) -> None:
        """发送即时动作消息
        
        Args:
            device_id: 设备ID
            actions: 动作列表
            topic: 发布主题（可选，默认使用vda5050格式）
        """
        import asyncio
        
        if not topic:
            topic = f"vda5050/{device_id}/instantActions"
        
        action_data = {"actions": actions}
        asyncio.create_task(
            self.publish_message(topic, action_data)
        )
    
    def request_state(self, device_id: str, topic: str = None) -> None:
        """请求设备状态
        
        Args:
            device_id: 设备ID
            topic: 发布主题（可选，默认使用vda5050格式）
        """
        import asyncio
        
        if not topic:
            topic = f"vda5050/{device_id}/state"
        
        state_request = {"request_type": "state"}
        asyncio.create_task(
            self.publish_message(topic, state_request)
        )
    
    def request_factsheet(self, device_id: str, topic: str = None) -> None:
        """请求设备信息
        
        Args:
            device_id: 设备ID
            topic: 发布主题（可选，默认使用vda5050格式）
        """
        import asyncio
        
        if not topic:
            topic = f"vda5050/{device_id}/factsheet"
        
        factsheet_request = {"request_type": "factsheet"}
        asyncio.create_task(
            self.publish_message(topic, factsheet_request)
        )
    
    # 保持兼容性的别名方法
    def send_vda5050_order(self, device_id: str, order_data: Dict[str, Any], topic: str = None) -> None:
        """发送VDA5050订单指令（兼容性方法）"""
        self.send_order(device_id, order_data, topic)
    
    def send_vda5050_instant_action(self, device_id: str, actions: List[Dict[str, Any]], topic: str = None) -> None:
        """发送VDA5050即时动作指令（兼容性方法）"""
        self.send_instant_action(device_id, actions, topic)
    
    def request_vda5050_state(self, device_id: str, topic: str = None) -> None:
        """请求VDA5050状态（兼容性方法）"""
        self.request_state(device_id, topic)
    
    @staticmethod
    def create_default_forward_handler():
        """创建默认的透传处理器"""
        def default_handler(message: DeviceMessage) -> Dict[str, Any]:
            """默认消息处理器：直接透传所有消息"""
            logger.info(f"默认处理器收到消息: device_id={message.device_id}, "
                       f"topic={message.topic}, payload大小={len(str(message.payload))}")
            
            # 透传所有消息到转发topics
            return {
                "forward": True,
                "payload": message.payload,
                "topics": message.target_topics
            }
        
        return default_handler
    


# 全局设备处理服务实例
_global_device_service = DeviceHandlerService()


def get_device_service() -> DeviceHandlerService:
    """获取全局设备处理服务实例"""
    return _global_device_service