191 lines
8.4 KiB
Python
191 lines
8.4 KiB
Python
# # -*- 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() |