add unilabos/workflow and entrypoint

(cherry picked from commit b1cdef9185)

.conda/recipe.yaml

@@ -1,6 +1,6 @@
 package:
   name: unilabos
-  version: 0.10.11
+  version: 0.10.12
 
 source:
   path: ../unilabos

recipes/msgs/recipe.yaml

@@ -1,6 +1,6 @@
 package:
   name: ros-humble-unilabos-msgs
-  version: 0.10.11
+  version: 0.10.12
 source:
   path: ../../unilabos_msgs
   target_directory: src

recipes/unilabos/recipe.yaml

@@ -1,6 +1,6 @@
 package:
   name: unilabos
-  version: "0.10.11"
+  version: "0.10.12"
 
 source:
   path: ../..

scripts/workflow.py

@@ -2,7 +2,6 @@ import json
 import logging
 import traceback
 import uuid
-import xml.etree.ElementTree as ET
 from typing import Any, Dict, List
 
 import networkx as nx
@@ -25,7 +24,15 @@ class SimpleGraph:
 
     def add_edge(self, source, target, **attrs):
         """添加边"""
-        edge = {"source": source, "target": target, **attrs}
+        # edge = {"source": source, "target": target, **attrs}
+        edge = {
+            "source": source, "target": target,
+            "source_node_uuid": source,
+            "target_node_uuid": target,
+            "source_handle_io": "source",
+            "target_handle_io": "target",
+            **attrs
+        }
         self.edges.append(edge)
 
     def to_dict(self):
@@ -42,6 +49,7 @@ class SimpleGraph:
             "multigraph": False,
             "graph": {},
             "nodes": nodes_list,
+            "edges": self.edges,
             "links": self.edges,
         }
 
@@ -58,495 +66,8 @@ def extract_json_from_markdown(text: str) -> str:
     return text
 
 
-def convert_to_type(val: str) -> Any:
-    """将字符串值转换为适当的数据类型"""
-    if val == "True":
-        return True
-    if val == "False":
-        return False
-    if val == "?":
-        return None
-    if val.endswith(" g"):
-        return float(val.split(" ")[0])
-    if val.endswith("mg"):
-        return float(val.split("mg")[0])
-    elif val.endswith("mmol"):
-        return float(val.split("mmol")[0]) / 1000
-    elif val.endswith("mol"):
-        return float(val.split("mol")[0])
-    elif val.endswith("ml"):
-        return float(val.split("ml")[0])
-    elif val.endswith("RPM"):
-        return float(val.split("RPM")[0])
-    elif val.endswith(" °C"):
-        return float(val.split(" ")[0])
-    elif val.endswith(" %"):
-        return float(val.split(" ")[0])
-    return val
 
 
-def refactor_data(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-    """统一的数据重构函数，根据操作类型自动选择模板"""
-    refactored_data = []
-
-    # 定义操作映射，包含生物实验和有机化学的所有操作
-    OPERATION_MAPPING = {
-        # 生物实验操作
-        "transfer_liquid": "SynBioFactory-liquid_handler.prcxi-transfer_liquid",
-        "transfer": "SynBioFactory-liquid_handler.biomek-transfer",
-        "incubation": "SynBioFactory-liquid_handler.biomek-incubation",
-        "move_labware": "SynBioFactory-liquid_handler.biomek-move_labware",
-        "oscillation": "SynBioFactory-liquid_handler.biomek-oscillation",
-        # 有机化学操作
-        "HeatChillToTemp": "SynBioFactory-workstation-HeatChillProtocol",
-        "StopHeatChill": "SynBioFactory-workstation-HeatChillStopProtocol",
-        "StartHeatChill": "SynBioFactory-workstation-HeatChillStartProtocol",
-        "HeatChill": "SynBioFactory-workstation-HeatChillProtocol",
-        "Dissolve": "SynBioFactory-workstation-DissolveProtocol",
-        "Transfer": "SynBioFactory-workstation-TransferProtocol",
-        "Evaporate": "SynBioFactory-workstation-EvaporateProtocol",
-        "Recrystallize": "SynBioFactory-workstation-RecrystallizeProtocol",
-        "Filter": "SynBioFactory-workstation-FilterProtocol",
-        "Dry": "SynBioFactory-workstation-DryProtocol",
-        "Add": "SynBioFactory-workstation-AddProtocol",
-    }
-
-    UNSUPPORTED_OPERATIONS = ["Purge", "Wait", "Stir", "ResetHandling"]
-
-    for step in data:
-        operation = step.get("action")
-        if not operation or operation in UNSUPPORTED_OPERATIONS:
-            continue
-
-        # 处理重复操作
-        if operation == "Repeat":
-            times = step.get("times", step.get("parameters", {}).get("times", 1))
-            sub_steps = step.get("steps", step.get("parameters", {}).get("steps", []))
-            for i in range(int(times)):
-                sub_data = refactor_data(sub_steps)
-                refactored_data.extend(sub_data)
-            continue
-
-        # 获取模板名称
-        template = OPERATION_MAPPING.get(operation)
-        if not template:
-            # 自动推断模板类型
-            if operation.lower() in ["transfer", "incubation", "move_labware", "oscillation"]:
-                template = f"SynBioFactory-liquid_handler.biomek-{operation}"
-            else:
-                template = f"SynBioFactory-workstation-{operation}Protocol"
-
-        # 创建步骤数据
-        step_data = {
-            "template": template,
-            "description": step.get("description", step.get("purpose", f"{operation} operation")),
-            "lab_node_type": "Device",
-            "parameters": step.get("parameters", step.get("action_args", {})),
-        }
-        refactored_data.append(step_data)
-
-    return refactored_data
-
-
-def build_protocol_graph(
-    labware_info: List[Dict[str, Any]], protocol_steps: List[Dict[str, Any]], workstation_name: str
-) -> SimpleGraph:
-    """统一的协议图构建函数，根据设备类型自动选择构建逻辑"""
-    G = SimpleGraph()
-    resource_last_writer = {}
-    LAB_NAME = "SynBioFactory"
-
-    protocol_steps = refactor_data(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:
-        # 生物实验协议图构建
-        for labware_id, labware in labware_info.items():
-            node_id = str(uuid.uuid4())
-
-            labware_attrs = labware.copy()
-            labware_id = labware_attrs.pop("id", labware_attrs.get("name", f"labware_{uuid.uuid4()}"))
-            labware_attrs["description"] = labware_id
-            labware_attrs["lab_node_type"] = (
-                "Reagent" if "Plate" in str(labware_id) else "Labware" if "Rack" in str(labware_id) else "Sample"
-            )
-            labware_attrs["device_id"] = workstation_name
-
-            G.add_node(node_id, template=f"{LAB_NAME}-host_node-create_resource", **labware_attrs)
-            resource_last_writer[labware_id] = f"{node_id}:labware"
-
-        # 处理协议步骤
-        prev_node = None
-        for i, step in enumerate(protocol_steps):
-            node_id = str(uuid.uuid4())
-            G.add_node(node_id, **step)
-
-            # 添加控制流边
-            if prev_node is not None:
-                G.add_edge(prev_node, node_id, source_port="ready", target_port="ready")
-            prev_node = node_id
-
-            # 处理物料流
-            params = step.get("parameters", {})
-            if "sources" in params and params["sources"] in resource_last_writer:
-                source_node, source_port = resource_last_writer[params["sources"]].split(":")
-                G.add_edge(source_node, node_id, source_port=source_port, target_port="labware")
-
-            if "targets" in params:
-                resource_last_writer[params["targets"]] = f"{node_id}:labware"
-
-        # 添加协议结束节点
-        end_id = str(uuid.uuid4())
-        G.add_node(end_id, template=f"{LAB_NAME}-liquid_handler.biomek-run_protocol")
-        if prev_node is not None:
-            G.add_edge(prev_node, end_id, source_port="ready", target_port="ready")
-
-    else:
-        # 有机化学协议图构建
-        WORKSTATION_ID = workstation_name
-
-        # 为所有labware创建资源节点
-        for item_id, item in labware_info.items():
-            # item_id = item.get("id") or item.get("name", f"item_{uuid.uuid4()}")
-            node_id = str(uuid.uuid4())
-
-            # 判断节点类型
-            if item.get("type") == "hardware" or "reactor" in str(item_id).lower():
-                if "reactor" not in str(item_id).lower():
-                    continue
-                lab_node_type = "Sample"
-                description = f"Prepare Reactor: {item_id}"
-                liquid_type = []
-                liquid_volume = []
-            else:
-                lab_node_type = "Reagent"
-                description = f"Add Reagent to Flask: {item_id}"
-                liquid_type = [item_id]
-                liquid_volume = [1e5]
-
-            G.add_node(
-                node_id,
-                template=f"{LAB_NAME}-host_node-create_resource",
-                description=description,
-                lab_node_type=lab_node_type,
-                res_id=item_id,
-                device_id=WORKSTATION_ID,
-                class_name="container",
-                parent=WORKSTATION_ID,
-                bind_locations={"x": 0.0, "y": 0.0, "z": 0.0},
-                liquid_input_slot=[-1],
-                liquid_type=liquid_type,
-                liquid_volume=liquid_volume,
-                slot_on_deck="",
-                role=item.get("role", ""),
-            )
-            resource_last_writer[item_id] = f"{node_id}:labware"
-
-        last_control_node_id = None
-
-        # 处理协议步骤
-        for step in protocol_steps:
-            node_id = str(uuid.uuid4())
-            G.add_node(node_id, **step)
-
-            # 控制流
-            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("parameters", {})
-            input_resources = {
-                "Vessel": params.get("vessel"),
-                "ToVessel": params.get("to_vessel"),
-                "FromVessel": params.get("from_vessel"),
-                "reagent": params.get("reagent"),
-                "solvent": params.get("solvent"),
-                "compound": params.get("compound"),
-                "sources": params.get("sources"),
-                "targets": params.get("targets"),
-            }
-
-            for target_port, resource_name in input_resources.items():
-                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)
-
-            output_resources = {
-                "VesselOut": params.get("vessel"),
-                "FromVesselOut": params.get("from_vessel"),
-                "ToVesselOut": params.get("to_vessel"),
-                "FiltrateOut": params.get("filtrate_vessel"),
-                "reagent": params.get("reagent"),
-                "solvent": params.get("solvent"),
-                "compound": params.get("compound"),
-                "sources_out": params.get("sources"),
-                "targets_out": params.get("targets"),
-            }
-
-            for source_port, resource_name in output_resources.items():
-                if resource_name:
-                    resource_last_writer[resource_name] = f"{node_id}:{source_port}"
-
-    return G
-
-
-def draw_protocol_graph(protocol_graph: SimpleGraph, output_path: str):
-    """
-    (辅助功能) 使用 networkx 和 matplotlib 绘制协议工作流图，用于可视化。
-    """
-    if not protocol_graph:
-        print("Cannot draw graph: Graph object is empty.")
-        return
-
-    G = nx.DiGraph()
-
-    for node_id, attrs in protocol_graph.nodes.items():
-        label = attrs.get("description", attrs.get("template", node_id[:8]))
-        G.add_node(node_id, label=label, **attrs)
-
-    for edge in protocol_graph.edges:
-        G.add_edge(edge["source"], edge["target"])
-
-    plt.figure(figsize=(20, 15))
-    try:
-        pos = nx.nx_agraph.graphviz_layout(G, prog="dot")
-    except Exception:
-        pos = nx.shell_layout(G)  # Fallback layout
-
-    node_labels = {node: data["label"] for node, data in G.nodes(data=True)}
-    nx.draw(
-        G,
-        pos,
-        with_labels=False,
-        node_size=2500,
-        node_color="skyblue",
-        node_shape="o",
-        edge_color="gray",
-        width=1.5,
-        arrowsize=15,
-    )
-    nx.draw_networkx_labels(G, pos, labels=node_labels, font_size=8, font_weight="bold")
-
-    plt.title("Chemical Protocol Workflow Graph", size=15)
-    plt.savefig(output_path, dpi=300, bbox_inches="tight")
-    plt.close()
-    print(f"  - Visualization saved to '{output_path}'")
-
-
-from networkx.drawing.nx_agraph import to_agraph
-import re
-
-COMPASS = {"n","e","s","w","ne","nw","se","sw","c"}
-
-def _is_compass(port: str) -> bool:
-    return isinstance(port, str) and port.lower() in COMPASS
-
-def draw_protocol_graph_with_ports(protocol_graph, output_path: str, rankdir: str = "LR"):
-    """
-    使用 Graphviz 端口语法绘制协议工作流图。
-    - 若边上的 source_port/target_port 是 compass（n/e/s/w/...），直接用 compass。
-    - 否则自动为节点创建 record 形状并定义命名端口 <portname>。
-    最终由 PyGraphviz 渲染并输出到 output_path（后缀决定格式，如 .png/.svg/.pdf）。
-    """
-    if not protocol_graph:
-        print("Cannot draw graph: Graph object is empty.")
-        return
-
-    # 1) 先用 networkx 搭建有向图，保留端口属性
-    G = nx.DiGraph()
-    for node_id, attrs in protocol_graph.nodes.items():
-        label = attrs.get("description", attrs.get("template", node_id[:8]))
-        # 保留一个干净的“中心标签”，用于放在 record 的中间槽
-        G.add_node(node_id, _core_label=str(label), **{k:v for k,v in attrs.items() if k not in ("label",)})
-
-    edges_data = []
-    in_ports_by_node  = {}  # 收集命名输入端口
-    out_ports_by_node = {}  # 收集命名输出端口
-
-    for edge in protocol_graph.edges:
-        u = edge["source"]
-        v = edge["target"]
-        sp = edge.get("source_port")
-        tp = edge.get("target_port")
-
-        # 记录到图里（保留原始端口信息）
-        G.add_edge(u, v, source_port=sp, target_port=tp)
-        edges_data.append((u, v, sp, tp))
-
-        # 如果不是 compass，就按“命名端口”先归类，等会儿给节点造 record
-        if sp and not _is_compass(sp):
-            out_ports_by_node.setdefault(u, set()).add(str(sp))
-        if tp and not _is_compass(tp):
-            in_ports_by_node.setdefault(v, set()).add(str(tp))
-
-    # 2) 转为 AGraph，使用 Graphviz 渲染
-    A = to_agraph(G)
-    A.graph_attr.update(rankdir=rankdir, splines="true", concentrate="false", fontsize="10")
-    A.node_attr.update(shape="box", style="rounded,filled", fillcolor="lightyellow", color="#999999", fontname="Helvetica")
-    A.edge_attr.update(arrowsize="0.8", color="#666666")
-
-    # 3) 为需要命名端口的节点设置 record 形状与 label
-    #    左列 = 输入端口；中间 = 核心标签；右列 = 输出端口
-    for n in A.nodes():
-        node = A.get_node(n)
-        core = G.nodes[n].get("_core_label", n)
-
-        in_ports  = sorted(in_ports_by_node.get(n, []))
-        out_ports = sorted(out_ports_by_node.get(n, []))
-
-        # 如果该节点涉及命名端口，则用 record；否则保留原 box
-        if in_ports or out_ports:
-            def port_fields(ports):
-                if not ports:
-                    return " "  # 必须留一个空槽占位
-                # 每个端口一个小格子，<p> name
-                return "|".join(f"<{re.sub(r'[^A-Za-z0-9_:.|-]', '_', p)}> {p}" for p in ports)
-
-            left  = port_fields(in_ports)
-            right = port_fields(out_ports)
-
-            # 三栏：左(入) | 中(节点名) | 右(出)
-            record_label = f"{{ {left} | {core} | {right} }}"
-            node.attr.update(shape="record", label=record_label)
-        else:
-            # 没有命名端口：普通盒子，显示核心标签
-            node.attr.update(label=str(core))
-
-    # 4) 给边设置 headport / tailport
-    #    - 若端口为 compass：直接用 compass（e.g., headport="e"）
-    #    - 若端口为命名端口：使用在 record 中定义的 <port> 名（同名即可）
-    for (u, v, sp, tp) in edges_data:
-        e = A.get_edge(u, v)
-
-        # Graphviz 属性：tail 是源，head 是目标
-        if sp:
-            if _is_compass(sp):
-                e.attr["tailport"] = sp.lower()
-            else:
-                # 与 record label 中 <port> 名一致；特殊字符已在 label 中做了清洗
-                e.attr["tailport"] = re.sub(r'[^A-Za-z0-9_:.|-]', '_', str(sp))
-
-        if tp:
-            if _is_compass(tp):
-                e.attr["headport"] = tp.lower()
-            else:
-                e.attr["headport"] = re.sub(r'[^A-Za-z0-9_:.|-]', '_', str(tp))
-
-        # 可选：若想让边更贴边缘，可设置 constraint/spline 等
-        # e.attr["arrowhead"] = "vee"
-
-    # 5) 输出
-    A.draw(output_path, prog="dot")
-    print(f"  - Port-aware workflow rendered to '{output_path}'")
-
-
-def flatten_xdl_procedure(procedure_elem: ET.Element) -> List[ET.Element]:
-    """展平嵌套的XDL程序结构"""
-    flattened_operations = []
-    TEMP_UNSUPPORTED_PROTOCOL = ["Purge", "Wait", "Stir", "ResetHandling"]
-
-    def extract_operations(element: ET.Element):
-        if element.tag not in ["Prep", "Reaction", "Workup", "Purification", "Procedure"]:
-            if element.tag not in TEMP_UNSUPPORTED_PROTOCOL:
-                flattened_operations.append(element)
-
-        for child in element:
-            extract_operations(child)
-
-    for child in procedure_elem:
-        extract_operations(child)
-
-    return flattened_operations
-
-
-def parse_xdl_content(xdl_content: str) -> tuple:
-    """解析XDL内容"""
-    try:
-        xdl_content_cleaned = "".join(c for c in xdl_content if c.isprintable())
-        root = ET.fromstring(xdl_content_cleaned)
-
-        synthesis_elem = root.find("Synthesis")
-        if synthesis_elem is None:
-            return None, None, None
-
-        # 解析硬件组件
-        hardware_elem = synthesis_elem.find("Hardware")
-        hardware = []
-        if hardware_elem is not None:
-            hardware = [{"id": c.get("id"), "type": c.get("type")} for c in hardware_elem.findall("Component")]
-
-        # 解析试剂
-        reagents_elem = synthesis_elem.find("Reagents")
-        reagents = []
-        if reagents_elem is not None:
-            reagents = [{"name": r.get("name"), "role": r.get("role", "")} for r in reagents_elem.findall("Reagent")]
-
-        # 解析程序
-        procedure_elem = synthesis_elem.find("Procedure")
-        if procedure_elem is None:
-            return None, None, None
-
-        flattened_operations = flatten_xdl_procedure(procedure_elem)
-        return hardware, reagents, flattened_operations
-
-    except ET.ParseError as e:
-        raise ValueError(f"Invalid XDL format: {e}")
-
-
-def convert_xdl_to_dict(xdl_content: str) -> Dict[str, Any]:
-    """
-    将XDL XML格式转换为标准的字典格式
-
-    Args:
-        xdl_content: XDL XML内容
-
-    Returns:
-        转换结果，包含步骤和器材信息
-    """
-    try:
-        hardware, reagents, flattened_operations = parse_xdl_content(xdl_content)
-        if hardware is None:
-            return {"error": "Failed to parse XDL content", "success": False}
-
-        # 将XDL元素转换为字典格式
-        steps_data = []
-        for elem in flattened_operations:
-            # 转换参数类型
-            parameters = {}
-            for key, val in elem.attrib.items():
-                converted_val = convert_to_type(val)
-                if converted_val is not None:
-                    parameters[key] = converted_val
-
-            step_dict = {
-                "operation": elem.tag,
-                "parameters": parameters,
-                "description": elem.get("purpose", f"Operation: {elem.tag}"),
-            }
-            steps_data.append(step_dict)
-
-        # 合并硬件和试剂为统一的labware_info格式
-        labware_data = []
-        labware_data.extend({"id": hw["id"], "type": "hardware", **hw} for hw in hardware)
-        labware_data.extend({"name": reagent["name"], "type": "reagent", **reagent} for reagent in reagents)
-
-        return {
-            "success": True,
-            "steps": steps_data,
-            "labware": labware_data,
-            "message": f"Successfully converted XDL to dict format. Found {len(steps_data)} steps and {len(labware_data)} labware items.",
-        }
-
-    except Exception as e:
-        error_msg = f"XDL conversion failed: {str(e)}"
-        logger.error(error_msg)
-        return {"error": error_msg, "success": False}
 
 
 def create_workflow(

setup.py

@@ -4,7 +4,7 @@ package_name = 'unilabos'
 
 setup(
     name=package_name,
-    version='0.10.11',
+    version='0.10.12',
     packages=find_packages(),
     include_package_data=True,
     install_requires=['setuptools'],

unilabos/__init__.py

@@ -1 +1 @@
-__version__ = "0.10.11"
+__version__ = "0.10.12"

unilabos/app/communication.py

@@ -141,7 +141,7 @@ class CommunicationClientFactory:
         """
         if cls._client_cache is None:
             cls._client_cache = cls.create_client(protocol)
-            logger.info(f"[CommunicationFactory] Created {type(cls._client_cache).__name__} client")
+            logger.trace(f"[CommunicationFactory] Created {type(cls._client_cache).__name__} client")
 
         return cls._client_cache
 

unilabos/app/main.py

@@ -49,6 +49,8 @@ def convert_argv_dashes_to_underscores(args: argparse.ArgumentParser):
 def parse_args():
     """解析命令行参数"""
     parser = argparse.ArgumentParser(description="Start Uni-Lab Edge server.")
+    subparsers = parser.add_subparsers(title="Valid subcommands", dest="command")
+
     parser.add_argument("-g", "--graph", help="Physical setup graph file path.")
     parser.add_argument("-c", "--controllers", default=None, help="Controllers config file path.")
     parser.add_argument(
@@ -153,6 +155,14 @@ def parse_args():
         default=False,
         help="Complete registry information",
     )
+
+    # label
+    workflow_parser = subparsers.add_parser(
+        "workflow_upload",
+        help="Upload workflow from xdl/json/python files",
+    )
+    workflow_parser.add_argument("-t", "--labeltype", default="singlepoint", type=str,
+                              help="QM calculation type, support 'singlepoint', 'optimize' and 'dimer' currently")
     return parser
 
 
@@ -163,6 +173,9 @@ def main():
     convert_argv_dashes_to_underscores(args)
     args_dict = vars(args.parse_args())
 
+    # 显示启动横幅
+    print_unilab_banner(args_dict)
+
     # 环境检查 - 检查并自动安装必需的包 (可选)
     if not args_dict.get("skip_env_check", False):
         from unilabos.utils.environment_check import check_environment
@@ -218,7 +231,7 @@ def main():
 
     if hasattr(BasicConfig, "log_level"):
         logger.info(f"Log level set to '{BasicConfig.log_level}' from config file.")
-        configure_logger(loglevel=BasicConfig.log_level)
+    configure_logger(loglevel=BasicConfig.log_level, working_dir=working_dir)
 
     if args_dict["addr"] == "test":
         print_status("使用测试环境地址", "info")
@@ -239,7 +252,18 @@ def main():
     if args_dict.get("sk", ""):
         BasicConfig.sk = args_dict.get("sk", "")
         print_status("传入了sk参数，优先采用传入参数！", "info")
+    BasicConfig.working_dir = working_dir
 
+    # 显示启动横幅
+    print_unilab_banner(args_dict)
+
+    #####################################
+    ######## 启动设备接入端(主入口) ########
+    #####################################
+    launch(args_dict)
+
+
+def launch(args_dict: Dict[str, Any]):
     # 使用远程资源启动
     if args_dict["use_remote_resource"]:
         print_status("使用远程资源启动", "info")
@@ -254,7 +278,6 @@ def main():
 
     BasicConfig.port = args_dict["port"] if args_dict["port"] else BasicConfig.port
     BasicConfig.disable_browser = args_dict["disable_browser"] or BasicConfig.disable_browser
-    BasicConfig.working_dir = working_dir
     BasicConfig.is_host_mode = not args_dict.get("is_slave", False)
     BasicConfig.slave_no_host = args_dict.get("slave_no_host", False)
     BasicConfig.upload_registry = args_dict.get("upload_registry", False)
@@ -278,9 +301,6 @@ def main():
     from unilabos.resources.graphio import modify_to_backend_format
     from unilabos.ros.nodes.resource_tracker import ResourceTreeSet, ResourceDict
 
-    # 显示启动横幅
-    print_unilab_banner(args_dict)
-
     # 注册表
     lab_registry = build_registry(
         args_dict["registry_path"], args_dict.get("complete_registry", False), args_dict["upload_registry"]

unilabos/app/oss_upload.py

@@ -34,14 +34,14 @@ def _get_oss_token(
         client = http_client
 
     # 构造scene参数: driver_name-exp_type
-    scene = f"{driver_name}-{exp_type}"
+    sub_path = f"{driver_name}-{exp_type}"
 
     # 构造请求URL，使用client的remote_addr（已包含/api/v1/）
     url = f"{client.remote_addr}/applications/token"
-    params = {"scene": scene, "filename": filename}
+    params = {"sub_path": sub_path, "filename": filename, "scene": "job"}
 
     try:
-        logger.info(f"[OSS] 请求预签名URL: scene={scene}, filename={filename}")
+        logger.info(f"[OSS] 请求预签名URL: sub_path={sub_path}, filename={filename}")
         response = requests.get(url, params=params, headers={"Authorization": f"Lab {client.auth}"}, timeout=10)
 
         if response.status_code == 200:

unilabos/app/ws_client.py

@@ -389,7 +389,7 @@ class MessageProcessor:
         self.is_running = True
         self.thread = threading.Thread(target=self._run, daemon=True, name="MessageProcessor")
         self.thread.start()
-        logger.info("[MessageProcessor] Started")
+        logger.trace("[MessageProcessor] Started")
 
     def stop(self) -> None:
         """停止消息处理线程"""
@@ -939,7 +939,7 @@ class QueueProcessor:
         # 事件通知机制
         self.queue_update_event = threading.Event()
 
-        logger.info("[QueueProcessor] Initialized")
+        logger.trace("[QueueProcessor] Initialized")
 
     def set_websocket_client(self, websocket_client: "WebSocketClient"):
         """设置WebSocket客户端引用"""
@@ -954,7 +954,7 @@ class QueueProcessor:
         self.is_running = True
         self.thread = threading.Thread(target=self._run, daemon=True, name="QueueProcessor")
         self.thread.start()
-        logger.info("[QueueProcessor] Started")
+        logger.trace("[QueueProcessor] Started")
 
     def stop(self) -> None:
         """停止队列处理线程"""
@@ -1314,3 +1314,19 @@ class WebSocketClient(BaseCommunicationClient):
             logger.info(f"[WebSocketClient] Job {job_log} cancelled successfully")
         else:
             logger.warning(f"[WebSocketClient] Failed to cancel job {job_log}")
+
+    def publish_host_ready(self) -> None:
+        """发布host_node ready信号"""
+        if self.is_disabled or not self.is_connected():
+            logger.debug("[WebSocketClient] Not connected, cannot publish host ready signal")
+            return
+
+        message = {
+            "action": "host_node_ready",
+            "data": {
+                "status": "ready",
+                "timestamp": time.time(),
+            },
+        }
+        self.message_processor.send_message(message)
+        logger.info("[WebSocketClient] Host node ready signal published")

unilabos/devices/liquid_handling/liquid_handler_abstract.py

@@ -147,6 +147,9 @@ class LiquidHandlerMiddleware(LiquidHandler):
         offsets: Optional[List[Coordinate]] = None,
         **backend_kwargs,
     ):
+        # 如果 use_channels 为 None，使用默认值（所有通道）
+        if use_channels is None:
+            use_channels = list(range(self.channel_num))
         if not offsets or (isinstance(offsets, list) and len(offsets) != len(use_channels)):
             offsets = [Coordinate.zero()] * len(use_channels)
         if self._simulator:
@@ -759,7 +762,7 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
                             blow_out_air_volume=current_dis_blow_out_air_volume,
                             spread=spread,
                         )
-                        if delays is not None:
+                        if delays is not None and len(delays) > 1:
                             await self.custom_delay(seconds=delays[1])
                         await self.touch_tip(current_targets)
                         await self.discard_tips()        
@@ -833,17 +836,19 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
                             spread=spread,
                         )
 
-                        if delays is not None:
+                        if delays is not None and len(delays) > 1:
                             await self.custom_delay(seconds=delays[1])
-                        await self.mix(
+                        # 只有在 mix_time 有效时才调用 mix
-                            targets=[targets[_]],
+                        if mix_time is not None and mix_time > 0:
-                            mix_time=mix_time,
+                            await self.mix(
-                            mix_vol=mix_vol,
+                                targets=[targets[_]],
-                            offsets=offsets if offsets else None,
+                                mix_time=mix_time,
-                            height_to_bottom=mix_liquid_height if mix_liquid_height else None,
+                                mix_vol=mix_vol,
-                            mix_rate=mix_rate if mix_rate else None,
+                                offsets=offsets if offsets else None,
-                        )
+                                height_to_bottom=mix_liquid_height if mix_liquid_height else None,
-                        if delays is not None:
+                                mix_rate=mix_rate if mix_rate else None,
+                            )
+                        if delays is not None and len(delays) > 1:
                             await self.custom_delay(seconds=delays[1])
                         await self.touch_tip(targets[_])
                         await self.discard_tips()
@@ -893,18 +898,20 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
                             blow_out_air_volume=current_dis_blow_out_air_volume,
                             spread=spread,
                         )
-                        if delays is not None:
+                        if delays is not None and len(delays) > 1:
                             await self.custom_delay(seconds=delays[1])
 
-                        await self.mix(
+                        # 只有在 mix_time 有效时才调用 mix
-                            targets=current_targets,
+                        if mix_time is not None and mix_time > 0:
-                            mix_time=mix_time,
+                            await self.mix(
-                            mix_vol=mix_vol,
+                                targets=current_targets,
-                            offsets=offsets if offsets else None,
+                                mix_time=mix_time,
-                            height_to_bottom=mix_liquid_height if mix_liquid_height else None,
+                                mix_vol=mix_vol,
-                            mix_rate=mix_rate if mix_rate else None,
+                                offsets=offsets if offsets else None,
-                        )
+                                height_to_bottom=mix_liquid_height if mix_liquid_height else None,
-                        if delays is not None:
+                                mix_rate=mix_rate if mix_rate else None,
+                            )
+                        if delays is not None and len(delays) > 1:
                             await self.custom_delay(seconds=delays[1])
                         await self.touch_tip(current_targets)
                         await self.discard_tips()
@@ -942,60 +949,146 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
         delays: Optional[List[int]] = None,
         none_keys: List[str] = [],
     ):
-        """Transfer liquid from each *source* well/plate to the corresponding *target*.
+        """Transfer liquid with automatic mode detection.
+
+        Supports three transfer modes:
+        1. One-to-many (1 source -> N targets): Distribute from one source to multiple targets
+        2. One-to-one (N sources -> N targets): Standard transfer, each source to corresponding target
+        3. Many-to-one (N sources -> 1 target): Combine multiple sources into one target
 
         Parameters
         ----------
         asp_vols, dis_vols
-            Single volume (µL) or list matching the number of transfers.
+            Single volume (µL) or list. Automatically expanded based on transfer mode.
         sources, targets
-            Same‑length sequences of containers (wells or plates). In 96‑well mode
+            Containers (wells or plates). Length determines transfer mode:
-            each must contain exactly one plate.
+            - len(sources) == 1, len(targets) > 1: One-to-many mode
+            - len(sources) == len(targets): One-to-one mode
+            - len(sources) > 1, len(targets) == 1: Many-to-one mode
         tip_racks
             One or more TipRacks providing fresh tips.
         is_96_well
             Set *True* to use the 96‑channel head.
         """
-
+        
-
+        # 确保 use_channels 有默认值
+        if use_channels is None:
+            use_channels = [0] if self.channel_num >= 1 else list(range(self.channel_num))
+        
         if is_96_well:
             pass  # This mode is not verified.
         else:
-            if len(asp_vols) != len(targets):
+            # 转换体积参数为列表
-                raise ValueError(f"Length of `asp_vols` {len(asp_vols)} must match `targets` {len(targets)}.")
+            if isinstance(asp_vols, (int, float)):
+                asp_vols = [float(asp_vols)]
+            else:
+                asp_vols = [float(v) for v in asp_vols]
+            
+            if isinstance(dis_vols, (int, float)):
+                dis_vols = [float(dis_vols)]
+            else:
+                dis_vols = [float(v) for v in dis_vols]
+            
+            # 识别传输模式
+            num_sources = len(sources)
+            num_targets = len(targets)
+            
+            if num_sources == 1 and num_targets > 1:
+                # 模式1: 一对多 (1 source -> N targets)
+                await self._transfer_one_to_many(
+                    sources[0], targets, tip_racks, use_channels,
+                    asp_vols, dis_vols, asp_flow_rates, dis_flow_rates,
+                    offsets, touch_tip, liquid_height, blow_out_air_volume,
+                    spread, mix_stage, mix_times, mix_vol, mix_rate,
+                    mix_liquid_height, delays
+                )
+            elif num_sources > 1 and num_targets == 1:
+                # 模式2: 多对一 (N sources -> 1 target)
+                await self._transfer_many_to_one(
+                    sources, targets[0], tip_racks, use_channels,
+                    asp_vols, dis_vols, asp_flow_rates, dis_flow_rates,
+                    offsets, touch_tip, liquid_height, blow_out_air_volume,
+                    spread, mix_stage, mix_times, mix_vol, mix_rate,
+                    mix_liquid_height, delays
+                )
+            elif num_sources == num_targets:
+                # 模式3: 一对一 (N sources -> N targets) - 原有逻辑
+                await self._transfer_one_to_one(
+                    sources, targets, tip_racks, use_channels,
+                    asp_vols, dis_vols, asp_flow_rates, dis_flow_rates,
+                    offsets, touch_tip, liquid_height, blow_out_air_volume,
+                    spread, mix_stage, mix_times, mix_vol, mix_rate,
+                    mix_liquid_height, delays
+                )
+            else:
+                raise ValueError(
+                    f"Unsupported transfer mode: {num_sources} sources -> {num_targets} targets. "
+                    "Supported modes: 1->N, N->1, or N->N."
+                )
 
-            # 首先应该对任务分组，然后每次1个/8个进行操作处理
+    async def _transfer_one_to_one(
-            if len(use_channels) == 1:
+        self,
-                for _ in range(len(targets)):
+        sources: Sequence[Container],
-                    tip = []
+        targets: Sequence[Container],
-                    for ___ in range(len(use_channels)):
+        tip_racks: Sequence[TipRack],
-                        tip.extend(next(self.current_tip))
+        use_channels: List[int],
-                    await self.pick_up_tips(tip)
+        asp_vols: List[float],
+        dis_vols: List[float],
+        asp_flow_rates: Optional[List[Optional[float]]],
+        dis_flow_rates: Optional[List[Optional[float]]],
+        offsets: Optional[List[Coordinate]],
+        touch_tip: bool,
+        liquid_height: Optional[List[Optional[float]]],
+        blow_out_air_volume: Optional[List[Optional[float]]],
+        spread: Literal["wide", "tight", "custom"],
+        mix_stage: Optional[Literal["none", "before", "after", "both"]],
+        mix_times: Optional[int],
+        mix_vol: Optional[int],
+        mix_rate: Optional[int],
+        mix_liquid_height: Optional[float],
+        delays: Optional[List[int]],
+    ):
+        """一对一传输模式：N sources -> N targets"""
+        # 验证参数长度
+        if len(asp_vols) != len(targets):
+            raise ValueError(f"Length of `asp_vols` {len(asp_vols)} must match `targets` {len(targets)}.")
+        if len(dis_vols) != len(targets):
+            raise ValueError(f"Length of `dis_vols` {len(dis_vols)} must match `targets` {len(targets)}.")
+        if len(sources) != len(targets):
+            raise ValueError(f"Length of `sources` {len(sources)} must match `targets` {len(targets)}.")
 
-                    await self.aspirate(
+        if len(use_channels) == 1:
-                        resources=[sources[_]],
+            for _ in range(len(targets)):
-                        vols=[asp_vols[_]],
+                tip = []
-                        use_channels=use_channels,
+                for ___ in range(len(use_channels)):
-                        flow_rates=[asp_flow_rates[0]] if asp_flow_rates else None,
+                    tip.extend(next(self.current_tip))
-                        offsets=[offsets[0]] if offsets else None,
+                await self.pick_up_tips(tip)
-                        liquid_height=[liquid_height[0]] if liquid_height else None,
+
-                        blow_out_air_volume=[blow_out_air_volume[0]] if blow_out_air_volume else None,
+                await self.aspirate(
-                        spread=spread,
+                    resources=[sources[_]],
-                    )
+                    vols=[asp_vols[_]],
-                    if delays is not None:
+                    use_channels=use_channels,
-                        await self.custom_delay(seconds=delays[0])
+                    flow_rates=[asp_flow_rates[_]] if asp_flow_rates and len(asp_flow_rates) > _ else None,
-                    await self.dispense(
+                    offsets=[offsets[_]] if offsets and len(offsets) > _ else None,
-                        resources=[targets[_]],
+                    liquid_height=[liquid_height[_]] if liquid_height and len(liquid_height) > _ else None,
-                        vols=[dis_vols[_]],
+                    blow_out_air_volume=[blow_out_air_volume[_]] if blow_out_air_volume and len(blow_out_air_volume) > _ else None,
-                        use_channels=use_channels,
+                    spread=spread,
-                        flow_rates=[dis_flow_rates[1]] if dis_flow_rates else None,
+                )
-                        offsets=[offsets[1]] if offsets else None,
+                if delays is not None:
-                        blow_out_air_volume=[blow_out_air_volume[1]] if blow_out_air_volume else None,
+                    await self.custom_delay(seconds=delays[0])
-                        liquid_height=[liquid_height[1]] if liquid_height else None,
+                await self.dispense(
-                        spread=spread,
+                    resources=[targets[_]],
-                    )
+                    vols=[dis_vols[_]],
-                    if delays is not None:
+                    use_channels=use_channels,
-                        await self.custom_delay(seconds=delays[1])
+                    flow_rates=[dis_flow_rates[_]] if dis_flow_rates and len(dis_flow_rates) > _ else None,
+                    offsets=[offsets[_]] if offsets and len(offsets) > _ else None,
+                    blow_out_air_volume=[blow_out_air_volume[_]] if blow_out_air_volume and len(blow_out_air_volume) > _ else None,
+                    liquid_height=[liquid_height[_]] if liquid_height and len(liquid_height) > _ else None,
+                    spread=spread,
+                )
+                if delays is not None and len(delays) > 1:
+                    await self.custom_delay(seconds=delays[1])
+                if mix_stage in ["after", "both"] and mix_times is not None and mix_times > 0:
                     await self.mix(
                         targets=[targets[_]],
                         mix_time=mix_times,
@@ -1004,63 +1097,60 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
                         height_to_bottom=mix_liquid_height if mix_liquid_height else None,
                         mix_rate=mix_rate if mix_rate else None,
                     )
-                    if delays is not None:
+                if delays is not None and len(delays) > 1:
-                        await self.custom_delay(seconds=delays[1])
+                    await self.custom_delay(seconds=delays[1])
-                    await self.touch_tip(targets[_])
+                await self.touch_tip(targets[_])
-                    await self.discard_tips()
+                await self.discard_tips(use_channels=use_channels)
 
-            elif len(use_channels) == 8:
+        elif len(use_channels) == 8:
-                # 对于8个的情况，需要判断此时任务是不是能被8通道移液站来成功处理
+            if len(targets) % 8 != 0:
-                if len(targets) % 8 != 0:
+                raise ValueError(f"Length of `targets` {len(targets)} must be a multiple of 8 for 8-channel mode.")
-                    raise ValueError(f"Length of `targets` {len(targets)} must be a multiple of 8 for 8-channel mode.")
 
-                # 8个8个来取任务序列
+            for i in range(0, len(targets), 8):
+                tip = []
+                for _ in range(len(use_channels)):
+                    tip.extend(next(self.current_tip))
+                await self.pick_up_tips(tip)
+                current_targets = targets[i:i + 8]
+                current_reagent_sources = sources[i:i + 8]
+                current_asp_vols = asp_vols[i:i + 8]
+                current_dis_vols = dis_vols[i:i + 8]
+                current_asp_flow_rates = asp_flow_rates[i:i + 8] if asp_flow_rates else None
+                current_asp_offset = offsets[i:i + 8] if offsets else [None] * 8
+                current_dis_offset = offsets[i:i + 8] if offsets else [None] * 8
+                current_asp_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8
+                current_dis_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8
+                current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
+                current_dis_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
+                current_dis_flow_rates = dis_flow_rates[i:i + 8] if dis_flow_rates else None
 
-                for i in range(0, len(targets), 8):
+                await self.aspirate(
-                    # 取出8个任务
+                    resources=current_reagent_sources,
-                    tip = []
+                    vols=current_asp_vols,
-                    for _ in range(len(use_channels)):
+                    use_channels=use_channels,
-                        tip.extend(next(self.current_tip))
+                    flow_rates=current_asp_flow_rates,
-                    await self.pick_up_tips(tip)
+                    offsets=current_asp_offset,
-                    current_targets = targets[i:i + 8]
+                    blow_out_air_volume=current_asp_blow_out_air_volume,
-                    current_reagent_sources = sources[i:i + 8]
+                    liquid_height=current_asp_liquid_height,
-                    current_asp_vols = asp_vols[i:i + 8]
+                    spread=spread,
-                    current_dis_vols = dis_vols[i:i + 8]
+                )
-                    current_asp_flow_rates = asp_flow_rates[i:i + 8]
-                    current_asp_offset = offsets[i:i + 8] if offsets else [None] * 8
-                    current_dis_offset = offsets[-i*8-8:len(offsets)-i*8] if offsets else [None] * 8
-                    current_asp_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8
-                    current_dis_liquid_height = liquid_height[-i*8-8:len(liquid_height)-i*8] if liquid_height else [None] * 8
-                    current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
-                    current_dis_blow_out_air_volume = blow_out_air_volume[-i*8-8:len(blow_out_air_volume)-i*8] if blow_out_air_volume else [None] * 8
-                    current_dis_flow_rates = dis_flow_rates[i:i + 8] if dis_flow_rates else [None] * 8
 
-                    await self.aspirate(
+                if delays is not None:
-                        resources=current_reagent_sources,
+                    await self.custom_delay(seconds=delays[0])
-                        vols=current_asp_vols,
+                await self.dispense(
-                        use_channels=use_channels,
+                    resources=current_targets,
-                        flow_rates=current_asp_flow_rates,
+                    vols=current_dis_vols,
-                        offsets=current_asp_offset,
+                    use_channels=use_channels,
-                        blow_out_air_volume=current_asp_blow_out_air_volume,
+                    flow_rates=current_dis_flow_rates,
-                        liquid_height=current_asp_liquid_height,
+                    offsets=current_dis_offset,
-                        spread=spread,
+                    blow_out_air_volume=current_dis_blow_out_air_volume,
-                    )
+                    liquid_height=current_dis_liquid_height,
-
+                    spread=spread,
-                    if delays is not None:
+                )
-                        await self.custom_delay(seconds=delays[0])
+                if delays is not None and len(delays) > 1:
-                    await self.dispense(
+                    await self.custom_delay(seconds=delays[1])
-                        resources=current_targets,
-                        vols=current_dis_vols,
-                        use_channels=use_channels,
-                        flow_rates=current_dis_flow_rates,
-                        offsets=current_dis_offset,
-                        blow_out_air_volume=current_dis_blow_out_air_volume,
-                        liquid_height=current_dis_liquid_height,
-                        spread=spread,
-                    )
-                    if delays is not None:
-                        await self.custom_delay(seconds=delays[1])
 
+                if mix_stage in ["after", "both"] and mix_times is not None and mix_times > 0:
                     await self.mix(
                         targets=current_targets,
                         mix_time=mix_times,
@@ -1069,10 +1159,363 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
                         height_to_bottom=mix_liquid_height if mix_liquid_height else None,
                         mix_rate=mix_rate if mix_rate else None,
                     )
-                    if delays is not None:
+                if delays is not None and len(delays) > 1:
-                        await self.custom_delay(seconds=delays[1])
+                    await self.custom_delay(seconds=delays[1])
+                await self.touch_tip(current_targets)
+                await self.discard_tips([0,1,2,3,4,5,6,7])
+
+    async def _transfer_one_to_many(
+        self,
+        source: Container,
+        targets: Sequence[Container],
+        tip_racks: Sequence[TipRack],
+        use_channels: List[int],
+        asp_vols: List[float],
+        dis_vols: List[float],
+        asp_flow_rates: Optional[List[Optional[float]]],
+        dis_flow_rates: Optional[List[Optional[float]]],
+        offsets: Optional[List[Coordinate]],
+        touch_tip: bool,
+        liquid_height: Optional[List[Optional[float]]],
+        blow_out_air_volume: Optional[List[Optional[float]]],
+        spread: Literal["wide", "tight", "custom"],
+        mix_stage: Optional[Literal["none", "before", "after", "both"]],
+        mix_times: Optional[int],
+        mix_vol: Optional[int],
+        mix_rate: Optional[int],
+        mix_liquid_height: Optional[float],
+        delays: Optional[List[int]],
+    ):
+        """一对多传输模式：1 source -> N targets"""
+        # 验证和扩展体积参数
+        if len(asp_vols) == 1:
+            # 如果只提供一个吸液体积，计算总吸液体积（所有分液体积之和）
+            total_asp_vol = sum(dis_vols)
+            asp_vol = asp_vols[0] if asp_vols[0] >= total_asp_vol else total_asp_vol
+        else:
+            raise ValueError("For one-to-many mode, `asp_vols` should be a single value or list with one element.")
+        
+        if len(dis_vols) != len(targets):
+            raise ValueError(f"Length of `dis_vols` {len(dis_vols)} must match `targets` {len(targets)}.")
+
+        if len(use_channels) == 1:
+            # 单通道模式：一次吸液，多次分液
+            tip = []
+            for _ in range(len(use_channels)):
+                tip.extend(next(self.current_tip))
+            await self.pick_up_tips(tip)
+
+            # 从源容器吸液（总体积）
+            await self.aspirate(
+                resources=[source],
+                vols=[asp_vol],
+                use_channels=use_channels,
+                flow_rates=[asp_flow_rates[0]] if asp_flow_rates and len(asp_flow_rates) > 0 else None,
+                offsets=[offsets[0]] if offsets and len(offsets) > 0 else None,
+                liquid_height=[liquid_height[0]] if liquid_height and len(liquid_height) > 0 else None,
+                blow_out_air_volume=[blow_out_air_volume[0]] if blow_out_air_volume and len(blow_out_air_volume) > 0 else None,
+                spread=spread,
+            )
+            
+            if delays is not None:
+                await self.custom_delay(seconds=delays[0])
+            
+            # 分多次分液到不同的目标容器
+            for idx, target in enumerate(targets):
+                await self.dispense(
+                    resources=[target],
+                    vols=[dis_vols[idx]],
+                    use_channels=use_channels,
+                    flow_rates=[dis_flow_rates[idx]] if dis_flow_rates and len(dis_flow_rates) > idx else None,
+                    offsets=[offsets[idx]] if offsets and len(offsets) > idx else None,
+                    blow_out_air_volume=[blow_out_air_volume[idx]] if blow_out_air_volume and len(blow_out_air_volume) > idx else None,
+                    liquid_height=[liquid_height[idx]] if liquid_height and len(liquid_height) > idx else None,
+                    spread=spread,
+                )
+                if delays is not None and len(delays) > 1:
+                    await self.custom_delay(seconds=delays[1])
+                if mix_stage in ["after", "both"] and mix_times is not None and mix_times > 0:
+                    await self.mix(
+                        targets=[target],
+                        mix_time=mix_times,
+                        mix_vol=mix_vol,
+                        offsets=offsets[idx:idx+1] if offsets else None,
+                        height_to_bottom=mix_liquid_height if mix_liquid_height else None,
+                        mix_rate=mix_rate if mix_rate else None,
+                    )
+                if touch_tip:
+                    await self.touch_tip([target])
+            
+            await self.discard_tips(use_channels=use_channels)
+        
+        elif len(use_channels) == 8:
+            # 8通道模式：需要确保目标数量是8的倍数
+            if len(targets) % 8 != 0:
+                raise ValueError(f"For 8-channel mode, number of targets {len(targets)} must be a multiple of 8.")
+            
+            # 每次处理8个目标
+            for i in range(0, len(targets), 8):
+                tip = []
+                for _ in range(len(use_channels)):
+                    tip.extend(next(self.current_tip))
+                await self.pick_up_tips(tip)
+                
+                current_targets = targets[i:i + 8]
+                current_dis_vols = dis_vols[i:i + 8]
+                
+                # 8个通道都从同一个源容器吸液，每个通道的吸液体积等于对应的分液体积
+                current_asp_flow_rates = asp_flow_rates[0:1] * 8 if asp_flow_rates and len(asp_flow_rates) > 0 else None
+                current_asp_offset = offsets[0:1] * 8 if offsets and len(offsets) > 0 else [None] * 8
+                current_asp_liquid_height = liquid_height[0:1] * 8 if liquid_height and len(liquid_height) > 0 else [None] * 8
+                current_asp_blow_out_air_volume = blow_out_air_volume[0:1] * 8 if blow_out_air_volume and len(blow_out_air_volume) > 0 else [None] * 8
+                
+                # 从源容器吸液（8个通道都从同一个源，但每个通道的吸液体积不同）
+                await self.aspirate(
+                    resources=[source] * 8,  # 8个通道都从同一个源
+                    vols=current_dis_vols,  # 每个通道的吸液体积等于对应的分液体积
+                    use_channels=use_channels,
+                    flow_rates=current_asp_flow_rates,
+                    offsets=current_asp_offset,
+                    liquid_height=current_asp_liquid_height,
+                    blow_out_air_volume=current_asp_blow_out_air_volume,
+                    spread=spread,
+                )
+                
+                if delays is not None:
+                    await self.custom_delay(seconds=delays[0])
+                
+                # 分液到8个目标
+                current_dis_flow_rates = dis_flow_rates[i:i + 8] if dis_flow_rates else None
+                current_dis_offset = offsets[i:i + 8] if offsets else [None] * 8
+                current_dis_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8
+                current_dis_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
+                
+                await self.dispense(
+                    resources=current_targets,
+                    vols=current_dis_vols,
+                    use_channels=use_channels,
+                    flow_rates=current_dis_flow_rates,
+                    offsets=current_dis_offset,
+                    blow_out_air_volume=current_dis_blow_out_air_volume,
+                    liquid_height=current_dis_liquid_height,
+                    spread=spread,
+                )
+                
+                if delays is not None and len(delays) > 1:
+                    await self.custom_delay(seconds=delays[1])
+                
+                if mix_stage in ["after", "both"] and mix_times is not None and mix_times > 0:
+                    await self.mix(
+                        targets=current_targets,
+                        mix_time=mix_times,
+                        mix_vol=mix_vol,
+                        offsets=offsets if offsets else None,
+                        height_to_bottom=mix_liquid_height if mix_liquid_height else None,
+                        mix_rate=mix_rate if mix_rate else None,
+                    )
+                
+                if touch_tip:
                     await self.touch_tip(current_targets)
-                    await self.discard_tips([0,1,2,3,4,5,6,7])
+                
+                await self.discard_tips([0,1,2,3,4,5,6,7])
+
+    async def _transfer_many_to_one(
+        self,
+        sources: Sequence[Container],
+        target: Container,
+        tip_racks: Sequence[TipRack],
+        use_channels: List[int],
+        asp_vols: List[float],
+        dis_vols: List[float],
+        asp_flow_rates: Optional[List[Optional[float]]],
+        dis_flow_rates: Optional[List[Optional[float]]],
+        offsets: Optional[List[Coordinate]],
+        touch_tip: bool,
+        liquid_height: Optional[List[Optional[float]]],
+        blow_out_air_volume: Optional[List[Optional[float]]],
+        spread: Literal["wide", "tight", "custom"],
+        mix_stage: Optional[Literal["none", "before", "after", "both"]],
+        mix_times: Optional[int],
+        mix_vol: Optional[int],
+        mix_rate: Optional[int],
+        mix_liquid_height: Optional[float],
+        delays: Optional[List[int]],
+    ):
+        """多对一传输模式：N sources -> 1 target（汇总/混合）"""
+        # 验证和扩展体积参数
+        if len(asp_vols) != len(sources):
+            raise ValueError(f"Length of `asp_vols` {len(asp_vols)} must match `sources` {len(sources)}.")
+        
+        # 支持两种模式：
+        # 1. dis_vols 为单个值：所有源汇总，使用总吸液体积或指定分液体积
+        # 2. dis_vols 长度等于 asp_vols：每个源按不同比例分液（按比例混合）
+        if len(dis_vols) == 1:
+            # 模式1：使用单个分液体积
+            total_dis_vol = sum(asp_vols)
+            dis_vol = dis_vols[0] if dis_vols[0] >= total_dis_vol else total_dis_vol
+            use_proportional_mixing = False
+        elif len(dis_vols) == len(asp_vols):
+            # 模式2：按不同比例混合
+            use_proportional_mixing = True
+        else:
+            raise ValueError(
+                f"For many-to-one mode, `dis_vols` should be a single value or list with length {len(asp_vols)} "
+                f"(matching `asp_vols`). Got length {len(dis_vols)}."
+            )
+
+        if len(use_channels) == 1:
+            # 单通道模式：多次吸液，一次分液
+            # 先混合前（如果需要）
+            if mix_stage in ["before", "both"] and mix_times is not None and mix_times > 0:
+                # 注意：在吸液前混合源容器通常不常见，这里跳过
+                pass
+            
+            # 从每个源容器吸液并分液到目标容器
+            for idx, source in enumerate(sources):
+                tip = []
+                for _ in range(len(use_channels)):
+                    tip.extend(next(self.current_tip))
+                await self.pick_up_tips(tip)
+
+                await self.aspirate(
+                    resources=[source],
+                    vols=[asp_vols[idx]],
+                    use_channels=use_channels,
+                    flow_rates=[asp_flow_rates[idx]] if asp_flow_rates and len(asp_flow_rates) > idx else None,
+                    offsets=[offsets[idx]] if offsets and len(offsets) > idx else None,
+                    liquid_height=[liquid_height[idx]] if liquid_height and len(liquid_height) > idx else None,
+                    blow_out_air_volume=[blow_out_air_volume[idx]] if blow_out_air_volume and len(blow_out_air_volume) > idx else None,
+                    spread=spread,
+                )
+                
+                if delays is not None:
+                    await self.custom_delay(seconds=delays[0])
+                
+                # 分液到目标容器
+                if use_proportional_mixing:
+                    # 按不同比例混合：使用对应的 dis_vols
+                    dis_vol = dis_vols[idx]
+                    dis_flow_rate = dis_flow_rates[idx] if dis_flow_rates and len(dis_flow_rates) > idx else None
+                    dis_offset = offsets[idx] if offsets and len(offsets) > idx else None
+                    dis_liquid_height = liquid_height[idx] if liquid_height and len(liquid_height) > idx else None
+                    dis_blow_out = blow_out_air_volume[idx] if blow_out_air_volume and len(blow_out_air_volume) > idx else None
+                else:
+                    # 标准模式：分液体积等于吸液体积
+                    dis_vol = asp_vols[idx]
+                    dis_flow_rate = dis_flow_rates[0] if dis_flow_rates and len(dis_flow_rates) > 0 else None
+                    dis_offset = offsets[0] if offsets and len(offsets) > 0 else None
+                    dis_liquid_height = liquid_height[0] if liquid_height and len(liquid_height) > 0 else None
+                    dis_blow_out = blow_out_air_volume[0] if blow_out_air_volume and len(blow_out_air_volume) > 0 else None
+                
+                await self.dispense(
+                    resources=[target],
+                    vols=[dis_vol],
+                    use_channels=use_channels,
+                    flow_rates=[dis_flow_rate] if dis_flow_rate is not None else None,
+                    offsets=[dis_offset] if dis_offset is not None else None,
+                    blow_out_air_volume=[dis_blow_out] if dis_blow_out is not None else None,
+                    liquid_height=[dis_liquid_height] if dis_liquid_height is not None else None,
+                    spread=spread,
+                )
+                
+                if delays is not None and len(delays) > 1:
+                    await self.custom_delay(seconds=delays[1])
+                
+                await self.discard_tips(use_channels=use_channels)
+            
+            # 最后在目标容器中混合（如果需要）
+            if mix_stage in ["after", "both"] and mix_times is not None and mix_times > 0:
+                await self.mix(
+                    targets=[target],
+                    mix_time=mix_times,
+                    mix_vol=mix_vol,
+                    offsets=offsets[0:1] if offsets else None,
+                    height_to_bottom=mix_liquid_height if mix_liquid_height else None,
+                    mix_rate=mix_rate if mix_rate else None,
+                )
+            
+            if touch_tip:
+                await self.touch_tip([target])
+        
+        elif len(use_channels) == 8:
+            # 8通道模式：需要确保源数量是8的倍数
+            if len(sources) % 8 != 0:
+                raise ValueError(f"For 8-channel mode, number of sources {len(sources)} must be a multiple of 8.")
+            
+            # 每次处理8个源
+            for i in range(0, len(sources), 8):
+                tip = []
+                for _ in range(len(use_channels)):
+                    tip.extend(next(self.current_tip))
+                await self.pick_up_tips(tip)
+                
+                current_sources = sources[i:i + 8]
+                current_asp_vols = asp_vols[i:i + 8]
+                current_asp_flow_rates = asp_flow_rates[i:i + 8] if asp_flow_rates else None
+                current_asp_offset = offsets[i:i + 8] if offsets else [None] * 8
+                current_asp_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8
+                current_asp_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
+                
+                # 从8个源容器吸液
+                await self.aspirate(
+                    resources=current_sources,
+                    vols=current_asp_vols,
+                    use_channels=use_channels,
+                    flow_rates=current_asp_flow_rates,
+                    offsets=current_asp_offset,
+                    blow_out_air_volume=current_asp_blow_out_air_volume,
+                    liquid_height=current_asp_liquid_height,
+                    spread=spread,
+                )
+                
+                if delays is not None:
+                    await self.custom_delay(seconds=delays[0])
+                
+                # 分液到目标容器（每个通道分液到同一个目标）
+                if use_proportional_mixing:
+                    # 按比例混合：使用对应的 dis_vols
+                    current_dis_vols = dis_vols[i:i + 8]
+                    current_dis_flow_rates = dis_flow_rates[i:i + 8] if dis_flow_rates else None
+                    current_dis_offset = offsets[i:i + 8] if offsets else [None] * 8
+                    current_dis_liquid_height = liquid_height[i:i + 8] if liquid_height else [None] * 8
+                    current_dis_blow_out_air_volume = blow_out_air_volume[i:i + 8] if blow_out_air_volume else [None] * 8
+                else:
+                    # 标准模式：每个通道分液体积等于其吸液体积
+                    current_dis_vols = current_asp_vols
+                    current_dis_flow_rates = dis_flow_rates[0:1] * 8 if dis_flow_rates else None
+                    current_dis_offset = offsets[0:1] * 8 if offsets else [None] * 8
+                    current_dis_liquid_height = liquid_height[0:1] * 8 if liquid_height else [None] * 8
+                    current_dis_blow_out_air_volume = blow_out_air_volume[0:1] * 8 if blow_out_air_volume else [None] * 8
+                
+                await self.dispense(
+                    resources=[target] * 8,  # 8个通道都分到同一个目标
+                    vols=current_dis_vols,
+                    use_channels=use_channels,
+                    flow_rates=current_dis_flow_rates,
+                    offsets=current_dis_offset,
+                    blow_out_air_volume=current_dis_blow_out_air_volume,
+                    liquid_height=current_dis_liquid_height,
+                    spread=spread,
+                )
+                
+                if delays is not None and len(delays) > 1:
+                    await self.custom_delay(seconds=delays[1])
+                
+                await self.discard_tips([0,1,2,3,4,5,6,7])
+            
+            # 最后在目标容器中混合（如果需要）
+            if mix_stage in ["after", "both"] and mix_times is not None and mix_times > 0:
+                await self.mix(
+                    targets=[target],
+                    mix_time=mix_times,
+                    mix_vol=mix_vol,
+                    offsets=offsets[0:1] if offsets else None,
+                    height_to_bottom=mix_liquid_height if mix_liquid_height else None,
+                    mix_rate=mix_rate if mix_rate else None,
+                )
+            
+            if touch_tip:
+                await self.touch_tip([target])
 
     # except Exception as e:
     #     traceback.print_exc()

unilabos/devices/virtual/virtual_multiway_valve.py

@@ -7,7 +7,7 @@ class VirtualMultiwayValve:
     """
     虚拟九通阀门 - 0号位连接transfer pump，1-8号位连接其他设备 🔄
     """
-    def __init__(self, port: str = "VIRTUAL", positions: int = 8):
+    def __init__(self, port: str = "VIRTUAL", positions: int = 8, **kwargs):
         self.port = port
         self.max_positions = positions  # 1-8号位
         self.total_positions = positions + 1  # 0-8号位，共9个位置

unilabos/devices/workstation/workstation_base.py

@@ -147,7 +147,7 @@ class WorkstationBase(ABC):
 
     def __init__(
         self,
-        deck: Deck,
+        deck: Optional[Deck],
         *args,
         **kwargs,  # 必须有kwargs
     ):
@@ -349,5 +349,5 @@ class WorkstationBase(ABC):
 
 
 class ProtocolNode(WorkstationBase):
-    def __init__(self, deck: Optional[PLRResource], *args, **kwargs):
+    def __init__(self, protocol_type: List[str], deck: Optional[PLRResource], *args, **kwargs):
         super().__init__(deck, *args, **kwargs)

unilabos/registry/devices/bioyond_dispensing_station.yaml

@@ -620,6 +620,56 @@ bioyond_dispensing_station:
           title: DispenStationSolnPrep
           type: object
         type: DispenStationSolnPrep
+      transfer_materials_to_reaction_station:
+        feedback: {}
+        goal:
+          target_device_id: target_device_id
+          transfer_groups: transfer_groups
+        goal_default:
+          target_device_id: ''
+          transfer_groups: ''
+        handles: {}
+        placeholder_keys:
+          target_device_id: unilabos_devices
+        result: {}
+        schema:
+          description: 将配液站完成的物料（溶液、样品等）转移到指定反应站的堆栈库位。支持配置多组转移任务，每组包含物料名称、目标堆栈和目标库位。
+          properties:
+            feedback: {}
+            goal:
+              properties:
+                target_device_id:
+                  description: 目标反应站设备ID（从设备列表中选择，所有转移组都使用同一个目标设备）
+                  type: string
+                transfer_groups:
+                  description: 转移任务组列表，每组包含物料名称、目标堆栈和目标库位，可以添加多组
+                  items:
+                    properties:
+                      materials:
+                        description: 物料名称（手动输入，系统将通过RPC查询验证）
+                        type: string
+                      target_sites:
+                        description: 目标库位（手动输入，如"A01"）
+                        type: string
+                      target_stack:
+                        description: 目标堆栈名称（手动输入，如"堆栈1左"）
+                        type: string
+                    required:
+                    - materials
+                    - target_stack
+                    - target_sites
+                    type: object
+                  type: array
+              required:
+              - target_device_id
+              - transfer_groups
+              type: object
+            result: {}
+          required:
+          - goal
+          title: transfer_materials_to_reaction_station参数
+          type: object
+        type: UniLabJsonCommand
       wait_for_multiple_orders_and_get_reports:
         feedback: {}
         goal:

unilabos/registry/devices/work_station.yaml

@@ -6036,7 +6036,12 @@ workstation:
       properties:
         deck:
           type: string
+        protocol_type:
+          items:
+            type: string
+          type: array
       required:
+      - protocol_type
       - deck
       type: object
     data:

unilabos/registry/resources/organic/container.yaml

@@ -2,7 +2,7 @@ container:
   category:
   - container
   class:
-    module: unilabos.resources.container:RegularContainer
+    module: unilabos.resources.container:get_regular_container
     type: pylabrobot
   description: regular organic container
   handles:

unilabos/resources/container.py

@@ -22,6 +22,13 @@ class RegularContainer(Container):
 
     def load_state(self, state: Dict[str, Any]):
         self.state = state
+
+
+def get_regular_container(name="container"):
+    r = RegularContainer(name=name)
+    r.category = "container"
+    return RegularContainer(name=name)
+
 #
 # class RegularContainer(object):
 #     # 第一个参数必须是id传入

unilabos/resources/graphio.py

@@ -45,10 +45,13 @@ def canonicalize_nodes_data(
     print_status(f"{len(nodes)} Resources loaded:", "info")
 
     # 第一步：基本预处理（处理graphml的label字段）
-    for node in nodes:
+    outer_host_node_id = None
+    for idx, node in enumerate(nodes):
         if node.get("label") is not None:
             node_id = node.pop("label")
             node["id"] = node["name"] = node_id
+        if node["id"] == "host_node":
+            outer_host_node_id = idx
         if not isinstance(node.get("config"), dict):
             node["config"] = {}
         if not node.get("type"):
@@ -58,25 +61,26 @@ def canonicalize_nodes_data(
             node["name"] = node.get("id")
             print_status(f"Warning: Node {node.get('id', 'unknown')} missing 'name', defaulting to {node['name']}", "warning")
         if not isinstance(node.get("position"), dict):
-            node["position"] = {"position": {}}
+            node["pose"] = {"position": {}}
             x = node.pop("x", None)
             if x is not None:
-                node["position"]["position"]["x"] = x
+                node["pose"]["position"]["x"] = x
             y = node.pop("y", None)
             if y is not None:
-                node["position"]["position"]["y"] = y
+                node["pose"]["position"]["y"] = y
             z = node.pop("z", None)
             if z is not None:
-                node["position"]["position"]["z"] = z
+                node["pose"]["position"]["z"] = z
         if "sample_id" in node:
             sample_id = node.pop("sample_id")
             if sample_id:
                 logger.error(f"{node}的sample_id参数已弃用，sample_id: {sample_id}")
         for k in list(node.keys()):
-            if k not in ["id", "uuid", "name", "description", "schema", "model", "icon", "parent_uuid", "parent", "type", "class", "position", "config", "data", "children"]:
+            if k not in ["id", "uuid", "name", "description", "schema", "model", "icon", "parent_uuid", "parent", "type", "class", "position", "config", "data", "children", "pose"]:
                 v = node.pop(k)
                 node["config"][k] = v
-
+    if outer_host_node_id is not None:
+        nodes.pop(outer_host_node_id)
     # 第二步：处理parent_relation
     id2idx = {node["id"]: idx for idx, node in enumerate(nodes)}
     for parent, children in parent_relation.items():
@@ -93,7 +97,7 @@ def canonicalize_nodes_data(
 
     for node in nodes:
         try:
-            print_status(f"DeviceId: {node['id']}, Class: {node['class']}", "info")
+            # print_status(f"DeviceId: {node['id']}, Class: {node['class']}", "info")
             # 使用标准化方法
             resource_instance = ResourceDictInstance.get_resource_instance_from_dict(node)
             known_nodes[node["id"]] = resource_instance
@@ -582,11 +586,15 @@ def resource_plr_to_ulab(resource_plr: "ResourcePLR", parent_name: str = None, w
             "tip_rack": "tip_rack",
             "warehouse": "warehouse",
             "container": "container",
+            "tube": "tube",
+            "bottle_carrier": "bottle_carrier",
+            "plate_adapter": "plate_adapter",
         }
         if source in replace_info:
             return replace_info[source]
         else:
-            logger.warning(f"转换pylabrobot的时候，出现未知类型: {source}")
+            if source is not None:
+                logger.warning(f"转换pylabrobot的时候，出现未知类型: {source}")
             return source
 
     def resource_plr_to_ulab_inner(d: dict, all_states: dict, child=True) -> dict:

unilabos/ros/device_node_wrapper.py

@@ -5,6 +5,7 @@ from unilabos.ros.msgs.message_converter import (
     get_action_type,
 )
 from unilabos.ros.nodes.base_device_node import init_wrapper, ROS2DeviceNode
+from unilabos.ros.nodes.resource_tracker import ResourceDictInstance
 
 # 定义泛型类型变量
 T = TypeVar("T")
@@ -18,12 +19,11 @@ class ROS2DeviceNodeWrapper(ROS2DeviceNode):
 
 def ros2_device_node(
     cls: Type[T],
-    device_config: Optional[Dict[str, Any]] = None,
+    device_config: Optional[ResourceDictInstance] = None,
     status_types: Optional[Dict[str, Any]] = None,
     action_value_mappings: Optional[Dict[str, Any]] = None,
     hardware_interface: Optional[Dict[str, Any]] = None,
     print_publish: bool = False,
-    children: Optional[Dict[str, Any]] = None,
 ) -> Type[ROS2DeviceNodeWrapper]:
     """Create a ROS2 Node class for a device class with properties and actions.
 
@@ -45,7 +45,7 @@ def ros2_device_node(
     if status_types is None:
         status_types = {}
     if device_config is None:
-        device_config = {}
+        raise ValueError("device_config cannot be None")
     if action_value_mappings is None:
         action_value_mappings = {}
     if hardware_interface is None:
@@ -82,7 +82,6 @@ def ros2_device_node(
                 action_value_mappings=action_value_mappings,
                 hardware_interface=hardware_interface,
                 print_publish=print_publish,
-                children=children,
                 *args,
                 **kwargs,
             ),

unilabos/ros/initialize_device.py

@@ -4,13 +4,14 @@ from typing import Optional
 from unilabos.registry.registry import lab_registry
 from unilabos.ros.device_node_wrapper import ros2_device_node
 from unilabos.ros.nodes.base_device_node import ROS2DeviceNode, DeviceInitError
+from unilabos.ros.nodes.resource_tracker import ResourceDictInstance
 from unilabos.utils import logger
 from unilabos.utils.exception import DeviceClassInvalid
 from unilabos.utils.import_manager import default_manager
 
 
 
-def initialize_device_from_dict(device_id, device_config) -> Optional[ROS2DeviceNode]:
+def initialize_device_from_dict(device_id, device_config: ResourceDictInstance) -> Optional[ROS2DeviceNode]:
     """Initializes a device based on its configuration.
 
     This function dynamically imports the appropriate device class and creates an instance of it using the provided device configuration.
@@ -24,15 +25,14 @@ def initialize_device_from_dict(device_id, device_config) -> Optional[ROS2Device
         None
     """
     d = None
-    original_device_config = copy.deepcopy(device_config)
+    device_class_config = device_config.res_content.klass
-    device_class_config = device_config["class"]
+    uid = device_config.res_content.uuid
-    uid = device_config["uuid"]
     if isinstance(device_class_config, str):  # 如果是字符串，则直接去lab_registry中查找，获取class
         if len(device_class_config) == 0:
             raise DeviceClassInvalid(f"Device [{device_id}] class cannot be an empty string. {device_config}")
         if device_class_config not in lab_registry.device_type_registry:
             raise DeviceClassInvalid(f"Device [{device_id}] class {device_class_config} not found. {device_config}")
-        device_class_config = device_config["class"] = lab_registry.device_type_registry[device_class_config]["class"]
+        device_class_config = lab_registry.device_type_registry[device_class_config]["class"]
     elif isinstance(device_class_config, dict):
         raise DeviceClassInvalid(f"Device [{device_id}] class config should be type 'str' but 'dict' got. {device_config}")
     if isinstance(device_class_config, dict):
@@ -41,17 +41,16 @@ def initialize_device_from_dict(device_id, device_config) -> Optional[ROS2Device
         DEVICE = ros2_device_node(
             DEVICE,
             status_types=device_class_config.get("status_types", {}),
-            device_config=original_device_config,
+            device_config=device_config,
             action_value_mappings=device_class_config.get("action_value_mappings", {}),
             hardware_interface=device_class_config.get(
                 "hardware_interface",
                 {"name": "hardware_interface", "write": "send_command", "read": "read_data", "extra_info": []},
-            ),
+            )
-            children=device_config.get("children", {})
         )
         try:
             d = DEVICE(
-                device_id=device_id, device_uuid=uid, driver_is_ros=device_class_config["type"] == "ros2", driver_params=device_config.get("config", {})
+                device_id=device_id, device_uuid=uid, driver_is_ros=device_class_config["type"] == "ros2", driver_params=device_config.res_content.config
             )
         except DeviceInitError as ex:
             return d

unilabos/ros/main_slave_run.py

@@ -192,7 +192,7 @@ def slave(
     for device_config in devices_config.root_nodes:
         device_id = device_config.res_content.id
         if device_config.res_content.type == "device":
-            d = initialize_device_from_dict(device_id, device_config.get_nested_dict())
+            d = initialize_device_from_dict(device_id, device_config)
             if d is not None:
                 devices_instances[device_id] = d
                 logger.info(f"Device {device_id} initialized.")

unilabos/ros/nodes/base_device_node.py

@@ -48,7 +48,7 @@ from unilabos_msgs.msg import Resource  # type: ignore
 from unilabos.ros.nodes.resource_tracker import (
     DeviceNodeResourceTracker,
     ResourceTreeSet,
-    ResourceTreeInstance,
+    ResourceTreeInstance, ResourceDictInstance,
 )
 from unilabos.ros.x.rclpyx import get_event_loop
 from unilabos.ros.utils.driver_creator import WorkstationNodeCreator, PyLabRobotCreator, DeviceClassCreator
@@ -133,12 +133,11 @@ def init_wrapper(
     device_id: str,
     device_uuid: str,
     driver_class: type[T],
-    device_config: Dict[str, Any],
+    device_config: ResourceTreeInstance,
     status_types: Dict[str, Any],
     action_value_mappings: Dict[str, Any],
     hardware_interface: Dict[str, Any],
     print_publish: bool,
-    children: Optional[list] = None,
     driver_params: Optional[Dict[str, Any]] = None,
     driver_is_ros: bool = False,
     *args,
@@ -147,8 +146,6 @@ def init_wrapper(
     """初始化设备节点的包装函数，和ROS2DeviceNode初始化保持一致"""
     if driver_params is None:
         driver_params = kwargs.copy()
-    if children is None:
-        children = []
     kwargs["device_id"] = device_id
     kwargs["device_uuid"] = device_uuid
     kwargs["driver_class"] = driver_class
@@ -157,7 +154,6 @@ def init_wrapper(
     kwargs["status_types"] = status_types
     kwargs["action_value_mappings"] = action_value_mappings
     kwargs["hardware_interface"] = hardware_interface
-    kwargs["children"] = children
     kwargs["print_publish"] = print_publish
     kwargs["driver_is_ros"] = driver_is_ros
     super(type(self), self).__init__(*args, **kwargs)
@@ -586,7 +582,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         except Exception as e:
             self.lab_logger().error(f"更新资源uuid失败: {e}")
             self.lab_logger().error(traceback.format_exc())
-        self.lab_logger().debug(f"资源更新结果: {response}")
+        self.lab_logger().trace(f"资源更新结果: {response}")
 
     async def get_resource(self, resources_uuid: List[str], with_children: bool = True) -> ResourceTreeSet:
         """
@@ -1144,7 +1140,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                             queried_resources = []
                             for resource_data in resource_inputs:
                                 plr_resource = await self.get_resource_with_dir(
-                                    resource_ids=resource_data["id"], with_children=True
+                                    resource_id=resource_data["id"], with_children=True
                                 )
                                 queried_resources.append(plr_resource)
 
@@ -1168,7 +1164,6 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                             execution_error = traceback.format_exc()
                             break
 
-            ##### self.lab_logger().info(f"准备执行: {action_kwargs}, 函数: {ACTION.__name__}")
             time_start = time.time()
             time_overall = 100
             future = None
@@ -1176,35 +1171,36 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                 # 将阻塞操作放入线程池执行
                 if asyncio.iscoroutinefunction(ACTION):
                     try:
-                        ##### self.lab_logger().info(f"异步执行动作 {ACTION}")
+                        self.lab_logger().trace(f"异步执行动作 {ACTION}")
-                        future = ROS2DeviceNode.run_async_func(ACTION, trace_error=False, **action_kwargs)
+                        def _handle_future_exception(fut: Future):
-
-                        def _handle_future_exception(fut):
                             nonlocal execution_error, execution_success, action_return_value
                             try:
                                 action_return_value = fut.result()
+                                if isinstance(action_return_value, BaseException):
+                                    raise action_return_value
                                 execution_success = True
-                            except Exception as e:
+                            except Exception as _:
                                 execution_error = traceback.format_exc()
                                 error(
                                     f"异步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{action_kwargs}"
                                 )
 
+                        future = ROS2DeviceNode.run_async_func(ACTION, trace_error=False, **action_kwargs)
                         future.add_done_callback(_handle_future_exception)
                     except Exception as e:
                         execution_error = traceback.format_exc()
                         execution_success = False
                         self.lab_logger().error(f"创建异步任务失败: {traceback.format_exc()}")
                 else:
-                    #####    self.lab_logger().info(f"同步执行动作 {ACTION}")
+                    self.lab_logger().trace(f"同步执行动作 {ACTION}")
                     future = self._executor.submit(ACTION, **action_kwargs)
 
-                    def _handle_future_exception(fut):
+                    def _handle_future_exception(fut: Future):
                         nonlocal execution_error, execution_success, action_return_value
                         try:
                             action_return_value = fut.result()
                             execution_success = True
-                        except Exception as e:
+                        except Exception as _:
                             execution_error = traceback.format_exc()
                             error(
                                 f"同步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{action_kwargs}"
@@ -1309,7 +1305,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                         get_result_info_str(execution_error, execution_success, action_return_value),
                     )
 
-            ##### self.lab_logger().info(f"动作 {action_name} 完成并返回结果")
+            self.lab_logger().trace(f"动作 {action_name} 完成并返回结果")
             return result_msg
 
         return execute_callback
@@ -1544,17 +1540,29 @@ class ROS2DeviceNode:
     这个类封装了设备类实例和ROS2节点的功能，提供ROS2接口。
     它不继承设备类，而是通过代理模式访问设备类的属性和方法。
     """
+    @staticmethod
+    async def safe_task_wrapper(trace_callback, func, **kwargs):
+        try:
+            if callable(trace_callback):
+                trace_callback(await func(**kwargs))
+            return await func(**kwargs)
+        except Exception as e:
+            if callable(trace_callback):
+                trace_callback(e)
+            return e
 
     @classmethod
-    def run_async_func(cls, func, trace_error=True, **kwargs) -> Task:
+    def run_async_func(cls, func, trace_error=True, inner_trace_callback=None, **kwargs) -> Task:
-        def _handle_future_exception(fut):
+        def _handle_future_exception(fut: Future):
             try:
-                fut.result()
+                ret = fut.result()
+                if isinstance(ret, BaseException):
+                    raise ret
             except Exception as e:
-                error(f"异步任务 {func.__name__} 报错了")
+                error(f"异步任务 {func.__name__} 获取结果失败")
                 error(traceback.format_exc())
 
-        future = rclpy.get_global_executor().create_task(func(**kwargs))
+        future = rclpy.get_global_executor().create_task(ROS2DeviceNode.safe_task_wrapper(inner_trace_callback, func, **kwargs))
         if trace_error:
             future.add_done_callback(_handle_future_exception)
         return future
@@ -1582,12 +1590,11 @@ class ROS2DeviceNode:
         device_id: str,
         device_uuid: str,
         driver_class: Type[T],
-        device_config: Dict[str, Any],
+        device_config: ResourceDictInstance,
         driver_params: Dict[str, Any],
         status_types: Dict[str, Any],
         action_value_mappings: Dict[str, Any],
         hardware_interface: Dict[str, Any],
-        children: Dict[str, Any],
         print_publish: bool = True,
         driver_is_ros: bool = False,
     ):
@@ -1598,7 +1605,7 @@ class ROS2DeviceNode:
             device_id: 设备标识符
             device_uuid: 设备uuid
             driver_class: 设备类
-            device_config: 原始初始化的json
+            device_config: 原始初始化的ResourceDictInstance
             driver_params: driver初始化的参数
             status_types: 状态类型映射
             action_value_mappings: 动作值映射
@@ -1612,6 +1619,7 @@ class ROS2DeviceNode:
         self._has_async_context = hasattr(driver_class, "__aenter__") and hasattr(driver_class, "__aexit__")
         self._driver_class = driver_class
         self.device_config = device_config
+        children: List[ResourceDictInstance] = device_config.children
         self.driver_is_ros = driver_is_ros
         self.driver_is_workstation = False
         self.resource_tracker = DeviceNodeResourceTracker()

unilabos/ros/nodes/presets/host_node.py

@@ -289,6 +289,12 @@ class HostNode(BaseROS2DeviceNode):
         self.lab_logger().info("[Host Node] Host node initialized.")
         HostNode._ready_event.set()
 
+        # 发送host_node ready信号到所有桥接器
+        for bridge in self.bridges:
+            if hasattr(bridge, "publish_host_ready"):
+                bridge.publish_host_ready()
+                self.lab_logger().debug(f"Host ready signal sent via {bridge.__class__.__name__}")
+
     def _send_re_register(self, sclient):
         sclient.wait_for_service()
         request = SerialCommand.Request()
@@ -532,7 +538,7 @@ class HostNode(BaseROS2DeviceNode):
         self.lab_logger().info(f"[Host Node] Initializing device: {device_id}")
 
         try:
-            d = initialize_device_from_dict(device_id, device_config.get_nested_dict())
+            d = initialize_device_from_dict(device_id, device_config)
         except DeviceClassInvalid as e:
             self.lab_logger().error(f"[Host Node] Device class invalid: {e}")
             d = None
@@ -712,7 +718,7 @@ class HostNode(BaseROS2DeviceNode):
             feedback_callback=lambda feedback_msg: self.feedback_callback(item, action_id, feedback_msg),
             goal_uuid=goal_uuid_obj,
         )
-        future.add_done_callback(lambda future: self.goal_response_callback(item, action_id, future))
+        future.add_done_callback(lambda f: self.goal_response_callback(item, action_id, f))
 
     def goal_response_callback(self, item: "QueueItem", action_id: str, future) -> None:
         """目标响应回调"""
@@ -723,9 +729,11 @@ class HostNode(BaseROS2DeviceNode):
 
         self.lab_logger().info(f"[Host Node] Goal {action_id} ({item.job_id}) accepted")
         self._goals[item.job_id] = goal_handle
-        goal_handle.get_result_async().add_done_callback(
+        goal_future = goal_handle.get_result_async()
-            lambda future: self.get_result_callback(item, action_id, future)
+        goal_future.add_done_callback(
+            lambda f: self.get_result_callback(item, action_id, f)
         )
+        goal_future.result()
 
     def feedback_callback(self, item: "QueueItem", action_id: str, feedback_msg) -> None:
         """反馈回调"""
@@ -794,6 +802,7 @@ class HostNode(BaseROS2DeviceNode):
             # 存储结果供 HTTP API 查询
             try:
                 from unilabos.app.web.controller import store_job_result
+
                 if goal_status == GoalStatus.STATUS_CANCELED:
                     store_job_result(job_id, status, return_info, {})
                 else:
@@ -1148,11 +1157,12 @@ class HostNode(BaseROS2DeviceNode):
             响应对象，包含查询到的资源
         """
         try:
+            from unilabos.app.web import http_client
             data = json.loads(request.command)
             if "uuid" in data and data["uuid"] is not None:
-                http_req = self.bridges[-1].resource_tree_get([data["uuid"]], data["with_children"])
+                http_req = http_client.resource_tree_get([data["uuid"]], data["with_children"])
             elif "id" in data and data["id"].startswith("/"):
-                http_req = self.bridges[-1].resource_get(data["id"], data["with_children"])
+                http_req = http_client.resource_get(data["id"], data["with_children"])
             else:
                 raise ValueError("没有使用正确的物料 id 或 uuid")
             response.response = json.dumps(http_req["data"])

unilabos/ros/nodes/presets/workstation.py

@@ -24,7 +24,7 @@ from unilabos.ros.msgs.message_converter import (
     convert_from_ros_msg_with_mapping,
 )
 from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, DeviceNodeResourceTracker, ROS2DeviceNode
-from unilabos.ros.nodes.resource_tracker import ResourceTreeSet
+from unilabos.ros.nodes.resource_tracker import ResourceTreeSet, ResourceDictInstance
 from unilabos.utils.type_check import get_result_info_str
 
 if TYPE_CHECKING:
@@ -47,7 +47,7 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
     def __init__(
         self,
         protocol_type: List[str],
-        children: Dict[str, Any],
+        children: List[ResourceDictInstance],
         *,
         driver_instance: "WorkstationBase",
         device_id: str,
@@ -81,10 +81,11 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
         # 初始化子设备
         self.communication_node_id_to_instance = {}
 
-        for device_id, device_config in self.children.items():
+        for device_config in self.children:
-            if device_config.get("type", "device") != "device":
+            device_id = device_config.res_content.id
+            if device_config.res_content.type != "device":
                 self.lab_logger().debug(
-                    f"[Protocol Node] Skipping type {device_config['type']} {device_id} already existed, skipping."
+                    f"[Protocol Node] Skipping type {device_config.res_content.type} {device_id} already existed, skipping."
                 )
                 continue
             try:
@@ -101,8 +102,9 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
                 self.communication_node_id_to_instance[device_id] = d
                 continue
 
-        for device_id, device_config in self.children.items():
+        for device_config in self.children:
-            if device_config.get("type", "device") != "device":
+            device_id = device_config.res_content.id
+            if device_config.res_content.type != "device":
                 continue
             # 设置硬件接口代理
             if device_id not in self.sub_devices:

unilabos/ros/nodes/resource_tracker.py

@@ -1,9 +1,11 @@
+import inspect
 import traceback
 import uuid
 from pydantic import BaseModel, field_serializer, field_validator
 from pydantic import Field
 from typing import List, Tuple, Any, Dict, Literal, Optional, cast, TYPE_CHECKING, Union
 
+from unilabos.resources.plr_additional_res_reg import register
 from unilabos.utils.log import logger
 
 if TYPE_CHECKING:
@@ -62,7 +64,6 @@ class ResourceDict(BaseModel):
     parent: Optional["ResourceDict"] = Field(description="Parent resource object", default=None, exclude=True)
     type: Union[Literal["device"], str] = Field(description="Resource type")
     klass: str = Field(alias="class", description="Resource class name")
-    position: ResourceDictPosition = Field(description="Resource position", default_factory=ResourceDictPosition)
     pose: ResourceDictPosition = Field(description="Resource position", default_factory=ResourceDictPosition)
     config: Dict[str, Any] = Field(description="Resource configuration")
     data: Dict[str, Any] = Field(description="Resource data")
@@ -146,15 +147,16 @@ class ResourceDictInstance(object):
         if not content.get("extra"):  # MagicCode
             content["extra"] = {}
         if "pose" not in content:
-            content["pose"] = content.get("position", {})
+            content["pose"] = content.pop("position", {})
         return ResourceDictInstance(ResourceDict.model_validate(content))
 
-    def get_nested_dict(self) -> Dict[str, Any]:
+    def get_plr_nested_dict(self) -> Dict[str, Any]:
         """获取资源实例的嵌套字典表示"""
         res_dict = self.res_content.model_dump(by_alias=True)
-        res_dict["children"] = {child.res_content.id: child.get_nested_dict() for child in self.children}
+        res_dict["children"] = {child.res_content.id: child.get_plr_nested_dict() for child in self.children}
         res_dict["parent"] = self.res_content.parent_instance_name
-        res_dict["position"] = self.res_content.position.position.model_dump()
+        res_dict["position"] = self.res_content.pose.position.model_dump()
+        del res_dict["pose"]
         return res_dict
 
 
@@ -429,9 +431,9 @@ class ResourceTreeSet(object):
         Returns:
             List[PLRResource]: PLR 资源实例列表
         """
+        register()
         from pylabrobot.resources import Resource as PLRResource
         from pylabrobot.utils.object_parsing import find_subclass
-        import inspect
 
         # 类型映射
         TYPE_MAP = {"plate": "Plate", "well": "Well", "deck": "Deck", "container": "RegularContainer"}
@@ -459,9 +461,9 @@ class ResourceTreeSet(object):
                 "size_y": res.config.get("size_y", 0),
                 "size_z": res.config.get("size_z", 0),
                 "location": {
-                    "x": res.position.position.x,
+                    "x": res.pose.position.x,
-                    "y": res.position.position.y,
+                    "y": res.pose.position.y,
-                    "z": res.position.position.z,
+                    "z": res.pose.position.z,
                     "type": "Coordinate",
                 },
                 "rotation": {"x": 0, "y": 0, "z": 0, "type": "Rotation"},

unilabos/ros/utils/driver_creator.py

@@ -9,10 +9,11 @@ import asyncio
 import inspect
 import traceback
 from abc import abstractmethod
-from typing import Type, Any, Dict, Optional, TypeVar, Generic
+from typing import Type, Any, Dict, Optional, TypeVar, Generic, List
 
 from unilabos.resources.graphio import nested_dict_to_list, resource_ulab_to_plr
-from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker
+from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker, ResourceTreeSet, ResourceDictInstance, \
+    ResourceTreeInstance
 from unilabos.utils import logger, import_manager
 from unilabos.utils.cls_creator import create_instance_from_config
 
@@ -33,7 +34,7 @@ class DeviceClassCreator(Generic[T]):
     这个类提供了从任意类创建实例的通用方法。
     """
 
-    def __init__(self, cls: Type[T], children: Dict[str, Any], resource_tracker: DeviceNodeResourceTracker):
+    def __init__(self, cls: Type[T], children: List[ResourceDictInstance], resource_tracker: DeviceNodeResourceTracker):
         """
         初始化设备类创建器
 
@@ -50,9 +51,9 @@ class DeviceClassCreator(Generic[T]):
         附加资源到设备类实例
         """
         if self.device_instance is not None:
-            for c in self.children.values():
+            for c in self.children:
-                if c["type"] != "device":
+                if c.res_content.type != "device":
-                    self.resource_tracker.add_resource(c)
+                    self.resource_tracker.add_resource(c.get_plr_nested_dict())
 
     def create_instance(self, data: Dict[str, Any]) -> T:
         """
@@ -94,7 +95,7 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
     这个类提供了针对PyLabRobot设备类的实例创建方法，特别处理deserialize方法。
     """
 
-    def __init__(self, cls: Type[T], children: Dict[str, Any], resource_tracker: DeviceNodeResourceTracker):
+    def __init__(self, cls: Type[T], children: List[ResourceDictInstance], resource_tracker: DeviceNodeResourceTracker):
         """
         初始化PyLabRobot设备类创建器
 
@@ -111,12 +112,12 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
     def attach_resource(self):
         pass  # 只能增加实例化物料，原来默认物料仅为字典查询
 
-    def _process_resource_mapping(self, resource, source_type):
+    # def _process_resource_mapping(self, resource, source_type):
-        if source_type == dict:
+    #     if source_type == dict:
-            from pylabrobot.resources.resource import Resource
+    #         from pylabrobot.resources.resource import Resource
-
+    #
-            return nested_dict_to_list(resource), Resource
+    #         return nested_dict_to_list(resource), Resource
-        return resource, source_type
+    #     return resource, source_type
 
     def _process_resource_references(
         self, data: Any, to_dict=False, states=None, prefix_path="", name_to_uuid=None
@@ -142,15 +143,21 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
         if isinstance(data, dict):
             if "_resource_child_name" in data:
                 child_name = data["_resource_child_name"]
-                if child_name in self.children:
+                resource: Optional[ResourceDictInstance] = None
-                    resource = self.children[child_name]
+                for child in self.children:
+                    if child.res_content.name == child_name:
+                        resource = child
+                if resource is not None:
                     if "_resource_type" in data:
                         type_path = data["_resource_type"]
                         try:
-                            target_type = import_manager.get_class(type_path)
+                            # target_type = import_manager.get_class(type_path)
-                            contain_model = not issubclass(target_type, Deck)
+                            # contain_model = not issubclass(target_type, Deck)
-                            resource, target_type = self._process_resource_mapping(resource, target_type)
+                            # resource, target_type = self._process_resource_mapping(resource, target_type)
-                            resource_instance: Resource = resource_ulab_to_plr(resource, contain_model)  # 带state
+                            res_tree = ResourceTreeInstance(resource)
+                            res_tree_set = ResourceTreeSet([res_tree])
+                            resource_instance: Resource = res_tree_set.to_plr_resources()[0]
+                            # resource_instance: Resource = resource_ulab_to_plr(resource, contain_model)  # 带state
                             states[prefix_path] = resource_instance.serialize_all_state()
                             # 使用 prefix_path 作为 key 存储资源状态
                             if to_dict:
@@ -202,12 +209,12 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
         stack = None
 
         # 递归遍历 children 构建 name_to_uuid 映射
-        def collect_name_to_uuid(children_dict: Dict[str, Any], result: Dict[str, str]):
+        def collect_name_to_uuid(children_list: List[ResourceDictInstance], result: Dict[str, str]):
             """递归遍历嵌套的 children 字典，收集 name 到 uuid 的映射"""
-            for child in children_dict.values():
+            for child in children_list:
-                if isinstance(child, dict):
+                if isinstance(child, ResourceDictInstance):
-                    result[child["name"]] = child["uuid"]
+                    result[child.res_content.name] = child.res_content.uuid
-                    collect_name_to_uuid(child["children"], result)
+                    collect_name_to_uuid(child.children, result)
 
         name_to_uuid = {}
         collect_name_to_uuid(self.children, name_to_uuid)
@@ -313,7 +320,7 @@ class WorkstationNodeCreator(DeviceClassCreator[T]):
     这个类提供了针对WorkstationNode设备类的实例创建方法，处理children参数。
     """
 
-    def __init__(self, cls: Type[T], children: Dict[str, Any], resource_tracker: DeviceNodeResourceTracker):
+    def __init__(self, cls: Type[T], children: List[ResourceDictInstance], resource_tracker: DeviceNodeResourceTracker):
         """
         初始化WorkstationNode设备类创建器
 
@@ -336,9 +343,9 @@ class WorkstationNodeCreator(DeviceClassCreator[T]):
         try:
             # 创建实例，额外补充一个给protocol node的字段，后面考虑取消
             data["children"] = self.children
-            for material_id, child in self.children.items():
+            for child in self.children:
-                if child["type"] != "device":
+                if child.res_content.type != "device":
-                    self.resource_tracker.add_resource(self.children[material_id])
+                    self.resource_tracker.add_resource(child.get_plr_nested_dict())
             deck_dict = data.get("deck")
             if deck_dict:
                 from pylabrobot.resources import Deck, Resource

unilabos/test/workflow/merge_workflow.py

@@ -9,7 +9,7 @@ if str(ROOT_DIR) not in sys.path:
 
 import pytest
 
-from scripts.workflow import build_protocol_graph, draw_protocol_graph, draw_protocol_graph_with_ports
+from unilabos.workflow.common import build_protocol_graph, draw_protocol_graph, draw_protocol_graph_with_ports
 
 
 ROOT_DIR = Path(__file__).resolve().parents[2]

unilabos/utils/log.py

@@ -124,11 +124,14 @@ class ColoredFormatter(logging.Formatter):
     def _format_basic(self, record):
         """基本格式化，不包含颜色"""
         datetime_str = datetime.fromtimestamp(record.created).strftime("%y-%m-%d [%H:%M:%S,%f")[:-3] + "]"
-        filename = os.path.basename(record.filename).rsplit(".", 1)[0]  # 提取文件名（不含路径和扩展名）
+        filename = record.filename.replace(".py", "").split("\\")[-1]  # 提取文件名（不含路径和扩展名）
+        if "/" in filename:
+            filename = filename.split("/")[-1]
         module_path = f"{record.name}.{filename}"
         func_line = f"{record.funcName}:{record.lineno}"
+        right_info = f" [{func_line}] [{module_path}]"
 
-        formatted_message = f"{datetime_str} [{record.levelname}] [{module_path}] [{func_line}]: {record.getMessage()}"
+        formatted_message = f"{datetime_str} [{record.levelname}] {record.getMessage()}{right_info}"
 
         if record.exc_info:
             exc_text = self.formatException(record.exc_info)
@@ -150,7 +153,7 @@ class ColoredFormatter(logging.Formatter):
 
 
 # 配置日志处理器
-def configure_logger(loglevel=None):
+def configure_logger(loglevel=None, working_dir=None):
     """配置日志记录器
 
     Args:
@@ -159,8 +162,9 @@ def configure_logger(loglevel=None):
     """
     # 获取根日志记录器
     root_logger = logging.getLogger()
-
+    root_logger.setLevel(TRACE_LEVEL)
     # 设置日志级别
+    numeric_level = logging.DEBUG
     if loglevel is not None:
         if isinstance(loglevel, str):
             # 将字符串转换为logging级别
@@ -170,12 +174,8 @@ def configure_logger(loglevel=None):
                 numeric_level = getattr(logging, loglevel.upper(), None)
                 if not isinstance(numeric_level, int):
                     print(f"警告: 无效的日志级别 '{loglevel}'，使用默认级别 DEBUG")
-                    numeric_level = logging.DEBUG
         else:
             numeric_level = loglevel
-        root_logger.setLevel(numeric_level)
-    else:
-        root_logger.setLevel(logging.DEBUG)  # 默认级别
 
     # 移除已存在的处理器
     for handler in root_logger.handlers[:]:
@@ -183,7 +183,7 @@ def configure_logger(loglevel=None):
 
     # 创建控制台处理器
     console_handler = logging.StreamHandler()
-    console_handler.setLevel(root_logger.level)  # 使用与根记录器相同的级别
+    console_handler.setLevel(numeric_level)  # 使用与根记录器相同的级别
 
     # 使用自定义的颜色格式化器
     color_formatter = ColoredFormatter()
@@ -191,9 +191,30 @@ def configure_logger(loglevel=None):
 
     # 添加处理器到根日志记录器
     root_logger.addHandler(console_handler)
+
+    # 如果指定了工作目录，添加文件处理器
+    if working_dir is not None:
+        logs_dir = os.path.join(working_dir, "logs")
+        os.makedirs(logs_dir, exist_ok=True)
+
+        # 生成日志文件名：日期 时间.log
+        log_filename = datetime.now().strftime("%Y-%m-%d %H-%M-%S") + ".log"
+        log_filepath = os.path.join(logs_dir, log_filename)
+
+        # 创建文件处理器
+        file_handler = logging.FileHandler(log_filepath, encoding="utf-8")
+        file_handler.setLevel(TRACE_LEVEL)
+
+        # 使用不带颜色的格式化器
+        file_formatter = ColoredFormatter(use_colors=False)
+        file_handler.setFormatter(file_formatter)
+
+        root_logger.addHandler(file_handler)
+
     logging.getLogger("asyncio").setLevel(logging.INFO)
     logging.getLogger("urllib3").setLevel(logging.INFO)
 
+
 # 配置日志系统
 configure_logger()
 

unilabos/workflow/common.py

@@ -0,0 +1,484 @@
+import re
+import uuid
+
+import networkx as nx
+from networkx.drawing.nx_agraph import to_agraph
+import matplotlib.pyplot as plt
+from typing import Dict, List, Any, Tuple, Optional
+
+Json = Dict[str, Any]
+
+# ---------------- Graph ----------------
+
+class WorkflowGraph:
+    """简单的有向图实现：使用 params 单层参数；inputs 内含连线；支持 node-link 导出"""
+
+    def __init__(self):
+        self.nodes: Dict[str, Dict[str, Any]] = {}
+        self.edges: List[Dict[str, Any]] = []
+
+    def add_node(self, node_id: str, **attrs):
+        self.nodes[node_id] = attrs
+
+    def add_edge(self, source: str, target: str, **attrs):
+        edge = {
+            "source": source,
+            "target": target,
+            "source_node_uuid": source,
+            "target_node_uuid": target,
+            "source_handle_io": attrs.pop("source_handle_io", "source"),
+            "target_handle_io": attrs.pop("target_handle_io", "target"),
+            **attrs
+        }
+        self.edges.append(edge)
+
+    def _materialize_wiring_into_inputs(self, obj: Any, inputs: Dict[str, Any],
+                                        variable_sources: Dict[str, Dict[str, Any]],
+                                        target_node_id: str, base_path: List[str]):
+        has_var = False
+
+        def walk(node: Any, path: List[str]):
+            nonlocal has_var
+            if isinstance(node, dict):
+                if "__var__" in node:
+                    has_var = True
+                    varname = node["__var__"]
+                    placeholder = f"${{{varname}}}"
+                    src = variable_sources.get(varname)
+                    if src:
+                        key = ".".join(path)  # e.g. "params.foo.bar.0"
+                        inputs[key] = {"node": src["node_id"], "output": src.get("output_name", "result")}
+                        self.add_edge(str(src["node_id"]), target_node_id,
+                                      source_handle_io=src.get("output_name", "result"),
+                                      target_handle_io=key)
+                    return placeholder
+                return {k: walk(v, path + [k]) for k, v in node.items()}
+            if isinstance(node, list):
+                return [walk(v, path + [str(i)]) for i, v in enumerate(node)]
+            return node
+
+        replaced = walk(obj, base_path[:])
+        return replaced, has_var
+
+    def add_workflow_node(self,
+                          node_id: int,
+                          *,
+                          device_key: Optional[str] = None,          # 实例名，如 "ser"
+                          resource_name: Optional[str] = None,       # registry key（原 device_class）
+                          module: Optional[str] = None,
+                          template_name: Optional[str] = None,       # 动作/模板名（原 action_key）
+                          params: Dict[str, Any],
+                          variable_sources: Dict[str, Dict[str, Any]],
+                          add_ready_if_no_vars: bool = True,
+                          prev_node_id: Optional[int] = None,
+                          **extra_attrs) -> None:
+        """添加工作流节点：params 单层；自动变量连线与 ready 串联；支持附加属性"""
+        node_id_str = str(node_id)
+        inputs: Dict[str, Any] = {}
+
+        params, has_var = self._materialize_wiring_into_inputs(
+            params, inputs, variable_sources, node_id_str, base_path=["params"]
+        )
+
+        if add_ready_if_no_vars and not has_var:
+            last_id = str(prev_node_id) if prev_node_id is not None else "-1"
+            inputs["ready"] = {"node": int(last_id), "output": "ready"}
+            self.add_edge(last_id, node_id_str, source_handle_io="ready", target_handle_io="ready")
+
+        node_obj = {
+            "device_key": device_key,
+            "resource_name": resource_name,   # ✅ 新名字
+            "module": module,
+            "template_name": template_name,   # ✅ 新名字
+            "params": params,
+            "inputs": inputs,
+        }
+        node_obj.update(extra_attrs or {})
+        self.add_node(node_id_str, parameters=node_obj)
+
+    # 顺序工作流导出（连线在 inputs，不返回 edges）
+    def to_dict(self) -> List[Dict[str, Any]]:
+        result = []
+        for node_id, attrs in self.nodes.items():
+            node = {"id": node_id}
+            params = dict(attrs.get("parameters", {}) or {})
+            flat = {k: v for k, v in attrs.items() if k != "parameters"}
+            flat.update(params)
+            node.update(flat)
+            result.append(node)
+        return sorted(result, key=lambda n: int(n["id"]) if str(n["id"]).isdigit() else n["id"])
+
+    # node-link 导出（含 edges）
+    def to_node_link_dict(self) -> Dict[str, Any]:
+        nodes_list = []
+        for node_id, attrs in self.nodes.items():
+            node_attrs = attrs.copy()
+            params = node_attrs.pop("parameters", {}) or {}
+            node_attrs.update(params)
+            nodes_list.append({"id": node_id, **node_attrs})
+        return {"directed": True, "multigraph": False, "graph": {}, "nodes": nodes_list, "edges": self.edges, "links": self.edges}
+
+
+def refactor_data(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """统一的数据重构函数，根据操作类型自动选择模板"""
+    refactored_data = []
+
+    # 定义操作映射，包含生物实验和有机化学的所有操作
+    OPERATION_MAPPING = {
+        # 生物实验操作
+        "transfer_liquid": "transfer_liquid",
+        "transfer": "transfer",
+        "incubation": "incubation",
+        "move_labware": "move_labware",
+        "oscillation": "oscillation",
+        # 有机化学操作
+        "HeatChillToTemp": "HeatChillProtocol",
+        "StopHeatChill": "HeatChillStopProtocol",
+        "StartHeatChill": "HeatChillStartProtocol",
+        "HeatChill": "HeatChillProtocol",
+        "Dissolve": "DissolveProtocol",
+        "Transfer": "TransferProtocol",
+        "Evaporate": "EvaporateProtocol",
+        "Recrystallize": "RecrystallizeProtocol",
+        "Filter": "FilterProtocol",
+        "Dry": "DryProtocol",
+        "Add": "AddProtocol",
+    }
+
+    UNSUPPORTED_OPERATIONS = ["Purge", "Wait", "Stir", "ResetHandling"]
+
+    for step in data:
+        operation = step.get("action")
+        if not operation or operation in UNSUPPORTED_OPERATIONS:
+            continue
+
+        # 处理重复操作
+        if operation == "Repeat":
+            times = step.get("times", step.get("parameters", {}).get("times", 1))
+            sub_steps = step.get("steps", step.get("parameters", {}).get("steps", []))
+            for i in range(int(times)):
+                sub_data = refactor_data(sub_steps)
+                refactored_data.extend(sub_data)
+            continue
+
+        # 获取模板名称
+        template = OPERATION_MAPPING.get(operation)
+        if not template:
+            # 自动推断模板类型
+            if operation.lower() in ["transfer", "incubation", "move_labware", "oscillation"]:
+                template = f"biomek-{operation}"
+            else:
+                template = f"{operation}Protocol"
+
+        # 创建步骤数据
+        step_data = {
+            "template": template,
+            "description": step.get("description", step.get("purpose", f"{operation} operation")),
+            "lab_node_type": "Device",
+            "parameters": step.get("parameters", step.get("action_args", {})),
+        }
+        refactored_data.append(step_data)
+
+    return refactored_data
+
+
+def build_protocol_graph(
+    labware_info: List[Dict[str, Any]], protocol_steps: List[Dict[str, Any]], workstation_name: str
+) -> WorkflowGraph:
+    """统一的协议图构建函数，根据设备类型自动选择构建逻辑"""
+    G = WorkflowGraph()
+    resource_last_writer = {}
+
+    protocol_steps = refactor_data(protocol_steps)
+    # 有机化学&移液站协议图构建
+    WORKSTATION_ID = workstation_name
+
+    # 为所有labware创建资源节点
+    for labware_id, item in labware_info.items():
+        # item_id = item.get("id") or item.get("name", f"item_{uuid.uuid4()}")
+        node_id = str(uuid.uuid4())
+
+        # 判断节点类型
+        if "Rack" in str(labware_id) or "Tip" in str(labware_id):
+            lab_node_type = "Labware"
+            description = f"Prepare Labware: {labware_id}"
+            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_type = []
+            liquid_volume = []
+        else:
+            lab_node_type = "Reagent"
+            description = f"Add Reagent to Flask: {labware_id}"
+            liquid_type = [labware_id]
+            liquid_volume = [1e5]
+
+        G.add_node(
+            node_id,
+            template_name=f"create_resource",
+            resource_name="host_node",
+            description=description,
+            lab_node_type=lab_node_type,
+            params={
+                "res_id": labware_id,
+                "device_id": WORKSTATION_ID,
+                "class_name": "container",
+                "parent": WORKSTATION_ID,
+                "bind_locations": {"x": 0.0, "y": 0.0, "z": 0.0},
+                "liquid_input_slot": [-1],
+                "liquid_type": liquid_type,
+                "liquid_volume": liquid_volume,
+                "slot_on_deck": "",
+            },
+            role=item.get("role", ""),
+        )
+        resource_last_writer[labware_id] = f"{node_id}:labware"
+
+    last_control_node_id = None
+
+    # 处理协议步骤
+    for step in protocol_steps:
+        node_id = str(uuid.uuid4())
+        G.add_node(node_id, **step)
+
+        # 控制流
+        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("parameters", {})
+        input_resources_possible_names = [
+            "vessel",
+            "to_vessel",
+            "from_vessel",
+            "reagent",
+            "solvent",
+            "compound",
+            "sources",
+            "targets",
+        ]
+
+        for target_port in input_resources_possible_names:
+            resource_name = params.get(target_port)
+            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)
+
+        output_resources = {
+            "vessel_out": params.get("vessel"),
+            "from_vessel_out": params.get("from_vessel"),
+            "to_vessel_out": params.get("to_vessel"),
+            "filtrate_out": params.get("filtrate_vessel"),
+            "reagent": params.get("reagent"),
+            "solvent": params.get("solvent"),
+            "compound": params.get("compound"),
+            "sources_out": params.get("sources"),
+            "targets_out": params.get("targets"),
+        }
+
+        for source_port, resource_name in output_resources.items():
+            if resource_name:
+                resource_last_writer[resource_name] = f"{node_id}:{source_port}"
+
+    return G
+
+
+def draw_protocol_graph(protocol_graph: WorkflowGraph, output_path: str):
+    """
+    (辅助功能) 使用 networkx 和 matplotlib 绘制协议工作流图，用于可视化。
+    """
+    if not protocol_graph:
+        print("Cannot draw graph: Graph object is empty.")
+        return
+
+    G = nx.DiGraph()
+
+    for node_id, attrs in protocol_graph.nodes.items():
+        label = attrs.get("description", attrs.get("template", node_id[:8]))
+        G.add_node(node_id, label=label, **attrs)
+
+    for edge in protocol_graph.edges:
+        G.add_edge(edge["source"], edge["target"])
+
+    plt.figure(figsize=(20, 15))
+    try:
+        pos = nx.nx_agraph.graphviz_layout(G, prog="dot")
+    except Exception:
+        pos = nx.shell_layout(G)  # Fallback layout
+
+    node_labels = {node: data["label"] for node, data in G.nodes(data=True)}
+    nx.draw(
+        G,
+        pos,
+        with_labels=False,
+        node_size=2500,
+        node_color="skyblue",
+        node_shape="o",
+        edge_color="gray",
+        width=1.5,
+        arrowsize=15,
+    )
+    nx.draw_networkx_labels(G, pos, labels=node_labels, font_size=8, font_weight="bold")
+
+    plt.title("Chemical Protocol Workflow Graph", size=15)
+    plt.savefig(output_path, dpi=300, bbox_inches="tight")
+    plt.close()
+    print(f"  - Visualization saved to '{output_path}'")
+
+
+COMPASS = {"n","e","s","w","ne","nw","se","sw","c"}
+
+def _is_compass(port: str) -> bool:
+    return isinstance(port, str) and port.lower() in COMPASS
+
+def draw_protocol_graph_with_ports(protocol_graph, output_path: str, rankdir: str = "LR"):
+    """
+    使用 Graphviz 端口语法绘制协议工作流图。
+    - 若边上的 source_port/target_port 是 compass（n/e/s/w/...），直接用 compass。
+    - 否则自动为节点创建 record 形状并定义命名端口 <portname>。
+    最终由 PyGraphviz 渲染并输出到 output_path（后缀决定格式，如 .png/.svg/.pdf）。
+    """
+    if not protocol_graph:
+        print("Cannot draw graph: Graph object is empty.")
+        return
+
+    # 1) 先用 networkx 搭建有向图，保留端口属性
+    G = nx.DiGraph()
+    for node_id, attrs in protocol_graph.nodes.items():
+        label = attrs.get("description", attrs.get("template", node_id[:8]))
+        # 保留一个干净的“中心标签”，用于放在 record 的中间槽
+        G.add_node(node_id, _core_label=str(label), **{k:v for k,v in attrs.items() if k not in ("label",)})
+
+    edges_data = []
+    in_ports_by_node  = {}  # 收集命名输入端口
+    out_ports_by_node = {}  # 收集命名输出端口
+
+    for edge in protocol_graph.edges:
+        u = edge["source"]
+        v = edge["target"]
+        sp = edge.get("source_port")
+        tp = edge.get("target_port")
+
+        # 记录到图里（保留原始端口信息）
+        G.add_edge(u, v, source_port=sp, target_port=tp)
+        edges_data.append((u, v, sp, tp))
+
+        # 如果不是 compass，就按“命名端口”先归类，等会儿给节点造 record
+        if sp and not _is_compass(sp):
+            out_ports_by_node.setdefault(u, set()).add(str(sp))
+        if tp and not _is_compass(tp):
+            in_ports_by_node.setdefault(v, set()).add(str(tp))
+
+    # 2) 转为 AGraph，使用 Graphviz 渲染
+    A = to_agraph(G)
+    A.graph_attr.update(rankdir=rankdir, splines="true", concentrate="false", fontsize="10")
+    A.node_attr.update(shape="box", style="rounded,filled", fillcolor="lightyellow", color="#999999", fontname="Helvetica")
+    A.edge_attr.update(arrowsize="0.8", color="#666666")
+
+    # 3) 为需要命名端口的节点设置 record 形状与 label
+    #    左列 = 输入端口；中间 = 核心标签；右列 = 输出端口
+    for n in A.nodes():
+        node = A.get_node(n)
+        core = G.nodes[n].get("_core_label", n)
+
+        in_ports  = sorted(in_ports_by_node.get(n, []))
+        out_ports = sorted(out_ports_by_node.get(n, []))
+
+        # 如果该节点涉及命名端口，则用 record；否则保留原 box
+        if in_ports or out_ports:
+            def port_fields(ports):
+                if not ports:
+                    return " "  # 必须留一个空槽占位
+                # 每个端口一个小格子，<p> name
+                return "|".join(f"<{re.sub(r'[^A-Za-z0-9_:.|-]', '_', p)}> {p}" for p in ports)
+
+            left  = port_fields(in_ports)
+            right = port_fields(out_ports)
+
+            # 三栏：左(入) | 中(节点名) | 右(出)
+            record_label = f"{{ {left} | {core} | {right} }}"
+            node.attr.update(shape="record", label=record_label)
+        else:
+            # 没有命名端口：普通盒子，显示核心标签
+            node.attr.update(label=str(core))
+
+    # 4) 给边设置 headport / tailport
+    #    - 若端口为 compass：直接用 compass（e.g., headport="e"）
+    #    - 若端口为命名端口：使用在 record 中定义的 <port> 名（同名即可）
+    for (u, v, sp, tp) in edges_data:
+        e = A.get_edge(u, v)
+
+        # Graphviz 属性：tail 是源，head 是目标
+        if sp:
+            if _is_compass(sp):
+                e.attr["tailport"] = sp.lower()
+            else:
+                # 与 record label 中 <port> 名一致；特殊字符已在 label 中做了清洗
+                e.attr["tailport"] = re.sub(r'[^A-Za-z0-9_:.|-]', '_', str(sp))
+
+        if tp:
+            if _is_compass(tp):
+                e.attr["headport"] = tp.lower()
+            else:
+                e.attr["headport"] = re.sub(r'[^A-Za-z0-9_:.|-]', '_', str(tp))
+
+        # 可选：若想让边更贴边缘，可设置 constraint/spline 等
+        # e.attr["arrowhead"] = "vee"
+
+    # 5) 输出
+    A.draw(output_path, prog="dot")
+    print(f"  - Port-aware workflow rendered to '{output_path}'")
+# ---------------- Registry Adapter ----------------
+
+
+class RegistryAdapter:
+    """根据 module 的类名（冒号右侧）反查 registry 的 resource_name（原 device_class），并抽取参数顺序"""
+    def __init__(self, device_registry: Dict[str, Any]):
+        self.device_registry = device_registry or {}
+        self.module_class_to_resource = self._build_module_class_index()
+
+    def _build_module_class_index(self) -> Dict[str, str]:
+        idx = {}
+        for resource_name, info in self.device_registry.items():
+            module = info.get("module")
+            if isinstance(module, str) and ":" in module:
+                cls = module.split(":")[-1]
+                idx[cls] = resource_name
+                idx[cls.lower()] = resource_name
+        return idx
+
+    def resolve_resource_by_classname(self, class_name: str) -> Optional[str]:
+        if not class_name:
+            return None
+        return (self.module_class_to_resource.get(class_name)
+                or self.module_class_to_resource.get(class_name.lower()))
+
+    def get_device_module(self, resource_name: Optional[str]) -> Optional[str]:
+        if not resource_name:
+            return None
+        return self.device_registry.get(resource_name, {}).get("module")
+
+    def get_actions(self, resource_name: Optional[str]) -> Dict[str, Any]:
+        if not resource_name:
+            return {}
+        return (self.device_registry.get(resource_name, {})
+                .get("class", {})
+                .get("action_value_mappings", {})) or {}
+
+    def get_action_schema(self, resource_name: Optional[str], template_name: str) -> Optional[Json]:
+        return (self.get_actions(resource_name).get(template_name) or {}).get("schema")
+
+    def get_action_goal_default(self, resource_name: Optional[str], template_name: str) -> Json:
+        return (self.get_actions(resource_name).get(template_name) or {}).get("goal_default", {}) or {}
+
+    def get_action_input_keys(self, resource_name: Optional[str], template_name: str) -> List[str]:
+        schema = self.get_action_schema(resource_name, template_name) or {}
+        goal = (schema.get("properties") or {}).get("goal") or {}
+        props = goal.get("properties") or {}
+        required = goal.get("required") or []
+        return list(dict.fromkeys(required + list(props.keys())))

unilabos/workflow/from_labwares_and_steps.py

@@ -0,0 +1,24 @@
+import json
+from os import PathLike
+
+from unilabos.workflow.common import build_protocol_graph
+
+
+def from_labwares_and_steps(data_path: PathLike):
+    with data_path.open("r", encoding="utf-8") as fp:
+        d = json.load(fp)
+
+    if "workflow" in d and "reagent" in d:
+        protocol_steps = d["workflow"]
+        labware_info = d["reagent"]
+    elif "steps_info" in d and "labware_info" in d:
+        protocol_steps = _normalize_steps(d["steps_info"])
+        labware_info = _normalize_labware(d["labware_info"])
+    else:
+        raise ValueError("Unsupported protocol format")
+
+    graph = build_protocol_graph(
+        labware_info=labware_info,
+        protocol_steps=protocol_steps,
+        workstation_name="PRCXi",
+    )

unilabos/workflow/from_python_script.py

@@ -0,0 +1,241 @@
+import ast
+import json
+from typing import Dict, List, Any, Tuple, Optional
+
+from .common import WorkflowGraph, RegistryAdapter
+
+Json = Dict[str, Any]
+
+# ---------------- Converter ----------------
+
+class DeviceMethodConverter:
+    """
+    - 字段统一：resource_name（原 device_class）、template_name（原 action_key）
+    - params 单层；inputs 使用 'params.' 前缀
+    - SimpleGraph.add_workflow_node 负责变量连线与边
+    """
+    def __init__(self, device_registry: Optional[Dict[str, Any]] = None):
+        self.graph = WorkflowGraph()
+        self.variable_sources: Dict[str, Dict[str, Any]] = {}        # var -> {node_id, output_name}
+        self.instance_to_resource: Dict[str, Optional[str]] = {}     # 实例名 -> resource_name
+        self.node_id_counter: int = 0
+        self.registry = RegistryAdapter(device_registry or {})
+
+    # ---- helpers ----
+    def _new_node_id(self) -> int:
+        nid = self.node_id_counter
+        self.node_id_counter += 1
+        return nid
+
+    def _assign_targets(self, targets) -> List[str]:
+        names: List[str] = []
+        import ast
+        if isinstance(targets, ast.Tuple):
+            for elt in targets.elts:
+                if isinstance(elt, ast.Name):
+                    names.append(elt.id)
+        elif isinstance(targets, ast.Name):
+            names.append(targets.id)
+        return names
+
+    def _extract_device_instantiation(self, node) -> Optional[Tuple[str, str]]:
+        import ast
+        if not isinstance(node.value, ast.Call):
+            return None
+        callee = node.value.func
+        if isinstance(callee, ast.Name):
+            class_name = callee.id
+        elif isinstance(callee, ast.Attribute) and isinstance(callee.value, ast.Name):
+            class_name = callee.attr
+        else:
+            return None
+        if isinstance(node.targets[0], ast.Name):
+            instance = node.targets[0].id
+            return instance, class_name
+        return None
+
+    def _extract_call(self, call) -> Tuple[str, str, Dict[str, Any], str]:
+        import ast
+        owner_name, method_name, call_kind = "", "", "func"
+        if isinstance(call.func, ast.Attribute):
+            method_name = call.func.attr
+            if isinstance(call.func.value, ast.Name):
+                owner_name = call.func.value.id
+                call_kind = "instance" if owner_name in self.instance_to_resource else "class_or_module"
+            elif isinstance(call.func.value, ast.Attribute) and isinstance(call.func.value.value, ast.Name):
+                owner_name = call.func.value.attr
+                call_kind = "class_or_module"
+        elif isinstance(call.func, ast.Name):
+            method_name = call.func.id
+            call_kind = "func"
+
+        def pack(node):
+            if isinstance(node, ast.Name):
+                return {"type": "variable", "value": node.id}
+            if isinstance(node, ast.Constant):
+                return {"type": "constant", "value": node.value}
+            if isinstance(node, ast.Dict):
+                return {"type": "dict", "value": self._parse_dict(node)}
+            if isinstance(node, ast.List):
+                return {"type": "list", "value": self._parse_list(node)}
+            return {"type": "raw", "value": ast.unparse(node) if hasattr(ast, "unparse") else str(node)}
+
+        args: Dict[str, Any] = {}
+        pos: List[Any] = []
+        for a in call.args:
+            pos.append(pack(a))
+        for kw in call.keywords:
+            args[kw.arg] = pack(kw.value)
+        if pos:
+            args["_positional"] = pos
+        return owner_name, method_name, args, call_kind
+
+    def _parse_dict(self, node) -> Dict[str, Any]:
+        import ast
+        out: Dict[str, Any] = {}
+        for k, v in zip(node.keys, node.values):
+            if isinstance(k, ast.Constant):
+                key = str(k.value)
+                if isinstance(v, ast.Name):
+                    out[key] = f"var:{v.id}"
+                elif isinstance(v, ast.Constant):
+                    out[key] = v.value
+                elif isinstance(v, ast.Dict):
+                    out[key] = self._parse_dict(v)
+                elif isinstance(v, ast.List):
+                    out[key] = self._parse_list(v)
+        return out
+
+    def _parse_list(self, node) -> List[Any]:
+        import ast
+        out: List[Any] = []
+        for elt in node.elts:
+            if isinstance(elt, ast.Name):
+                out.append(f"var:{elt.id}")
+            elif isinstance(elt, ast.Constant):
+                out.append(elt.value)
+            elif isinstance(elt, ast.Dict):
+                out.append(self._parse_dict(elt))
+            elif isinstance(elt, ast.List):
+                out.append(self._parse_list(elt))
+        return out
+
+    def _normalize_var_tokens(self, x: Any) -> Any:
+        if isinstance(x, str) and x.startswith("var:"):
+            return {"__var__": x[4:]}
+        if isinstance(x, list):
+            return [self._normalize_var_tokens(i) for i in x]
+        if isinstance(x, dict):
+            return {k: self._normalize_var_tokens(v) for k, v in x.items()}
+        return x
+
+    def _make_params_payload(self, resource_name: Optional[str], template_name: str, call_args: Dict[str, Any]) -> Dict[str, Any]:
+        input_keys = self.registry.get_action_input_keys(resource_name, template_name) if resource_name else []
+        defaults = self.registry.get_action_goal_default(resource_name, template_name) if resource_name else {}
+        params: Dict[str, Any] = dict(defaults)
+
+        def unpack(p):
+            t, v = p.get("type"), p.get("value")
+            if t == "variable":
+                return {"__var__": v}
+            if t == "dict":
+                return self._normalize_var_tokens(v)
+            if t == "list":
+                return self._normalize_var_tokens(v)
+            return v
+
+        for k, p in call_args.items():
+            if k == "_positional":
+                continue
+            params[k] = unpack(p)
+
+        pos = call_args.get("_positional", [])
+        if pos:
+            if input_keys:
+                for i, p in enumerate(pos):
+                    if i >= len(input_keys):
+                        break
+                    name = input_keys[i]
+                    if name in params:
+                        continue
+                    params[name] = unpack(p)
+            else:
+                for i, p in enumerate(pos):
+                    params[f"arg_{i}"] = unpack(p)
+        return params
+
+    # ---- handlers ----
+    def _on_assign(self, stmt):
+        import ast
+        inst = self._extract_device_instantiation(stmt)
+        if inst:
+            instance, code_class = inst
+            resource_name = self.registry.resolve_resource_by_classname(code_class)
+            self.instance_to_resource[instance] = resource_name
+            return
+
+        if isinstance(stmt.value, ast.Call):
+            owner, method, call_args, kind = self._extract_call(stmt.value)
+            if kind == "instance":
+                device_key = owner
+                resource_name = self.instance_to_resource.get(owner)
+            else:
+                device_key = owner
+                resource_name = self.registry.resolve_resource_by_classname(owner)
+
+            module = self.registry.get_device_module(resource_name)
+            params = self._make_params_payload(resource_name, method, call_args)
+
+            nid = self._new_node_id()
+            self.graph.add_workflow_node(
+                nid,
+                device_key=device_key,
+                resource_name=resource_name,      # ✅
+                module=module,
+                template_name=method,             # ✅
+                params=params,
+                variable_sources=self.variable_sources,
+                add_ready_if_no_vars=True,
+                prev_node_id=(nid - 1) if nid > 0 else None,
+            )
+
+            out_vars = self._assign_targets(stmt.targets[0])
+            for var in out_vars:
+                self.variable_sources[var] = {"node_id": nid, "output_name": "result"}
+
+    def _on_expr(self, stmt):
+        import ast
+        if not isinstance(stmt.value, ast.Call):
+            return
+        owner, method, call_args, kind = self._extract_call(stmt.value)
+        if kind == "instance":
+            device_key = owner
+            resource_name = self.instance_to_resource.get(owner)
+        else:
+            device_key = owner
+            resource_name = self.registry.resolve_resource_by_classname(owner)
+
+        module = self.registry.get_device_module(resource_name)
+        params = self._make_params_payload(resource_name, method, call_args)
+
+        nid = self._new_node_id()
+        self.graph.add_workflow_node(
+            nid,
+            device_key=device_key,
+            resource_name=resource_name,      # ✅
+            module=module,
+            template_name=method,             # ✅
+            params=params,
+            variable_sources=self.variable_sources,
+            add_ready_if_no_vars=True,
+            prev_node_id=(nid - 1) if nid > 0 else None,
+        )
+
+    def convert(self, python_code: str):
+        tree = ast.parse(python_code)
+        for stmt in tree.body:
+            if isinstance(stmt, ast.Assign):
+                self._on_assign(stmt)
+            elif isinstance(stmt, ast.Expr):
+                self._on_expr(stmt)
+        return self

unilabos/workflow/from_xdl.py

@@ -0,0 +1,131 @@
+from typing import List, Any, Dict
+import xml.etree.ElementTree as ET
+
+
+def convert_to_type(val: str) -> Any:
+    """将字符串值转换为适当的数据类型"""
+    if val == "True":
+        return True
+    if val == "False":
+        return False
+    if val == "?":
+        return None
+    if val.endswith(" g"):
+        return float(val.split(" ")[0])
+    if val.endswith("mg"):
+        return float(val.split("mg")[0])
+    elif val.endswith("mmol"):
+        return float(val.split("mmol")[0]) / 1000
+    elif val.endswith("mol"):
+        return float(val.split("mol")[0])
+    elif val.endswith("ml"):
+        return float(val.split("ml")[0])
+    elif val.endswith("RPM"):
+        return float(val.split("RPM")[0])
+    elif val.endswith(" °C"):
+        return float(val.split(" ")[0])
+    elif val.endswith(" %"):
+        return float(val.split(" ")[0])
+    return val
+
+
+def flatten_xdl_procedure(procedure_elem: ET.Element) -> List[ET.Element]:
+    """展平嵌套的XDL程序结构"""
+    flattened_operations = []
+    TEMP_UNSUPPORTED_PROTOCOL = ["Purge", "Wait", "Stir", "ResetHandling"]
+
+    def extract_operations(element: ET.Element):
+        if element.tag not in ["Prep", "Reaction", "Workup", "Purification", "Procedure"]:
+            if element.tag not in TEMP_UNSUPPORTED_PROTOCOL:
+                flattened_operations.append(element)
+
+        for child in element:
+            extract_operations(child)
+
+    for child in procedure_elem:
+        extract_operations(child)
+
+    return flattened_operations
+
+
+def parse_xdl_content(xdl_content: str) -> tuple:
+    """解析XDL内容"""
+    try:
+        xdl_content_cleaned = "".join(c for c in xdl_content if c.isprintable())
+        root = ET.fromstring(xdl_content_cleaned)
+
+        synthesis_elem = root.find("Synthesis")
+        if synthesis_elem is None:
+            return None, None, None
+
+        # 解析硬件组件
+        hardware_elem = synthesis_elem.find("Hardware")
+        hardware = []
+        if hardware_elem is not None:
+            hardware = [{"id": c.get("id"), "type": c.get("type")} for c in hardware_elem.findall("Component")]
+
+        # 解析试剂
+        reagents_elem = synthesis_elem.find("Reagents")
+        reagents = []
+        if reagents_elem is not None:
+            reagents = [{"name": r.get("name"), "role": r.get("role", "")} for r in reagents_elem.findall("Reagent")]
+
+        # 解析程序
+        procedure_elem = synthesis_elem.find("Procedure")
+        if procedure_elem is None:
+            return None, None, None
+
+        flattened_operations = flatten_xdl_procedure(procedure_elem)
+        return hardware, reagents, flattened_operations
+
+    except ET.ParseError as e:
+        raise ValueError(f"Invalid XDL format: {e}")
+
+
+def convert_xdl_to_dict(xdl_content: str) -> Dict[str, Any]:
+    """
+    将XDL XML格式转换为标准的字典格式
+
+    Args:
+        xdl_content: XDL XML内容
+
+    Returns:
+        转换结果，包含步骤和器材信息
+    """
+    try:
+        hardware, reagents, flattened_operations = parse_xdl_content(xdl_content)
+        if hardware is None:
+            return {"error": "Failed to parse XDL content", "success": False}
+
+        # 将XDL元素转换为字典格式
+        steps_data = []
+        for elem in flattened_operations:
+            # 转换参数类型
+            parameters = {}
+            for key, val in elem.attrib.items():
+                converted_val = convert_to_type(val)
+                if converted_val is not None:
+                    parameters[key] = converted_val
+
+            step_dict = {
+                "operation": elem.tag,
+                "parameters": parameters,
+                "description": elem.get("purpose", f"Operation: {elem.tag}"),
+            }
+            steps_data.append(step_dict)
+
+        # 合并硬件和试剂为统一的labware_info格式
+        labware_data = []
+        labware_data.extend({"id": hw["id"], "type": "hardware", **hw} for hw in hardware)
+        labware_data.extend({"name": reagent["name"], "type": "reagent", **reagent} for reagent in reagents)
+
+        return {
+            "success": True,
+            "steps": steps_data,
+            "labware": labware_data,
+            "message": f"Successfully converted XDL to dict format. Found {len(steps_data)} steps and {len(labware_data)} labware items.",
+        }
+
+    except Exception as e:
+        error_msg = f"XDL conversion failed: {str(e)}"
+        return {"error": error_msg, "success": False}

unilabos_msgs/package.xml

@@ -2,7 +2,7 @@
 <?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
 <package format="3">
   <name>unilabos_msgs</name>
-  <version>0.10.11</version>
+  <version>0.10.12</version>
   <description>ROS2 Messages package for unilabos devices</description>
   <maintainer email="changjh@pku.edu.cn">Junhan Chang</maintainer>
   <maintainer email="18435084+Xuwznln@users.noreply.github.com">Xuwznln</maintainer>