Bump version to 0.10.4

Fix and tested new create_resource.

Update create_resource to resource tree mode.

.conda/recipe.yaml

@@ -1,6 +1,6 @@
 package:
   name: unilabos
-  version: 0.10.15
+  version: 0.10.14
 
 source:
   path: ../unilabos

.github/workflows/conda-pack-build.yml

@@ -24,7 +24,7 @@ jobs:
             platform: linux-64
             env_file: unilabos-linux-64.yaml
             script_ext: sh
-          - os: macos-15 # Intel (via Rosetta)
+          - os: macos-13 # Intel
             platform: osx-64
             env_file: unilabos-osx-64.yaml
             script_ext: sh

.github/workflows/multi-platform-build.yml

@@ -27,7 +27,7 @@ jobs:
           - os: ubuntu-latest
             platform: linux-64
             env_file: unilabos-linux-64.yaml
-          - os: macos-15 # Intel (via Rosetta)
+          - os: macos-13 # Intel
             platform: osx-64
             env_file: unilabos-osx-64.yaml
           - os: macos-latest # ARM64

.github/workflows/unilabos-conda-build.yml

@@ -26,7 +26,7 @@ jobs:
         include:
           - os: ubuntu-latest
             platform: linux-64
-          - os: macos-15 # Intel (via Rosetta)
+          - os: macos-13 # Intel
             platform: osx-64
           - os: macos-latest # ARM64
             platform: osx-arm64

README.md

@@ -31,9 +31,7 @@ Detailed documentation can be found at:
 
 ## Quick Start
 
-1. Setup Conda Environment
-
-Uni-Lab-OS recommends using `mamba` for environment management:
+Uni-Lab-OS recommends using `mamba` for environment management. Choose the appropriate environment file for your operating system:
 
 ```bash
 # Create new environment
@@ -42,7 +40,7 @@ mamba activate unilab
 mamba install -n unilab uni-lab::unilabos -c robostack-staging -c conda-forge
 ```
 
-2. Install Dev Uni-Lab-OS
+## Install Dev Uni-Lab-OS
 
 ```bash
 # Clone the repository
@@ -53,21 +51,17 @@ cd Uni-Lab-OS
 pip install .
 ```
 
-3. Start Uni-Lab System
+3. Start Uni-Lab System:
 
 Please refer to [Documentation - Boot Examples](https://deepmodeling.github.io/Uni-Lab-OS/boot_examples/index.html)
 
-4. Best Practice
-
-See [Best Practice Guide](https://deepmodeling.github.io/Uni-Lab-OS/user_guide/best_practice.html)
-
 ## Message Format
 
 Uni-Lab-OS uses pre-built `unilabos_msgs` for system communication. You can find the built versions on the [GitHub Releases](https://github.com/deepmodeling/Uni-Lab-OS/releases) page.
 
 ## Citation
 
-If you use [Uni-Lab-OS](https://arxiv.org/abs/2512.21766) in academic research, please cite:
+If you use Uni-Lab-OS in academic research, please cite:
 
 ```bibtex
 @article{gao2025unilabos,

README_zh.md

@@ -53,21 +53,17 @@ cd Uni-Lab-OS
 pip install .
 ```
 
-3. 启动 Uni-Lab 系统
+3. 启动 Uni-Lab 系统:
 
 请见[文档-启动样例](https://deepmodeling.github.io/Uni-Lab-OS/boot_examples/index.html)
 
-4. 最佳实践
-
-请见[最佳实践指南](https://deepmodeling.github.io/Uni-Lab-OS/user_guide/best_practice.html)
-
 ## 消息格式
 
 Uni-Lab-OS 使用预构建的 `unilabos_msgs` 进行系统通信。您可以在 [GitHub Releases](https://github.com/deepmodeling/Uni-Lab-OS/releases) 页面找到已构建的版本。
 
 ## 引用
 
-如果您在学术研究中使用 [Uni-Lab-OS](https://arxiv.org/abs/2512.21766)，请引用：
+如果您在学术研究中使用 Uni-Lab-OS，请引用：
 
 ```bibtex
 @article{gao2025unilabos,

docs/conf.py

@@ -24,7 +24,7 @@ extensions = [
     "sphinx.ext.autodoc",
     "sphinx.ext.napoleon",  # 如果您使用 Google 或 NumPy 风格的 docstrings
     "sphinx_rtd_theme",
-    "sphinxcontrib.mermaid",
+    "sphinxcontrib.mermaid"
 ]
 
 source_suffix = {
@@ -58,7 +58,7 @@ html_theme = "sphinx_rtd_theme"
 
 # sphinx-book-theme 主题选项
 html_theme_options = {
-    "repository_url": "https://github.com/deepmodeling/Uni-Lab-OS",
+    "repository_url": "https://github.com/用户名/Uni-Lab",
     "use_repository_button": True,
     "use_issues_button": True,
     "use_edit_page_button": True,

docs/requirements.txt

@@ -12,7 +12,3 @@ sphinx-copybutton>=0.5.0
 
 # 用于自动摘要生成
 sphinx-autobuild>=2024.2.4
-
-# 用于PDF导出 (rinohtype方案，纯Python无需LaTeX)
-rinohtype>=0.5.4
-sphinx-simplepdf>=1.6.0

docs/user_guide/graph_files.md

@@ -463,7 +463,7 @@ Uni-Lab 使用 `ResourceDictInstance.get_resource_instance_from_dict()` 方法
 ### 使用示例
 
 ```python
-from unilabos.resources.resource_tracker import ResourceDictInstance
+from unilabos.ros.nodes.resource_tracker import ResourceDictInstance
 
 # 旧格式节点
 old_format_node = {
@@ -477,10 +477,10 @@ old_format_node = {
 instance = ResourceDictInstance.get_resource_instance_from_dict(old_format_node)
 
 # 访问标准化后的数据
-print(instance.res_content.id)  # "pump_1"
-print(instance.res_content.uuid)  # 自动生成的 UUID
+print(instance.res_content.id)      # "pump_1"
+print(instance.res_content.uuid)    # 自动生成的 UUID
 print(instance.res_content.config)  # {}
-print(instance.res_content.data)  # {}
+print(instance.res_content.data)    # {}
 ```
 
 ### 格式迁移建议

recipes/msgs/recipe.yaml

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

recipes/unilabos/recipe.yaml

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

scripts/workflow.py

@@ -2,6 +2,7 @@ import json
 import logging
 import traceback
 import uuid
+import xml.etree.ElementTree as ET
 from typing import Any, Dict, List
 
 import networkx as nx
@@ -24,15 +25,7 @@ class SimpleGraph:
 
     def add_edge(self, source, 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
-        }
+        edge = {"source": source, "target": target, **attrs}
         self.edges.append(edge)
 
     def to_dict(self):
@@ -49,7 +42,6 @@ class SimpleGraph:
             "multigraph": False,
             "graph": {},
             "nodes": nodes_list,
-            "edges": self.edges,
             "links": self.edges,
         }
 
@@ -66,8 +58,495 @@ 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.15',
+    version='0.10.14',
     packages=find_packages(),
     include_package_data=True,
     install_requires=['setuptools'],

tests/resources/test_resourcetreeset.py

@@ -2,8 +2,9 @@ import pytest
 import json
 import os
 
+from pylabrobot.resources import Resource as ResourcePLR
 from unilabos.resources.graphio import resource_bioyond_to_plr
-from unilabos.resources.resource_tracker import ResourceTreeSet
+from unilabos.ros.nodes.resource_tracker import ResourceTreeSet
 from unilabos.registry.registry import lab_registry
 
 from unilabos.resources.bioyond.decks import BIOYOND_PolymerReactionStation_Deck

unilabos/__init__.py

@@ -1 +1 @@
-__version__ = "0.10.15"
+__version__ = "0.10.14"

unilabos/app/backend.py

@@ -1,6 +1,6 @@
 import threading
 
-from unilabos.resources.resource_tracker import ResourceTreeSet
+from unilabos.ros.nodes.resource_tracker import ResourceTreeSet
 from unilabos.utils import logger
 
 

unilabos/app/main.py

@@ -19,12 +19,6 @@ if unilabos_dir not in sys.path:
 
 from unilabos.utils.banner_print import print_status, print_unilab_banner
 from unilabos.config.config import load_config, BasicConfig, HTTPConfig
-from unilabos.app.utils import cleanup_for_restart
-
-# Global restart flags (used by ws_client and web/server)
-_restart_requested: bool = False
-_restart_reason: str = ""
-
 
 def load_config_from_file(config_path):
     if config_path is None:
@@ -47,7 +41,7 @@ def convert_argv_dashes_to_underscores(args: argparse.ArgumentParser):
     for i, arg in enumerate(sys.argv):
         for option_string in option_strings:
             if arg.startswith(option_string):
-                new_arg = arg[:2] + arg[2 : len(option_string)].replace("-", "_") + arg[len(option_string) :]
+                new_arg = arg[:2] + arg[2:len(option_string)].replace("-", "_") + arg[len(option_string):]
                 sys.argv[i] = new_arg
                 break
 
@@ -55,8 +49,6 @@ 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(
@@ -161,44 +153,6 @@ def parse_args():
         default=False,
         help="Complete registry information",
     )
-    parser.add_argument(
-        "--no_update_feedback",
-        action="store_true",
-        help="Disable sending update feedback to server",
-    )
-    # workflow upload subcommand
-    workflow_parser = subparsers.add_parser(
-        "workflow_upload",
-        aliases=["wf"],
-        help="Upload workflow from xdl/json/python files",
-    )
-    workflow_parser.add_argument(
-        "-f",
-        "--workflow_file",
-        type=str,
-        required=True,
-        help="Path to the workflow file (JSON format)",
-    )
-    workflow_parser.add_argument(
-        "-n",
-        "--workflow_name",
-        type=str,
-        default=None,
-        help="Workflow name, if not provided will use the name from file or filename",
-    )
-    workflow_parser.add_argument(
-        "--tags",
-        type=str,
-        nargs="*",
-        default=[],
-        help="Tags for the workflow (space-separated)",
-    )
-    workflow_parser.add_argument(
-        "--published",
-        action="store_true",
-        default=False,
-        help="Whether to publish the workflow (default: False)",
-    )
     return parser
 
 
@@ -214,6 +168,7 @@ def main():
     if not args_dict.get("skip_env_check", False):
         from unilabos.utils.environment_check import check_environment
 
+        print_status("正在进行环境依赖检查...", "info")
         if not check_environment(auto_install=True):
             print_status("环境检查失败，程序退出", "error")
             os._exit(1)
@@ -286,12 +241,9 @@ def main():
     if args_dict.get("sk", ""):
         BasicConfig.sk = args_dict.get("sk", "")
         print_status("传入了sk参数，优先采用传入参数！", "info")
-    BasicConfig.working_dir = working_dir
-
-    workflow_upload = args_dict.get("command") in ("workflow_upload", "wf")
 
     # 使用远程资源启动
-    if not workflow_upload and args_dict["use_remote_resource"]:
+    if args_dict["use_remote_resource"]:
         print_status("使用远程资源启动", "info")
         from unilabos.app.web import http_client
 
@@ -304,10 +256,10 @@ 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)
-    BasicConfig.no_update_feedback = args_dict.get("no_update_feedback", False)
     BasicConfig.communication_protocol = "websocket"
     machine_name = os.popen("hostname").read().strip()
     machine_name = "".join([c if c.isalnum() or c == "_" else "_" for c in machine_name])
@@ -326,38 +278,16 @@ def main():
     from unilabos.app.web import start_server
     from unilabos.app.register import register_devices_and_resources
     from unilabos.resources.graphio import modify_to_backend_format
-    from unilabos.resources.resource_tracker import ResourceTreeSet, ResourceDict
+    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), BasicConfig.upload_registry
+        args_dict["registry_path"], args_dict.get("complete_registry", False), args_dict["upload_registry"]
     )
 
-    if BasicConfig.upload_registry:
-        # 设备注册到服务端 - 需要 ak 和 sk
-        if BasicConfig.ak and BasicConfig.sk:
-            print_status("开始注册设备到服务端...", "info")
-            try:
-                register_devices_and_resources(lab_registry)
-                print_status("设备注册完成", "info")
-            except Exception as e:
-                print_status(f"设备注册失败: {e}", "error")
-        else:
-            print_status("未提供 ak 和 sk，跳过设备注册", "info")
-    else:
-        print_status("本次启动注册表不报送云端，如果您需要联网调试，请在启动命令增加--upload_registry", "warning")
-
-    # 处理 workflow_upload 子命令
-    if workflow_upload:
-        from unilabos.workflow.wf_utils import handle_workflow_upload_command
-
-        handle_workflow_upload_command(args_dict)
-        print_status("工作流上传完成，程序退出", "info")
-        os._exit(0)
-
     if not BasicConfig.ak or not BasicConfig.sk:
         print_status("后续运行必须拥有一个实验室，请前往 https://uni-lab.bohrium.com 注册实验室！", "warning")
         os._exit(1)
@@ -438,6 +368,20 @@ def main():
     args_dict["devices_config"] = resource_tree_set
     args_dict["graph"] = graph_res.physical_setup_graph
 
+    if BasicConfig.upload_registry:
+        # 设备注册到服务端 - 需要 ak 和 sk
+        if BasicConfig.ak and BasicConfig.sk:
+            print_status("开始注册设备到服务端...", "info")
+            try:
+                register_devices_and_resources(lab_registry)
+                print_status("设备注册完成", "info")
+            except Exception as e:
+                print_status(f"设备注册失败: {e}", "error")
+        else:
+            print_status("未提供 ak 和 sk，跳过设备注册", "info")
+    else:
+        print_status("本次启动注册表不报送云端，如果您需要联网调试，请在启动命令增加--upload_registry", "warning")
+
     if args_dict["controllers"] is not None:
         args_dict["controllers_config"] = yaml.safe_load(open(args_dict["controllers"], encoding="utf-8"))
     else:
@@ -452,7 +396,6 @@ def main():
         comm_client = get_communication_client()
         if "websocket" in args_dict["app_bridges"]:
             args_dict["bridges"].append(comm_client)
-
             def _exit(signum, frame):
                 comm_client.stop()
                 sys.exit(0)
@@ -494,13 +437,16 @@ def main():
                 resource_visualization.start()
             except OSError as e:
                 if "AMENT_PREFIX_PATH" in str(e):
-                    print_status(f"ROS 2环境未正确设置，跳过3D可视化启动。错误详情: {e}", "warning")
+                    print_status(
+                        f"ROS 2环境未正确设置，跳过3D可视化启动。错误详情: {e}",
+                        "warning"
+                    )
                     print_status(
                         "建议解决方案：\n"
                         "1. 激活Conda环境: conda activate unilab\n"
                         "2. 或使用 --backend simple 参数\n"
                         "3. 或使用 --visual disable 参数禁用可视化",
-                        "info",
+                        "info"
                     )
                 else:
                     raise
@@ -508,19 +454,13 @@ def main():
                 time.sleep(1)
         else:
             start_backend(**args_dict)
-            restart_requested = start_server(
+            start_server(
                 open_browser=not args_dict["disable_browser"],
                 port=BasicConfig.port,
             )
-            if restart_requested:
-                print_status("[Main] Restart requested, cleaning up...", "info")
-                cleanup_for_restart()
-                return
     else:
         start_backend(**args_dict)
-
-        # 启动服务器（默认支持WebSocket触发重启）
-        restart_requested = start_server(
+        start_server(
             open_browser=not args_dict["disable_browser"],
             port=BasicConfig.port,
         )

unilabos/app/utils.py

@@ -1,144 +0,0 @@
-"""
-UniLabOS 应用工具函数
-
-提供清理、重启等工具函数
-"""
-
-import gc
-import os
-import threading
-import time
-
-from unilabos.utils.banner_print import print_status
-
-
-def cleanup_for_restart() -> bool:
-    """
-    Clean up all resources for restart without exiting the process.
-
-    This function prepares the system for re-initialization by:
-    1. Stopping all communication clients
-    2. Destroying ROS nodes
-    3. Resetting singletons
-    4. Waiting for threads to finish
-
-    Returns:
-        bool: True if cleanup was successful, False otherwise
-    """
-    print_status("[Restart] Starting cleanup for restart...", "info")
-
-    # Step 1: Stop WebSocket communication client
-    print_status("[Restart] Step 1: Stopping WebSocket client...", "info")
-    try:
-        from unilabos.app.communication import get_communication_client
-
-        comm_client = get_communication_client()
-        if comm_client is not None:
-            comm_client.stop()
-            print_status("[Restart] WebSocket client stopped", "info")
-    except Exception as e:
-        print_status(f"[Restart] Error stopping WebSocket: {e}", "warning")
-
-    # Step 2: Get HostNode and cleanup ROS
-    print_status("[Restart] Step 2: Cleaning up ROS nodes...", "info")
-    try:
-        from unilabos.ros.nodes.presets.host_node import HostNode
-        import rclpy
-        from rclpy.timer import Timer
-
-        host_instance = HostNode.get_instance(timeout=5)
-        if host_instance is not None:
-            print_status(f"[Restart] Found HostNode: {host_instance.device_id}", "info")
-
-            # Gracefully shutdown background threads
-            print_status("[Restart] Shutting down background threads...", "info")
-            HostNode.shutdown_background_threads(timeout=5.0)
-            print_status("[Restart] Background threads shutdown complete", "info")
-
-            # Stop discovery timer
-            if hasattr(host_instance, "_discovery_timer") and isinstance(host_instance._discovery_timer, Timer):
-                host_instance._discovery_timer.cancel()
-                print_status("[Restart] Discovery timer cancelled", "info")
-
-            # Destroy device nodes
-            device_count = len(host_instance.devices_instances)
-            print_status(f"[Restart] Destroying {device_count} device instances...", "info")
-            for device_id, device_node in list(host_instance.devices_instances.items()):
-                try:
-                    if hasattr(device_node, "ros_node_instance") and device_node.ros_node_instance is not None:
-                        device_node.ros_node_instance.destroy_node()
-                        print_status(f"[Restart] Device {device_id} destroyed", "info")
-                except Exception as e:
-                    print_status(f"[Restart] Error destroying device {device_id}: {e}", "warning")
-
-            # Clear devices instances
-            host_instance.devices_instances.clear()
-            host_instance.devices_names.clear()
-
-            # Destroy host node
-            try:
-                host_instance.destroy_node()
-                print_status("[Restart] HostNode destroyed", "info")
-            except Exception as e:
-                print_status(f"[Restart] Error destroying HostNode: {e}", "warning")
-
-            # Reset HostNode state
-            HostNode.reset_state()
-            print_status("[Restart] HostNode state reset", "info")
-
-        # Shutdown executor first (to stop executor.spin() gracefully)
-        if hasattr(rclpy, "__executor") and rclpy.__executor is not None:
-            try:
-                rclpy.__executor.shutdown()
-                rclpy.__executor = None  # Clear for restart
-                print_status("[Restart] ROS executor shutdown complete", "info")
-            except Exception as e:
-                print_status(f"[Restart] Error shutting down executor: {e}", "warning")
-
-        # Shutdown rclpy
-        if rclpy.ok():
-            rclpy.shutdown()
-            print_status("[Restart] rclpy shutdown complete", "info")
-
-    except ImportError as e:
-        print_status(f"[Restart] ROS modules not available: {e}", "warning")
-    except Exception as e:
-        print_status(f"[Restart] Error in ROS cleanup: {e}", "warning")
-        return False
-
-    # Step 3: Reset communication client singleton
-    print_status("[Restart] Step 3: Resetting singletons...", "info")
-    try:
-        from unilabos.app import communication
-
-        if hasattr(communication, "_communication_client"):
-            communication._communication_client = None
-            print_status("[Restart] Communication client singleton reset", "info")
-    except Exception as e:
-        print_status(f"[Restart] Error resetting communication singleton: {e}", "warning")
-
-    # Step 4: Wait for threads to finish
-    print_status("[Restart] Step 4: Waiting for threads to finish...", "info")
-    time.sleep(3)  # Give threads time to finish
-
-    # Check remaining threads
-    remaining_threads = []
-    for t in threading.enumerate():
-        if t.name != "MainThread" and t.is_alive():
-            remaining_threads.append(t.name)
-
-    if remaining_threads:
-        print_status(
-            f"[Restart] Warning: {len(remaining_threads)} threads still running: {remaining_threads}", "warning"
-        )
-    else:
-        print_status("[Restart] All threads stopped", "info")
-
-    # Step 5: Force garbage collection
-    print_status("[Restart] Step 5: Running garbage collection...", "info")
-    gc.collect()
-    gc.collect()  # Run twice for weak references
-    print_status("[Restart] Garbage collection complete", "info")
-
-    print_status("[Restart] Cleanup complete. Ready for re-initialization.", "info")
-    return True

unilabos/app/web/client.py

@@ -6,10 +6,12 @@ HTTP客户端模块
 
 import json
 import os
+import time
+from threading import Thread
 from typing import List, Dict, Any, Optional
 
 import requests
-from unilabos.resources.resource_tracker import ResourceTreeSet
+from unilabos.ros.nodes.resource_tracker import ResourceTreeSet
 from unilabos.utils.log import info
 from unilabos.config.config import HTTPConfig, BasicConfig
 from unilabos.utils import logger
@@ -74,8 +76,7 @@ class HTTPClient:
             Dict[str, str]: 旧UUID到新UUID的映射关系 {old_uuid: new_uuid}
         """
         with open(os.path.join(BasicConfig.working_dir, "req_resource_tree_add.json"), "w", encoding="utf-8") as f:
-            payload = {"nodes": [x for xs in resources.dump() for x in xs], "mount_uuid": mount_uuid}
-            f.write(json.dumps(payload, indent=4))
+            f.write(json.dumps({"nodes": [x for xs in resources.dump() for x in xs], "mount_uuid": mount_uuid}, indent=4))
         # 从序列化数据中提取所有节点的UUID（保存旧UUID）
         old_uuids = {n.res_content.uuid: n for n in resources.all_nodes}
         if not self.initialized or first_add:
@@ -334,67 +335,6 @@ class HTTPClient:
                 logger.error(f"响应内容: {response.text}")
         return None
 
-    def workflow_import(
-        self,
-        name: str,
-        workflow_uuid: str,
-        workflow_name: str,
-        nodes: List[Dict[str, Any]],
-        edges: List[Dict[str, Any]],
-        tags: Optional[List[str]] = None,
-        published: bool = False,
-    ) -> Dict[str, Any]:
-        """
-        导入工作流到服务器
-
-        Args:
-            name: 工作流名称（顶层）
-            workflow_uuid: 工作流UUID
-            workflow_name: 工作流名称（data内部）
-            nodes: 工作流节点列表
-            edges: 工作流边列表
-            tags: 工作流标签列表，默认为空列表
-            published: 是否发布工作流，默认为False
-
-        Returns:
-            Dict: API响应数据，包含 code 和 data (uuid, name)
-        """
-        # target_lab_uuid 暂时使用默认值，后续由后端根据 ak/sk 获取
-        payload = {
-            "target_lab_uuid": "28c38bb0-63f6-4352-b0d8-b5b8eb1766d5",
-            "name": name,
-            "data": {
-                "workflow_uuid": workflow_uuid,
-                "workflow_name": workflow_name,
-                "nodes": nodes,
-                "edges": edges,
-                "tags": tags if tags is not None else [],
-                "published": published,
-            },
-        }
-        # 保存请求到文件
-        with open(os.path.join(BasicConfig.working_dir, "req_workflow_upload.json"), "w", encoding="utf-8") as f:
-            f.write(json.dumps(payload, indent=4, ensure_ascii=False))
-
-        response = requests.post(
-            f"{self.remote_addr}/lab/workflow/owner/import",
-            json=payload,
-            headers={"Authorization": f"Lab {self.auth}"},
-            timeout=60,
-        )
-        # 保存响应到文件
-        with open(os.path.join(BasicConfig.working_dir, "res_workflow_upload.json"), "w", encoding="utf-8") as f:
-            f.write(f"{response.status_code}" + "\n" + response.text)
-
-        if response.status_code == 200:
-            res = response.json()
-            if "code" in res and res["code"] != 0:
-                logger.error(f"导入工作流失败: {response.text}")
-            return res
-        else:
-            logger.error(f"导入工作流失败: {response.status_code}, {response.text}")
-            return {"code": response.status_code, "message": response.text}
-
 
 # 创建默认客户端实例
 http_client = HTTPClient()

unilabos/app/web/server.py

@@ -6,6 +6,7 @@ Web服务器模块
 
 import webbrowser
 
+import uvicorn
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from starlette.responses import Response
@@ -95,7 +96,7 @@ def setup_server() -> FastAPI:
     return app
 
 
-def start_server(host: str = "0.0.0.0", port: int = 8002, open_browser: bool = True) -> bool:
+def start_server(host: str = "0.0.0.0", port: int = 8002, open_browser: bool = True) -> None:
     """
     启动服务器
 
@@ -103,14 +104,7 @@ def start_server(host: str = "0.0.0.0", port: int = 8002, open_browser: bool = T
         host: 服务器主机
         port: 服务器端口
         open_browser: 是否自动打开浏览器
-
-    Returns:
-        bool: True if restart was requested, False otherwise
     """
-    import threading
-    import time
-    from uvicorn import Config, Server
-
     # 设置服务器
     setup_server()
 
@@ -129,37 +123,7 @@ def start_server(host: str = "0.0.0.0", port: int = 8002, open_browser: bool = T
 
     # 启动服务器
     info(f"[Web] 启动FastAPI服务器: {host}:{port}")
-
-    # 使用支持重启的模式
-    config = Config(app=app, host=host, port=port, log_config=log_config)
-    server = Server(config)
-
-    # 启动服务器线程
-    server_thread = threading.Thread(target=server.run, daemon=True, name="uvicorn_server")
-    server_thread.start()
-
-    info("[Web] Server started, monitoring for restart requests...")
-
-    # 监控重启标志
-    import unilabos.app.main as main_module
-
-    while server_thread.is_alive():
-        if hasattr(main_module, "_restart_requested") and main_module._restart_requested:
-            info(
-                f"[Web] Restart requested via WebSocket, reason: {getattr(main_module, '_restart_reason', 'unknown')}"
-            )
-            main_module._restart_requested = False
-
-            # 停止服务器
-            server.should_exit = True
-            server_thread.join(timeout=5)
-
-            info("[Web] Server stopped, ready for restart")
-            return True
-
-        time.sleep(1)
-
-    return False
+    uvicorn.run(app, host=host, port=port, log_config=log_config)
 
 
 # 当脚本直接运行时启动服务器

unilabos/app/ws_client.py

@@ -359,7 +359,7 @@ class MessageProcessor:
         self.device_manager = device_manager
         self.queue_processor = None  # 延迟设置
         self.websocket_client = None  # 延迟设置
-        self.session_id = str(uuid.uuid4())[:6]  # 产生一个随机的session_id
+        self.session_id = ""
 
         # WebSocket连接
         self.websocket = None
@@ -438,7 +438,7 @@ class MessageProcessor:
                     self.connected = True
                     self.reconnect_count = 0
 
-                    logger.trace(f"[MessageProcessor] Connected to {self.websocket_url}")
+                    logger.info(f"[MessageProcessor] Connected to {self.websocket_url}")
 
                     # 启动发送协程
                     send_task = asyncio.create_task(self._send_handler())
@@ -488,16 +488,7 @@ class MessageProcessor:
             async for message in self.websocket:
                 try:
                     data = json.loads(message)
-                    message_type = data.get("action", "")
-                    message_data = data.get("data")
-                    if self.session_id and self.session_id == data.get("edge_session"):
-                        await self._process_message(message_type, message_data)
-                    else:
-                        if message_type.endswith("_material"):
-                            logger.trace(f"[MessageProcessor] 收到一条归属 {data.get('edge_session')} 的旧消息：{data}")
-                            logger.debug(f"[MessageProcessor] 跳过了一条归属 {data.get('edge_session')} 的旧消息: {data.get('action')}")
-                        else:
-                            await self._process_message(message_type, message_data)
+                    await self._process_message(data)
                 except json.JSONDecodeError:
                     logger.error(f"[MessageProcessor] Invalid JSON received: {message}")
                 except Exception as e:
@@ -512,7 +503,7 @@ class MessageProcessor:
 
     async def _send_handler(self):
         """处理发送队列中的消息"""
-        logger.trace("[MessageProcessor] Send handler started")
+        logger.debug("[MessageProcessor] Send handler started")
 
         try:
             while self.connected and self.websocket:
@@ -563,8 +554,11 @@ class MessageProcessor:
         finally:
             logger.debug("[MessageProcessor] Send handler stopped")
 
-    async def _process_message(self, message_type: str, message_data: Dict[str, Any]):
+    async def _process_message(self, data: Dict[str, Any]):
         """处理收到的消息"""
+        message_type = data.get("action", "")
+        message_data = data.get("data")
+
         logger.debug(f"[MessageProcessor] Processing message: {message_type}")
 
         try:
@@ -577,19 +571,16 @@ class MessageProcessor:
             elif message_type == "cancel_action" or message_type == "cancel_task":
                 await self._handle_cancel_action(message_data)
             elif message_type == "add_material":
-                # noinspection PyTypeChecker
                 await self._handle_resource_tree_update(message_data, "add")
             elif message_type == "update_material":
-                # noinspection PyTypeChecker
                 await self._handle_resource_tree_update(message_data, "update")
             elif message_type == "remove_material":
-                # noinspection PyTypeChecker
                 await self._handle_resource_tree_update(message_data, "remove")
-            # elif message_type == "session_id":
-            #     self.session_id = message_data.get("session_id")
-            #     logger.info(f"[MessageProcessor] Session ID: {self.session_id}")
-            elif message_type == "request_restart":
-                await self._handle_request_restart(message_data)
+            elif message_type == "session_id":
+                self.session_id = message_data.get("session_id")
+                logger.info(f"[MessageProcessor] Session ID: {self.session_id}")
+            elif message_type == "request_reload":
+                await self._handle_request_reload(message_data)
             else:
                 logger.debug(f"[MessageProcessor] Unknown message type: {message_type}")
 
@@ -899,48 +890,19 @@ class MessageProcessor:
             )
             thread.start()
 
-    async def _handle_request_restart(self, data: Dict[str, Any]):
+    async def _handle_request_reload(self, data: Dict[str, Any]):
         """
-        处理重启请求
+        处理重载请求
         
-        当LabGo发送request_restart时，执行清理并触发重启
+        当LabGo发送request_reload时，重新发送设备注册信息
         """
         reason = data.get("reason", "unknown")
-        delay = data.get("delay", 2)  # 默认延迟2秒
-        logger.info(f"[MessageProcessor] Received restart request, reason: {reason}, delay: {delay}s")
+        logger.info(f"[MessageProcessor] Received reload request, reason: {reason}")
         
-        # 发送确认消息
+        # 重新发送host_node_ready信息
         if self.websocket_client:
-            await self.websocket_client.send_message({
-                "action": "restart_acknowledged",
-                "data": {"reason": reason, "delay": delay}
-            })
-        
-        # 设置全局重启标志
-        import unilabos.app.main as main_module
-        main_module._restart_requested = True
-        main_module._restart_reason = reason
-        
-        # 延迟后执行清理
-        await asyncio.sleep(delay)
-        
-        # 在新线程中执行清理，避免阻塞当前事件循环
-        def do_cleanup():
-            import time
-            time.sleep(0.5)  # 给当前消息处理完成的时间
-            logger.info(f"[MessageProcessor] Starting cleanup for restart, reason: {reason}")
-            try:
-                from unilabos.app.utils import cleanup_for_restart
-                if cleanup_for_restart():
-                    logger.info("[MessageProcessor] Cleanup successful, main() will restart")
-                else:
-                    logger.error("[MessageProcessor] Cleanup failed")
-            except Exception as e:
-                logger.error(f"[MessageProcessor] Error during cleanup: {e}")
-        
-        cleanup_thread = threading.Thread(target=do_cleanup, name="RestartCleanupThread", daemon=True)
-        cleanup_thread.start()
-        logger.info(f"[MessageProcessor] Restart cleanup scheduled")
+            self.websocket_client.publish_host_ready()
+            logger.info("[MessageProcessor] Re-sent host_node_ready after reload request")
 
     async def _send_action_state_response(
         self, device_id: str, action_name: str, task_id: str, job_id: str, typ: str, free: bool, need_more: int
@@ -1019,7 +981,7 @@ class QueueProcessor:
 
     def _run(self):
         """运行队列处理主循环"""
-        logger.trace("[QueueProcessor] Queue processor started")
+        logger.debug("[QueueProcessor] Queue processor started")
 
         while self.is_running:
             try:
@@ -1229,6 +1191,7 @@ class WebSocketClient(BaseCommunicationClient):
         else:
             url = f"{scheme}://{parsed.netloc}/api/v1/ws/schedule"
 
+        logger.debug(f"[WebSocketClient] URL: {url}")
         return url
 
     def start(self) -> None:
@@ -1241,11 +1204,13 @@ class WebSocketClient(BaseCommunicationClient):
             logger.error("[WebSocketClient] WebSocket URL not configured")
             return
 
+        logger.info(f"[WebSocketClient] Starting connection to {self.websocket_url}")
+
         # 启动两个核心线程
         self.message_processor.start()
         self.queue_processor.start()
 
-        logger.trace("[WebSocketClient] All threads started")
+        logger.info("[WebSocketClient] All threads started")
 
     def stop(self) -> None:
         """停止WebSocket客户端"""

unilabos/config/config.py

@@ -16,14 +16,12 @@ class BasicConfig:
     upload_registry = False
     machine_name = "undefined"
     vis_2d_enable = False
-    no_update_feedback = False
     enable_resource_load = True
     communication_protocol = "websocket"
     startup_json_path = None  # 填写绝对路径
     disable_browser = False  # 禁止浏览器自动打开
     port = 8002  # 本地HTTP服务
-    # 'TRACE', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'
-    log_level: Literal["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "DEBUG"
+    log_level: Literal['TRACE', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] = "DEBUG"  # 'TRACE', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'
 
     @classmethod
     def auth_secret(cls):
@@ -67,14 +65,13 @@ def _update_config_from_module(module):
                     if not attr.startswith("_"):
                         setattr(obj, attr, getattr(getattr(module, name), attr))
 
-
 def _update_config_from_env():
     prefix = "UNILABOS_"
     for env_key, env_value in os.environ.items():
         if not env_key.startswith(prefix):
             continue
         try:
-            key_path = env_key[len(prefix) :]  # Remove UNILAB_ prefix
+            key_path = env_key[len(prefix):]  # Remove UNILAB_ prefix
             class_field = key_path.upper().split("_", 1)
             if len(class_field) != 2:
                 logger.warning(f"[ENV] 环境变量格式不正确：{env_key}")

unilabos/device_comms/coin_cell_assembly_workstation.py

@@ -6,7 +6,7 @@ Coin Cell Assembly Workstation
 """
 from typing import Dict, Any, List, Optional, Union
 
-from unilabos.resources.resource_tracker import DeviceNodeResourceTracker
+from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker
 from unilabos.device_comms.workstation_base import WorkstationBase, WorkflowInfo
 from unilabos.device_comms.workstation_communication import (
     WorkstationCommunicationBase, CommunicationConfig, CommunicationProtocol, CoinCellCommunication
@@ -61,7 +61,7 @@ class CoinCellAssemblyWorkstation(WorkstationBase):
         
         # 创建资源跟踪器（如果没有提供）
         if resource_tracker is None:
-            from unilabos.resources.resource_tracker import DeviceNodeResourceTracker
+            from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker
             resource_tracker = DeviceNodeResourceTracker()
         
         # 初始化基类

unilabos/devices/liquid_handling/biomek.py

@@ -13,7 +13,7 @@ from pylabrobot.resources import (
 import copy
 from unilabos_msgs.msg import Resource
 
-from unilabos.resources.resource_tracker import DeviceNodeResourceTracker  # type: ignore
+from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker  # type: ignore
 
 
 class LiquidHandlerBiomek:

unilabos/registry/devices/liquid_handler.yaml

@@ -9737,34 +9737,7 @@ liquid_handler.prcxi:
           touch_tip: false
           use_channels:
           - 0
-        handles:
-          input:
-          - data_key: liquid
-            data_source: handle
-            data_type: resource
-            handler_key: sources
-            label: sources
-          - data_key: liquid
-            data_source: executor
-            data_type: resource
-            handler_key: targets
-            label: targets
-          - data_key: liquid
-            data_source: executor
-            data_type: resource
-            handler_key: tip_rack
-            label: tip_rack
-          output:
-          - data_key: liquid
-            data_source: handle
-            data_type: resource
-            handler_key: sources_out
-            label: sources
-          - data_key: liquid
-            data_source: executor
-            data_type: resource
-            handler_key: targets_out
-            label: targets
+        handles: {}
         placeholder_keys:
           sources: unilabos_resources
           targets: unilabos_resources

unilabos/registry/registry.py

@@ -222,7 +222,7 @@ class Registry:
         abs_path = Path(path).absolute()
         resource_path = abs_path / "resources"
         files = list(resource_path.glob("*/*.yaml"))
-        logger.trace(f"[UniLab Registry] load resources? {resource_path.exists()}, total: {len(files)}")
+        logger.debug(f"[UniLab Registry] resources: {resource_path.exists()}, total: {len(files)}")
         current_resource_number = len(self.resource_type_registry) + 1
         for i, file in enumerate(files):
             with open(file, encoding="utf-8", mode="r") as f:

unilabos/resources/graphio.py

@@ -13,7 +13,7 @@ from unilabos.config.config import BasicConfig
 from unilabos.resources.container import RegularContainer
 from unilabos.resources.itemized_carrier import ItemizedCarrier, BottleCarrier
 from unilabos.ros.msgs.message_converter import convert_to_ros_msg
-from unilabos.resources.resource_tracker import (
+from unilabos.ros.nodes.resource_tracker import (
     ResourceDictInstance,
     ResourceTreeSet,
 )
@@ -42,7 +42,7 @@ def canonicalize_nodes_data(
     Returns:
         ResourceTreeSet: 标准化后的资源树集合
     """
-    print_status(f"{len(nodes)} Resources loaded", "info")
+    print_status(f"{len(nodes)} Resources loaded:", "info")
 
     # 第一步：基本预处理（处理graphml的label字段）
     outer_host_node_id = None

unilabos/resources/itemized_carrier.py

@@ -149,7 +149,6 @@ class ItemizedCarrier(ResourcePLR):
 
     if not reassign and self.sites[idx] is not None:
       raise ValueError(f"a site with index {idx} already exists")
-    location = list(self.child_locations.values())[idx]
     super().assign_child_resource(resource, location=location, reassign=reassign)
     self.sites[idx] = resource
 

unilabos/ros/device_node_wrapper.py

@@ -5,7 +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.resources.resource_tracker import ResourceDictInstance
+from unilabos.ros.nodes.resource_tracker import ResourceDictInstance
 
 # 定义泛型类型变量
 T = TypeVar("T")

unilabos/ros/initialize_device.py

@@ -1,9 +1,10 @@
+import copy
 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.resources.resource_tracker import ResourceDictInstance
+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

unilabos/ros/main_slave_run.py

@@ -1,5 +1,4 @@
 import json
-
 # from nt import device_encoding
 import threading
 import time
@@ -11,7 +10,7 @@ from unilabos_msgs.srv._serial_command import SerialCommand_Response
 
 from unilabos.app.register import register_devices_and_resources
 from unilabos.ros.nodes.presets.resource_mesh_manager import ResourceMeshManager
-from unilabos.resources.resource_tracker import DeviceNodeResourceTracker, ResourceTreeSet
+from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker, ResourceTreeSet
 from unilabos.devices.ros_dev.liquid_handler_joint_publisher import LiquidHandlerJointPublisher
 from unilabos_msgs.srv import SerialCommand  # type: ignore
 from rclpy.executors import MultiThreadedExecutor
@@ -56,11 +55,7 @@ def main(
 ) -> None:
     """主函数"""
 
-    # Support restart - check if rclpy is already initialized
-    if not rclpy.ok():
-        rclpy.init(args=rclpy_init_args)
-    else:
-        logger.info("[ROS] rclpy already initialized, reusing context")
+    rclpy.init(args=rclpy_init_args)
     executor = rclpy.__executor = MultiThreadedExecutor()
     # 创建主机节点
     host_node = HostNode(
@@ -93,7 +88,7 @@ def main(
         joint_republisher = JointRepublisher("joint_republisher", host_node.resource_tracker)
         # lh_joint_pub = LiquidHandlerJointPublisher(
         #     resources_config=resources_list, resource_tracker=host_node.resource_tracker
-        # )
+        # ) 
         executor.add_node(resource_mesh_manager)
         executor.add_node(joint_republisher)
         # executor.add_node(lh_joint_pub)

unilabos/ros/nodes/base_device_node.py

@@ -1,3 +1,4 @@
+import copy
 import inspect
 import io
 import json
@@ -12,6 +13,7 @@ import asyncio
 
 import rclpy
 import yaml
+from msgcenterpy import ROS2MessageInstance
 from rclpy.node import Node
 from rclpy.action import ActionServer, ActionClient
 from rclpy.action.server import ServerGoalHandle
@@ -20,13 +22,15 @@ from rclpy.callback_groups import ReentrantCallbackGroup
 from rclpy.service import Service
 from unilabos_msgs.action import SendCmd
 from unilabos_msgs.srv._serial_command import SerialCommand_Request, SerialCommand_Response
-
-from unilabos.config.config import BasicConfig
 from unilabos.utils.decorator import get_topic_config, get_all_subscriptions
 
 from unilabos.resources.container import RegularContainer
 from unilabos.resources.graphio import (
+    resource_ulab_to_plr,
     initialize_resources,
+    dict_to_tree,
+    resource_plr_to_ulab,
+    tree_to_list,
 )
 from unilabos.resources.plr_additional_res_reg import register
 from unilabos.ros.msgs.message_converter import (
@@ -43,7 +47,7 @@ from unilabos_msgs.srv import (
 )  # type: ignore
 from unilabos_msgs.msg import Resource  # type: ignore
 
-from unilabos.resources.resource_tracker import (
+from unilabos.ros.nodes.resource_tracker import (
     DeviceNodeResourceTracker,
     ResourceTreeSet,
     ResourceTreeInstance,
@@ -359,6 +363,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
             return res
 
         async def append_resource(req: SerialCommand_Request, res: SerialCommand_Response):
+            from pylabrobot.resources.resource import Resource as ResourcePLR
             from pylabrobot.resources.deck import Deck
             from pylabrobot.resources import Coordinate
             from pylabrobot.resources import Plate
@@ -792,7 +797,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
 
         def _handle_update(
             plr_resources: List[Union[ResourcePLR, ResourceDictInstance]], tree_set: ResourceTreeSet, additional_add_params: Dict[str, Any]
-        ) -> Tuple[Dict[str, Any], List[ResourcePLR]]:
+        ) -> Dict[str, Any]:
             """
             处理资源更新操作的内部函数
 
@@ -804,7 +809,6 @@ class BaseROS2DeviceNode(Node, Generic[T]):
             Returns:
                 操作结果字典
             """
-            original_instances = []
             for plr_resource, tree in zip(plr_resources, tree_set.trees):
                 if isinstance(plr_resource, ResourceDictInstance):
                     self._lab_logger.info(f"跳过 非资源{plr_resource.res_content.name} 的更新")
@@ -847,16 +851,6 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                     and original_parent_resource is not None
                 ):
                     self.transfer_to_new_resource(original_instance, tree, additional_add_params)
-                else:
-                    # 判断是否变更了resource_site
-                    target_site = original_instance.unilabos_extra.get("update_resource_site")
-                    sites = original_instance.parent.sites if original_instance.parent is not None and hasattr(original_instance.parent, "sites") else None
-                    site_names = list(original_instance.parent._ordering.keys()) if original_instance.parent is not None and hasattr(original_instance.parent, "sites") else []
-                    if target_site is not None and sites is not None and site_names is not None:
-                        site_index = sites.index(original_instance)
-                        site_name = site_names[site_index]
-                        if site_name != target_site:
-                            self.transfer_to_new_resource(original_instance, tree, additional_add_params)
 
                 # 加载状态
                 original_instance.load_all_state(states)
@@ -864,14 +858,13 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                 self.lab_logger().info(
                     f"更新了资源属性 {plr_resource}[{tree.root_node.res_content.uuid}] " f"及其子节点 {child_count} 个"
                 )
-                original_instances.append(original_instance)
 
             # 调用driver的update回调
             func = getattr(self.driver_instance, "resource_tree_update", None)
             if callable(func):
-                func(original_instances)
+                func(plr_resources)
 
-            return {"success": True, "action": "update"}, original_instances
+            return {"success": True, "action": "update"}
 
         try:
             data = json.loads(req.command)
@@ -895,13 +888,6 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                             raise ValueError("tree_set不能为None")
                         plr_resources = tree_set.to_plr_resources()
                         result = _handle_add(plr_resources, tree_set, additional_add_params)
-                        new_tree_set = ResourceTreeSet.from_plr_resources(plr_resources)
-                        r = SerialCommand.Request()
-                        r.command = json.dumps(
-                            {"data": {"data": new_tree_set.dump()}, "action": "update"})  # 和Update Resource一致
-                        response: SerialCommand_Response = await self._resource_clients[
-                            "c2s_update_resource_tree"].call_async(r)  # type: ignore
-                        self.lab_logger().info(f"确认资源云端 Add 结果: {response.response}")
                         results.append(result)
                     elif action == "update":
                         if tree_set is None:
@@ -912,15 +898,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                                 plr_resources.append(tree.root_node)
                             else:
                                 plr_resources.append(ResourceTreeSet([tree]).to_plr_resources()[0])
-                        result, original_instances = _handle_update(plr_resources, tree_set, additional_add_params)
-                        if not BasicConfig.no_update_feedback:
-                            new_tree_set = ResourceTreeSet.from_plr_resources(original_instances)
-                            r = SerialCommand.Request()
-                            r.command = json.dumps(
-                                {"data": {"data": new_tree_set.dump()}, "action": "update"})  # 和Update Resource一致
-                            response: SerialCommand_Response = await self._resource_clients[
-                                "c2s_update_resource_tree"].call_async(r)  # type: ignore
-                            self.lab_logger().info(f"确认资源云端 Update 结果: {response.response}")
+                        result = _handle_update(plr_resources, tree_set, additional_add_params)
                         results.append(result)
                     elif action == "remove":
                         result = _handle_remove(resources_uuid)
@@ -1787,7 +1765,6 @@ class ROS2DeviceNode:
             or driver_class.__name__ == "LiquidHandlerBiomek"
             or driver_class.__name__ == "PRCXI9300Handler"
             or driver_class.__name__ == "TransformXYZHandler"
-            or driver_class.__name__ == "OpcUaClient"
         )
 
         # 创建设备类实例

unilabos/ros/nodes/presets/host_node.py

@@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Optional, Dict, Any, List, ClassVar, Set, Type
 from action_msgs.msg import GoalStatus
 from geometry_msgs.msg import Point
 from rclpy.action import ActionClient, get_action_server_names_and_types_by_node
+from rclpy.callback_groups import ReentrantCallbackGroup
 from rclpy.service import Service
 from unilabos_msgs.msg import Resource  # type: ignore
 from unilabos_msgs.srv import (
@@ -18,6 +19,7 @@ from unilabos_msgs.srv import (
     ResourceUpdate,
     ResourceList,
     SerialCommand,
+    ResourceGet,
 )  # type: ignore
 from unilabos_msgs.srv._serial_command import SerialCommand_Request, SerialCommand_Response
 from unique_identifier_msgs.msg import UUID
@@ -35,7 +37,7 @@ from unilabos.ros.msgs.message_converter import (
 )
 from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode, DeviceNodeResourceTracker
 from unilabos.ros.nodes.presets.controller_node import ControllerNode
-from unilabos.resources.resource_tracker import (
+from unilabos.ros.nodes.resource_tracker import (
     ResourceDict,
     ResourceDictInstance,
     ResourceTreeSet,
@@ -70,8 +72,6 @@ class HostNode(BaseROS2DeviceNode):
 
     _instance: ClassVar[Optional["HostNode"]] = None
     _ready_event: ClassVar[threading.Event] = threading.Event()
-    _shutting_down: ClassVar[bool] = False  # Flag to signal shutdown to background threads
-    _background_threads: ClassVar[List[threading.Thread]] = []  # Track all background threads for cleanup
     _device_action_status: ClassVar[collections.defaultdict[str, DeviceActionStatus]] = collections.defaultdict(
         DeviceActionStatus
     )
@@ -83,48 +83,6 @@ class HostNode(BaseROS2DeviceNode):
             return cls._instance
         return None
 
-    @classmethod
-    def shutdown_background_threads(cls, timeout: float = 5.0) -> None:
-        """
-        Gracefully shutdown all background threads for clean exit or restart.
-
-        This method:
-        1. Sets shutdown flag to stop background operations
-        2. Waits for background threads to finish with timeout
-        3. Cleans up finished threads from tracking list
-
-        Args:
-            timeout: Maximum time to wait for each thread (seconds)
-        """
-        cls._shutting_down = True
-
-        # Wait for background threads to finish
-        active_threads = []
-        for t in cls._background_threads:
-            if t.is_alive():
-                t.join(timeout=timeout)
-                if t.is_alive():
-                    active_threads.append(t.name)
-
-        if active_threads:
-            logger.warning(f"[Host Node] Some background threads still running: {active_threads}")
-
-        # Clear the thread list
-        cls._background_threads.clear()
-        logger.info(f"[Host Node] Background threads shutdown complete")
-
-    @classmethod
-    def reset_state(cls) -> None:
-        """
-        Reset the HostNode singleton state for restart or clean exit.
-        Call this after destroying the instance.
-        """
-        cls._instance = None
-        cls._ready_event.clear()
-        cls._shutting_down = False
-        cls._background_threads.clear()
-        logger.info("[Host Node] State reset complete")
-
     def __init__(
         self,
         device_id: str,
@@ -338,37 +296,12 @@ class HostNode(BaseROS2DeviceNode):
                 bridge.publish_host_ready()
                 self.lab_logger().debug(f"Host ready signal sent via {bridge.__class__.__name__}")
 
-    def _send_re_register(self, sclient, device_namespace: str):
-        """
-        Send re-register command to a device. This is a one-time operation.
-
-        Args:
-            sclient: The service client
-            device_namespace: The device namespace for logging
-        """
-        try:
-            # Use timeout to prevent indefinite blocking
-            if not sclient.wait_for_service(timeout_sec=10.0):
-                self.lab_logger().debug(f"[Host Node] Re-register timeout for {device_namespace}")
-                return
-
-            # Check shutdown flag after wait
-            if self._shutting_down:
-                self.lab_logger().debug(f"[Host Node] Re-register aborted for {device_namespace} (shutdown)")
-                return
-
-            request = SerialCommand.Request()
-            request.command = ""
-            future = sclient.call_async(request)
-            # Use timeout for result as well
-            future.result(timeout_sec=5.0)
-            self.lab_logger().debug(f"[Host Node] Re-register completed for {device_namespace}")
-        except Exception as e:
-            # Gracefully handle destruction during shutdown
-            if "destruction was requested" in str(e) or self._shutting_down:
-                self.lab_logger().debug(f"[Host Node] Re-register aborted for {device_namespace} (cleanup)")
-            else:
-                self.lab_logger().warning(f"[Host Node] Re-register failed for {device_namespace}: {e}")
+    def _send_re_register(self, sclient):
+        sclient.wait_for_service()
+        request = SerialCommand.Request()
+        request.command = ""
+        future = sclient.call_async(request)
+        response = future.result()
 
     def _discover_devices(self) -> None:
         """
@@ -400,27 +333,23 @@ class HostNode(BaseROS2DeviceNode):
                 self._create_action_clients_for_device(device_id, namespace)
                 self._online_devices.add(device_key)
                 sclient = self.create_client(SerialCommand, f"/srv{namespace}/re_register_device")
-                t = threading.Thread(
+                threading.Thread(
                     target=self._send_re_register,
-                    args=(sclient, namespace),
+                    args=(sclient,),
                     daemon=True,
                     name=f"ROSDevice{self.device_id}_re_register_device_{namespace}",
-                )
-                self._background_threads.append(t)
-                t.start()
+                ).start()
             elif device_key not in self._online_devices:
                 # 设备重新上线
                 self.lab_logger().info(f"[Host Node] Device reconnected: {device_key}")
                 self._online_devices.add(device_key)
                 sclient = self.create_client(SerialCommand, f"/srv{namespace}/re_register_device")
-                t = threading.Thread(
+                threading.Thread(
                     target=self._send_re_register,
-                    args=(sclient, namespace),
+                    args=(sclient,),
                     daemon=True,
                     name=f"ROSDevice{self.device_id}_re_register_device_{namespace}",
-                )
-                self._background_threads.append(t)
-                t.start()
+                ).start()
 
         # 检测离线设备
         offline_devices = self._online_devices - current_devices
@@ -778,14 +707,13 @@ class HostNode(BaseROS2DeviceNode):
             raise ValueError(f"ActionClient {action_id} not found.")
 
         action_client: ActionClient = self._action_clients[action_id]
-
         # 遍历action_kwargs下的所有子dict，将"sample_uuid"的值赋给"sample_id"
         def assign_sample_id(obj):
             if isinstance(obj, dict):
                 if "sample_uuid" in obj:
                     obj["sample_id"] = obj["sample_uuid"]
                     obj.pop("sample_uuid")
-                for k, v in obj.items():
+                for k,v in obj.items():
                     if k != "unilabos_extra":
                         assign_sample_id(v)
             elif isinstance(obj, list):
@@ -816,7 +744,9 @@ 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_future = goal_handle.get_result_async()
-        goal_future.add_done_callback(lambda f: self.get_result_callback(item, action_id, f))
+        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:
@@ -1238,13 +1168,11 @@ 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 = http_client.resource_tree_get([data["uuid"]], data["with_children"])
+                http_req = self.bridges[-1].resource_tree_get([data["uuid"]], data["with_children"])
             elif "id" in data and data["id"].startswith("/"):
-                http_req = http_client.resource_get(data["id"], data["with_children"])
+                http_req = self.bridges[-1].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

@@ -12,10 +12,11 @@ from unilabos_msgs.srv import ResourceUpdate
 from unilabos.messages import *  # type: ignore  # protocol names
 from rclpy.action import ActionServer, ActionClient
 from rclpy.action.server import ServerGoalHandle
+from rclpy.callback_groups import ReentrantCallbackGroup
 from unilabos_msgs.srv._serial_command import SerialCommand_Request, SerialCommand_Response
 
 from unilabos.compile import action_protocol_generators
-from unilabos.resources.graphio import nested_dict_to_list
+from unilabos.resources.graphio import list_to_nested_dict, nested_dict_to_list
 from unilabos.ros.initialize_device import initialize_device_from_dict
 from unilabos.ros.msgs.message_converter import (
     get_action_type,
@@ -23,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.resources.resource_tracker import ResourceTreeSet, ResourceDictInstance
+from unilabos.ros.nodes.resource_tracker import ResourceTreeSet, ResourceDictInstance
 from unilabos.utils.type_check import get_result_info_str
 
 if TYPE_CHECKING:

unilabos/ros/nodes/resource_tracker.py

@@ -14,9 +14,9 @@ if TYPE_CHECKING:
 
 
 class ResourceDictPositionSize(BaseModel):
-    depth: float = Field(description="Depth", default=0.0)  # z
-    width: float = Field(description="Width", default=0.0)  # x
-    height: float = Field(description="Height", default=0.0)  # y
+    depth: float = Field(description="Depth", default=0.0)
+    width: float = Field(description="Width", default=0.0)
+    height: float = Field(description="Height", default=0.0)
 
 
 class ResourceDictPositionScale(BaseModel):
@@ -66,8 +66,8 @@ class ResourceDict(BaseModel):
     klass: str = Field(alias="class", description="Resource class name")
     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, eg: container liquid data")
-    extra: Dict[str, Any] = Field(description="Extra data, eg: slot index")
+    data: Dict[str, Any] = Field(description="Resource data")
+    extra: Dict[str, Any] = Field(description="Extra data")
 
     @field_serializer("parent_uuid")
     def _serialize_parent(self, parent_uuid: Optional["ResourceDict"]):
@@ -469,9 +469,9 @@ class ResourceTreeSet(object):
                 **res.config,
                 "name": res.name,
                 "type": res.config.get("type", plr_type),
-                "size_x": res.pose.size.width,
-                "size_y": res.pose.size.height,
-                "size_z": res.pose.size.depth,
+                "size_x": res.config.get("size_x", 0),
+                "size_y": res.config.get("size_y", 0),
+                "size_z": res.config.get("size_z", 0),
                 "location": {
                     "x": res.pose.position.x,
                     "y": res.pose.position.y,

unilabos/ros/utils/driver_creator.py

@@ -11,9 +11,10 @@ import traceback
 from abc import abstractmethod
 from typing import Type, Any, Dict, Optional, TypeVar, Generic, List
 
-from unilabos.resources.resource_tracker import DeviceNodeResourceTracker, ResourceTreeSet, ResourceDictInstance, \
+from unilabos.resources.graphio import nested_dict_to_list, resource_ulab_to_plr
+from unilabos.ros.nodes.resource_tracker import DeviceNodeResourceTracker, ResourceTreeSet, ResourceDictInstance, \
     ResourceTreeInstance
-from unilabos.utils import logger
+from unilabos.utils import logger, import_manager
 from unilabos.utils.cls_creator import create_instance_from_config
 
 # 定义泛型类型变量
@@ -134,7 +135,7 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
         Returns:
             处理后的数据
         """
-        from pylabrobot.resources import Resource
+        from pylabrobot.resources import Deck, Resource
 
         if states is None:
             states = {}

unilabos/workflow/common.py

@@ -1,547 +0,0 @@
-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):
-        # 将 source_port/target_port 映射为服务端期望的 source_handle_key/target_handle_key
-        source_handle_key = attrs.pop("source_port", "") or attrs.pop("source_handle_key", "")
-        target_handle_key = attrs.pop("target_port", "") or attrs.pop("target_handle_key", "")
-
-        edge = {
-            "source": source,
-            "target": target,
-            "source_node_uuid": source,
-            "target_node_uuid": target,
-            "source_handle_key": source_handle_key,
-            "source_handle_io": attrs.pop("source_handle_io", "source"),
-            "target_handle_key": target_handle_key,
-            "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 = {"uuid": 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["uuid"]) if str(n["uuid"]).isdigit() else n["uuid"])
-
-    # 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({"uuid": 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]],
-    action_resource_mapping: Optional[Dict[str, str]] = None,
-) -> List[Dict[str, Any]]:
-    """统一的数据重构函数，根据操作类型自动选择模板
-
-    Args:
-        data: 原始步骤数据列表
-        action_resource_mapping: action 到 resource_name 的映射字典，可选
-    """
-    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, action_resource_mapping)
-                refactored_data.extend(sub_data)
-            continue
-
-        # 获取模板名称
-        template_name = OPERATION_MAPPING.get(operation)
-        if not template_name:
-            # 自动推断模板类型
-            if operation.lower() in ["transfer", "incubation", "move_labware", "oscillation"]:
-                template_name = f"biomek-{operation}"
-            else:
-                template_name = f"{operation}Protocol"
-
-        # 获取 resource_name
-        resource_name = f"device.{operation.lower()}"
-        if action_resource_mapping:
-            resource_name = action_resource_mapping.get(operation, resource_name)
-
-        # 获取步骤编号，生成 name 字段
-        step_number = step.get("step_number")
-        name = f"Step {step_number}" if step_number is not None else None
-
-        # 创建步骤数据
-        step_data = {
-            "template_name": template_name,
-            "resource_name": resource_name,
-            "description": step.get("description", step.get("purpose", f"{operation} operation")),
-            "lab_node_type": "Device",
-            "param": step.get("parameters", step.get("action_args", {})),
-            "footer": f"{template_name}-{resource_name}",
-        }
-        if name:
-            step_data["name"] = name
-        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,
-    action_resource_mapping: Optional[Dict[str, str]] = None,
-) -> WorkflowGraph:
-    """统一的协议图构建函数，根据设备类型自动选择构建逻辑
-
-    Args:
-        labware_info: labware 信息字典
-        protocol_steps: 协议步骤列表
-        workstation_name: 工作站名称
-        action_resource_mapping: action 到 resource_name 的映射字典，可选
-    """
-    G = WorkflowGraph()
-    resource_last_writer = {}
-
-    protocol_steps = refactor_data(protocol_steps, action_resource_mapping)
-    # 有机化学&移液站协议图构建
-    WORKSTATION_ID = workstation_name
-
-    # 为所有labware创建资源节点
-    res_index = 0
-    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]
-
-        res_index += 1
-        G.add_node(
-            node_id,
-            template_name="create_resource",
-            resource_name="host_node",
-            name=f"Res {res_index}",
-            description=description,
-            lab_node_type=lab_node_type,
-            footer="create_resource-host_node",
-            param={
-                "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": "",
-            },
-        )
-        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("param", {})
-        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_name", 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_name", 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_handle_key") or edge.get("source_port")
-        tp = edge.get("target_handle_key") or edge.get("target_port")
-
-        # 记录到图里（保留原始端口信息）
-        G.add_edge(u, v, source_handle_key=sp, target_handle_key=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/convert_from_json.py

@@ -1,356 +0,0 @@
-"""
-JSON 工作流转换模块
-
-提供从多种 JSON 格式转换为统一工作流格式的功能。
-支持的格式：
-1. workflow/reagent 格式
-2. steps_info/labware_info 格式
-"""
-
-import json
-from os import PathLike
-from pathlib import Path
-from typing import Any, Dict, List, Optional, Set, Tuple, Union
-
-from unilabos.workflow.common import WorkflowGraph, build_protocol_graph
-from unilabos.registry.registry import lab_registry
-
-
-def get_action_handles(resource_name: str, template_name: str) -> Dict[str, List[str]]:
-    """
-    从 registry 获取指定设备和动作的 handles 配置
-
-    Args:
-        resource_name: 设备资源名称，如 "liquid_handler.prcxi"
-        template_name: 动作模板名称，如 "transfer_liquid"
-
-    Returns:
-        包含 source 和 target handler_keys 的字典:
-        {"source": ["sources_out", "targets_out", ...], "target": ["sources", "targets", ...]}
-    """
-    result = {"source": [], "target": []}
-
-    device_info = lab_registry.device_type_registry.get(resource_name, {})
-    if not device_info:
-        return result
-
-    action_mappings = device_info.get("class", {}).get("action_value_mappings", {})
-    action_config = action_mappings.get(template_name, {})
-    handles = action_config.get("handles", {})
-
-    if isinstance(handles, dict):
-        # 处理 input handles (作为 target)
-        for handle in handles.get("input", []):
-            handler_key = handle.get("handler_key", "")
-            if handler_key:
-                result["source"].append(handler_key)
-        # 处理 output handles (作为 source)
-        for handle in handles.get("output", []):
-            handler_key = handle.get("handler_key", "")
-            if handler_key:
-                result["target"].append(handler_key)
-
-    return result
-
-
-def validate_workflow_handles(graph: WorkflowGraph) -> Tuple[bool, List[str]]:
-    """
-    校验工作流图中所有边的句柄配置是否正确
-
-    Args:
-        graph: 工作流图对象
-
-    Returns:
-        (is_valid, errors): 是否有效，错误信息列表
-    """
-    errors = []
-    nodes = graph.nodes
-
-    for edge in graph.edges:
-        left_uuid = edge.get("source")
-        right_uuid = edge.get("target")
-        # target_handle_key是target, right的输入节点（入节点）
-        # source_handle_key是source, left的输出节点（出节点）
-        right_source_conn_key = edge.get("target_handle_key", "")
-        left_target_conn_key = edge.get("source_handle_key", "")
-
-        # 获取源节点和目标节点信息
-        left_node = nodes.get(left_uuid, {})
-        right_node = nodes.get(right_uuid, {})
-
-        left_res_name = left_node.get("resource_name", "")
-        left_template_name = left_node.get("template_name", "")
-        right_res_name = right_node.get("resource_name", "")
-        right_template_name = right_node.get("template_name", "")
-
-        # 获取源节点的 output handles
-        left_node_handles = get_action_handles(left_res_name, left_template_name)
-        target_valid_keys = left_node_handles.get("target", [])
-        target_valid_keys.append("ready")
-
-        # 获取目标节点的 input handles
-        right_node_handles = get_action_handles(right_res_name, right_template_name)
-        source_valid_keys = right_node_handles.get("source", [])
-        source_valid_keys.append("ready")
-
-        # 如果节点配置了 output handles，则 source_port 必须有效
-        if not right_source_conn_key:
-            node_name = left_node.get("name", left_uuid[:8])
-            errors.append(f"源节点 '{node_name}' 的 source_handle_key 为空，" f"应设置为: {source_valid_keys}")
-        elif right_source_conn_key not in source_valid_keys:
-            node_name = left_node.get("name", left_uuid[:8])
-            errors.append(
-                f"源节点 '{node_name}' 的 source 端点 '{right_source_conn_key}' 不存在，" f"支持的端点: {source_valid_keys}"
-            )
-
-        # 如果节点配置了 input handles，则 target_port 必须有效
-        if not left_target_conn_key:
-            node_name = right_node.get("name", right_uuid[:8])
-            errors.append(f"目标节点 '{node_name}' 的 target_handle_key 为空，" f"应设置为: {target_valid_keys}")
-        elif left_target_conn_key not in target_valid_keys:
-            node_name = right_node.get("name", right_uuid[:8])
-            errors.append(
-                f"目标节点 '{node_name}' 的 target 端点 '{left_target_conn_key}' 不存在，"
-                f"支持的端点: {target_valid_keys}"
-            )
-
-    return len(errors) == 0, errors
-
-
-# action 到 resource_name 的映射
-ACTION_RESOURCE_MAPPING: Dict[str, str] = {
-    # 生物实验操作
-    "transfer_liquid": "liquid_handler.prcxi",
-    "transfer": "liquid_handler.prcxi",
-    "incubation": "incubator.prcxi",
-    "move_labware": "labware_mover.prcxi",
-    "oscillation": "shaker.prcxi",
-    # 有机化学操作
-    "HeatChillToTemp": "heatchill.chemputer",
-    "StopHeatChill": "heatchill.chemputer",
-    "StartHeatChill": "heatchill.chemputer",
-    "HeatChill": "heatchill.chemputer",
-    "Dissolve": "stirrer.chemputer",
-    "Transfer": "liquid_handler.chemputer",
-    "Evaporate": "rotavap.chemputer",
-    "Recrystallize": "reactor.chemputer",
-    "Filter": "filter.chemputer",
-    "Dry": "dryer.chemputer",
-    "Add": "liquid_handler.chemputer",
-}
-
-
-def normalize_steps(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-    """
-    将不同格式的步骤数据规范化为统一格式
-
-    支持的输入格式：
-    - action + parameters
-    - action + action_args
-    - operation + parameters
-
-    Args:
-        data: 原始步骤数据列表
-
-    Returns:
-        规范化后的步骤列表，格式为 [{"action": str, "parameters": dict, "description": str?, "step_number": int?}, ...]
-    """
-    normalized = []
-    for idx, step in enumerate(data):
-        # 获取动作名称（支持 action 或 operation 字段）
-        action = step.get("action") or step.get("operation")
-        if not action:
-            continue
-
-        # 获取参数（支持 parameters 或 action_args 字段）
-        raw_params = step.get("parameters") or step.get("action_args") or {}
-        params = dict(raw_params)
-
-        # 规范化 source/target -> sources/targets
-        if "source" in raw_params and "sources" not in raw_params:
-            params["sources"] = raw_params["source"]
-        if "target" in raw_params and "targets" not in raw_params:
-            params["targets"] = raw_params["target"]
-
-        # 获取描述（支持 description 或 purpose 字段）
-        description = step.get("description") or step.get("purpose")
-
-        # 获取步骤编号（优先使用原始数据中的 step_number，否则使用索引+1）
-        step_number = step.get("step_number", idx + 1)
-
-        step_dict = {"action": action, "parameters": params, "step_number": step_number}
-        if description:
-            step_dict["description"] = description
-
-        normalized.append(step_dict)
-
-    return normalized
-
-
-def normalize_labware(data: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
-    """
-    将不同格式的 labware 数据规范化为统一的字典格式
-
-    支持的输入格式：
-    - reagent_name + material_name + positions
-    - name + labware + slot
-
-    Args:
-        data: 原始 labware 数据列表
-
-    Returns:
-        规范化后的 labware 字典，格式为 {name: {"slot": int, "labware": str, "well": list, "type": str, "role": str, "name": str}, ...}
-    """
-    labware = {}
-    for item in data:
-        # 获取 key 名称（优先使用 reagent_name，其次是 material_name 或 name）
-        reagent_name = item.get("reagent_name")
-        key = reagent_name or item.get("material_name") or item.get("name")
-        if not key:
-            continue
-
-        key = str(key)
-
-        # 处理重复 key，自动添加后缀
-        idx = 1
-        original_key = key
-        while key in labware:
-            idx += 1
-            key = f"{original_key}_{idx}"
-
-        labware[key] = {
-            "slot": item.get("positions") or item.get("slot"),
-            "labware": item.get("material_name") or item.get("labware"),
-            "well": item.get("well", []),
-            "type": item.get("type", "reagent"),
-            "role": item.get("role", ""),
-            "name": key,
-        }
-
-    return labware
-
-
-def convert_from_json(
-    data: Union[str, PathLike, Dict[str, Any]],
-    workstation_name: str = "PRCXi",
-    validate: bool = True,
-) -> WorkflowGraph:
-    """
-    从 JSON 数据或文件转换为 WorkflowGraph
-
-    支持的 JSON 格式：
-    1. {"workflow": [...], "reagent": {...}} - 直接格式
-    2. {"steps_info": [...], "labware_info": [...]} - 需要规范化的格式
-
-    Args:
-        data: JSON 文件路径、字典数据、或 JSON 字符串
-        workstation_name: 工作站名称，默认 "PRCXi"
-        validate: 是否校验句柄配置，默认 True
-
-    Returns:
-        WorkflowGraph: 构建好的工作流图
-
-    Raises:
-        ValueError: 不支持的 JSON 格式 或 句柄校验失败
-        FileNotFoundError: 文件不存在
-        json.JSONDecodeError: JSON 解析失败
-    """
-    # 处理输入数据
-    if isinstance(data, (str, PathLike)):
-        path = Path(data)
-        if path.exists():
-            with path.open("r", encoding="utf-8") as fp:
-                json_data = json.load(fp)
-        elif isinstance(data, str):
-            # 尝试作为 JSON 字符串解析
-            json_data = json.loads(data)
-        else:
-            raise FileNotFoundError(f"文件不存在: {data}")
-    elif isinstance(data, dict):
-        json_data = data
-    else:
-        raise TypeError(f"不支持的数据类型: {type(data)}")
-
-    # 根据格式解析数据
-    if "workflow" in json_data and "reagent" in json_data:
-        # 格式1: workflow/reagent（已经是规范格式）
-        protocol_steps = json_data["workflow"]
-        labware_info = json_data["reagent"]
-    elif "steps_info" in json_data and "labware_info" in json_data:
-        # 格式2: steps_info/labware_info（需要规范化）
-        protocol_steps = normalize_steps(json_data["steps_info"])
-        labware_info = normalize_labware(json_data["labware_info"])
-    elif "steps" in json_data and "labware" in json_data:
-        # 格式3: steps/labware（另一种常见格式）
-        protocol_steps = normalize_steps(json_data["steps"])
-        if isinstance(json_data["labware"], list):
-            labware_info = normalize_labware(json_data["labware"])
-        else:
-            labware_info = json_data["labware"]
-    else:
-        raise ValueError(
-            "不支持的 JSON 格式。支持的格式：\n"
-            "1. {'workflow': [...], 'reagent': {...}}\n"
-            "2. {'steps_info': [...], 'labware_info': [...]}\n"
-            "3. {'steps': [...], 'labware': [...]}"
-        )
-
-    # 构建工作流图
-    graph = build_protocol_graph(
-        labware_info=labware_info,
-        protocol_steps=protocol_steps,
-        workstation_name=workstation_name,
-        action_resource_mapping=ACTION_RESOURCE_MAPPING,
-    )
-
-    # 校验句柄配置
-    if validate:
-        is_valid, errors = validate_workflow_handles(graph)
-        if not is_valid:
-            import warnings
-
-            for error in errors:
-                warnings.warn(f"句柄校验警告: {error}")
-
-    return graph
-
-
-def convert_json_to_node_link(
-    data: Union[str, PathLike, Dict[str, Any]],
-    workstation_name: str = "PRCXi",
-) -> Dict[str, Any]:
-    """
-    将 JSON 数据转换为 node-link 格式的字典
-
-    Args:
-        data: JSON 文件路径、字典数据、或 JSON 字符串
-        workstation_name: 工作站名称，默认 "PRCXi"
-
-    Returns:
-        Dict: node-link 格式的工作流数据
-    """
-    graph = convert_from_json(data, workstation_name)
-    return graph.to_node_link_dict()
-
-
-def convert_json_to_workflow_list(
-    data: Union[str, PathLike, Dict[str, Any]],
-    workstation_name: str = "PRCXi",
-) -> List[Dict[str, Any]]:
-    """
-    将 JSON 数据转换为工作流列表格式
-
-    Args:
-        data: JSON 文件路径、字典数据、或 JSON 字符串
-        workstation_name: 工作站名称，默认 "PRCXi"
-
-    Returns:
-        List: 工作流节点列表
-    """
-    graph = convert_from_json(data, workstation_name)
-    return graph.to_dict()
-
-
-# 为了向后兼容，保留下划线前缀的别名
-_normalize_steps = normalize_steps
-_normalize_labware = normalize_labware

unilabos/workflow/from_python_script.py

@@ -1,241 +0,0 @@
-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

@@ -1,131 +0,0 @@
-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.15</version>
+  <version>0.10.14</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>