workflow upload & set liquid fix & add set liquid with plate

2026-02-04 21:35:09 +00:00 · 2026-02-02 18:23:33 +08:00
parent b0da149252
commit 23ce145f74
6 changed files with 1060 additions and 506 deletions
--- a/unilabos/workflow/common.py
+++ b/unilabos/workflow/common.py
@@ -1,3 +1,89 @@
+"""
+工作流转换模块 - JSON 到 WorkflowGraph 的转换流程
+
+==================== 输入格式 (JSON) ====================
+
+{
+    "workflow": [
+        {"action": "transfer_liquid", "action_args": {"sources": "cell_lines", "targets": "Liquid_1", "asp_vol": 100.0, "dis_vol": 74.75, ...}},
+        ...
+    ],
+    "reagent": {
+        "cell_lines": {"slot": 4, "well": ["A1", "A3", "A5"], "labware": "DRUG + YOYO-MEDIA"},
+        "Liquid_1": {"slot": 1, "well": ["A4", "A7", "A10"], "labware": "rep 1"},
+        ...
+    }
+}
+
+==================== 转换步骤 ====================
+
+第一步: 按 slot 去重创建 create_resource 节点（创建板子）
+--------------------------------------------------------------------------------
+- 遍历所有 reagent，按 slot 去重，为每个唯一的 slot 创建一个板子
+- 生成参数:
+    res_id: plate_slot_{slot}
+    device_id: /PRCXI
+    class_name: PRCXI_BioER_96_wellplate
+    parent: /PRCXI/PRCXI_Deck/T{slot}
+    slot_on_deck: "{slot}"
+- 输出端口: labware（用于连接 set_liquid_from_plate）
+- 控制流: create_resource 之间通过 ready 端口串联
+
+示例: slot=1, slot=4 -> 创建 2 个 create_resource 节点
+
+第二步: 为每个 reagent 创建 set_liquid_from_plate 节点（设置液体）
+--------------------------------------------------------------------------------
+- 遍历所有 reagent，为每个试剂创建 set_liquid_from_plate 节点
+- 生成参数:
+    plate: []（通过连接传递，来自 create_resource 的 labware）
+    well_names: ["A1", "A3", "A5"]（来自 reagent 的 well 数组）
+    liquid_names: ["cell_lines", "cell_lines", "cell_lines"]（与 well 数量一致）
+    volumes: [1e5, 1e5, 1e5]（与 well 数量一致，默认体积）
+- 输入连接: create_resource (labware) -> set_liquid_from_plate (input_plate)
+- 输出端口: output_wells（用于连接 transfer_liquid）
+- 控制流: set_liquid_from_plate 连接在所有 create_resource 之后，通过 ready 端口串联
+
+第三步: 解析 workflow，创建 transfer_liquid 等动作节点
+--------------------------------------------------------------------------------
+- 遍历 workflow 数组，为每个动作创建步骤节点
+- 参数重命名: asp_vol -> asp_vols, dis_vol -> dis_vols, asp_flow_rate -> asp_flow_rates, dis_flow_rate -> dis_flow_rates
+- 参数扩展: 根据 targets 的 wells 数量，将单值扩展为数组
+    例: asp_vol=100.0, targets 有 3 个 wells -> asp_vols=[100.0, 100.0, 100.0]
+- 连接处理: 如果 sources/targets 已通过 set_liquid_from_plate 连接，参数值改为 []
+- 输入连接: set_liquid_from_plate (output_wells) -> transfer_liquid (sources_identifier / targets_identifier)
+- 输出端口: sources_out, targets_out（用于连接下一个 transfer_liquid）
+
+==================== 连接关系图 ====================
+
+控制流 (ready 端口串联):
+    create_resource_1 -> create_resource_2 -> ... -> set_liquid_1 -> set_liquid_2 -> ... -> transfer_liquid_1 -> transfer_liquid_2 -> ...
+
+物料流:
+    [create_resource] --labware--> [set_liquid_from_plate] --output_wells--> [transfer_liquid] --sources_out/targets_out--> [下一个 transfer_liquid]
+          (slot=1)                    (cell_lines)           (input_plate)     (sources_identifier)                          (sources_identifier)
+          (slot=4)                    (Liquid_1)                               (targets_identifier)                          (targets_identifier)
+
+==================== 端口映射 ====================
+
+create_resource:
+    输出: labware
+
+set_liquid_from_plate:
+    输入: input_plate
+    输出: output_plate, output_wells
+
+transfer_liquid:
+    输入: sources -> sources_identifier, targets -> targets_identifier
+    输出: sources -> sources_out, targets -> targets_out
+
+==================== 校验规则 ====================
+
+- 检查 sources/targets 是否在 reagent 中定义
+- 检查 sources 和 targets 的 wells 数量是否匹配
+- 检查参数数组长度是否与 wells 数量一致
+- 如有问题，在 footer 中添加 [WARN: ...] 标记
+"""
+
 import re
 import uuid

@@ -14,13 +100,21 @@ Json = Dict[str, Any]
 # create_resource 节点默认参数
 CREATE_RESOURCE_DEFAULTS = {
    "device_id": "/PRCXI",
-    "parent": "/PRCXI/PRCXI_Deck",
+    "parent_template": "/PRCXI/PRCXI_Deck/T{slot}",  # {slot} 会被替换为实际的 slot 值
    "class_name": "PRCXI_BioER_96_wellplate",
 }

 # 默认液体体积 (uL)
 DEFAULT_LIQUID_VOLUME = 1e5

+# 参数重命名映射：单数 -> 复数（用于 transfer_liquid 等动作）
+PARAM_RENAME_MAPPING = {
+    "asp_vol": "asp_vols",
+    "dis_vol": "dis_vols",
+    "asp_flow_rate": "asp_flow_rates",
+    "dis_flow_rate": "dis_flow_rates",
+}
+

 # ---------------- Graph ----------------

@@ -256,73 +350,108 @@ def build_protocol_graph(
        action_resource_mapping: action 到 resource_name 的映射字典，可选
    """
    G = WorkflowGraph()
-    resource_last_writer = {}
+    resource_last_writer = {}  # reagent_name -> "node_id:port"
+    slot_to_create_resource = {}  # slot -> create_resource node_id

    protocol_steps = refactor_data(protocol_steps, action_resource_mapping)
-    # 有机化学&移液站协议图构建
-    WORKSTATION_ID = workstation_name

-    # 为所有labware创建资源节点
-    res_index = 0
+    # ==================== 第一步：按 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 节点
+    res_index = 0
+    last_create_resource_id = None
+    for slot, info in slots_info.items():
        node_id = str(uuid.uuid4())
-
-        # res_id 不能有空格，替换为下划线
-        res_id = str(labware_id).replace(" ", "_")
-
-        # 从 reagent 数据中获取 well 信息
-        wells = item.get("well", [])
-        slot = str(item.get("slot", ""))  # slot_on_deck 是字符串
-        well_count = len(wells) if wells else 1
-
-        # 判断节点类型
-        if "Rack" in str(labware_id) or "Tip" in str(labware_id):
-            lab_node_type = "Labware"
-            description = f"Prepare Labware: {labware_id}"
-            liquid_input_slot = wells if wells else [-1]
-            liquid_type = []
-            liquid_volume = []
-        elif item.get("type") == "hardware" or "reactor" in str(labware_id).lower():
-            if "reactor" not in str(labware_id).lower():
-                continue
-            lab_node_type = "Sample"
-            description = f"Prepare Reactor: {labware_id}"
-            liquid_input_slot = wells if wells else [-1]
-            liquid_type = []
-            liquid_volume = []
-        else:
-            lab_node_type = "Reagent"
-            description = f"Add Reagent to Flask: {labware_id}"
-            # liquid_input_slot, liquid_type, liquid_volume 数量与 wells 保持一致
-            liquid_input_slot = wells if wells else [-1]
-            liquid_type = [res_id] * well_count
-            liquid_volume = [DEFAULT_LIQUID_VOLUME] * well_count
+        res_id = info["res_id"]

        res_index += 1
        G.add_node(
            node_id,
            template_name="create_resource",
            resource_name="host_node",
-            name=f"Res {res_index}",
-            description=description,
-            lab_node_type=lab_node_type,
+            name=f"Plate {res_index}",
+            description=f"Create plate on slot {slot}",
+            lab_node_type="Labware",
            footer="create_resource-host_node",
            param={
                "res_id": res_id,
                "device_id": CREATE_RESOURCE_DEFAULTS["device_id"],
                "class_name": CREATE_RESOURCE_DEFAULTS["class_name"],
-                "parent": CREATE_RESOURCE_DEFAULTS["parent"],
+                "parent": CREATE_RESOURCE_DEFAULTS["parent_template"].format(slot=slot),
                "bind_locations": {"x": 0.0, "y": 0.0, "z": 0.0},
-                "liquid_input_slot": liquid_input_slot,
-                "liquid_type": liquid_type,
-                "liquid_volume": liquid_volume,
                "slot_on_deck": slot,
            },
        )
-        # create_resource 节点输出 liquid_slots，用于连接 transfer_liquid 的 sources/targets
-        resource_last_writer[labware_id] = f"{node_id}:liquid_slots"
+        slot_to_create_resource[slot] = node_id

-    last_control_node_id = None
+        # create_resource 之间通过 ready 串联
+        if last_create_resource_id is not None:
+            G.add_edge(last_create_resource_id, node_id, source_port="ready", target_port="ready")
+        last_create_resource_id = node_id
+
+    # ==================== 第二步：为每个 reagent 创建 set_liquid_from_plate 节点 ====================
+    set_liquid_index = 0
+    last_set_liquid_id = last_create_resource_id  # set_liquid_from_plate 连接在 create_resource 之后
+
+    for labware_id, item in labware_info.items():
+        # 跳过 Tip/Rack 类型
+        if "Rack" in str(labware_id) or "Tip" in str(labware_id):
+            continue
+        if item.get("type") == "hardware":
+            continue
+
+        slot = str(item.get("slot", ""))
+        wells = item.get("well", [])
+        if not wells or not slot:
+            continue
+
+        # res_id 不能有空格
+        res_id = str(labware_id).replace(" ", "_")
+        well_count = len(wells)
+
+        node_id = str(uuid.uuid4())
+        set_liquid_index += 1
+
+        G.add_node(
+            node_id,
+            template_name="set_liquid_from_plate",
+            resource_name="liquid_handler.prcxi",
+            name=f"SetLiquid {set_liquid_index}",
+            description=f"Set liquid: {labware_id}",
+            lab_node_type="Reagent",
+            footer="set_liquid_from_plate-liquid_handler.prcxi",
+            param={
+                "plate": [],  # 通过连接传递
+                "well_names": wells,  # 孔位名数组，如 ["A1", "A3", "A5"]
+                "liquid_names": [res_id] * well_count,
+                "volumes": [DEFAULT_LIQUID_VOLUME] * well_count,
+            },
+        )
+
+        # ready 连接：上一个节点 -> set_liquid_from_plate
+        if last_set_liquid_id is not None:
+            G.add_edge(last_set_liquid_id, node_id, source_port="ready", target_port="ready")
+        last_set_liquid_id = node_id
+
+        # 物料流：create_resource 的 labware -> set_liquid_from_plate 的 input_plate
+        create_res_node_id = slot_to_create_resource.get(slot)
+        if create_res_node_id:
+            G.add_edge(create_res_node_id, node_id, source_port="labware", target_port="input_plate")
+
+        # set_liquid_from_plate 的输出 output_wells 用于连接 transfer_liquid
+        resource_last_writer[labware_id] = f"{node_id}:output_wells"
+
+    last_control_node_id = last_set_liquid_id

    # 端口名称映射：JSON 字段名 -> 实际 handle key
    INPUT_PORT_MAPPING = {
@@ -348,18 +477,20 @@ def build_protocol_graph(
        "compound": "compound",
    }

+    # 需要根据 wells 数量扩展的参数列表（复数形式）
+    EXPAND_BY_WELLS_PARAMS = ["asp_vols", "dis_vols", "asp_flow_rates", "dis_flow_rates"]
+
    # 处理协议步骤
    for step in protocol_steps:
        node_id = str(uuid.uuid4())
-        G.add_node(node_id, **step)
+        params = step.get("param", {}).copy()  # 复制一份，避免修改原数据
+        connected_params = set()  # 记录被连接的参数
+        warnings = []  # 收集警告信息

-        # 控制流
-        if last_control_node_id is not None:
-            G.add_edge(last_control_node_id, node_id, source_port="ready", target_port="ready")
-        last_control_node_id = node_id
-
-        # 物料流
-        params = step.get("param", {})
+        # 参数重命名：单数 -> 复数
+        for old_name, new_name in PARAM_RENAME_MAPPING.items():
+            if old_name in params:
+                params[new_name] = params.pop(old_name)

        # 处理输入连接
        for param_key, target_port in INPUT_PORT_MAPPING.items():
@@ -367,10 +498,71 @@ def build_protocol_graph(
            if resource_name and resource_name in resource_last_writer:
                source_node, source_port = resource_last_writer[resource_name].split(":")
                G.add_edge(source_node, node_id, source_port=source_port, target_port=target_port)
+                connected_params.add(param_key)
+            elif resource_name and resource_name not in resource_last_writer:
+                # 资源名在 labware_info 中不存在
+                warnings.append(f"{param_key}={resource_name} 未找到")
+
+        # 获取 targets 对应的 wells 数量，用于扩展参数
+        targets_name = params.get("targets")
+        sources_name = params.get("sources")
+        targets_wells_count = 1
+        sources_wells_count = 1
+
+        if targets_name and targets_name in labware_info:
+            target_wells = labware_info[targets_name].get("well", [])
+            targets_wells_count = len(target_wells) if target_wells else 1
+        elif targets_name:
+            warnings.append(f"targets={targets_name} 未在 reagent 中定义")
+
+        if sources_name and sources_name in labware_info:
+            source_wells = labware_info[sources_name].get("well", [])
+            sources_wells_count = len(source_wells) if source_wells else 1
+        elif sources_name:
+            warnings.append(f"sources={sources_name} 未在 reagent 中定义")
+
+        # 检查 sources 和 targets 的 wells 数量是否匹配
+        if targets_wells_count != sources_wells_count and targets_name and sources_name:
+            warnings.append(f"wells 数量不匹配: sources={sources_wells_count}, targets={targets_wells_count}")
+
+        # 使用 targets 的 wells 数量来扩展参数
+        wells_count = targets_wells_count
+
+        # 扩展单值参数为数组（根据 targets 的 wells 数量）
+        for expand_param in EXPAND_BY_WELLS_PARAMS:
+            if expand_param in params:
+                value = params[expand_param]
+                # 如果是单个值，扩展为数组
+                if not isinstance(value, list):
+                    params[expand_param] = [value] * wells_count
+                # 如果已经是数组但长度不对，记录警告
+                elif len(value) != wells_count:
+                    warnings.append(f"{expand_param} 数量({len(value)})与 wells({wells_count})不匹配")
+
+        # 如果 sources/targets 已通过连接传递，将参数值改为空数组
+        for param_key in connected_params:
+            if param_key in params:
+                params[param_key] = []
+
+        # 更新 step 的 param 和 footer
+        step_copy = step.copy()
+        step_copy["param"] = params
+
+        # 如果有警告，修改 footer 添加警告标记（警告放前面）
+        if warnings:
+            original_footer = step.get("footer", "")
+            step_copy["footer"] = f"[WARN: {'; '.join(warnings)}] {original_footer}"
+
+        G.add_node(node_id, **step_copy)
+
+        # 控制流
+        if last_control_node_id is not None:
+            G.add_edge(last_control_node_id, node_id, source_port="ready", target_port="ready")
+        last_control_node_id = node_id

        # 处理输出：更新 resource_last_writer
        for param_key, output_port in OUTPUT_PORT_MAPPING.items():
-            resource_name = params.get(param_key)
+            resource_name = step.get("param", {}).get(param_key)  # 使用原始参数值
            if resource_name:
                resource_last_writer[resource_name] = f"{node_id}:{output_port}"