100 lines
2.2 KiB
Python
100 lines
2.2 KiB
Python
from abc import ABC, abstractmethod
|
||
from typing import Any, Dict, Optional
|
||
from enum import Enum
|
||
|
||
|
||
class DeviceStatus(Enum):
|
||
"""设备状态枚举"""
|
||
UNKNOWN = "unknown"
|
||
ONLINE = "online"
|
||
OFFLINE = "offline"
|
||
ERROR = "error"
|
||
BUSY = "busy"
|
||
|
||
|
||
class BaseDevice(ABC):
|
||
"""
|
||
设备接口抽象基类
|
||
定义所有设备需要实现的基本方法
|
||
"""
|
||
|
||
def __init__(self, device_id: str, device_type: str, address: str, bus: str):
|
||
"""
|
||
初始化设备
|
||
|
||
Args:
|
||
device_id: 设备唯一标识
|
||
device_type: 设备类型
|
||
address: 设备地址
|
||
bus: 所在总线
|
||
"""
|
||
self.device_id = device_id
|
||
self.device_type = device_type
|
||
self.address = address
|
||
self.bus = bus
|
||
self.status = DeviceStatus.UNKNOWN
|
||
|
||
@abstractmethod
|
||
def connect(self) -> bool:
|
||
"""
|
||
连接设备
|
||
|
||
Returns:
|
||
bool: 连接是否成功
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def disconnect(self) -> None:
|
||
"""
|
||
断开设备连接
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def read_data(self) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
读取设备数据
|
||
|
||
Returns:
|
||
dict: 设备数据，格式为 {数据名: 数据值}，失败时返回None
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def write_data(self, data: Dict[str, Any]) -> bool:
|
||
"""
|
||
向设备写入数据
|
||
|
||
Args:
|
||
data: 要写入的数据，格式为 {数据名: 数据值}
|
||
|
||
Returns:
|
||
bool: 写入是否成功
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def get_status(self) -> DeviceStatus:
|
||
"""
|
||
获取设备状态
|
||
|
||
Returns:
|
||
DeviceStatus: 设备状态
|
||
"""
|
||
pass
|
||
|
||
def get_info(self) -> Dict[str, Any]:
|
||
"""
|
||
获取设备信息
|
||
|
||
Returns:
|
||
dict: 设备信息
|
||
"""
|
||
return {
|
||
"device_id": self.device_id,
|
||
"device_type": self.device_type,
|
||
"address": self.address,
|
||
"bus": self.bus,
|
||
"status": self.status.value
|
||
} |