100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
"""
|
||
Tan mingyan
|
||
2026/3/9
|
||
|
||
2. 具体驱动实现（以 1553B 为例，bus_1553b.py）
|
||
其他总线（CAN/UART/AD 等）按相同逻辑实现，仅需重写基类接口：
|
||
"""
|
||
# midware/drivers/bus_1553b.py
|
||
from .base_driver import BaseBusDriver
|
||
from typing import Optional, Dict, Any
|
||
from loguru import logger
|
||
import time
|
||
|
||
class Bus1553BDriver(BaseBusDriver):
|
||
"""1553B总线驱动实现（适配实际1553B板卡/仿真器）"""
|
||
def _init_config(self):
|
||
"""初始化1553B默认配置"""
|
||
# 默认配置（可从yaml文件加载）
|
||
self.config.setdefault("board_id", 0) # 板卡ID
|
||
self.config.setdefault("bc_address", 0x01) # BC地址
|
||
self.config.setdefault("rt_address", 0x02) # RT地址
|
||
self.config.setdefault("baudrate", 1000000) # 波特率1Mbps
|
||
logger.info(f"1553B驱动配置初始化完成：{self.config}")
|
||
|
||
def connect(self) -> bool:
|
||
"""连接1553B板卡/仿真器"""
|
||
try:
|
||
# 实际逻辑：调用1553B板卡SDK的连接接口
|
||
# 示例：self.board = SDK_1553B.connect(board_id=self.config["board_id"])
|
||
time.sleep(0.1) # 模拟连接耗时
|
||
self.is_connected = True
|
||
logger.info("1553B总线连接成功")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"1553B总线连接失败：{str(e)}")
|
||
self.is_connected = False
|
||
return False
|
||
|
||
def disconnect(self) -> bool:
|
||
"""断开1553B连接"""
|
||
try:
|
||
# 实际逻辑：调用SDK的断开接口
|
||
# self.board.disconnect()
|
||
self.is_connected = False
|
||
logger.info("1553B总线断开成功")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"1553B总线断开失败：{str(e)}")
|
||
return False
|
||
|
||
def send(self, data: bytes, **kwargs) -> bool:
|
||
"""发送1553B数据帧"""
|
||
if not self.is_connected:
|
||
logger.error("1553B驱动未连接，发送失败")
|
||
return False
|
||
|
||
try:
|
||
# 扩展参数：优先级、子地址等
|
||
priority = kwargs.get("priority", 1) # 1553B最高优先级
|
||
sub_address = kwargs.get("sub_address", 0x00)
|
||
|
||
# 实际逻辑：调用SDK发送1553B帧
|
||
logger.info(
|
||
f"1553B发送数据｜优先级：{priority}｜子地址：{sub_address}｜"
|
||
f"数据：{data.hex()}｜长度：{len(data)}字节"
|
||
)
|
||
# self.board.send_frame(
|
||
# bc_addr=self.config["bc_address"],
|
||
# rt_addr=self.config["rt_address"],
|
||
# sub_addr=sub_address,
|
||
# data=data,
|
||
# priority=priority
|
||
# )
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"1553B数据发送失败：{str(e)}")
|
||
return False
|
||
|
||
def recv(self, timeout: float = 0.1, **kwargs) -> Optional[bytes]:
|
||
"""接收1553B数据帧"""
|
||
if not self.is_connected:
|
||
logger.error("1553B驱动未连接，接收失败")
|
||
return None
|
||
|
||
try:
|
||
# 实际逻辑：调用SDK接收1553B帧
|
||
# frame = self.board.recv_frame(timeout=timeout)
|
||
# if frame:
|
||
# return frame.data
|
||
# return None
|
||
|
||
# 模拟接收（替换为真实逻辑）
|
||
time.sleep(0.001) # 模拟接收耗时
|
||
# mock_data = b"\x01\x02\x03\x04\x05" # 模拟1553B数据
|
||
# logger.debug(f"1553B接收数据：{mock_data.hex()}")
|
||
#return mock_data
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"1553B数据接收失败：{str(e)}")
|
||
return None |