open_pub/Uni-Lab-OS
1
0
Fork 0
mirror of https://github.com/dptech-corp/Uni-Lab-OS.git synced 2026-02-13 03:05:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

registry update & workflow update

Browse Source
This commit is contained in:
Xuwznln
2026-02-10 22:45:51 +08:00
parent 8a0116c852
commit ee6307a568
6 changed files with 41 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
unilabos/workflow/common.py
View File

@@ -362,14 +362,16 @@ def build_protocol_graph(
protocol_steps: List[Dict[str, Any]],
workstation_name: str,
action_resource_mapping: Optional[Dict[str, str]] = None,
labware_defs: Optional[List[Dict[str, Any]]] = None,
) -> WorkflowGraph:
"""统一的协议图构建函数，根据设备类型自动选择构建逻辑
Args:
labware_info: labware 信息字典，格式为 {name: {slot, well, labware, ...}, ...}
labware_info: reagent 信息字典，格式为 {name: {slot, well}, ...}，用于 set_liquid 和 well 查找
protocol_steps: 协议步骤列表
workstation_name: 工作站名称
action_resource_mapping: action 到 resource_name 的映射字典，可选
labware_defs: labware 定义列表，格式为 [{"name": "...", "slot": "1", "type": "lab_xxx"}, ...]
"""
G = WorkflowGraph()
resource_last_writer = {} # reagent_name -> "node_id:port"
@@ -377,18 +379,7 @@ def build_protocol_graph(
protocol_steps = refactor_data(protocol_steps, action_resource_mapping)
# ==================== 第一步：按 slot 去重创建 create_resource 节点 ====================
# 收集所有唯一的 slot
slots_info = {} # slot -> {labware, res_id}
for labware_id, item in labware_info.items():
slot = str(item.get("slot", ""))
if slot and slot not in slots_info:
res_id = f"plate_slot_{slot}"
slots_info[slot] = {
"labware": item.get("labware", ""),
"res_id": res_id,
}
# ==================== 第一步：按 slot 创建 create_resource 节点 ====================
# 创建 Group 节点，包含所有 create_resource 节点
group_node_id = str(uuid.uuid4())
G.add_node(
@@ -404,29 +395,35 @@ def build_protocol_graph(
param=None,
)
# 为每个唯一的 slot 创建 create_resource 节点
# 直接使用 JSON 中的 labware 定义，每个 slot 一条记录，type 即 class_name
res_index = 0
for slot, info in slots_info.items():
node_id = str(uuid.uuid4())
res_id = info["res_id"]
for lw in (labware_defs or []):
slot = str(lw.get("slot", ""))
if not slot or slot in slot_to_create_resource:
continue # 跳过空 slot 或已处理的 slot
lw_name = lw.get("name", f"slot {slot}")
lw_type = lw.get("type", CREATE_RESOURCE_DEFAULTS["class_name"])
res_id = f"plate_slot_{slot}"
res_index += 1
node_id = str(uuid.uuid4())
G.add_node(
node_id,
template_name="create_resource",
resource_name="host_node",
name=f"Plate {res_index}",
description=f"Create plate on slot {slot}",
name=lw_name,
description=f"Create {lw_name}",
lab_node_type="Labware",
footer="create_resource-host_node",
device_name=DEVICE_NAME_HOST,
type=NODE_TYPE_DEFAULT,
parent_uuid=group_node_id, # 指向 Group 节点
minimized=True, # 折叠显示
parent_uuid=group_node_id,
minimized=True,
param={
"res_id": res_id,
"device_id": CREATE_RESOURCE_DEFAULTS["device_id"],
"class_name": CREATE_RESOURCE_DEFAULTS["class_name"],
"class_name": lw_type,
"parent": CREATE_RESOURCE_DEFAULTS["parent_template"].format(slot=slot),
"bind_locations": {"x": 0.0, "y": 0.0, "z": 0.0},
"slot_on_deck": slot,
@@ -434,8 +431,6 @@ def build_protocol_graph(
)
slot_to_create_resource[slot] = node_id
# create_resource 之间不需要 ready 连接
# ==================== 第二步：为每个 reagent 创建 set_liquid_from_plate 节点 ====================
# 创建 Group 节点，包含所有 set_liquid_from_plate 节点
set_liquid_group_id = str(uuid.uuid4())