# # -*- coding: utf-8 -*- """ Author； Tan Mingyan @Time : 2026/2/27 """ """ UDP数据协议分发处理器 功能：解析udp_server.recv_multi_udp_hex()返回的数据，按protocol字段分发到对应处理函数 支持协议：UART、SYNC、CAN、1553B、AD、OC、网络 """ import sys import io import os # 设置标准输出和标准错误的编码为UTF-8 # sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') # sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8') from loguru import logger from typing import List, Dict, Any from pathlib import Path # 获取当前文件的目录 current_dir = os.path.dirname(os.path.abspath(__file__)) # 获取上级目录（项目根目录） parent_dir = os.path.dirname(current_dir) # 将项目根目录添加到系统路径 sys.path.append(parent_dir) sys.path.append(str(Path(__file__).parent.parent.parent)) import midware.config.base_config as base_config #from midware.config.base_config import DEVICE_CONFIG_DICT, UDP_CONFIG # ==================== 协议类型常量（统一管理，避免硬编码）==================== # 与你要求的协议类型严格对应 PROTOCOL_UART = "UART" PROTOCOL_SYNC = "SYNC" PROTOCOL_CAN = "CAN" PROTOCOL_1553B = "1553B" PROTOCOL_AD = "AD" PROTOCOL_OC = "OC" PROTOCOL_NETWORK = "网络" # 支持的协议列表（用于合法性校验） SUPPORTED_PROTOCOLS = [ PROTOCOL_UART, PROTOCOL_SYNC, PROTOCOL_CAN, PROTOCOL_1553B, PROTOCOL_AD, PROTOCOL_OC, PROTOCOL_NETWORK ] # ==================== 各协议处理函数（预留接口，按需填充逻辑）==================== def handle_uart(data: Dict[str, Any]) -> None: """处理UART协议数据""" logger.info(f"[UART处理] 外设：{data.get('dev_name', '未知')} | 端口：{data.get('port', '未知')} | 数据长度：{len(data.get('raw_data', b''))}字节") # TODO: 补充UART数据解析、转发、故障注入等业务逻辑 pass def handle_sync(data: Dict[str, Any]) -> None: """处理SYNC（同步）协议数据""" logger.info(f"[SYNC处理] 外设：{data.get('dev_name', '未知')} | 端口：{data.get('port', '未知')} | 数据长度：{len(data.get('raw_data', b''))}字节") # TODO: 补充SYNC同步数据解析、时序控制等业务逻辑 pass def handle_can(data: Dict[str, Any]) -> None: """处理CAN协议数据""" logger.info(f"[CAN处理] 外设：{data.get('dev_name', '未知')} | 端口：{data.get('port', '未知')} | 数据长度：{len(data.get('raw_data', b''))}字节") # TODO: 补充CAN帧解析、总线转发等业务逻辑 pass def handle_1553b(data: Dict[str, Any]) -> None: """处理1553B协议数据""" logger.info(f"[1553B处理] 外设：{data.get('dev_name', '未知')} | 端口：{data.get('port', '未知')} | 数据长度：{len(data.get('raw_data', b''))}字节") # TODO: 补充1553B总线数据解析、RT/BC交互等业务逻辑 pass def handle_ad(data: Dict[str, Any]) -> None: """处理AD（模拟量输入）协议数据""" logger.info(f"[AD处理] 外设：{data.get('dev_name', '未知')} | 端口：{data.get('port', '未知')} | 数据长度：{len(data.get('raw_data', b''))}字节") # TODO: 补充AD采样值解析、校准、越限告警等业务逻辑 pass def handle_oc(data: Dict[str, Any]) -> None: """处理OC（数字量输出）协议数据""" logger.info(f"[OC处理] 外设：{data.get('dev_name', '未知')} | 端口：{data.get('port', '未知')} | 数据长度：{len(data.get('raw_data', b''))}字节") # TODO: 补充OC通道控制、状态反馈、故障模拟等业务逻辑 pass def handle_network(data: Dict[str, Any]) -> None: """处理网络协议数据""" logger.info(f"[网络处理] 外设：{data.get('dev_name', '未知')} | 端口：{data.get('port', '未知')} | 数据长度：{len(data.get('raw_data', b''))}字节") # TODO: 补充网络包转发、IP/TCP解析、网络故障注入等业务逻辑 pass def handle_unknown_protocol(data: Dict[str, Any]) -> None: """处理未知协议数据（容错兜底）""" unknown_protocol = data.get('protocol', '未知') logger.warning( f"[未知协议] 外设：{data.get('dev_name', '未知')} | 端口：{data.get('port', '未知')} | " f"协议类型：{unknown_protocol} | 支持的协议列表：{SUPPORTED_PROTOCOLS}" ) # ==================== 核心分发函数（对外暴露的唯一入口）==================== def dispatch_udp_data(recv_data: List[Dict[str, Any]]) -> None: """ 核心分发函数：遍历recv_data，按protocol分发到对应处理函数 :param recv_data: udp_server.recv_multi_udp_hex()返回的原始数据列表，每条数据格式： {"port": 端口号, "dev_name": 外设名, "protocol": 协议类型, "raw_data": 原始二进制数据} """ # 第一步：校验输入数据类型 if not isinstance(recv_data, list): logger.error("分发失败：输入的recv_data不是列表类型") return # 第二步：空数据直接返回（避免无意义日志） if not recv_data: return # 第三步：构建协议-处理函数映射表（核心：新增/修改协议只需改此表） protocol_handler_map = { PROTOCOL_UART: handle_uart, PROTOCOL_SYNC: handle_sync, PROTOCOL_CAN: handle_can, PROTOCOL_1553B: handle_1553b, PROTOCOL_AD: handle_ad, PROTOCOL_OC: handle_oc, PROTOCOL_NETWORK: handle_network } # 第四步：遍历每条数据，按协议分发处理 for single_data in recv_data: try: # 容错：确保protocol字段存在且为字符串 protocol = single_data.get("protocol", "").strip().upper() port = single_data["port"] dev_name = single_data["dev_name"] # protocol = data["protocol"] raw_data = single_data["raw_data"] #根据dev_name，获取内存基地址，偏移地址，发送缓存区大小，接收缓存区大小 #dev_info = DEVICE_CONFIG_DICT.get(dev_name, {}) if dev_name not in base_config.DEVICE_CONFIG_DICT: logger.error(f"UART外设不存在：{dev_name}") return False # 获取外设基地址、偏移地址 base_addr = base_config.DEVICE_CONFIG_DICT[device_name]["base_addr"] offset_addr = base_config.DEVICE_CONFIG_DICT[device_name]["offset_addr"] print(f"sysbus WriteDoubleWord {base_addr + offset_addr} {raw_data}") # 兼容协议名大小写（如"can"→"CAN"、"sync"→"SYNC"） if protocol in protocol_handler_map: # 调用对应处理函数 protocol_handler_map[protocol](single_data) else: # 未知协议调用兜底函数 handle_unknown_protocol(single_data) except Exception as e: # 单条数据处理异常不影响整体，记录错误日志 logger.error( f"[数据处理异常] 外设：{single_data.get('dev_name', '未知')} | 错误信息：{str(e)}", exc_info=True ) # ==================== 测试用例（验证分发逻辑是否正常）==================== def test_dispatcher(): """模拟udp_server.recv_multi_udp_hex()返回的数据，测试分发逻辑""" # 模拟接收数据 mock_recv_data = [ {"port": 8880, "dev_name": "Uart0", "protocol": "uart", "raw_data": b"UART_TEST_DATA"}, {"port": 8881, "dev_name": "Sync0", "protocol": "SYNC", "raw_data": b"SYNC_TEST_DATA"}, {"port": 8882, "dev_name": "Can1", "protocol": "Can", "raw_data": b"CAN_TEST_DATA"}, {"port": 8883, "dev_name": "1553B0", "protocol": "1553B", "raw_data": b"1553B_TEST_DATA"}, {"port": 8884, "dev_name": "AD0", "protocol": "ad", "raw_data": b"AD_TEST_DATA"}, {"port": 8885, "dev_name": "OC0", "protocol": "OC", "raw_data": b"OC_TEST_DATA"}, {"port": 8886, "dev_name": "Net0", "protocol": "网络", "raw_data": b"NETWORK_TEST_DATA"}, {"port": 8887, "dev_name": "Unknown0", "protocol": "SPI", "raw_data": b"UNKNOWN_TEST_DATA"}, ] # 执行分发 logger.info("开始测试协议分发逻辑...") dispatch_udp_data(mock_recv_data) logger.info("协议分发测试完成") if __name__ == "__main__": # 运行测试用例，验证分发逻辑 test_dispatcher()