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

修改prcxi连线

Browse Source
This commit is contained in:
Xianwei Qi
2025-10-22 14:00:38 +08:00
parent 97212be8b7
commit 12ba110569
3 changed files with 13 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
scripts/workflow.py
View File

@@ -92,7 +92,7 @@ def refactor_data(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
# 定义操作映射，包含生物实验和有机化学的所有操作 # 定义操作映射，包含生物实验和有机化学的所有操作
OPERATION_MAPPING = { OPERATION_MAPPING = {
# 生物实验操作 # 生物实验操作
"transfer_liquid": "SynBioFactory-liquid_handler.transfer_liquid", "transfer_liquid": "SynBioFactory-liquid_handler.prcxi-transfer_liquid",
"transfer": "SynBioFactory-liquid_handler.biomek-transfer", "transfer": "SynBioFactory-liquid_handler.biomek-transfer",
"incubation": "SynBioFactory-liquid_handler.biomek-incubation", "incubation": "SynBioFactory-liquid_handler.biomek-incubation",
"move_labware": "SynBioFactory-liquid_handler.biomek-move_labware", "move_labware": "SynBioFactory-liquid_handler.biomek-move_labware",
@@ -159,7 +159,10 @@ def build_protocol_graph(
protocol_steps = refactor_data(protocol_steps) protocol_steps = refactor_data(protocol_steps)
# 检查协议步骤中的模板来判断协议类型 # 检查协议步骤中的模板来判断协议类型
has_biomek_template = any("biomek" in step.get("template", "") for step in protocol_steps) has_biomek_template = any(
("biomek" in step.get("template", "")) or ("prcxi" in step.get("template", ""))
for step in protocol_steps
)
if has_biomek_template: if has_biomek_template:
# 生物实验协议图构建 # 生物实验协议图构建
@@ -178,12 +181,14 @@ def build_protocol_graph(
resource_last_writer[labware_id] = f"{node_id}:labware" resource_last_writer[labware_id] = f"{node_id}:labware"
# 处理协议步骤 # 处理协议步骤
prev_node = None
for i, step in enumerate(protocol_steps): for i, step in enumerate(protocol_steps):
node_id = str(uuid.uuid4()) node_id = str(uuid.uuid4())
G.add_node(node_id, **step) G.add_node(node_id, **step)
# 添加控制流边 # 添加控制流边
G.add_edge(prev_node, node_id, source_port="ready", target_port="ready") if prev_node is not None:
G.add_edge(prev_node, node_id, source_port="ready", target_port="ready")
prev_node = node_id prev_node = node_id
# 处理物料流 # 处理物料流
@@ -198,7 +203,8 @@ def build_protocol_graph(
# 添加协议结束节点 # 添加协议结束节点
end_id = str(uuid.uuid4()) end_id = str(uuid.uuid4())
G.add_node(end_id, template=f"{LAB_NAME}-liquid_handler.biomek-run_protocol") G.add_node(end_id, template=f"{LAB_NAME}-liquid_handler.biomek-run_protocol")
G.add_edge(prev_node, end_id, source_port="ready", target_port="ready") if prev_node is not None:
G.add_edge(prev_node, end_id, source_port="ready", target_port="ready")
else: else:
# 有机化学协议图构建 # 有机化学协议图构建

BIN
test/workflow/example_prcxi_graph_20251022_1359.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 117 KiB

4
test/workflow/merge_workflow.py
View File

@@ -1,5 +1,6 @@
import json import json
import sys import sys
from datetime import datetime
from pathlib import Path from pathlib import Path
ROOT_DIR = Path(__file__).resolve().parents[2] ROOT_DIR = Path(__file__).resolve().parents[2]
@@ -87,6 +88,7 @@ def test_build_protocol_graph(protocol_name):
protocol_steps=protocol_steps, protocol_steps=protocol_steps,
workstation_name="PRCXi", workstation_name="PRCXi",
) )
output_path = data_path.with_name(f"{protocol_name}_graph.png") timestamp = datetime.now().strftime("%Y%m%d_%H%M")
output_path = data_path.with_name(f"{protocol_name}_graph_{timestamp}.png")
draw_protocol_graph_with_ports(graph, str(output_path)) draw_protocol_graph_with_ports(graph, str(output_path))
print(graph) print(graph)