open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2025-12-17 13:01:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(workstation): 移除旧版bioyond设备配置并优化扣电组装工作站- 删除bioyond.yaml和bioyond_dispensing_station.yaml旧设备配置文件- 优化扣电组装工作站配置，移除不必要的子资源引用- 更新Modbus通信地址和端口配置- 简化CoinCellAssemblyWorkstation类的初始化参数- 移除冗余的deck资源创建逻辑

Browse Source 
- 更新反应站配置文件中drip_back命令的位置
- 添加新的Modbus寄存器和线圈定义
- 移除workstation_base.py基类文件
This commit is contained in:
calvincao
2025-10-29 10:44:30 +08:00
parent a97781c4eb
commit a130c03ebd
7 changed files with 88 additions and 1220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
unilabos/devices/workstation/coin_cell_assembly/coin_cell_assembly.py
View File

@@ -8,34 +8,34 @@ from typing import Any, Dict, Optional
from pylabrobot.resources import Resource as PLRResource
from unilabos_msgs.msg import Resource
from unilabos.device_comms.modbus_plc.client import ModbusTcpClient
from unilabos.devices.workstation.coin_cell_assembly.button_battery_station import MaterialHole, MaterialPlate
from unilabos.devices.workstation.workstation_base import WorkstationBase
from unilabos.device_comms.modbus_plc.client import TCPClient, ModbusNode, PLCWorkflow, ModbusWorkflow, WorkflowAction, BaseClient
from unilabos.device_comms.modbus_plc.modbus import DeviceType, Base as ModbusNodeBase, DataType, WorderOrder
from unilabos.devices.workstation.coin_cell_assembly.button_battery_station import *
from unilabos.ros.nodes.base_device_node import ROS2DeviceNode, BaseROS2DeviceNode
from unilabos.ros.nodes.presets.workstation import ROS2WorkstationNode
from unilabos.devices.workstation.coin_cell_assembly.button_battery_station import CoincellDeck
#构建物料系统
class CoinCellAssemblyWorkstation(WorkstationBase):
def __init__(
self,
station_resource: CoincellDeck,
deck: Deck=None,
address: str = "172.21.32.20",
port: str = "502",
debug_mode: bool = True,
debug_mode: bool = False,
*args,
**kwargs,
):
super().__init__(
#桌子
station_resource=station_resource,
deck=deck,
*args,
**kwargs,
)
self.debug_mode = debug_mode
self.station_resource = station_resource
self.deck = deck
""" 连接初始化 """
modbus_client = TCPClient(addr=address, port=port)
print("modbus_client", modbus_client)
@@ -75,7 +75,7 @@ class CoinCellAssemblyWorkstation(WorkstationBase):
self._ros_node = ros_node
#self.deck = create_a_coin_cell_deck()
ROS2DeviceNode.run_async_func(self._ros_node.update_resource, True, **{
"resources": [self.station_resource]
"resources": [self.deck]
})
# 批量操作在这里写
@@ -85,7 +85,7 @@ class CoinCellAssemblyWorkstation(WorkstationBase):
async def fill_plate(self):
plate_1: MaterialPlate = self.station_resource.children[0].children[0]
plate_1: MaterialPlate = self.deck.children[0].children[0]
#plate_1
return await self._ros_node.update_resource(plate_1)
@@ -342,7 +342,7 @@ class CoinCellAssemblyWorkstation(WorkstationBase):
def modify_deck_name(self, resource_name: str):
# figure_res = self._ros_node.resource_tracker.figure_resource({"name": resource_name})
# print(f"!!! figure_res: {type(figure_res)}")
self.station_resource.children[1]
self.deck.children[1]
return
@property
@@ -1105,8 +1105,17 @@ class CoinCellAssemblyWorkstation(WorkstationBase):
if __name__ == "__main__":
from pylabrobot.resources import Resource
Coin_Cell = CoinCellAssemblyWorkstation(Resource("1", 1, 1, 1), debug_mode=True)
# modbus_client = TCPClient(addr="172.21.32.20", port="502")
# # modbus_client.client.connect()
# nodes = BaseClient.load_csv(os.path.join(os.path.dirname(__file__), 'coin_cell_assembly_a.csv'))
# client = modbus_client.register_node_list(nodes)
# # print("modbus_client", modbus_client)
# while True:
# time.sleep(1)
# cmd_feedback, read_err = modbus_client.use_node('COIL_SYS_AUTO_CMD').read(1)
# print("modbus_client", cmd_feedback)
Coin_Cell = CoinCellAssemblyWorkstation()
print(Coin_Cell.deck)
#Coin_Cell.func_pack_device_init()
#Coin_Cell.func_pack_device_auto()
#Coin_Cell.func_pack_device_start()
@@ -1121,7 +1130,7 @@ if __name__ == "__main__":
#print("success")
#创建一个物料台面
deck = create_a_coin_cell_deck()
# deck = create_a_coin_cell_deck()
#deck = create_a_full_coin_cell_deck()
@@ -1169,4 +1178,6 @@ if __name__ == "__main__":
#print(resources)
http_client.remote_addr = "https://uni-lab.test.bohrium.com/api/v1"
http_client.resource_add(resources)
http_client.resource_add(resources)

18
unilabos/devices/workstation/coin_cell_assembly/coin_cell_assembly_a.csv
View File

@@ -43,3 +43,21 @@ REG_DATA_ELECTROLYTE_USE_NUM,INT16,,,,hold_register,10000,
UNILAB_SEND_FINISHED_CMD,BOOL,,,,coil,8730,
UNILAB_RECE_FINISHED_CMD,BOOL,,,,coil,8530,
REG_DATA_ASSEMBLY_TYPE,INT16,,,,hold_register,10018,ASSEMBLY_TYPE7or8
COIL_ALUMINUM_FOIL,BOOL,,ʹ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>,,coil,8340,
REG_MSG_NE_PLATE_MATRIX,INT16,,<EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƭ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>λ,,hold_register,440,
REG_MSG_SEPARATOR_PLATE_MATRIX,INT16,,<EFBFBD><EFBFBD>Ĥ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>λ,,hold_register,450,
REG_MSG_TIP_BOX_MATRIX,INT16,,<EFBFBD><EFBFBD>Һǹͷ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>λ,,hold_register,480,
REG_MSG_NE_PLATE_NUM,INT16,,<EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƭ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>,,hold_register,443,
REG_MSG_SEPARATOR_PLATE_NUM,INT16,,<EFBFBD><EFBFBD>Ĥ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>,,hold_register,453,
REG_MSG_PRESS_MODE,BOOL,,ѹ<EFBFBD><EFBFBD>ģʽ<EFBFBD><EFBFBD>false:ѹ<><D1B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ģʽ<C4A3><CABD>True:<3A><><EFBFBD><EFBFBD>ģʽ<C4A3><CABD>,,coil,8360,<EFBFBD><EFBFBD><EFBFBD><EFBFBD>ѹ<EFBFBD><EFBFBD>ģʽ
,,,,,,,
,BOOL,,<EFBFBD>Ӿ<EFBFBD><EFBFBD><EFBFBD>λ<EFBFBD><EFBFBD>false:ʹ<>ã<EFBFBD>true:<3A><><EFBFBD>ԣ<EFBFBD>,,coil,8300,<EFBFBD>Ӿ<EFBFBD><EFBFBD><EFBFBD>λ
,BOOL,,<EFBFBD><EFBFBD><EFBFBD>죨false:ʹ<>ã<EFBFBD>true:<3A><><EFBFBD>ԣ<EFBFBD>,,coil,8310,<EFBFBD>Ӿ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
,BOOL,,<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>_<EFBFBD><EFBFBD><EFBFBD>֣<EFBFBD>false:ʹ<>ã<EFBFBD>true:<3A><><EFBFBD>ԣ<EFBFBD>,,coil,8320,<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
,BOOL,,<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>_<EFBFBD>Ҳ֣<EFBFBD>false:ʹ<>ã<EFBFBD>true:<3A><><EFBFBD>ԣ<EFBFBD>,,coil,8420,<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ҳ<EFBFBD>
,BOOL,,<EFBFBD><EFBFBD><EFBFBD>ռ<EFBFBD>֪<EFBFBD><EFBFBD>false:ʹ<>ã<EFBFBD>true:<3A><><EFBFBD>ԣ<EFBFBD>,,coil,8350,<EFBFBD><EFBFBD><EFBFBD>ռ<EFBFBD>֪
,BOOL,,<EFBFBD><EFBFBD><EFBFBD><EFBFBD>Һ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>ģʽ<EFBFBD><EFBFBD>false:<3A><><EFBFBD>ε<EFBFBD>Һ<EFBFBD><D2BA>true:<3A><><EFBFBD>ε<EFBFBD>Һ<EFBFBD><D2BA>,,coil,8370,<EFBFBD><EFBFBD>Һģʽ
,BOOL,,<EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƭ<EFBFBD><EFBFBD><EFBFBD>أ<EFBFBD>false:ʹ<>ã<EFBFBD>true:<3A><><EFBFBD>ԣ<EFBFBD>,,coil,8380,<EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƭ<EFBFBD><EFBFBD><EFBFBD><EFBFBD>
,BOOL,,<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƭ<EFBFBD><EFBFBD>װ<EFBFBD><EFBFBD>ʽ<EFBFBD><EFBFBD>false:<3A><>װ<EFBFBD><D7B0>true:<3A><>װ<EFBFBD><D7B0>,,coil,8390,<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>װ
,BOOL,,ѹ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ࣨfalse:ʹ<>ã<EFBFBD>true:<3A><><EFBFBD>ԣ<EFBFBD>,,coil,8400,ѹ<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
,BOOL,,<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>̰<EFBFBD><EFBFBD>̷<EFBFBD>ʽ<EFBFBD><EFBFBD>false:ˮƽ<CBAE><C6BD><EFBFBD>̣<EFBFBD>true:<3A>ѵ<EFBFBD><D1B5><EFBFBD><EFBFBD>̣<EFBFBD>,,coil,8410,<EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƭ<EFBFBD><EFBFBD><EFBFBD>̷<EFBFBD>ʽ
1 Name DataType InitValue Comment Attribute DeviceType Address
43 UNILAB_SEND_FINISHED_CMD BOOL coil 8730
44 UNILAB_RECE_FINISHED_CMD BOOL coil 8530
45 REG_DATA_ASSEMBLY_TYPE INT16 hold_register 10018 ASSEMBLY_TYPE7or8
46 COIL_ALUMINUM_FOIL BOOL ʹ coil 8340
47 REG_MSG_NE_PLATE_MATRIX INT16 Ƭλ hold_register 440
48 REG_MSG_SEPARATOR_PLATE_MATRIX INT16 Ĥλ hold_register 450
49 REG_MSG_TIP_BOX_MATRIX INT16 Һǹͷλ hold_register 480
50 REG_MSG_NE_PLATE_NUM INT16 Ƭ hold_register 443
51 REG_MSG_SEPARATOR_PLATE_NUM INT16 Ĥ hold_register 453
52 REG_MSG_PRESS_MODE BOOL ѹģʽfalse:ѹģʽTrue:ģʽ coil 8360 ѹģʽ
53
54 BOOL Ӿλfalse:ʹãtrue:ԣ coil 8300 Ӿλ
55 BOOL 죨false:ʹãtrue:ԣ coil 8310 Ӿ
56 BOOL _֣false:ʹãtrue:ԣ coil 8320
57 BOOL _Ҳ֣false:ʹãtrue:ԣ coil 8420 Ҳ
58 BOOL ռ֪false:ʹãtrue:ԣ coil 8350 ռ֪
59 BOOL Һģʽfalse:εҺtrue:εҺ coil 8370 Һģʽ
60 BOOL Ƭأfalse:ʹãtrue:ԣ coil 8380 Ƭ
61 BOOL Ƭװʽfalse:װtrue:װ coil 8390 װ
62 BOOL ѹࣨfalse:ʹãtrue:ԣ coil 8400 ѹ
63 BOOL ̷̰ʽfalse:ˮƽ̣true:ѵ̣ coil 8410 Ƭ̷ʽ

489
unilabos/devices/workstation/coin_cell_assembly/workstation_base.py
View File

@@ -1,489 +0,0 @@
"""
工作站基类
Workstation Base Class - 简化版
基于PLR Deck的简化工作站架构
专注于核心物料系统和工作流管理
"""
import collections
import time
from typing import Dict, Any, List, Optional, Union
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from pylabrobot.resources import Deck, Plate, Resource as PLRResource
from pylabrobot.resources.coordinate import Coordinate
from unilabos.ros.nodes.presets.workstation import ROS2WorkstationNode
from unilabos.utils.log import logger
class WorkflowStatus(Enum):
"""工作流状态"""
IDLE = "idle"
INITIALIZING = "initializing"
RUNNING = "running"
PAUSED = "paused"
STOPPING = "stopping"
STOPPED = "stopped"
ERROR = "error"
COMPLETED = "completed"
@dataclass
class WorkflowInfo:
"""工作流信息"""
name: str
description: str
estimated_duration: float # 预估持续时间（秒）
required_materials: List[str] # 所需物料类型
output_product: str # 输出产品类型
parameters_schema: Dict[str, Any] # 参数架构
class WorkStationContainer(Plate):
"""
WorkStation 专用 Container 类，继承自 Plate和TipRack
注意这个物料必须通过plr_additional_res_reg.py注册到edge，才能正常序列化
"""
def __init__(
self,
name: str,
size_x: float,
size_y: float,
size_z: float,
category: str,
ordering: collections.OrderedDict,
model: Optional[str] = None,
):
"""
这里的初始化入参要和plr的保持一致
"""
super().__init__(name, size_x, size_y, size_z, category=category, ordering=ordering, model=model)
self._unilabos_state = {} # 必须有此行，自己的类描述的是物料的
def load_state(self, state: Dict[str, Any]) -> None:
"""从给定的状态加载工作台信息。"""
super().load_state(state)
self._unilabos_state = state
def serialize_state(self) -> Dict[str, Dict[str, Any]]:
data = super().serialize_state()
data.update(
self._unilabos_state
) # Container自身的信息，云端物料将保存这一data，本地也通过这里的data进行读写，当前类用来表示这个物料的长宽高大小的属性，而data（state用来表示物料的内容，细节等）
return data
def get_workstation_plate_resource(name: str) -> PLRResource: # 要给定一个返回plr的方法
"""
用于获取一些模板，例如返回一个带有特定信息/子物料的 Plate，这里需要到注册表注册，例如unilabos/registry/resources/organic/workstation.yaml
可以直接运行该函数或者利用注册表补全机制，来检查是否资源出错
:param name: 资源名称
:return: Resource对象
"""
plate = WorkStationContainer(
name, size_x=50, size_y=50, size_z=10, category="plate", ordering=collections.OrderedDict()
)
tip_rack = WorkStationContainer(
"tip_rack_inside_plate",
size_x=50,
size_y=50,
size_z=10,
category="tip_rack",
ordering=collections.OrderedDict(),
)
plate.assign_child_resource(tip_rack, Coordinate.zero())
return plate
class ResourceSynchronizer(ABC):
"""资源同步器基类
负责与外部物料系统的同步，并对 self.deck 做修改
"""
def __init__(self, workstation: "WorkstationBase"):
self.workstation = workstation
@abstractmethod
async def sync_from_external(self) -> bool:
"""从外部系统同步物料到本地deck"""
pass
@abstractmethod
async def sync_to_external(self, plr_resource: PLRResource) -> bool:
"""将本地物料同步到外部系统"""
pass
@abstractmethod
async def handle_external_change(self, change_info: Dict[str, Any]) -> bool:
"""处理外部系统的变更通知"""
pass
class WorkstationBase(ABC):
"""工作站基类 - 简化版
核心功能：
1. 基于 PLR Deck 的物料系统，支持格式转换
2. 可选的资源同步器支持外部物料系统
3. 简化的工作流管理
"""
_ros_node: ROS2WorkstationNode
@property
def _children(self) -> Dict[str, Any]: # 不要删除这个下划线，不然会自动导入注册表，后面改成装饰器识别
return self._ros_node.children
async def update_resource_example(self):
return await self._ros_node.update_resource([get_workstation_plate_resource("test")])
def __init__(
self,
station_resource: PLRResource,
*args,
**kwargs, # 必须有kwargs
):
# 基本配置
print(station_resource)
self.deck_config = station_resource
# PLR 物料系统
self.deck: Optional[Deck] = None
self.plr_resources: Dict[str, PLRResource] = {}
# 资源同步器（可选）
# self.resource_synchronizer = ResourceSynchronizer(self) # 要在driver中自行初始化，只有workstation用
# 硬件接口
self.hardware_interface: Union[Any, str] = None
# 工作流状态
self.current_workflow_status = WorkflowStatus.IDLE
self.current_workflow_info = None
self.workflow_start_time = None
self.workflow_parameters = {}
# 支持的工作流（静态预定义）
self.supported_workflows: Dict[str, WorkflowInfo] = {}
# 初始化物料系统
self._initialize_material_system()
# 注册支持的工作流
# self._register_supported_workflows()
# logger.info(f"工作站 {device_id} 初始化完成（简化版）")
def _initialize_material_system(self):
"""初始化物料系统 - 使用 graphio 转换"""
try:
from unilabos.resources.graphio import resource_ulab_to_plr
# # 1. 合并 deck_config 和 children 创建完整的资源树
# complete_resource_config = self._create_complete_resource_config()
# # 2. 使用 graphio 转换为 PLR 资源
# self.deck = resource_ulab_to_plr(complete_resource_config, plr_model=True)
# # 3. 建立资源映射
# self._build_resource_mappings(self.deck)
# # 4. 如果有资源同步器，执行初始同步
# if self.resource_synchronizer:
# # 这里可以异步执行，暂时跳过
# pass
# logger.info(f"工作站 {self.device_id} 物料系统初始化成功，创建了 {len(self.plr_resources)} 个资源")
pass
except Exception as e:
# logger.error(f"工作站 {self.device_id} 物料系统初始化失败: {e}")
raise
def _create_complete_resource_config(self) -> Dict[str, Any]:
"""创建完整的资源配置 - 合并 deck_config 和 children"""
# 创建主 deck 配置
deck_resource = {
"id": f"{self.device_id}_deck",
"name": f"{self.device_id}_deck",
"type": "deck",
"position": {"x": 0, "y": 0, "z": 0},
"config": {
"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", 100.0),
**{k: v for k, v in self.deck_config.items() if k not in ["size_x", "size_y", "size_z"]},
},
"data": {},
"children": [],
"parent": None,
}
# 添加子资源
if self._children:
children_list = []
for child_id, child_config in self._children.items():
child_resource = self._normalize_child_resource(child_id, child_config, deck_resource["id"])
children_list.append(child_resource)
deck_resource["children"] = children_list
return deck_resource
def _normalize_child_resource(self, resource_id: str, config: Dict[str, Any], parent_id: str) -> Dict[str, Any]:
"""标准化子资源配置"""
return {
"id": resource_id,
"name": config.get("name", resource_id),
"type": config.get("type", "container"),
"position": self._normalize_position(config.get("position", {})),
"config": config.get("config", {}),
"data": config.get("data", {}),
"children": [], # 简化版本：只支持一层子资源
"parent": parent_id,
}
def _normalize_position(self, position: Any) -> Dict[str, float]:
"""标准化位置信息"""
if isinstance(position, dict):
return {
"x": float(position.get("x", 0)),
"y": float(position.get("y", 0)),
"z": float(position.get("z", 0)),
}
elif isinstance(position, (list, tuple)) and len(position) >= 2:
return {
"x": float(position[0]),
"y": float(position[1]),
"z": float(position[2]) if len(position) > 2 else 0.0,
}
else:
return {"x": 0.0, "y": 0.0, "z": 0.0}
def _build_resource_mappings(self, deck: Deck):
"""递归构建资源映射"""
def add_resource_recursive(resource: PLRResource):
if hasattr(resource, "name"):
self.plr_resources[resource.name] = resource
if hasattr(resource, "children"):
for child in resource.children:
add_resource_recursive(child)
add_resource_recursive(deck)
# ============ 硬件接口管理 ============
def set_hardware_interface(self, hardware_interface: Union[Any, str]):
"""设置硬件接口"""
self.hardware_interface = hardware_interface
logger.info(f"工作站 {self.device_id} 硬件接口设置: {type(hardware_interface).__name__}")
def set_workstation_node(self, workstation_node: "ROS2WorkstationNode"):
"""设置协议节点引用（用于代理模式）"""
self._ros_node = workstation_node
logger.info(f"工作站 {self.device_id} 关联协议节点")
# ============ 设备操作接口 ============
def call_device_method(self, method: str, *args, **kwargs) -> Any:
"""调用设备方法的统一接口"""
# 1. 代理模式：通过协议节点转发
if isinstance(self.hardware_interface, str) and self.hardware_interface.startswith("proxy:"):
if not self._ros_node:
raise RuntimeError("代理模式需要设置workstation_node")
device_id = self.hardware_interface[6:] # 移除 "proxy:" 前缀
return self._ros_node.call_device_method(device_id, method, *args, **kwargs)
# 2. 直接模式：直接调用硬件接口方法
elif self.hardware_interface and hasattr(self.hardware_interface, method):
return getattr(self.hardware_interface, method)(*args, **kwargs)
else:
raise AttributeError(f"硬件接口不支持方法: {method}")
def get_device_status(self) -> Dict[str, Any]:
"""获取设备状态"""
try:
return self.call_device_method("get_status")
except AttributeError:
# 如果设备不支持get_status方法，返回基础状态
return {
"status": "unknown",
"interface_type": type(self.hardware_interface).__name__,
"timestamp": time.time(),
}
def is_device_available(self) -> bool:
"""检查设备是否可用"""
try:
self.get_device_status()
return True
except:
return False
# ============ 物料系统接口 ============
def get_deck(self) -> Deck:
"""获取主 Deck"""
return self.deck
def get_all_resources(self) -> Dict[str, PLRResource]:
"""获取所有 PLR 资源"""
return self.plr_resources.copy()
def find_resource_by_name(self, name: str) -> Optional[PLRResource]:
"""按名称查找资源"""
return self.plr_resources.get(name)
def find_resources_by_type(self, resource_type: type) -> List[PLRResource]:
"""按类型查找资源"""
return [res for res in self.plr_resources.values() if isinstance(res, resource_type)]
async def sync_with_external_system(self) -> bool:
"""与外部物料系统同步"""
if not self.resource_synchronizer:
logger.info(f"工作站 {self.device_id} 没有配置资源同步器")
return True
try:
success = await self.resource_synchronizer.sync_from_external()
if success:
logger.info(f"工作站 {self.device_id} 外部同步成功")
else:
logger.warning(f"工作站 {self.device_id} 外部同步失败")
return success
except Exception as e:
logger.error(f"工作站 {self.device_id} 外部同步异常: {e}")
return False
# ============ 简化的工作流控制 ============
def execute_workflow(self, workflow_name: str, parameters: Dict[str, Any]) -> bool:
"""执行工作流"""
try:
# 设置工作流状态
self.current_workflow_status = WorkflowStatus.INITIALIZING
self.workflow_parameters = parameters
self.workflow_start_time = time.time()
# 委托给子类实现
success = self._execute_workflow_impl(workflow_name, parameters)
if success:
self.current_workflow_status = WorkflowStatus.RUNNING
logger.info(f"工作站 {self.device_id} 工作流 {workflow_name} 启动成功")
else:
self.current_workflow_status = WorkflowStatus.ERROR
logger.error(f"工作站 {self.device_id} 工作流 {workflow_name} 启动失败")
return success
except Exception as e:
self.current_workflow_status = WorkflowStatus.ERROR
logger.error(f"工作站 {self.device_id} 执行工作流失败: {e}")
return False
def stop_workflow(self, emergency: bool = False) -> bool:
"""停止工作流"""
try:
if self.current_workflow_status in [WorkflowStatus.IDLE, WorkflowStatus.STOPPED]:
logger.warning(f"工作站 {self.device_id} 没有正在运行的工作流")
return True
self.current_workflow_status = WorkflowStatus.STOPPING
# 委托给子类实现
success = self._stop_workflow_impl(emergency)
if success:
self.current_workflow_status = WorkflowStatus.STOPPED
logger.info(f"工作站 {self.device_id} 工作流停止成功 (紧急: {emergency})")
else:
self.current_workflow_status = WorkflowStatus.ERROR
logger.error(f"工作站 {self.device_id} 工作流停止失败")
return success
except Exception as e:
self.current_workflow_status = WorkflowStatus.ERROR
logger.error(f"工作站 {self.device_id} 停止工作流失败: {e}")
return False
# ============ 状态属性 ============
@property
def workflow_status(self) -> WorkflowStatus:
"""获取当前工作流状态"""
return self.current_workflow_status
@property
def is_busy(self) -> bool:
"""检查工作站是否忙碌"""
return self.current_workflow_status in [
WorkflowStatus.INITIALIZING,
WorkflowStatus.RUNNING,
WorkflowStatus.STOPPING,
]
@property
def workflow_runtime(self) -> float:
"""获取工作流运行时间（秒）"""
if self.workflow_start_time is None:
return 0.0
return time.time() - self.workflow_start_time
# ============ 抽象方法 - 子类必须实现 ============
# @abstractmethod
# def _register_supported_workflows(self):
# """注册支持的工作流 - 子类必须实现"""
# pass
# @abstractmethod
# def _execute_workflow_impl(self, workflow_name: str, parameters: Dict[str, Any]) -> bool:
# """执行工作流的具体实现 - 子类必须实现"""
# pass
# @abstractmethod
# def _stop_workflow_impl(self, emergency: bool = False) -> bool:
# """停止工作流的具体实现 - 子类必须实现"""
# pass
class WorkstationExample(WorkstationBase):
"""工作站示例实现"""
def _register_supported_workflows(self):
"""注册支持的工作流"""
self.supported_workflows["example_workflow"] = WorkflowInfo(
name="example_workflow",
description="这是一个示例工作流",
estimated_duration=300.0,
required_materials=["sample_plate"],
output_product="processed_plate",
parameters_schema={"param1": "string", "param2": "integer"},
)
def _execute_workflow_impl(self, workflow_name: str, parameters: Dict[str, Any]) -> bool:
"""执行工作流的具体实现"""
if workflow_name not in self.supported_workflows:
logger.error(f"工作站 {self.device_id} 不支持工作流: {workflow_name}")
return False
# 这里添加实际的工作流逻辑
logger.info(f"工作站 {self.device_id} 正在执行工作流: {workflow_name} with parameters {parameters}")
return True
def _stop_workflow_impl(self, emergency: bool = False) -> bool:
"""停止工作流的具体实现"""
# 这里添加实际的停止逻辑
logger.info(f"工作站 {self.device_id} 正在停止工作流 (紧急: {emergency})")
return True