73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
"""
|
||
Tan mingyan
|
||
2026/3/9
|
||
|
||
oc 驱动示例（bus_oc.py）—— 补充参考
|
||
"""
|
||
# midware/drivers/bus_can.py
|
||
from .base_driver import BaseBusDriver
|
||
from typing import Optional, Dict, Any
|
||
from loguru import logger
|
||
import time
|
||
|
||
class BusOCDriver(BaseBusDriver):
|
||
"""uart总线驱动实现（适配）"""
|
||
def _init_config(self):
|
||
self.config.setdefault("channel", 0) # CAN通道
|
||
self.config.setdefault("baudrate", 500000) # 500kbps
|
||
self.config.setdefault("frame_format", "standard") # 标准帧/扩展帧
|
||
logger.info(f"UART驱动配置初始化完成：{self.config}")
|
||
|
||
def connect(self) -> bool:
|
||
try:
|
||
# 实际逻辑：打开CAN通道
|
||
time.sleep(0.05)
|
||
self.is_connected = True
|
||
logger.info("CAN总线连接成功")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"CAN总线连接失败：{str(e)}")
|
||
return False
|
||
|
||
def disconnect(self) -> bool:
|
||
try:
|
||
self.is_connected = False
|
||
logger.info("CAN总线断开成功")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"CAN总线断开失败：{str(e)}")
|
||
return False
|
||
|
||
def send(self, data: bytes, **kwargs) -> bool:
|
||
if not self.is_connected:
|
||
logger.error("CAN驱动未连接，发送失败")
|
||
return False
|
||
|
||
try:
|
||
frame_id = kwargs.get("frame_id", 0x123) # CAN帧ID
|
||
is_extended = kwargs.get("is_extended", False) # 是否扩展帧
|
||
logger.info(
|
||
f"CAN发送数据｜帧ID：0x{frame_id:X}｜扩展帧：{is_extended}｜"
|
||
f"数据：{data.hex()}｜长度：{len(data)}字节"
|
||
)
|
||
# 实际逻辑：调用CAN驱动发送接口
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"CAN数据发送失败：{str(e)}")
|
||
return False
|
||
|
||
def recv(self, timeout: float = 0.1, **kwargs) -> Optional[bytes]:
|
||
if not self.is_connected:
|
||
logger.error("CAN驱动未连接，接收失败")
|
||
return None
|
||
|
||
try:
|
||
# 模拟接收
|
||
time.sleep(0.001)
|
||
# mock_data = b"\x11\x22\x33\x44"
|
||
# logger.debug(f"CAN接收数据：{mock_data.hex()}")
|
||
# return mock_data
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"CAN数据接收失败：{str(e)}")
|
||
return None |