no opcua installation on macos

fix possible crash

fix deck & host_node

set liquid with tube

add test_resource_schema

fix test resource schema

registry update & workflow update

add test mode

support description & tags upload

fix config load

fix log

add registry name & add always free

correct config organic synthesis

Adapt to new scheduler, sampels, and edge upload format (#230)

* add sample_material

* adapt to new samples sys

* fix pump transfer. fix resource update when protocol & ros callback

* Adapt to new scheduler.

Feat/samples (#229)

* add sample_material

* adapt to new samples sys

adapt to new samples sys

adapt to new edge format

workflow upload & prcxi transfer liquid

lh liquid

speed up registry load

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

fix upload workflow json

Author: Xuwznln

.conda/base/recipe.yaml

@@ -46,7 +46,9 @@ requirements:
   - jinja2
   - requests
   - uvicorn
-  - opcua
+  - if: not osx
+    then:
+    - opcua
   - pyserial
   - pandas
   - pymodbus

docs/user_guide/best_practice.md

@@ -452,8 +452,9 @@ unilab --ak your_ak --sk your_sk -g test/experiments/mock_devices/mock_all.json
 **操作步骤：**
 
 1. 将两个 `container` 拖拽到 `workstation` 中
-2. 将 `virtual_transfer_pump` 拖拽到 `workstation` 中
-3. 在画布上连接它们（建立父子关系）
+2. 将 `virtual_multiway_valve` 拖拽到 `workstation` 中
+3. 将 `virtual_transfer_pump` 拖拽到 `workstation` 中
+4. 在画布上连接它们（建立父子关系）
 
 ![设备连接](image/links.png)
 

docs/user_guide/image/links.png

@@ -171,6 +171,12 @@ def parse_args():
         action="store_true",
         help="Disable sending update feedback to server",
     )
+    parser.add_argument(
+        "--test_mode",
+        action="store_true",
+        default=False,
+        help="Test mode: all actions simulate execution and return mock results without running real hardware",
+    )
     # workflow upload subcommand
     workflow_parser = subparsers.add_parser(
         "workflow_upload",
@@ -204,6 +210,12 @@ def parse_args():
         default=False,
         help="Whether to publish the workflow (default: False)",
     )
+    workflow_parser.add_argument(
+        "--description",
+        type=str,
+        default="",
+        help="Workflow description, used when publishing the workflow",
+    )
     return parser
 
 
@@ -231,52 +243,60 @@ def main():
     # 加载配置文件，优先加载config，然后从env读取
     config_path = args_dict.get("config")
 
-    if check_mode:
-        args_dict["working_dir"] = os.path.abspath(os.getcwd())
-    # 当 skip_env_check 时，默认使用当前目录作为 working_dir
-    if skip_env_check and not args_dict.get("working_dir") and not config_path:
+    # === 解析 working_dir ===
+    # 规则1: working_dir 传入 → 检测 unilabos_data 子目录，已是则不修改
+    # 规则2: 仅 config_path 传入 → 用其父目录作为 working_dir
+    # 规则4: 两者都传入 → 各用各的，但 working_dir 仍做 unilabos_data 子目录检测
+    raw_working_dir = args_dict.get("working_dir")
+    if raw_working_dir:
+        working_dir = os.path.abspath(raw_working_dir)
+    elif config_path and os.path.exists(config_path):
+        working_dir = os.path.dirname(os.path.abspath(config_path))
+    else:
         working_dir = os.path.abspath(os.getcwd())
-        print_status(f"跳过环境检查模式：使用当前目录作为工作目录 {working_dir}", "info")
-        # 检查当前目录是否有 local_config.py
-        local_config_in_cwd = os.path.join(working_dir, "local_config.py")
-        if os.path.exists(local_config_in_cwd):
-            config_path = local_config_in_cwd
+
+    # unilabos_data 子目录自动检测
+    if os.path.basename(working_dir) != "unilabos_data":
+        unilabos_data_sub = os.path.join(working_dir, "unilabos_data")
+        if os.path.isdir(unilabos_data_sub):
+            working_dir = unilabos_data_sub
+        elif not raw_working_dir and not (config_path and os.path.exists(config_path)):
+            # 未显式指定路径，默认使用 cwd/unilabos_data
+            working_dir = os.path.abspath(os.path.join(os.getcwd(), "unilabos_data"))
+
+    # === 解析 config_path ===
+    if config_path and not os.path.exists(config_path):
+        # config_path 传入但不存在，尝试在 working_dir 中查找
+        candidate = os.path.join(working_dir, "local_config.py")
+        if os.path.exists(candidate):
+            config_path = candidate
+            print_status(f"在工作目录中发现配置文件: {config_path}", "info")
+        else:
+            print_status(
+                f"配置文件 {config_path} 不存在，工作目录 {working_dir} 中也未找到 local_config.py，"
+                f"请通过 --config 传入 local_config.py 文件路径",
+                "error",
+            )
+            os._exit(1)
+    elif not config_path:
+        # 规则3: 未传入 config_path，尝试 working_dir/local_config.py
+        candidate = os.path.join(working_dir, "local_config.py")
+        if os.path.exists(candidate):
+            config_path = candidate
             print_status(f"发现本地配置文件: {config_path}", "info")
         else:
             print_status(f"未指定config路径，可通过 --config 传入 local_config.py 文件路径", "info")
-    elif os.getcwd().endswith("unilabos_data"):
-        working_dir = os.path.abspath(os.getcwd())
-    else:
-        working_dir = os.path.abspath(os.path.join(os.getcwd(), "unilabos_data"))
-
-    if args_dict.get("working_dir"):
-        working_dir = args_dict.get("working_dir", "")
-        if config_path and not os.path.exists(config_path):
-            config_path = os.path.join(working_dir, "local_config.py")
-            if not os.path.exists(config_path):
-                print_status(
-                    f"当前工作目录 {working_dir} 未找到local_config.py，请通过 --config 传入 local_config.py 文件路径",
-                    "error",
+            print_status(f"您是否为第一次使用？并将当前路径 {working_dir} 作为工作目录？ (Y/n)", "info")
+            if check_mode or input() != "n":
+                os.makedirs(working_dir, exist_ok=True)
+                config_path = os.path.join(working_dir, "local_config.py")
+                shutil.copy(
+                    os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", "example_config.py"),
+                    config_path,
                 )
+                print_status(f"已创建 local_config.py 路径： {config_path}", "info")
+            else:
                 os._exit(1)
-    elif config_path and os.path.exists(config_path):
-        working_dir = os.path.dirname(config_path)
-    elif os.path.exists(working_dir) and os.path.exists(os.path.join(working_dir, "local_config.py")):
-        config_path = os.path.join(working_dir, "local_config.py")
-    elif not skip_env_check and not config_path and (
-        not os.path.exists(working_dir) or not os.path.exists(os.path.join(working_dir, "local_config.py"))
-    ):
-        print_status(f"未指定config路径，可通过 --config 传入 local_config.py 文件路径", "info")
-        print_status(f"您是否为第一次使用？并将当前路径 {working_dir} 作为工作目录？ (Y/n)", "info")
-        if input() != "n":
-            os.makedirs(working_dir, exist_ok=True)
-            config_path = os.path.join(working_dir, "local_config.py")
-            shutil.copy(
-                os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", "example_config.py"), config_path
-            )
-            print_status(f"已创建 local_config.py 路径： {config_path}", "info")
-        else:
-            os._exit(1)
 
     # 加载配置文件 (check_mode 跳过)
     print_status(f"当前工作目录为 {working_dir}", "info")
@@ -288,7 +308,9 @@ def main():
 
     if hasattr(BasicConfig, "log_level"):
         logger.info(f"Log level set to '{BasicConfig.log_level}' from config file.")
-    configure_logger(loglevel=BasicConfig.log_level, working_dir=working_dir)
+    file_path = configure_logger(loglevel=BasicConfig.log_level, working_dir=working_dir)
+    if file_path is not None:
+        logger.info(f"[LOG_FILE] {file_path}")
 
     if args.addr != parser.get_default("addr"):
         if args.addr == "test":
@@ -332,6 +354,9 @@ def main():
     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.test_mode = args_dict.get("test_mode", False)
+    if BasicConfig.test_mode:
+        print_status("启用测试模式：所有动作将模拟执行，不调用真实硬件", "warning")
     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])

unilabos/app/main.py

@@ -54,6 +54,7 @@ class JobAddReq(BaseModel):
     action_type: str = Field(
         examples=["unilabos_msgs.action._str_single_input.StrSingleInput"], description="action type", default=""
     )
+    sample_material: dict = Field(examples=[{"string": "string"}], description="sample uuid to material uuid")
     action_args: dict = Field(examples=[{"string": "string"}], description="action arguments", default_factory=dict)
     task_id: str = Field(examples=["task_id"], description="task uuid (auto-generated if empty)", default="")
     job_id: str = Field(examples=["job_id"], description="goal uuid (auto-generated if empty)", default="")

unilabos/app/model.py

@@ -38,9 +38,9 @@ def register_devices_and_resources(lab_registry, gather_only=False) -> Optional[
             response = http_client.resource_registry({"resources": list(devices_to_register.values())})
             cost_time = time.time() - start_time
             if response.status_code in [200, 201]:
-                logger.info(f"[UniLab Register] 成功注册 {len(devices_to_register)} 个设备 {cost_time}ms")
+                logger.info(f"[UniLab Register] 成功注册 {len(devices_to_register)} 个设备 {cost_time}s")
             else:
-                logger.error(f"[UniLab Register] 设备注册失败: {response.status_code}, {response.text} {cost_time}ms")
+                logger.error(f"[UniLab Register] 设备注册失败: {response.status_code}, {response.text} {cost_time}s")
         except Exception as e:
             logger.error(f"[UniLab Register] 设备注册异常: {e}")
 
@@ -51,9 +51,9 @@ def register_devices_and_resources(lab_registry, gather_only=False) -> Optional[
             response = http_client.resource_registry({"resources": list(resources_to_register.values())})
             cost_time = time.time() - start_time
             if response.status_code in [200, 201]:
-                logger.info(f"[UniLab Register] 成功注册 {len(resources_to_register)} 个资源 {cost_time}ms")
+                logger.info(f"[UniLab Register] 成功注册 {len(resources_to_register)} 个资源 {cost_time}s")
             else:
-                logger.error(f"[UniLab Register] 资源注册失败: {response.status_code}, {response.text} {cost_time}ms")
+                logger.error(f"[UniLab Register] 资源注册失败: {response.status_code}, {response.text} {cost_time}s")
         except Exception as e:
             logger.error(f"[UniLab Register] 资源注册异常: {e}")
 

unilabos/app/register.py

@@ -343,9 +343,10 @@ def workflow_import(
         edges: List[Dict[str, Any]],
         tags: Optional[List[str]] = None,
         published: bool = False,
+        description: str = "",
     ) -> Dict[str, Any]:
         """
-        导入工作流到服务器
+        导入工作流到服务器，如果 published 为 True，则额外发起发布请求
 
         Args:
             name: 工作流名称（顶层）
@@ -355,6 +356,7 @@ def workflow_import(
             edges: 工作流边列表
             tags: 工作流标签列表，默认为空列表
             published: 是否发布工作流，默认为False
+            description: 工作流描述，发布时使用
 
         Returns:
             Dict: API响应数据，包含 code 和 data (uuid, name)
@@ -367,7 +369,6 @@ def workflow_import(
                 "nodes": nodes,
                 "edges": edges,
                 "tags": tags if tags is not None else [],
-                "published": published,
             },
         }
         # 保存请求到文件
@@ -388,11 +389,51 @@ def workflow_import(
             res = response.json()
             if "code" in res and res["code"] != 0:
                 logger.error(f"导入工作流失败: {response.text}")
+                return res
+            # 导入成功后，如果需要发布则额外发起发布请求
+            if published:
+                imported_uuid = res.get("data", {}).get("uuid", workflow_uuid)
+                publish_res = self.workflow_publish(imported_uuid, description)
+                res["publish_result"] = publish_res
             return res
         else:
             logger.error(f"导入工作流失败: {response.status_code}, {response.text}")
             return {"code": response.status_code, "message": response.text}
 
+    def workflow_publish(self, workflow_uuid: str, description: str = "") -> Dict[str, Any]:
+        """
+        发布工作流
+
+        Args:
+            workflow_uuid: 工作流UUID
+            description: 工作流描述
+
+        Returns:
+            Dict: API响应数据
+        """
+        payload = {
+            "uuid": workflow_uuid,
+            "description": description,
+            "published": True,
+        }
+        logger.info(f"正在发布工作流: {workflow_uuid}")
+        response = requests.patch(
+            f"{self.remote_addr}/lab/workflow/owner",
+            json=payload,
+            headers={"Authorization": f"Lab {self.auth}"},
+            timeout=60,
+        )
+        if response.status_code == 200:
+            res = response.json()
+            if "code" in res and res["code"] != 0:
+                logger.error(f"发布工作流失败: {response.text}")
+            else:
+                logger.info(f"工作流发布成功: {workflow_uuid}")
+            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/client.py

@@ -327,6 +327,7 @@ def job_add(req: JobAddReq) -> JobData:
             queue_item,
             action_type=action_type,
             action_kwargs=action_args,
+            sample_material=req.sample_material,
             server_info=server_info,
         )