mirror of
https://github.com/dptech-corp/Uni-Lab-OS.git
synced 2025-12-18 13:31:20 +00:00
refactor: workstation_base 重构为仅含业务逻辑，通信和子设备管理交给 ProtocolNode
This commit is contained in:
Junhan Chang
File diff suppressed because it is too large
Load Diff
@@ -1,600 +0,0 @@
|
||||
"""
|
||||
工作站通信基类
|
||||
Workstation Communication Base Class
|
||||
|
||||
从具体设备驱动中抽取通用通信模式
|
||||
"""
|
||||
import json
|
||||
import time
|
||||
import threading
|
||||
from typing import Dict, Any, Optional, Callable, Union, List
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
|
||||
from unilabos.device_comms.modbus_plc.client import TCPClient as ModbusTCPClient
|
||||
from unilabos.device_comms.modbus_plc.node.modbus import DataType, WorderOrder
|
||||
from unilabos.utils.log import logger
|
||||
|
||||
|
||||
class CommunicationProtocol(Enum):
|
||||
"""通信协议类型"""
|
||||
MODBUS_TCP = "modbus_tcp"
|
||||
MODBUS_RTU = "modbus_rtu"
|
||||
SERIAL = "serial"
|
||||
ETHERNET = "ethernet"
|
||||
|
||||
|
||||
@dataclass
|
||||
class CommunicationConfig:
|
||||
"""通信配置"""
|
||||
protocol: CommunicationProtocol
|
||||
host: str
|
||||
port: int
|
||||
timeout: float = 5.0
|
||||
retry_count: int = 3
|
||||
extra_params: Dict[str, Any] = None
|
||||
|
||||
|
||||
class WorkstationCommunicationBase(ABC):
|
||||
"""工作站通信基类
|
||||
|
||||
定义工作站通信的标准接口：
|
||||
1. 状态查询 - 定期获取设备状态
|
||||
2. 命令下发 - 发送控制指令
|
||||
3. 数据采集 - 收集生产数据
|
||||
4. 紧急控制 - 单点调试控制
|
||||
"""
|
||||
|
||||
def __init__(self, communication_config: CommunicationConfig):
|
||||
self.config = communication_config
|
||||
self.client = None
|
||||
self.is_connected = False
|
||||
self.last_status = {}
|
||||
self.data_export_thread = None
|
||||
self.data_export_running = False
|
||||
|
||||
# 状态缓存
|
||||
self._status_cache = {}
|
||||
self._last_update_time = 0
|
||||
self._cache_timeout = 1.0 # 缓存1秒
|
||||
|
||||
self._initialize_communication()
|
||||
|
||||
@abstractmethod
|
||||
def _initialize_communication(self):
|
||||
"""初始化通信连接"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _load_address_mapping(self) -> Dict[str, Any]:
|
||||
"""加载地址映射表"""
|
||||
pass
|
||||
|
||||
def connect(self) -> bool:
|
||||
"""建立连接"""
|
||||
try:
|
||||
if self.config.protocol == CommunicationProtocol.MODBUS_TCP:
|
||||
self.client = ModbusTCPClient(
|
||||
addr=self.config.host,
|
||||
port=self.config.port
|
||||
)
|
||||
self.client.client.connect()
|
||||
|
||||
# 等待连接建立
|
||||
count = 100
|
||||
while count > 0:
|
||||
count -= 1
|
||||
if self.client.client.is_socket_open():
|
||||
self.is_connected = True
|
||||
logger.info(f"工作站通信连接成功: {self.config.host}:{self.config.port}")
|
||||
return True
|
||||
time.sleep(0.1)
|
||||
|
||||
if not self.client.client.is_socket_open():
|
||||
raise ConnectionError(f"无法连接到工作站: {self.config.host}:{self.config.port}")
|
||||
|
||||
else:
|
||||
raise NotImplementedError(f"协议 {self.config.protocol} 暂未实现")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"工作站通信连接失败: {e}")
|
||||
self.is_connected = False
|
||||
return False
|
||||
|
||||
def disconnect(self):
|
||||
"""断开连接"""
|
||||
try:
|
||||
if self.client and hasattr(self.client, 'client'):
|
||||
self.client.client.close()
|
||||
self.is_connected = False
|
||||
logger.info("工作站通信连接已断开")
|
||||
except Exception as e:
|
||||
logger.error(f"断开连接时出错: {e}")
|
||||
|
||||
# ============ 标准工作流接口 ============
|
||||
|
||||
def start_workflow(self, workflow_type: str, parameters: Dict[str, Any] = None) -> bool:
|
||||
"""启动工作流"""
|
||||
try:
|
||||
if not self.is_connected:
|
||||
logger.error("通信未连接，无法启动工作流")
|
||||
return False
|
||||
|
||||
logger.info(f"启动工作流: {workflow_type}, 参数: {parameters}")
|
||||
return self._execute_start_workflow(workflow_type, parameters or {})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"启动工作流失败: {e}")
|
||||
return False
|
||||
|
||||
def stop_workflow(self, emergency: bool = False) -> bool:
|
||||
"""停止工作流"""
|
||||
try:
|
||||
if not self.is_connected:
|
||||
logger.error("通信未连接，无法停止工作流")
|
||||
return False
|
||||
|
||||
logger.info(f"停止工作流 (紧急: {emergency})")
|
||||
return self._execute_stop_workflow(emergency)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"停止工作流失败: {e}")
|
||||
return False
|
||||
|
||||
def get_workflow_status(self) -> Dict[str, Any]:
|
||||
"""获取工作流状态"""
|
||||
try:
|
||||
if not self.is_connected:
|
||||
return {"error": "通信未连接"}
|
||||
|
||||
return self._query_workflow_status()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询工作流状态失败: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
# ============ 设备状态查询接口 ============
|
||||
|
||||
def get_device_status(self, force_refresh: bool = False) -> Dict[str, Any]:
|
||||
"""获取设备状态（带缓存）"""
|
||||
current_time = time.time()
|
||||
|
||||
if not force_refresh and (current_time - self._last_update_time) < self._cache_timeout:
|
||||
return self._status_cache
|
||||
|
||||
try:
|
||||
if not self.is_connected:
|
||||
return {"error": "通信未连接"}
|
||||
|
||||
status = self._query_device_status()
|
||||
self._status_cache = status
|
||||
self._last_update_time = current_time
|
||||
return status
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询设备状态失败: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
def get_production_data(self) -> Dict[str, Any]:
|
||||
"""获取生产数据"""
|
||||
try:
|
||||
if not self.is_connected:
|
||||
return {"error": "通信未连接"}
|
||||
|
||||
return self._query_production_data()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询生产数据失败: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
# ============ 单点控制接口（调试用）============
|
||||
|
||||
def write_register(self, register_name: str, value: Any, data_type: DataType = None, word_order: WorderOrder = None) -> bool:
|
||||
"""写寄存器（单点控制）"""
|
||||
try:
|
||||
if not self.is_connected:
|
||||
logger.error("通信未连接，无法写寄存器")
|
||||
return False
|
||||
|
||||
return self._write_single_register(register_name, value, data_type, word_order)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"写寄存器失败: {e}")
|
||||
return False
|
||||
|
||||
def read_register(self, register_name: str, count: int = 1, data_type: DataType = None, word_order: WorderOrder = None) -> tuple:
|
||||
"""读寄存器（单点控制）"""
|
||||
try:
|
||||
if not self.is_connected:
|
||||
logger.error("通信未连接，无法读寄存器")
|
||||
return None, True
|
||||
|
||||
return self._read_single_register(register_name, count, data_type, word_order)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"读寄存器失败: {e}")
|
||||
return None, True
|
||||
|
||||
# ============ 数据导出功能 ============
|
||||
|
||||
def start_data_export(self, file_path: str, export_interval: float = 1.0) -> bool:
|
||||
"""开始数据导出"""
|
||||
try:
|
||||
if self.data_export_running:
|
||||
logger.warning("数据导出已在运行")
|
||||
return False
|
||||
|
||||
self.data_export_file = file_path
|
||||
self.data_export_interval = export_interval
|
||||
self.data_export_running = True
|
||||
|
||||
# 创建CSV文件并写入表头
|
||||
self._initialize_export_file(file_path)
|
||||
|
||||
# 启动数据收集线程
|
||||
self.data_export_thread = threading.Thread(target=self._data_export_worker)
|
||||
self.data_export_thread.daemon = True
|
||||
self.data_export_thread.start()
|
||||
|
||||
logger.info(f"数据导出已启动: {file_path}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"启动数据导出失败: {e}")
|
||||
return False
|
||||
|
||||
def stop_data_export(self) -> bool:
|
||||
"""停止数据导出"""
|
||||
try:
|
||||
if not self.data_export_running:
|
||||
logger.warning("数据导出未运行")
|
||||
return False
|
||||
|
||||
self.data_export_running = False
|
||||
|
||||
if self.data_export_thread and self.data_export_thread.is_alive():
|
||||
self.data_export_thread.join(timeout=5.0)
|
||||
|
||||
logger.info("数据导出已停止")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"停止数据导出失败: {e}")
|
||||
return False
|
||||
|
||||
def _data_export_worker(self):
|
||||
"""数据导出工作线程"""
|
||||
while self.data_export_running:
|
||||
try:
|
||||
data = self.get_production_data()
|
||||
self._append_to_export_file(data)
|
||||
time.sleep(self.data_export_interval)
|
||||
except Exception as e:
|
||||
logger.error(f"数据导出工作线程错误: {e}")
|
||||
|
||||
# ============ 抽象方法 - 子类必须实现 ============
|
||||
|
||||
@abstractmethod
|
||||
def _execute_start_workflow(self, workflow_type: str, parameters: Dict[str, Any]) -> bool:
|
||||
"""执行启动工作流命令"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _execute_stop_workflow(self, emergency: bool) -> bool:
|
||||
"""执行停止工作流命令"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _query_workflow_status(self) -> Dict[str, Any]:
|
||||
"""查询工作流状态"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _query_device_status(self) -> Dict[str, Any]:
|
||||
"""查询设备状态"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _query_production_data(self) -> Dict[str, Any]:
|
||||
"""查询生产数据"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _write_single_register(self, register_name: str, value: Any, data_type: DataType, word_order: WorderOrder) -> bool:
|
||||
"""写单个寄存器"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _read_single_register(self, register_name: str, count: int, data_type: DataType, word_order: WorderOrder) -> tuple:
|
||||
"""读单个寄存器"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _initialize_export_file(self, file_path: str):
|
||||
"""初始化导出文件"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def _append_to_export_file(self, data: Dict[str, Any]):
|
||||
"""追加数据到导出文件"""
|
||||
pass
|
||||
|
||||
|
||||
class CoinCellCommunication(WorkstationCommunicationBase):
|
||||
"""纽扣电池组装系统通信类
|
||||
|
||||
从 coin_cell_assembly_system 抽取的通信功能
|
||||
"""
|
||||
|
||||
def __init__(self, communication_config: CommunicationConfig, csv_path: str = "./coin_cell_assembly.csv"):
|
||||
self.csv_path = csv_path
|
||||
super().__init__(communication_config)
|
||||
|
||||
def _initialize_communication(self):
|
||||
"""初始化通信连接"""
|
||||
# 加载节点映射
|
||||
try:
|
||||
nodes = self.client.load_csv(self.csv_path) if self.client else []
|
||||
if self.client:
|
||||
self.client.register_node_list(nodes)
|
||||
except Exception as e:
|
||||
logger.error(f"加载节点映射失败: {e}")
|
||||
|
||||
def _load_address_mapping(self) -> Dict[str, Any]:
|
||||
"""加载地址映射表"""
|
||||
# 从CSV文件加载地址映射
|
||||
return {}
|
||||
|
||||
def _execute_start_workflow(self, workflow_type: str, parameters: Dict[str, Any]) -> bool:
|
||||
"""执行启动工作流命令"""
|
||||
if workflow_type == "battery_manufacturing":
|
||||
# 发送电池制造启动命令
|
||||
return self._start_battery_manufacturing(parameters)
|
||||
else:
|
||||
logger.error(f"不支持的工作流类型: {workflow_type}")
|
||||
return False
|
||||
|
||||
def _start_battery_manufacturing(self, parameters: Dict[str, Any]) -> bool:
|
||||
"""启动电池制造工作流"""
|
||||
try:
|
||||
# 1. 设置参数
|
||||
if "electrolyte_num" in parameters:
|
||||
self.client.use_node('REG_MSG_ELECTROLYTE_NUM').write(parameters["electrolyte_num"])
|
||||
|
||||
if "electrolyte_volume" in parameters:
|
||||
self.client.use_node('REG_MSG_ELECTROLYTE_VOLUME').write(
|
||||
parameters["electrolyte_volume"],
|
||||
data_type=DataType.FLOAT32,
|
||||
word_order=WorderOrder.LITTLE
|
||||
)
|
||||
|
||||
if "assembly_pressure" in parameters:
|
||||
self.client.use_node('REG_MSG_ASSEMBLY_PRESSURE').write(
|
||||
parameters["assembly_pressure"],
|
||||
data_type=DataType.FLOAT32,
|
||||
word_order=WorderOrder.LITTLE
|
||||
)
|
||||
|
||||
# 2. 发送启动命令
|
||||
self.client.use_node('COIL_SYS_START_CMD').write(True)
|
||||
|
||||
# 3. 确认启动成功
|
||||
time.sleep(0.5)
|
||||
status, read_err = self.client.use_node('COIL_SYS_START_STATUS').read(1)
|
||||
return not read_err and status[0]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"启动电池制造工作流失败: {e}")
|
||||
return False
|
||||
|
||||
def _execute_stop_workflow(self, emergency: bool) -> bool:
|
||||
"""执行停止工作流命令"""
|
||||
try:
|
||||
if emergency:
|
||||
# 紧急停止
|
||||
self.client.use_node('COIL_SYS_RESET_CMD').write(True)
|
||||
else:
|
||||
# 正常停止
|
||||
self.client.use_node('COIL_SYS_STOP_CMD').write(True)
|
||||
|
||||
time.sleep(0.5)
|
||||
status, read_err = self.client.use_node('COIL_SYS_STOP_STATUS').read(1)
|
||||
return not read_err and status[0]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"停止工作流失败: {e}")
|
||||
return False
|
||||
|
||||
def _query_workflow_status(self) -> Dict[str, Any]:
|
||||
"""查询工作流状态"""
|
||||
try:
|
||||
status = {}
|
||||
|
||||
# 读取系统状态
|
||||
start_status, _ = self.client.use_node('COIL_SYS_START_STATUS').read(1)
|
||||
stop_status, _ = self.client.use_node('COIL_SYS_STOP_STATUS').read(1)
|
||||
auto_status, _ = self.client.use_node('COIL_SYS_AUTO_STATUS').read(1)
|
||||
init_status, _ = self.client.use_node('COIL_SYS_INIT_STATUS').read(1)
|
||||
|
||||
status.update({
|
||||
"is_running": start_status[0] if start_status else False,
|
||||
"is_stopped": stop_status[0] if stop_status else False,
|
||||
"is_auto_mode": auto_status[0] if auto_status else False,
|
||||
"is_initialized": init_status[0] if init_status else False,
|
||||
})
|
||||
|
||||
return status
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询工作流状态失败: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
def _query_device_status(self) -> Dict[str, Any]:
|
||||
"""查询设备状态"""
|
||||
try:
|
||||
status = {}
|
||||
|
||||
# 读取位置信息
|
||||
x_pos, _ = self.client.use_node('REG_DATA_AXIS_X_POS').read(2, word_order=WorderOrder.LITTLE)
|
||||
y_pos, _ = self.client.use_node('REG_DATA_AXIS_Y_POS').read(2, word_order=WorderOrder.LITTLE)
|
||||
z_pos, _ = self.client.use_node('REG_DATA_AXIS_Z_POS').read(2, word_order=WorderOrder.LITTLE)
|
||||
|
||||
# 读取环境数据
|
||||
pressure, _ = self.client.use_node('REG_DATA_GLOVE_BOX_PRESSURE').read(2, word_order=WorderOrder.LITTLE)
|
||||
o2_content, _ = self.client.use_node('REG_DATA_GLOVE_BOX_O2_CONTENT').read(2, word_order=WorderOrder.LITTLE)
|
||||
water_content, _ = self.client.use_node('REG_DATA_GLOVE_BOX_WATER_CONTENT').read(2, word_order=WorderOrder.LITTLE)
|
||||
|
||||
status.update({
|
||||
"axis_position": {
|
||||
"x": x_pos[0] if x_pos else 0.0,
|
||||
"y": y_pos[0] if y_pos else 0.0,
|
||||
"z": z_pos[0] if z_pos else 0.0,
|
||||
},
|
||||
"environment": {
|
||||
"glove_box_pressure": pressure[0] if pressure else 0.0,
|
||||
"o2_content": o2_content[0] if o2_content else 0.0,
|
||||
"water_content": water_content[0] if water_content else 0.0,
|
||||
}
|
||||
})
|
||||
|
||||
return status
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询设备状态失败: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
def _query_production_data(self) -> Dict[str, Any]:
|
||||
"""查询生产数据"""
|
||||
try:
|
||||
data = {}
|
||||
|
||||
# 读取生产统计
|
||||
coin_cell_num, _ = self.client.use_node('REG_DATA_ASSEMBLY_COIN_CELL_NUM').read(1)
|
||||
assembly_time, _ = self.client.use_node('REG_DATA_ASSEMBLY_TIME').read(2, word_order=WorderOrder.LITTLE)
|
||||
voltage, _ = self.client.use_node('REG_DATA_OPEN_CIRCUIT_VOLTAGE').read(2, word_order=WorderOrder.LITTLE)
|
||||
|
||||
# 读取当前产品信息
|
||||
coin_cell_code, _ = self.client.use_node('REG_DATA_COIN_CELL_CODE').read(20) # 假设是字符串
|
||||
electrolyte_code, _ = self.client.use_node('REG_DATA_ELECTROLYTE_CODE').read(20)
|
||||
|
||||
data.update({
|
||||
"production_count": coin_cell_num[0] if coin_cell_num else 0,
|
||||
"assembly_time": assembly_time[0] if assembly_time else 0.0,
|
||||
"open_circuit_voltage": voltage[0] if voltage else 0.0,
|
||||
"current_battery_code": self._decode_string(coin_cell_code) if coin_cell_code else "",
|
||||
"current_electrolyte_code": self._decode_string(electrolyte_code) if electrolyte_code else "",
|
||||
"timestamp": time.time(),
|
||||
})
|
||||
|
||||
return data
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询生产数据失败: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
def _write_single_register(self, register_name: str, value: Any, data_type: DataType = None, word_order: WorderOrder = None) -> bool:
|
||||
"""写单个寄存器"""
|
||||
try:
|
||||
kwargs = {"value": value}
|
||||
if data_type:
|
||||
kwargs["data_type"] = data_type
|
||||
if word_order:
|
||||
kwargs["word_order"] = word_order
|
||||
|
||||
result = self.client.use_node(register_name).write(**kwargs)
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"写寄存器 {register_name} 失败: {e}")
|
||||
return False
|
||||
|
||||
def _read_single_register(self, register_name: str, count: int = 1, data_type: DataType = None, word_order: WorderOrder = None) -> tuple:
|
||||
"""读单个寄存器"""
|
||||
try:
|
||||
kwargs = {"count": count}
|
||||
if data_type:
|
||||
kwargs["data_type"] = data_type
|
||||
if word_order:
|
||||
kwargs["word_order"] = word_order
|
||||
|
||||
value, error = self.client.use_node(register_name).read(**kwargs)
|
||||
return value, error
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"读寄存器 {register_name} 失败: {e}")
|
||||
return None, True
|
||||
|
||||
def _initialize_export_file(self, file_path: str):
|
||||
"""初始化导出文件"""
|
||||
import csv
|
||||
try:
|
||||
with open(file_path, 'w', newline='', encoding='utf-8') as csvfile:
|
||||
fieldnames = [
|
||||
'timestamp', 'production_count', 'assembly_time',
|
||||
'open_circuit_voltage', 'battery_code', 'electrolyte_code',
|
||||
'axis_x', 'axis_y', 'axis_z', 'glove_box_pressure',
|
||||
'o2_content', 'water_content'
|
||||
]
|
||||
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
|
||||
writer.writeheader()
|
||||
except Exception as e:
|
||||
logger.error(f"初始化导出文件失败: {e}")
|
||||
|
||||
def _append_to_export_file(self, data: Dict[str, Any]):
|
||||
"""追加数据到导出文件"""
|
||||
import csv
|
||||
try:
|
||||
with open(self.data_export_file, 'a', newline='', encoding='utf-8') as csvfile:
|
||||
fieldnames = [
|
||||
'timestamp', 'production_count', 'assembly_time',
|
||||
'open_circuit_voltage', 'battery_code', 'electrolyte_code',
|
||||
'axis_x', 'axis_y', 'axis_z', 'glove_box_pressure',
|
||||
'o2_content', 'water_content'
|
||||
]
|
||||
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
|
||||
|
||||
row = {
|
||||
'timestamp': data.get('timestamp', time.time()),
|
||||
'production_count': data.get('production_count', 0),
|
||||
'assembly_time': data.get('assembly_time', 0.0),
|
||||
'open_circuit_voltage': data.get('open_circuit_voltage', 0.0),
|
||||
'battery_code': data.get('current_battery_code', ''),
|
||||
'electrolyte_code': data.get('current_electrolyte_code', ''),
|
||||
}
|
||||
|
||||
# 添加位置数据
|
||||
axis_pos = data.get('axis_position', {})
|
||||
row.update({
|
||||
'axis_x': axis_pos.get('x', 0.0),
|
||||
'axis_y': axis_pos.get('y', 0.0),
|
||||
'axis_z': axis_pos.get('z', 0.0),
|
||||
})
|
||||
|
||||
# 添加环境数据
|
||||
env = data.get('environment', {})
|
||||
row.update({
|
||||
'glove_box_pressure': env.get('glove_box_pressure', 0.0),
|
||||
'o2_content': env.get('o2_content', 0.0),
|
||||
'water_content': env.get('water_content', 0.0),
|
||||
})
|
||||
|
||||
writer.writerow(row)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"追加数据到导出文件失败: {e}")
|
||||
|
||||
def _decode_string(self, data_list: List[int]) -> str:
|
||||
"""将寄存器数据解码为字符串"""
|
||||
try:
|
||||
# 假设每个寄存器包含2个字符（16位）
|
||||
chars = []
|
||||
for value in data_list:
|
||||
if value == 0:
|
||||
break
|
||||
chars.append(chr(value & 0xFF))
|
||||
if (value >> 8) & 0xFF != 0:
|
||||
chars.append(chr((value >> 8) & 0xFF))
|
||||
return ''.join(chars).rstrip('\x00')
|
||||
except:
|
||||
return ""
|
||||
@@ -1,605 +0,0 @@
|
||||
"""
|
||||
工作站HTTP服务模块
|
||||
Workstation HTTP Service Module
|
||||
|
||||
统一的工作站报送接收服务，基于LIMS协议规范：
|
||||
1. 步骤完成报送 - POST /report/step_finish
|
||||
2. 通量完成报送 - POST /report/sample_finish
|
||||
3. 任务完成报送 - POST /report/order_finish
|
||||
4. 批量更新报送 - POST /report/batch_update
|
||||
5. 物料变更报送 - POST /report/material_change
|
||||
6. 错误处理报送 - POST /report/error_handling
|
||||
7. 健康检查和状态查询
|
||||
|
||||
统一使用LIMS协议字段规范，简化接口避免功能重复
|
||||
"""
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
from typing import Dict, Any, Optional, List
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
from urllib.parse import urlparse
|
||||
from dataclasses import dataclass, asdict
|
||||
from datetime import datetime
|
||||
|
||||
from unilabos.utils.log import logger
|
||||
|
||||
|
||||
@dataclass
|
||||
class WorkstationReportRequest:
|
||||
"""统一工作站报送请求（基于LIMS协议规范）"""
|
||||
token: str # 授权令牌
|
||||
request_time: str # 请求时间，格式：2024-12-12 12:12:12.xxx
|
||||
data: Dict[str, Any] # 报送数据
|
||||
|
||||
|
||||
@dataclass
|
||||
class MaterialUsage:
|
||||
"""物料使用记录"""
|
||||
materialId: str # 物料Id（GUID）
|
||||
locationId: str # 库位Id（GUID）
|
||||
typeMode: str # 物料类型（样品1、试剂2、耗材0）
|
||||
usedQuantity: float # 使用的数量（数字）
|
||||
|
||||
|
||||
@dataclass
|
||||
class HttpResponse:
|
||||
"""HTTP响应"""
|
||||
success: bool
|
||||
message: str
|
||||
data: Optional[Dict[str, Any]] = None
|
||||
acknowledgment_id: Optional[str] = None
|
||||
|
||||
|
||||
class WorkstationHTTPHandler(BaseHTTPRequestHandler):
|
||||
"""工作站HTTP请求处理器"""
|
||||
|
||||
def __init__(self, workstation_instance, *args, **kwargs):
|
||||
self.workstation = workstation_instance
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def do_POST(self):
|
||||
"""处理POST请求 - 统一的工作站报送接口"""
|
||||
try:
|
||||
# 解析请求路径
|
||||
parsed_path = urlparse(self.path)
|
||||
endpoint = parsed_path.path
|
||||
|
||||
# 读取请求体
|
||||
content_length = int(self.headers.get('Content-Length', 0))
|
||||
if content_length > 0:
|
||||
post_data = self.rfile.read(content_length)
|
||||
request_data = json.loads(post_data.decode('utf-8'))
|
||||
else:
|
||||
request_data = {}
|
||||
|
||||
logger.info(f"收到工作站报送: {endpoint} - {request_data.get('token', 'unknown')}")
|
||||
|
||||
# 统一的报送端点路由（基于LIMS协议规范）
|
||||
if endpoint == '/report/step_finish':
|
||||
response = self._handle_step_finish_report(request_data)
|
||||
elif endpoint == '/report/sample_finish':
|
||||
response = self._handle_sample_finish_report(request_data)
|
||||
elif endpoint == '/report/order_finish':
|
||||
response = self._handle_order_finish_report(request_data)
|
||||
elif endpoint == '/report/batch_update':
|
||||
response = self._handle_batch_update_report(request_data)
|
||||
# 扩展报送端点
|
||||
elif endpoint == '/report/material_change':
|
||||
response = self._handle_material_change_report(request_data)
|
||||
elif endpoint == '/report/error_handling':
|
||||
response = self._handle_error_handling_report(request_data)
|
||||
# 保留LIMS协议端点以兼容现有系统
|
||||
elif endpoint == '/LIMS/step_finish':
|
||||
response = self._handle_step_finish_report(request_data)
|
||||
elif endpoint == '/LIMS/preintake_finish':
|
||||
response = self._handle_sample_finish_report(request_data)
|
||||
elif endpoint == '/LIMS/order_finish':
|
||||
response = self._handle_order_finish_report(request_data)
|
||||
else:
|
||||
response = HttpResponse(
|
||||
success=False,
|
||||
message=f"不支持的报送端点: {endpoint}",
|
||||
data={"supported_endpoints": [
|
||||
"/report/step_finish",
|
||||
"/report/sample_finish",
|
||||
"/report/order_finish",
|
||||
"/report/batch_update",
|
||||
"/report/material_change",
|
||||
"/report/error_handling"
|
||||
]}
|
||||
)
|
||||
|
||||
# 发送响应
|
||||
self._send_response(response)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理工作站报送失败: {e}\\n{traceback.format_exc()}")
|
||||
error_response = HttpResponse(
|
||||
success=False,
|
||||
message=f"请求处理失败: {str(e)}"
|
||||
)
|
||||
self._send_response(error_response)
|
||||
|
||||
def do_GET(self):
|
||||
"""处理GET请求 - 健康检查和状态查询"""
|
||||
try:
|
||||
parsed_path = urlparse(self.path)
|
||||
endpoint = parsed_path.path
|
||||
|
||||
if endpoint == '/status':
|
||||
response = self._handle_status_check()
|
||||
elif endpoint == '/health':
|
||||
response = HttpResponse(success=True, message="服务健康")
|
||||
else:
|
||||
response = HttpResponse(
|
||||
success=False,
|
||||
message=f"不支持的查询端点: {endpoint}",
|
||||
data={"supported_endpoints": ["/status", "/health"]}
|
||||
)
|
||||
|
||||
self._send_response(response)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"GET请求处理失败: {e}")
|
||||
error_response = HttpResponse(
|
||||
success=False,
|
||||
message=f"GET请求处理失败: {str(e)}"
|
||||
)
|
||||
self._send_response(error_response)
|
||||
|
||||
def _handle_step_finish_report(self, request_data: Dict[str, Any]) -> HttpResponse:
|
||||
"""处理步骤完成报送（统一LIMS协议规范）"""
|
||||
try:
|
||||
# 验证基本字段
|
||||
required_fields = ['token', 'request_time', 'data']
|
||||
if missing_fields := [field for field in required_fields if field not in request_data]:
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"缺少必要字段: {', '.join(missing_fields)}"
|
||||
)
|
||||
|
||||
# 验证data字段内容
|
||||
data = request_data['data']
|
||||
data_required_fields = ['orderCode', 'orderName', 'stepName', 'stepId', 'sampleId', 'startTime', 'endTime']
|
||||
if data_missing_fields := [field for field in data_required_fields if field not in data]:
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"data字段缺少必要内容: {', '.join(data_missing_fields)}"
|
||||
)
|
||||
|
||||
# 创建统一请求对象
|
||||
report_request = WorkstationReportRequest(
|
||||
token=request_data['token'],
|
||||
request_time=request_data['request_time'],
|
||||
data=data
|
||||
)
|
||||
|
||||
# 调用工作站处理方法
|
||||
result = self.workstation.process_step_finish_report(report_request)
|
||||
|
||||
return HttpResponse(
|
||||
success=True,
|
||||
message=f"步骤完成报送已处理: {data['stepName']} ({data['orderCode']})",
|
||||
acknowledgment_id=f"STEP_{int(time.time() * 1000)}_{data['stepId']}",
|
||||
data=result
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理步骤完成报送失败: {e}")
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"步骤完成报送处理失败: {str(e)}"
|
||||
)
|
||||
|
||||
def _handle_sample_finish_report(self, request_data: Dict[str, Any]) -> HttpResponse:
|
||||
"""处理通量完成报送（统一LIMS协议规范）"""
|
||||
try:
|
||||
# 验证基本字段
|
||||
required_fields = ['token', 'request_time', 'data']
|
||||
if missing_fields := [field for field in required_fields if field not in request_data]:
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"缺少必要字段: {', '.join(missing_fields)}"
|
||||
)
|
||||
|
||||
# 验证data字段内容
|
||||
data = request_data['data']
|
||||
data_required_fields = ['orderCode', 'orderName', 'sampleId', 'startTime', 'endTime', 'Status']
|
||||
if data_missing_fields := [field for field in data_required_fields if field not in data]:
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"data字段缺少必要内容: {', '.join(data_missing_fields)}"
|
||||
)
|
||||
|
||||
# 创建统一请求对象
|
||||
report_request = WorkstationReportRequest(
|
||||
token=request_data['token'],
|
||||
request_time=request_data['request_time'],
|
||||
data=data
|
||||
)
|
||||
|
||||
# 调用工作站处理方法
|
||||
result = self.workstation.process_sample_finish_report(report_request)
|
||||
|
||||
status_names = {
|
||||
"0": "待生产", "2": "进样", "10": "开始",
|
||||
"20": "完成", "-2": "异常停止", "-3": "人工停止"
|
||||
}
|
||||
status_desc = status_names.get(str(data['Status']), f"状态{data['Status']}")
|
||||
|
||||
return HttpResponse(
|
||||
success=True,
|
||||
message=f"通量完成报送已处理: {data['sampleId']} ({data['orderCode']}) - {status_desc}",
|
||||
acknowledgment_id=f"SAMPLE_{int(time.time() * 1000)}_{data['sampleId']}",
|
||||
data=result
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理通量完成报送失败: {e}")
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"通量完成报送处理失败: {str(e)}"
|
||||
)
|
||||
|
||||
def _handle_order_finish_report(self, request_data: Dict[str, Any]) -> HttpResponse:
|
||||
"""处理任务完成报送（统一LIMS协议规范）"""
|
||||
try:
|
||||
# 验证基本字段
|
||||
required_fields = ['token', 'request_time', 'data']
|
||||
if missing_fields := [field for field in required_fields if field not in request_data]:
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"缺少必要字段: {', '.join(missing_fields)}"
|
||||
)
|
||||
|
||||
# 验证data字段内容
|
||||
data = request_data['data']
|
||||
data_required_fields = ['orderCode', 'orderName', 'startTime', 'endTime', 'status']
|
||||
if data_missing_fields := [field for field in data_required_fields if field not in data]:
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"data字段缺少必要内容: {', '.join(data_missing_fields)}"
|
||||
)
|
||||
|
||||
# 处理物料使用记录
|
||||
used_materials = []
|
||||
if 'usedMaterials' in data:
|
||||
for material_data in data['usedMaterials']:
|
||||
material = MaterialUsage(
|
||||
materialId=material_data.get('materialId', ''),
|
||||
locationId=material_data.get('locationId', ''),
|
||||
typeMode=material_data.get('typeMode', ''),
|
||||
usedQuantity=material_data.get('usedQuantity', 0.0)
|
||||
)
|
||||
used_materials.append(material)
|
||||
|
||||
# 创建统一请求对象
|
||||
report_request = WorkstationReportRequest(
|
||||
token=request_data['token'],
|
||||
request_time=request_data['request_time'],
|
||||
data=data
|
||||
)
|
||||
|
||||
# 调用工作站处理方法
|
||||
result = self.workstation.process_order_finish_report(report_request, used_materials)
|
||||
|
||||
status_names = {"30": "完成", "-11": "异常停止", "-12": "人工停止"}
|
||||
status_desc = status_names.get(str(data['status']), f"状态{data['status']}")
|
||||
|
||||
return HttpResponse(
|
||||
success=True,
|
||||
message=f"任务完成报送已处理: {data['orderName']} ({data['orderCode']}) - {status_desc}",
|
||||
acknowledgment_id=f"ORDER_{int(time.time() * 1000)}_{data['orderCode']}",
|
||||
data=result
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理任务完成报送失败: {e}")
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"任务完成报送处理失败: {str(e)}"
|
||||
)
|
||||
|
||||
def _handle_batch_update_report(self, request_data: Dict[str, Any]) -> HttpResponse:
|
||||
"""处理批量报送"""
|
||||
try:
|
||||
step_updates = request_data.get('step_updates', [])
|
||||
sample_updates = request_data.get('sample_updates', [])
|
||||
order_updates = request_data.get('order_updates', [])
|
||||
|
||||
results = {
|
||||
'step_results': [],
|
||||
'sample_results': [],
|
||||
'order_results': [],
|
||||
'total_processed': 0,
|
||||
'total_failed': 0
|
||||
}
|
||||
|
||||
# 处理批量步骤更新
|
||||
for step_data in step_updates:
|
||||
try:
|
||||
step_data['token'] = request_data.get('token', step_data.get('token'))
|
||||
step_data['request_time'] = request_data.get('request_time', step_data.get('request_time'))
|
||||
result = self._handle_step_finish_report(step_data)
|
||||
results['step_results'].append(result)
|
||||
if result.success:
|
||||
results['total_processed'] += 1
|
||||
else:
|
||||
results['total_failed'] += 1
|
||||
except Exception as e:
|
||||
results['step_results'].append(HttpResponse(success=False, message=str(e)))
|
||||
results['total_failed'] += 1
|
||||
|
||||
# 处理批量通量更新
|
||||
for sample_data in sample_updates:
|
||||
try:
|
||||
sample_data['token'] = request_data.get('token', sample_data.get('token'))
|
||||
sample_data['request_time'] = request_data.get('request_time', sample_data.get('request_time'))
|
||||
result = self._handle_sample_finish_report(sample_data)
|
||||
results['sample_results'].append(result)
|
||||
if result.success:
|
||||
results['total_processed'] += 1
|
||||
else:
|
||||
results['total_failed'] += 1
|
||||
except Exception as e:
|
||||
results['sample_results'].append(HttpResponse(success=False, message=str(e)))
|
||||
results['total_failed'] += 1
|
||||
|
||||
# 处理批量任务更新
|
||||
for order_data in order_updates:
|
||||
try:
|
||||
order_data['token'] = request_data.get('token', order_data.get('token'))
|
||||
order_data['request_time'] = request_data.get('request_time', order_data.get('request_time'))
|
||||
result = self._handle_order_finish_report(order_data)
|
||||
results['order_results'].append(result)
|
||||
if result.success:
|
||||
results['total_processed'] += 1
|
||||
else:
|
||||
results['total_failed'] += 1
|
||||
except Exception as e:
|
||||
results['order_results'].append(HttpResponse(success=False, message=str(e)))
|
||||
results['total_failed'] += 1
|
||||
|
||||
return HttpResponse(
|
||||
success=results['total_failed'] == 0,
|
||||
message=f"批量报送处理完成: {results['total_processed']} 成功, {results['total_failed']} 失败",
|
||||
acknowledgment_id=f"BATCH_{int(time.time() * 1000)}",
|
||||
data=results
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理批量报送失败: {e}")
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"批量报送处理失败: {str(e)}"
|
||||
)
|
||||
|
||||
def _handle_material_change_report(self, request_data: Dict[str, Any]) -> HttpResponse:
|
||||
"""处理物料变更报送"""
|
||||
try:
|
||||
# 验证必需字段
|
||||
required_fields = ['workstation_id', 'timestamp', 'resource_id', 'change_type']
|
||||
if missing_fields := [field for field in required_fields if field not in request_data]:
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"缺少必要字段: {', '.join(missing_fields)}"
|
||||
)
|
||||
|
||||
# 调用工作站的处理方法
|
||||
result = self.workstation.process_material_change_report(request_data)
|
||||
|
||||
return HttpResponse(
|
||||
success=True,
|
||||
message=f"物料变更报送已处理: {request_data['resource_id']} ({request_data['change_type']})",
|
||||
acknowledgment_id=f"MATERIAL_{int(time.time() * 1000)}_{request_data['resource_id']}",
|
||||
data=result
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理物料变更报送失败: {e}")
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"物料变更报送处理失败: {str(e)}"
|
||||
)
|
||||
|
||||
def _handle_error_handling_report(self, request_data: Dict[str, Any]) -> HttpResponse:
|
||||
"""处理错误处理报送"""
|
||||
try:
|
||||
# 验证必需字段
|
||||
required_fields = ['workstation_id', 'timestamp', 'error_type', 'error_message']
|
||||
if missing_fields := [field for field in required_fields if field not in request_data]:
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"缺少必要字段: {', '.join(missing_fields)}"
|
||||
)
|
||||
|
||||
# 调用工作站的处理方法
|
||||
result = self.workstation.handle_external_error(request_data)
|
||||
|
||||
return HttpResponse(
|
||||
success=True,
|
||||
message=f"错误处理报送已处理: {request_data['error_type']} - {request_data['error_message']}",
|
||||
acknowledgment_id=f"ERROR_{int(time.time() * 1000)}_{request_data.get('action_id', 'unknown')}",
|
||||
data=result
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理错误处理报送失败: {e}")
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"错误处理报送处理失败: {str(e)}"
|
||||
)
|
||||
|
||||
def _handle_status_check(self) -> HttpResponse:
|
||||
"""处理状态查询"""
|
||||
try:
|
||||
return HttpResponse(
|
||||
success=True,
|
||||
message="工作站报送服务正常运行",
|
||||
data={
|
||||
"workstation_id": self.workstation.device_id,
|
||||
"service_type": "unified_reporting_service",
|
||||
"uptime": time.time() - getattr(self.workstation, '_start_time', time.time()),
|
||||
"reports_received": getattr(self.workstation, '_reports_received_count', 0),
|
||||
"supported_endpoints": [
|
||||
"POST /report/step_finish",
|
||||
"POST /report/sample_finish",
|
||||
"POST /report/order_finish",
|
||||
"POST /report/batch_update",
|
||||
"POST /report/material_change",
|
||||
"POST /report/error_handling",
|
||||
"GET /status",
|
||||
"GET /health"
|
||||
]
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"处理状态查询失败: {e}")
|
||||
return HttpResponse(
|
||||
success=False,
|
||||
message=f"状态查询失败: {str(e)}"
|
||||
)
|
||||
|
||||
def _send_response(self, response: HttpResponse):
|
||||
"""发送响应"""
|
||||
try:
|
||||
# 设置响应状态码
|
||||
status_code = 200 if response.success else 400
|
||||
self.send_response(status_code)
|
||||
|
||||
# 设置响应头
|
||||
self.send_header('Content-Type', 'application/json; charset=utf-8')
|
||||
self.send_header('Access-Control-Allow-Origin', '*')
|
||||
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
|
||||
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
|
||||
self.end_headers()
|
||||
|
||||
# 发送响应体
|
||||
response_json = json.dumps(asdict(response), ensure_ascii=False, indent=2)
|
||||
self.wfile.write(response_json.encode('utf-8'))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"发送响应失败: {e}")
|
||||
|
||||
def log_message(self, format, *args):
|
||||
"""重写日志方法"""
|
||||
logger.debug(f"HTTP请求: {format % args}")
|
||||
|
||||
|
||||
class WorkstationHTTPService:
|
||||
"""工作站HTTP服务"""
|
||||
|
||||
def __init__(self, workstation_instance, host: str = "127.0.0.1", port: int = 8080):
|
||||
self.workstation = workstation_instance
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.server = None
|
||||
self.server_thread = None
|
||||
self.running = False
|
||||
|
||||
# 初始化统计信息
|
||||
self.workstation._start_time = time.time()
|
||||
self.workstation._reports_received_count = 0
|
||||
|
||||
def start(self):
|
||||
"""启动HTTP服务"""
|
||||
try:
|
||||
# 创建处理器工厂函数
|
||||
def handler_factory(*args, **kwargs):
|
||||
return WorkstationHTTPHandler(self.workstation, *args, **kwargs)
|
||||
|
||||
# 创建HTTP服务器
|
||||
self.server = HTTPServer((self.host, self.port), handler_factory)
|
||||
|
||||
# 在单独线程中运行服务器
|
||||
self.server_thread = threading.Thread(
|
||||
target=self._run_server,
|
||||
daemon=True,
|
||||
name=f"WorkstationHTTP-{self.workstation.device_id}"
|
||||
)
|
||||
|
||||
self.running = True
|
||||
self.server_thread.start()
|
||||
|
||||
logger.info(f"工作站HTTP报送服务已启动: http://{self.host}:{self.port}")
|
||||
logger.info("统一的报送端点 (基于LIMS协议规范):")
|
||||
logger.info(" - POST /report/step_finish # 步骤完成报送")
|
||||
logger.info(" - POST /report/sample_finish # 通量完成报送")
|
||||
logger.info(" - POST /report/order_finish # 任务完成报送")
|
||||
logger.info(" - POST /report/batch_update # 批量更新报送")
|
||||
logger.info("扩展报送端点:")
|
||||
logger.info(" - POST /report/material_change # 物料变更报送")
|
||||
logger.info(" - POST /report/error_handling # 错误处理报送")
|
||||
logger.info("兼容端点:")
|
||||
logger.info(" - POST /LIMS/step_finish # 兼容LIMS步骤完成")
|
||||
logger.info(" - POST /LIMS/preintake_finish # 兼容LIMS通量完成")
|
||||
logger.info(" - POST /LIMS/order_finish # 兼容LIMS任务完成")
|
||||
logger.info("服务端点:")
|
||||
logger.info(" - GET /status # 服务状态查询")
|
||||
logger.info(" - GET /health # 健康检查")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"启动HTTP服务失败: {e}")
|
||||
raise
|
||||
|
||||
def stop(self):
|
||||
"""停止HTTP服务"""
|
||||
try:
|
||||
if self.running and self.server:
|
||||
self.running = False
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
|
||||
if self.server_thread and self.server_thread.is_alive():
|
||||
self.server_thread.join(timeout=5.0)
|
||||
|
||||
logger.info("工作站HTTP报送服务已停止")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"停止HTTP服务失败: {e}")
|
||||
|
||||
def _run_server(self):
|
||||
"""运行HTTP服务器"""
|
||||
try:
|
||||
while self.running:
|
||||
self.server.handle_request()
|
||||
except Exception as e:
|
||||
if self.running: # 只在非正常停止时记录错误
|
||||
logger.error(f"HTTP服务运行错误: {e}")
|
||||
|
||||
@property
|
||||
def is_running(self) -> bool:
|
||||
"""检查服务是否正在运行"""
|
||||
return self.running and self.server_thread and self.server_thread.is_alive()
|
||||
|
||||
@property
|
||||
def service_url(self) -> str:
|
||||
"""获取服务URL"""
|
||||
return f"http://{self.host}:{self.port}"
|
||||
|
||||
|
||||
# 导出主要类 - 保持向后兼容
|
||||
@dataclass
|
||||
class MaterialChangeReport:
|
||||
"""已废弃：物料变更报送，请使用统一的WorkstationReportRequest"""
|
||||
pass
|
||||
|
||||
|
||||
@dataclass
|
||||
class TaskExecutionReport:
|
||||
"""已废弃：任务执行报送，请使用统一的WorkstationReportRequest"""
|
||||
pass
|
||||
|
||||
|
||||
# 导出列表
|
||||
__all__ = [
|
||||
'WorkstationReportRequest',
|
||||
'MaterialUsage',
|
||||
'HttpResponse',
|
||||
'WorkstationHTTPService',
|
||||
# 向后兼容
|
||||
'MaterialChangeReport',
|
||||
'TaskExecutionReport'
|
||||
]
|
||||
@@ -1,583 +0,0 @@
|
||||
"""
|
||||
工作站物料管理基类
|
||||
Workstation Material Management Base Class
|
||||
|
||||
基于PyLabRobot的物料管理系统
|
||||
"""
|
||||
from typing import Dict, Any, List, Optional, Union, Type
|
||||
from abc import ABC, abstractmethod
|
||||
import json
|
||||
|
||||
from pylabrobot.resources import (
|
||||
Resource as PLRResource,
|
||||
Container,
|
||||
Deck,
|
||||
Coordinate as PLRCoordinate,
|
||||
)
|
||||
|
||||
from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker
|
||||
from unilabos.utils.log import logger
|
||||
from unilabos.resources.graphio import resource_plr_to_ulab, resource_ulab_to_plr
|
||||
|
||||
|
||||
class MaterialManagementBase(ABC):
|
||||
"""物料管理基类
|
||||
|
||||
定义工作站物料管理的标准接口：
|
||||
1. 物料初始化 - 根据配置创建物料资源
|
||||
2. 物料追踪 - 实时跟踪物料位置和状态
|
||||
3. 物料查找 - 按类型、位置、状态查找物料
|
||||
4. 物料转换 - PyLabRobot与UniLab资源格式转换
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
device_id: str,
|
||||
deck_config: Dict[str, Any],
|
||||
resource_tracker: DeviceNodeResourceTracker,
|
||||
children_config: Dict[str, Dict[str, Any]] = None
|
||||
):
|
||||
self.device_id = device_id
|
||||
self.deck_config = deck_config
|
||||
self.resource_tracker = resource_tracker
|
||||
self.children_config = children_config or {}
|
||||
|
||||
# 创建主台面
|
||||
self.plr_deck = self._create_deck()
|
||||
|
||||
# 扩展ResourceTracker
|
||||
self._extend_resource_tracker()
|
||||
|
||||
# 注册deck到resource tracker
|
||||
self.resource_tracker.add_resource(self.plr_deck)
|
||||
|
||||
# 初始化子资源
|
||||
self.plr_resources = {}
|
||||
self._initialize_materials()
|
||||
|
||||
def _create_deck(self) -> Deck:
|
||||
"""创建主台面"""
|
||||
return Deck(
|
||||
name=f"{self.device_id}_deck",
|
||||
size_x=self.deck_config.get("size_x", 1000.0),
|
||||
size_y=self.deck_config.get("size_y", 1000.0),
|
||||
size_z=self.deck_config.get("size_z", 500.0),
|
||||
origin=PLRCoordinate(0, 0, 0)
|
||||
)
|
||||
|
||||
def _extend_resource_tracker(self):
|
||||
"""扩展ResourceTracker以支持PyLabRobot特定功能"""
|
||||
|
||||
def find_by_type(resource_type):
|
||||
"""按类型查找资源"""
|
||||
return self._find_resources_by_type_recursive(self.plr_deck, resource_type)
|
||||
|
||||
def find_by_category(category: str):
|
||||
"""按类别查找资源"""
|
||||
found = []
|
||||
for resource in self._get_all_resources():
|
||||
if hasattr(resource, 'category') and resource.category == category:
|
||||
found.append(resource)
|
||||
return found
|
||||
|
||||
def find_by_name_pattern(pattern: str):
|
||||
"""按名称模式查找资源"""
|
||||
import re
|
||||
found = []
|
||||
for resource in self._get_all_resources():
|
||||
if re.search(pattern, resource.name):
|
||||
found.append(resource)
|
||||
return found
|
||||
|
||||
# 动态添加方法到resource_tracker
|
||||
self.resource_tracker.find_by_type = find_by_type
|
||||
self.resource_tracker.find_by_category = find_by_category
|
||||
self.resource_tracker.find_by_name_pattern = find_by_name_pattern
|
||||
|
||||
def _find_resources_by_type_recursive(self, resource, target_type):
|
||||
"""递归查找指定类型的资源"""
|
||||
found = []
|
||||
if isinstance(resource, target_type):
|
||||
found.append(resource)
|
||||
|
||||
# 递归查找子资源
|
||||
children = getattr(resource, "children", [])
|
||||
for child in children:
|
||||
found.extend(self._find_resources_by_type_recursive(child, target_type))
|
||||
|
||||
return found
|
||||
|
||||
def _get_all_resources(self) -> List[PLRResource]:
|
||||
"""获取所有资源"""
|
||||
all_resources = []
|
||||
|
||||
def collect_resources(resource):
|
||||
all_resources.append(resource)
|
||||
children = getattr(resource, "children", [])
|
||||
for child in children:
|
||||
collect_resources(child)
|
||||
|
||||
collect_resources(self.plr_deck)
|
||||
return all_resources
|
||||
|
||||
def _initialize_materials(self):
|
||||
"""初始化物料"""
|
||||
try:
|
||||
# 确定创建顺序，确保父资源先于子资源创建
|
||||
creation_order = self._determine_creation_order()
|
||||
|
||||
# 按顺序创建资源
|
||||
for resource_id in creation_order:
|
||||
config = self.children_config[resource_id]
|
||||
self._create_plr_resource(resource_id, config)
|
||||
|
||||
logger.info(f"物料管理系统初始化完成，共创建 {len(self.plr_resources)} 个资源")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"物料初始化失败: {e}")
|
||||
|
||||
def _determine_creation_order(self) -> List[str]:
|
||||
"""确定资源创建顺序"""
|
||||
order = []
|
||||
visited = set()
|
||||
|
||||
def visit(resource_id: str):
|
||||
if resource_id in visited:
|
||||
return
|
||||
visited.add(resource_id)
|
||||
|
||||
config = self.children_config.get(resource_id, {})
|
||||
parent_id = config.get("parent")
|
||||
|
||||
# 如果有父资源，先访问父资源
|
||||
if parent_id and parent_id in self.children_config:
|
||||
visit(parent_id)
|
||||
|
||||
order.append(resource_id)
|
||||
|
||||
for resource_id in self.children_config:
|
||||
visit(resource_id)
|
||||
|
||||
return order
|
||||
|
||||
def _create_plr_resource(self, resource_id: str, config: Dict[str, Any]):
|
||||
"""创建PyLabRobot资源"""
|
||||
try:
|
||||
resource_type = config.get("type", "unknown")
|
||||
data = config.get("data", {})
|
||||
location_config = config.get("location", {})
|
||||
|
||||
# 创建位置坐标
|
||||
location = PLRCoordinate(
|
||||
x=location_config.get("x", 0.0),
|
||||
y=location_config.get("y", 0.0),
|
||||
z=location_config.get("z", 0.0)
|
||||
)
|
||||
|
||||
# 根据类型创建资源
|
||||
resource = self._create_resource_by_type(resource_id, resource_type, config, data, location)
|
||||
|
||||
if resource:
|
||||
# 设置父子关系
|
||||
parent_id = config.get("parent")
|
||||
if parent_id and parent_id in self.plr_resources:
|
||||
parent_resource = self.plr_resources[parent_id]
|
||||
parent_resource.assign_child_resource(resource, location)
|
||||
else:
|
||||
# 直接放在deck上
|
||||
self.plr_deck.assign_child_resource(resource, location)
|
||||
|
||||
# 保存资源引用
|
||||
self.plr_resources[resource_id] = resource
|
||||
|
||||
# 注册到resource tracker
|
||||
self.resource_tracker.add_resource(resource)
|
||||
|
||||
logger.debug(f"创建资源成功: {resource_id} ({resource_type})")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"创建资源失败 {resource_id}: {e}")
|
||||
|
||||
@abstractmethod
|
||||
def _create_resource_by_type(
|
||||
self,
|
||||
resource_id: str,
|
||||
resource_type: str,
|
||||
config: Dict[str, Any],
|
||||
data: Dict[str, Any],
|
||||
location: PLRCoordinate
|
||||
) -> Optional[PLRResource]:
|
||||
"""根据类型创建资源 - 子类必须实现"""
|
||||
pass
|
||||
|
||||
# ============ 物料查找接口 ============
|
||||
|
||||
def find_materials_by_type(self, material_type: str) -> List[PLRResource]:
|
||||
"""按材料类型查找物料"""
|
||||
return self.resource_tracker.find_by_category(material_type)
|
||||
|
||||
def find_material_by_id(self, resource_id: str) -> Optional[PLRResource]:
|
||||
"""按ID查找物料"""
|
||||
return self.plr_resources.get(resource_id)
|
||||
|
||||
def find_available_positions(self, position_type: str) -> List[PLRResource]:
|
||||
"""查找可用位置"""
|
||||
positions = self.resource_tracker.find_by_category(position_type)
|
||||
available = []
|
||||
|
||||
for pos in positions:
|
||||
if hasattr(pos, 'is_available') and pos.is_available():
|
||||
available.append(pos)
|
||||
elif hasattr(pos, 'children') and len(pos.children) == 0:
|
||||
available.append(pos)
|
||||
|
||||
return available
|
||||
|
||||
def get_material_inventory(self) -> Dict[str, int]:
|
||||
"""获取物料库存统计"""
|
||||
inventory = {}
|
||||
|
||||
for resource in self._get_all_resources():
|
||||
if hasattr(resource, 'category'):
|
||||
category = resource.category
|
||||
inventory[category] = inventory.get(category, 0) + 1
|
||||
|
||||
return inventory
|
||||
|
||||
# ============ 物料状态更新接口 ============
|
||||
|
||||
def update_material_location(self, material_id: str, new_location: PLRCoordinate) -> bool:
|
||||
"""更新物料位置"""
|
||||
try:
|
||||
material = self.find_material_by_id(material_id)
|
||||
if material:
|
||||
material.location = new_location
|
||||
return True
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"更新物料位置失败: {e}")
|
||||
return False
|
||||
|
||||
def move_material(self, material_id: str, target_container_id: str) -> bool:
|
||||
"""移动物料到目标容器"""
|
||||
try:
|
||||
material = self.find_material_by_id(material_id)
|
||||
target = self.find_material_by_id(target_container_id)
|
||||
|
||||
if material and target:
|
||||
# 从原位置移除
|
||||
if material.parent:
|
||||
material.parent.unassign_child_resource(material)
|
||||
|
||||
# 添加到新位置
|
||||
target.assign_child_resource(material)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"移动物料失败: {e}")
|
||||
return False
|
||||
|
||||
# ============ 资源转换接口 ============
|
||||
|
||||
def convert_to_unilab_format(self, plr_resource: PLRResource) -> Dict[str, Any]:
|
||||
"""将PyLabRobot资源转换为UniLab格式"""
|
||||
return resource_plr_to_ulab(plr_resource)
|
||||
|
||||
def convert_from_unilab_format(self, unilab_resource: Dict[str, Any]) -> PLRResource:
|
||||
"""将UniLab格式转换为PyLabRobot资源"""
|
||||
return resource_ulab_to_plr(unilab_resource)
|
||||
|
||||
def get_deck_state(self) -> Dict[str, Any]:
|
||||
"""获取Deck状态"""
|
||||
try:
|
||||
return {
|
||||
"deck_info": {
|
||||
"name": self.plr_deck.name,
|
||||
"size": {
|
||||
"x": self.plr_deck.size_x,
|
||||
"y": self.plr_deck.size_y,
|
||||
"z": self.plr_deck.size_z
|
||||
},
|
||||
"children_count": len(self.plr_deck.children)
|
||||
},
|
||||
"resources": {
|
||||
resource_id: self.convert_to_unilab_format(resource)
|
||||
for resource_id, resource in self.plr_resources.items()
|
||||
},
|
||||
"inventory": self.get_material_inventory()
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"获取Deck状态失败: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
# ============ 数据持久化接口 ============
|
||||
|
||||
def save_state_to_file(self, file_path: str) -> bool:
|
||||
"""保存状态到文件"""
|
||||
try:
|
||||
state = self.get_deck_state()
|
||||
with open(file_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(state, f, indent=2, ensure_ascii=False)
|
||||
logger.info(f"状态已保存到: {file_path}")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"保存状态失败: {e}")
|
||||
return False
|
||||
|
||||
def load_state_from_file(self, file_path: str) -> bool:
|
||||
"""从文件加载状态"""
|
||||
try:
|
||||
with open(file_path, 'r', encoding='utf-8') as f:
|
||||
state = json.load(f)
|
||||
|
||||
# 重新创建资源
|
||||
self._recreate_resources_from_state(state)
|
||||
logger.info(f"状态已从文件加载: {file_path}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"加载状态失败: {e}")
|
||||
return False
|
||||
|
||||
def _recreate_resources_from_state(self, state: Dict[str, Any]):
|
||||
"""从状态重新创建资源"""
|
||||
# 清除现有资源
|
||||
self.plr_resources.clear()
|
||||
self.plr_deck.children.clear()
|
||||
|
||||
# 从状态重新创建
|
||||
resources_data = state.get("resources", {})
|
||||
for resource_id, resource_data in resources_data.items():
|
||||
try:
|
||||
plr_resource = self.convert_from_unilab_format(resource_data)
|
||||
self.plr_resources[resource_id] = plr_resource
|
||||
self.plr_deck.assign_child_resource(plr_resource)
|
||||
except Exception as e:
|
||||
logger.error(f"重新创建资源失败 {resource_id}: {e}")
|
||||
|
||||
|
||||
class CoinCellMaterialManagement(MaterialManagementBase):
|
||||
"""纽扣电池物料管理类
|
||||
|
||||
从 button_battery_station 抽取的物料管理功能
|
||||
"""
|
||||
|
||||
def _create_resource_by_type(
|
||||
self,
|
||||
resource_id: str,
|
||||
resource_type: str,
|
||||
config: Dict[str, Any],
|
||||
data: Dict[str, Any],
|
||||
location: PLRCoordinate
|
||||
) -> Optional[PLRResource]:
|
||||
"""根据类型创建纽扣电池相关资源"""
|
||||
|
||||
# 导入纽扣电池资源类
|
||||
from unilabos.device_comms.button_battery_station import (
|
||||
MaterialPlate, PlateSlot, ClipMagazine, BatteryPressSlot,
|
||||
TipBox64, WasteTipBox, BottleRack, Battery, ElectrodeSheet
|
||||
)
|
||||
|
||||
try:
|
||||
if resource_type == "material_plate":
|
||||
return self._create_material_plate(resource_id, config, data, location)
|
||||
|
||||
elif resource_type == "plate_slot":
|
||||
return self._create_plate_slot(resource_id, config, data, location)
|
||||
|
||||
elif resource_type == "clip_magazine":
|
||||
return self._create_clip_magazine(resource_id, config, data, location)
|
||||
|
||||
elif resource_type == "battery_press_slot":
|
||||
return self._create_battery_press_slot(resource_id, config, data, location)
|
||||
|
||||
elif resource_type == "tip_box":
|
||||
return self._create_tip_box(resource_id, config, data, location)
|
||||
|
||||
elif resource_type == "waste_tip_box":
|
||||
return self._create_waste_tip_box(resource_id, config, data, location)
|
||||
|
||||
elif resource_type == "bottle_rack":
|
||||
return self._create_bottle_rack(resource_id, config, data, location)
|
||||
|
||||
elif resource_type == "battery":
|
||||
return self._create_battery(resource_id, config, data, location)
|
||||
|
||||
else:
|
||||
logger.warning(f"未知的资源类型: {resource_type}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"创建资源失败 {resource_id} ({resource_type}): {e}")
|
||||
return None
|
||||
|
||||
def _create_material_plate(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
||||
"""创建料板"""
|
||||
from unilabos.device_comms.button_battery_station import MaterialPlate, ElectrodeSheet
|
||||
|
||||
plate = MaterialPlate(
|
||||
name=resource_id,
|
||||
size_x=config.get("size_x", 80.0),
|
||||
size_y=config.get("size_y", 80.0),
|
||||
size_z=config.get("size_z", 10.0),
|
||||
hole_diameter=config.get("hole_diameter", 15.0),
|
||||
hole_depth=config.get("hole_depth", 8.0),
|
||||
hole_spacing_x=config.get("hole_spacing_x", 20.0),
|
||||
hole_spacing_y=config.get("hole_spacing_y", 20.0),
|
||||
number=data.get("number", "")
|
||||
)
|
||||
plate.location = location
|
||||
|
||||
# 如果有预填充的极片数据，创建极片
|
||||
electrode_sheets = data.get("electrode_sheets", [])
|
||||
for i, sheet_data in enumerate(electrode_sheets):
|
||||
if i < len(plate.children): # 确保不超过洞位数量
|
||||
hole = plate.children[i]
|
||||
sheet = ElectrodeSheet(
|
||||
name=f"{resource_id}_sheet_{i}",
|
||||
diameter=sheet_data.get("diameter", 14.0),
|
||||
thickness=sheet_data.get("thickness", 0.1),
|
||||
mass=sheet_data.get("mass", 0.01),
|
||||
material_type=sheet_data.get("material_type", "cathode"),
|
||||
info=sheet_data.get("info", "")
|
||||
)
|
||||
hole.place_electrode_sheet(sheet)
|
||||
|
||||
return plate
|
||||
|
||||
def _create_plate_slot(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
||||
"""创建板槽位"""
|
||||
from unilabos.device_comms.button_battery_station import PlateSlot
|
||||
|
||||
slot = PlateSlot(
|
||||
name=resource_id,
|
||||
max_plates=config.get("max_plates", 8)
|
||||
)
|
||||
slot.location = location
|
||||
return slot
|
||||
|
||||
def _create_clip_magazine(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
||||
"""创建子弹夹"""
|
||||
from unilabos.device_comms.button_battery_station import ClipMagazine
|
||||
|
||||
magazine = ClipMagazine(
|
||||
name=resource_id,
|
||||
size_x=config.get("size_x", 150.0),
|
||||
size_y=config.get("size_y", 100.0),
|
||||
size_z=config.get("size_z", 50.0),
|
||||
hole_diameter=config.get("hole_diameter", 15.0),
|
||||
hole_depth=config.get("hole_depth", 40.0),
|
||||
hole_spacing=config.get("hole_spacing", 25.0),
|
||||
max_sheets_per_hole=config.get("max_sheets_per_hole", 100)
|
||||
)
|
||||
magazine.location = location
|
||||
return magazine
|
||||
|
||||
def _create_battery_press_slot(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
||||
"""创建电池压制槽"""
|
||||
from unilabos.device_comms.button_battery_station import BatteryPressSlot
|
||||
|
||||
slot = BatteryPressSlot(
|
||||
name=resource_id,
|
||||
diameter=config.get("diameter", 20.0),
|
||||
depth=config.get("depth", 15.0)
|
||||
)
|
||||
slot.location = location
|
||||
return slot
|
||||
|
||||
def _create_tip_box(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
||||
"""创建枪头盒"""
|
||||
from unilabos.device_comms.button_battery_station import TipBox64
|
||||
|
||||
tip_box = TipBox64(
|
||||
name=resource_id,
|
||||
size_x=config.get("size_x", 127.8),
|
||||
size_y=config.get("size_y", 85.5),
|
||||
size_z=config.get("size_z", 60.0),
|
||||
with_tips=data.get("with_tips", True)
|
||||
)
|
||||
tip_box.location = location
|
||||
return tip_box
|
||||
|
||||
def _create_waste_tip_box(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
||||
"""创建废枪头盒"""
|
||||
from unilabos.device_comms.button_battery_station import WasteTipBox
|
||||
|
||||
waste_box = WasteTipBox(
|
||||
name=resource_id,
|
||||
size_x=config.get("size_x", 127.8),
|
||||
size_y=config.get("size_y", 85.5),
|
||||
size_z=config.get("size_z", 60.0),
|
||||
max_tips=config.get("max_tips", 100)
|
||||
)
|
||||
waste_box.location = location
|
||||
return waste_box
|
||||
|
||||
def _create_bottle_rack(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
||||
"""创建瓶架"""
|
||||
from unilabos.device_comms.button_battery_station import BottleRack
|
||||
|
||||
rack = BottleRack(
|
||||
name=resource_id,
|
||||
size_x=config.get("size_x", 210.0),
|
||||
size_y=config.get("size_y", 140.0),
|
||||
size_z=config.get("size_z", 100.0),
|
||||
bottle_diameter=config.get("bottle_diameter", 30.0),
|
||||
bottle_height=config.get("bottle_height", 100.0),
|
||||
position_spacing=config.get("position_spacing", 35.0)
|
||||
)
|
||||
rack.location = location
|
||||
return rack
|
||||
|
||||
def _create_battery(self, resource_id: str, config: Dict[str, Any], data: Dict[str, Any], location: PLRCoordinate):
|
||||
"""创建电池"""
|
||||
from unilabos.device_comms.button_battery_station import Battery
|
||||
|
||||
battery = Battery(
|
||||
name=resource_id,
|
||||
diameter=config.get("diameter", 20.0),
|
||||
height=config.get("height", 3.2),
|
||||
max_volume=config.get("max_volume", 100.0),
|
||||
barcode=data.get("barcode", "")
|
||||
)
|
||||
battery.location = location
|
||||
return battery
|
||||
|
||||
# ============ 纽扣电池特定查找方法 ============
|
||||
|
||||
def find_material_plates(self):
|
||||
"""查找所有料板"""
|
||||
from unilabos.device_comms.button_battery_station import MaterialPlate
|
||||
return self.resource_tracker.find_by_type(MaterialPlate)
|
||||
|
||||
def find_batteries(self):
|
||||
"""查找所有电池"""
|
||||
from unilabos.device_comms.button_battery_station import Battery
|
||||
return self.resource_tracker.find_by_type(Battery)
|
||||
|
||||
def find_electrode_sheets(self):
|
||||
"""查找所有极片"""
|
||||
found = []
|
||||
plates = self.find_material_plates()
|
||||
for plate in plates:
|
||||
for hole in plate.children:
|
||||
if hasattr(hole, 'has_electrode_sheet') and hole.has_electrode_sheet():
|
||||
found.append(hole._electrode_sheet)
|
||||
return found
|
||||
|
||||
def find_plate_slots(self):
|
||||
"""查找所有板槽位"""
|
||||
from unilabos.device_comms.button_battery_station import PlateSlot
|
||||
return self.resource_tracker.find_by_type(PlateSlot)
|
||||
|
||||
def find_clip_magazines(self):
|
||||
"""查找所有子弹夹"""
|
||||
from unilabos.device_comms.button_battery_station import ClipMagazine
|
||||
return self.resource_tracker.find_by_type(ClipMagazine)
|
||||
|
||||
def find_press_slots(self):
|
||||
"""查找所有压制槽"""
|
||||
from unilabos.device_comms.button_battery_station import BatteryPressSlot
|
||||
return self.resource_tracker.find_by_type(BatteryPressSlot)
|
||||
Reference in New Issue
Block a user