78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
# midware/drivers/base_driver.py
|
||
|
||
"""
|
||
Tan mingyan
|
||
2026/3/9
|
||
驱动基类（base_driver.py）—— 定义标准接口（核心）
|
||
所有总线驱动继承该基类，保证接口统一，解耦 bus_udp_decoupler.py 与具体驱动实现：
|
||
"""
|
||
from abc import ABC, abstractmethod
|
||
from typing import Optional, Dict, Any
|
||
from loguru import logger
|
||
|
||
class BaseBusDriver(ABC):
|
||
"""
|
||
总线驱动基类：定义所有总线必须实现的标准接口
|
||
遵循「开闭原则」：新增总线只需继承该类实现接口，无需修改现有代码
|
||
"""
|
||
def __init__(self, config: Optional[Dict[str, Any]] = None):
|
||
"""
|
||
初始化驱动
|
||
:param config: 驱动配置（如波特率、端口、IP等）
|
||
"""
|
||
self.config = config or {}
|
||
self.is_connected = False # 驱动连接状态
|
||
self._init_config() # 初始化配置
|
||
|
||
def _init_config(self):
|
||
"""初始化默认配置（子类可重写）"""
|
||
logger.info(f"初始化 {self.__class__.__name__} 默认配置")
|
||
|
||
@abstractmethod
|
||
def connect(self) -> bool:
|
||
"""
|
||
连接总线（如打开串口、建立CAN通道、连接1553B板卡）
|
||
:return: 连接成功返回True，失败返回False
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def disconnect(self) -> bool:
|
||
"""
|
||
断开总线连接
|
||
:return: 断开成功返回True，失败返回False
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def send(self, data: bytes, **kwargs) -> bool:
|
||
"""
|
||
发送数据到总线
|
||
:param data: 待发送的原始字节数据
|
||
:param kwargs: 扩展参数（如优先级、地址、帧ID等）
|
||
:return: 发送成功返回True，失败返回False
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def recv(self, timeout: float = 0.1, **kwargs) -> Optional[bytes]:
|
||
"""
|
||
从总线接收数据
|
||
:param timeout: 接收超时时间（秒）
|
||
:param kwargs: 扩展参数（如过滤条件、地址等）
|
||
:return: 接收到的字节数据，超时/失败返回None
|
||
"""
|
||
pass
|
||
|
||
def check_status(self) -> bool:
|
||
"""
|
||
检查驱动状态（默认实现，子类可重写）
|
||
:return: 驱动正常返回True，异常返回False
|
||
"""
|
||
return self.is_connected
|
||
|
||
def __del__(self):
|
||
"""析构函数：自动断开连接"""
|
||
if self.is_connected:
|
||
self.disconnect()
|
||
logger.info(f"{self.__class__.__name__} 自动断开连接") |