Update organic syn station.

update "laiyu" missing init file.

fix "laiyu" missing init file.

fix "🐛 fix"

🐛 fix: config file is overwrited by default args even if not be set.

mix

修改了mix，仿真流程报错问题

README.md

@@ -39,7 +39,9 @@ Uni-Lab-OS recommends using `mamba` for environment management. Choose the appro
 
 ```bash
 # Create new environment
-mamba create -n unilab uni-lab::unilabos -c robostack-staging -c conda-forge
+mamba create -n unilab python=3.11.11
+mamba activate unilab
+mamba install -n unilab uni-lab::unilabos -c robostack-staging -c conda-forge
 ```
 
 ## Install Dev Uni-Lab-OS

README_zh.md

@@ -41,7 +41,9 @@ Uni-Lab-OS 建议使用 `mamba` 管理环境。根据您的操作系统选择适
 
 ```bash
 # 创建新环境
-mamba create -n unilab uni-lab::unilabos -c robostack-staging -c conda-forge
+mamba create -n unilab python=3.11.11
+mamba activate unilab
+mamba install -n unilab uni-lab::unilabos -c robostack-staging -c conda-forge
 ```
 
 2. 安装开发版 Uni-Lab-OS:

docs/user_guide/installation.md

@@ -317,45 +317,6 @@ unilab --help
 
 如果所有命令都正常输出，说明开发环境配置成功！
 
-### 开发工具推荐
-
-#### IDE
-
-- **PyCharm Professional**: 强大的 Python IDE，支持远程调试
-- **VS Code**: 轻量级，配合 Python 扩展使用
-- **Vim/Emacs**: 适合终端开发
-
-#### 推荐的 VS Code 扩展
-
-- Python
-- Pylance
-- ROS
-- URDF
-- YAML
-
-#### 调试工具
-
-```bash
-# 安装调试工具
-pip install ipdb pytest pytest-cov -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
-
-# 代码质量检查
-pip install black flake8 mypy -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
-```
-
-### 设置 pre-commit 钩子（可选）
-
-```bash
-# 安装 pre-commit
-pip install pre-commit -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
-
-# 设置钩子
-pre-commit install
-
-# 手动运行检查
-pre-commit run --all-files
-```
-
 ---
 
 ## 验证安装

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(

unilabos/app/main.py

@@ -49,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(
@@ -155,26 +153,16 @@ def parse_args():
         default=False,
         help="Complete registry information",
     )
-
-    # label
-    workflow_parser = subparsers.add_parser(
-        "workflow_upload",
-        help="Upload workflow from xdl/json/python files",
-    )
-    workflow_parser.add_argument("-t", "--labeltype", default="singlepoint", type=str,
-                              help="QM calculation type, support 'singlepoint', 'optimize' and 'dimer' currently")
     return parser
 
 
 def main():
     """主函数"""
     # 解析命令行参数
-    args = parse_args()
-    convert_argv_dashes_to_underscores(args)
-    args_dict = vars(args.parse_args())
-
-    # 显示启动横幅
-    print_unilab_banner(args_dict)
+    parser = parse_args()
+    convert_argv_dashes_to_underscores(parser)
+    args = parser.parse_args()
+    args_dict = vars(args)
 
     # 环境检查 - 检查并自动安装必需的包 (可选)
     if not args_dict.get("skip_env_check", False):
@@ -233,17 +221,18 @@ def main():
         logger.info(f"Log level set to '{BasicConfig.log_level}' from config file.")
     configure_logger(loglevel=BasicConfig.log_level, working_dir=working_dir)
 
-    if args_dict["addr"] == "test":
-        print_status("使用测试环境地址", "info")
-        HTTPConfig.remote_addr = "https://uni-lab.test.bohrium.com/api/v1"
-    elif args_dict["addr"] == "uat":
-        print_status("使用uat环境地址", "info")
-        HTTPConfig.remote_addr = "https://uni-lab.uat.bohrium.com/api/v1"
-    elif args_dict["addr"] == "local":
-        print_status("使用本地环境地址", "info")
-        HTTPConfig.remote_addr = "http://127.0.0.1:48197/api/v1"
-    else:
-        HTTPConfig.remote_addr = args_dict.get("addr", "")
+    if args.addr != parser.get_default("addr"):
+        if args.addr == "test":
+            print_status("使用测试环境地址", "info")
+            HTTPConfig.remote_addr = "https://uni-lab.test.bohrium.com/api/v1"
+        elif args.addr == "uat":
+            print_status("使用uat环境地址", "info")
+            HTTPConfig.remote_addr = "https://uni-lab.uat.bohrium.com/api/v1"
+        elif args.addr == "local":
+            print_status("使用本地环境地址", "info")
+            HTTPConfig.remote_addr = "http://127.0.0.1:48197/api/v1"
+        else:
+            HTTPConfig.remote_addr = args.addr
 
     # 设置BasicConfig参数
     if args_dict.get("ak", ""):
@@ -252,18 +241,7 @@ def main():
     if args_dict.get("sk", ""):
         BasicConfig.sk = args_dict.get("sk", "")
         print_status("传入了sk参数，优先采用传入参数！", "info")
-    BasicConfig.working_dir = working_dir
 
-    # 显示启动横幅
-    print_unilab_banner(args_dict)
-
-    #####################################
-    ######## 启动设备接入端(主入口) ########
-    #####################################
-    launch(args_dict)
-
-
-def launch(args_dict: Dict[str, Any]):
     # 使用远程资源启动
     if args_dict["use_remote_resource"]:
         print_status("使用远程资源启动", "info")
@@ -278,6 +256,7 @@ def launch(args_dict: Dict[str, Any]):
 
     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)
@@ -301,6 +280,9 @@ def launch(args_dict: Dict[str, Any]):
     from unilabos.resources.graphio import modify_to_backend_format
     from unilabos.ros.nodes.resource_tracker import ResourceTreeSet, ResourceDict
 
+    # 显示启动横幅
+    print_unilab_banner(args_dict)
+
     # 注册表
     lab_registry = build_registry(
         args_dict["registry_path"], args_dict.get("complete_registry", False), args_dict["upload_registry"]
@@ -347,6 +329,10 @@ def launch(args_dict: Dict[str, Any]):
     for ind, i in enumerate(resource_edge_info[::-1]):
         source_node: ResourceDict = nodes[i["source"]]
         target_node: ResourceDict = nodes[i["target"]]
+        if "sourceHandle" not in source_node:
+            continue
+        if "targetHandle" not in target_node:
+            continue
         source_handle = i["sourceHandle"]
         target_handle = i["targetHandle"]
         source_handler_keys = [

unilabos/config/config.py

@@ -41,7 +41,7 @@ class WSConfig:
 
 # HTTP配置
 class HTTPConfig:
-    remote_addr = "http://127.0.0.1:48197/api/v1"
+    remote_addr = "https://uni-lab.bohrium.com/api/v1"
 
 
 # ROS配置

unilabos/devices/liquid_handling/liquid_handler_abstract.py

@@ -988,6 +988,18 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
                 dis_vols = [float(dis_vols)]
             else:
                 dis_vols = [float(v) for v in dis_vols]
+
+        # 统一混合次数为标量，防止数组/列表与 int 比较时报错
+        if mix_times is not None and not isinstance(mix_times, (int, float)):
+            try:
+                mix_times = mix_times[0] if len(mix_times) > 0 else None
+            except Exception:
+                try:
+                    mix_times = next(iter(mix_times))
+                except Exception:
+                    pass
+        if mix_times is not None:
+            mix_times = int(mix_times)
             
             # 识别传输模式
             num_sources = len(sources)

unilabos/devices/liquid_handling/prcxi/prcxi.py

@@ -5,6 +5,7 @@ import json
 import os
 import socket
 import time
+import uuid
 from typing import Any, List, Dict, Optional, Tuple, TypedDict, Union, Sequence, Iterator, Literal
 
 from pylabrobot.liquid_handling import (
@@ -856,7 +857,30 @@ class PRCXI9300Api:
 
     def _raw_request(self, payload: str) -> str:
         if self.debug:
-            return " "
+            # 调试/仿真模式下直接返回可解析的模拟 JSON，避免后续 json.loads 报错
+            try:
+                req = json.loads(payload)
+                method = req.get("MethodName")
+            except Exception:
+                method = None
+
+            data: Any = True
+            if method in {"AddSolution"}:
+                data = str(uuid.uuid4())
+            elif method in {"AddWorkTabletMatrix", "AddWorkTabletMatrix2"}:
+                data = {"Success": True, "Message": "debug mock"}
+            elif method in {"GetErrorCode"}:
+                data = ""
+            elif method in {"RemoveErrorCodet", "Reset", "Start", "LoadSolution", "Pause", "Resume", "Stop"}:
+                data = True
+            elif method in {"GetStepStateList", "GetStepStatus", "GetStepState"}:
+                data = []
+            elif method in {"GetLocation"}:
+                data = {"X": 0, "Y": 0, "Z": 0}
+            elif method in {"GetResetStatus"}:
+                data = False
+
+            return json.dumps({"Success": True, "Msg": "debug mock", "Data": data})
         with contextlib.closing(socket.socket()) as sock:
             sock.settimeout(self.timeout)
             sock.connect((self.host, self.port))

unilabos/registry/devices/bioyond_dispensing_station.yaml

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

unilabos/resources/graphio.py

@@ -284,10 +284,18 @@ def modify_to_backend_format(data: list[dict[str, Any]]) -> list[dict[str, Any]]
             edge["sourceHandle"] = port[source]
         elif "source_port" in edge:
             edge["sourceHandle"] = edge.pop("source_port")
+        else:
+            typ = edge.get("type")
+            if typ == "communication":
+                continue
         if target in port:
             edge["targetHandle"] = port[target]
         elif "target_port" in edge:
             edge["targetHandle"] = edge.pop("target_port")
+        else:
+            typ = edge.get("type")
+            if typ == "communication":
+                continue
         edge["id"] = f"reactflow__edge-{source}-{edge['sourceHandle']}-{target}-{edge['targetHandle']}"
         for key in ["source_port", "target_port"]:
             if key in edge:

unilabos/ros/nodes/presets/host_node.py

@@ -1157,12 +1157,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/test/experiments/Grignard_flow_batchreact_single_pumpvalve.json

@@ -69,7 +69,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "device",
-            "class": "syringepump.runze",
+            "class": "syringe_pump_with_valve.runze.SY03B-T08",
             "position": {
                 "x": 620.6111111111111,
                 "y": 171,
@@ -93,7 +93,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 430.4087301587302,
                 "y": 428,
@@ -117,7 +117,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 295.36944444444447,
                 "y": 428,
@@ -141,7 +141,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 165.36944444444444,
                 "y": 428,
@@ -165,7 +165,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 165.36944444444444,
                 "y": 428,
@@ -189,7 +189,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 35,
                 "y": 428,
@@ -213,7 +213,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 698.1111111111111,
                 "y": 428,
@@ -255,7 +255,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "device",
-            "class": "syringepump.runze",
+            "class": "syringe_pump_with_valve.runze.SY03B-T08",
             "position": {
                 "x": 1195.611507936508,
                 "y": 686,
@@ -279,7 +279,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 1587.703373015873,
                 "y": 1172.5,
@@ -299,7 +299,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "device",
-            "class": "separator_controller",
+            "class": "separator.homemade",
             "position": {
                 "x": 1624.4027777777778,
                 "y": 665.5,
@@ -320,7 +320,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 1614.404365079365,
                 "y": 948,
@@ -340,7 +340,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 1915.7035714285714,
                 "y": 665.5,
@@ -360,7 +360,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 1785.7035714285714,
                 "y": 665.5,
@@ -384,7 +384,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 2054.0650793650793,
                 "y": 665.5,
@@ -408,7 +408,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "device",
-            "class": "syringepump.runze",
+            "class": "syringe_pump_with_valve.runze.SY03B-T08",
             "position": {
                 "x": 1630.6527777777778,
                 "y": 448.5,
@@ -432,7 +432,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "device",
-            "class": "rotavap",
+            "class": "rotavap.one",
             "position": {
                 "x": 1339.7031746031746,
                 "y": 968.5,
@@ -453,7 +453,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 1339.7031746031746,
                 "y": 1152,
@@ -473,7 +473,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 909.722619047619,
                 "y": 948,
@@ -493,7 +493,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 867.972619047619,
                 "y": 1152,
@@ -513,7 +513,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 742.722619047619,
                 "y": 948,
@@ -533,7 +533,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 1206.722619047619,
                 "y": 948,
@@ -553,7 +553,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "container",
-            "class": null,
+            "class": "container",
             "position": {
                 "x": 1148.222619047619,
                 "y": 1152,
@@ -573,7 +573,7 @@
             "children": [],
             "parent": "YugongStation",
             "type": "device",
-            "class": "syringepump.runze",
+            "class": "syringe_pump_with_valve.runze.SY03B-T08",
             "position": {
                 "x": 1469.7031746031746,
                 "y": 968.5,

unilabos/test/workflow/merge_workflow.py

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

unilabos/workflow/common.py

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

unilabos/workflow/from_labwares_and_steps.py

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

unilabos/workflow/from_python_script.py

@@ -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}