new registry sys

exp. support with add device

.conda/base/recipe.yaml

@@ -3,7 +3,7 @@
 
 package:
   name: unilabos
-  version: 0.10.17
+  version: 0.10.18
 
 source:
   path: ../../unilabos
@@ -46,13 +46,15 @@ requirements:
   - jinja2
   - requests
   - uvicorn
-  - opcua # [not osx]
+  - if: not osx
+    then:
+    - opcua
   - pyserial
   - pandas
   - pymodbus
   - matplotlib
   - pylibftdi
-  - uni-lab::unilabos-env ==0.10.17
+  - uni-lab::unilabos-env ==0.10.18
 
 about:
   repository: https://github.com/deepmodeling/Uni-Lab-OS

.conda/environment/recipe.yaml

@@ -2,7 +2,7 @@
 
 package:
   name: unilabos-env
-  version: 0.10.17
+  version: 0.10.18
 
 build:
   noarch: generic

.conda/full/recipe.yaml

@@ -3,7 +3,7 @@
 
 package:
   name: unilabos-full
-  version: 0.10.17
+  version: 0.10.18
 
 build:
   noarch: generic
@@ -11,7 +11,7 @@ build:
 requirements:
   run:
   # Base unilabos package (includes unilabos-env)
-  - uni-lab::unilabos ==0.10.17
+  - uni-lab::unilabos ==0.10.18
   # Documentation tools
   - sphinx
   - sphinx_rtd_theme

.github/workflows/ci-check.yml

@@ -49,7 +49,7 @@ jobs:
           uv pip uninstall enum34 || echo enum34 not installed, skipping
           uv pip install .
 
-      - name: Run check mode (complete_registry)
+      - name: Run check mode (AST registry validation)
         run: |
           call conda activate check-env
           echo Running check mode...

.gitignore

@@ -5,6 +5,7 @@ output/
 unilabos_data/
 pyrightconfig.json
 .cursorignore
+device_package*/
 ## Python
 
 # Byte-compiled / optimized / DLL files

AGENTS.md

@@ -0,0 +1,87 @@
+# AGENTS.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+Also follow the monorepo-level rules in `../AGENTS.md`.
+
+## Build & Development
+
+```bash
+# Install in editable mode (requires mamba env with python 3.11)
+pip install -e .
+uv pip install -r unilabos/utils/requirements.txt
+
+# Run with a device graph
+unilab --graph <graph.json> --config <config.py> --backend ros
+unilab --graph <graph.json> --config <config.py> --backend simple  # no ROS2 needed
+
+# Common CLI flags
+unilab --app_bridges websocket fastapi    # communication bridges
+unilab --test_mode                        # simulate hardware, no real execution
+unilab --check_mode                       # CI validation of registry imports
+unilab --skip_env_check                   # skip auto-install of dependencies
+unilab --visual rviz|web|disable          # visualization mode
+unilab --is_slave                         # run as slave node
+
+# Workflow upload subcommand
+unilab workflow_upload -f <workflow.json> -n <name> --tags tag1 tag2
+
+# Tests
+pytest tests/                              # all tests
+pytest tests/resources/test_resourcetreeset.py  # single test file
+pytest tests/resources/test_resourcetreeset.py::TestClassName::test_method  # single test
+```
+
+## Architecture
+
+### Startup Flow
+
+`unilab` CLI → `unilabos/app/main.py:main()` → loads config → builds registry → reads device graph (JSON/GraphML) → starts backend thread (ROS2/simple) → starts FastAPI web server + WebSocket client.
+
+### Core Layers
+
+**Registry** (`unilabos/registry/`): Singleton `Registry` class discovers and catalogs all device types, resource types, and communication devices from YAML definitions. Device types live in `registry/devices/*.yaml`, resources in `registry/resources/`, comms in `registry/device_comms/`. The registry resolves class paths to actual Python classes via `utils/import_manager.py`.
+
+**Resource Tracking** (`unilabos/resources/resource_tracker.py`): Pydantic-based `ResourceDict` → `ResourceDictInstance` → `ResourceTreeSet` hierarchy. `ResourceTreeSet` is the canonical in-memory representation of all devices and resources, used throughout the system. Graph I/O is in `resources/graphio.py` (reads JSON/GraphML device topology files into `nx.Graph` + `ResourceTreeSet`).
+
+**Device Drivers** (`unilabos/devices/`): 30+ hardware drivers organized by device type (liquid_handling, hplc, balance, arm, etc.). Each driver is a Python class that gets wrapped by `ros/device_node_wrapper.py:ros2_device_node()` to become a ROS2 node with publishers, subscribers, and action servers.
+
+**ROS2 Layer** (`unilabos/ros/`): `device_node_wrapper.py` dynamically wraps any device class into `ROS2DeviceNode` (defined in `ros/nodes/base_device_node.py`). Preset node types in `ros/nodes/presets/` include `host_node`, `controller_node`, `workstation`, `serial_node`, `camera`. Messages use custom `unilabos_msgs` (pre-built, distributed via releases).
+
+**Protocol Compilation** (`unilabos/compile/`): 20+ protocol compilers (add, centrifuge, dissolve, filter, heatchill, stir, pump, etc.) that transform YAML protocol definitions into executable sequences.
+
+**Communication** (`unilabos/device_comms/`): Hardware communication adapters — OPC-UA client, Modbus PLC, RPC, and a universal driver. `app/communication.py` provides a factory pattern for WebSocket client connections to the cloud.
+
+**Web/API** (`unilabos/app/web/`): FastAPI server with REST API (`api.py`), Jinja2 template pages (`pages.py`), and HTTP client for cloud communication (`client.py`). Runs on port 8002 by default.
+
+### Configuration System
+
+- **Config classes** in `unilabos/config/config.py`: `BasicConfig`, `WSConfig`, `HTTPConfig`, `ROSConfig` — all class-level attributes, loaded from Python config files
+- Config files are `.py` files with matching class names (see `config/example_config.py`)
+- Environment variables override with prefix `UNILABOS_` (e.g., `UNILABOS_BASICCONFIG_PORT=9000`)
+- Device topology defined in graph files (JSON with node-link format, or GraphML)
+
+### Key Data Flow
+
+1. Graph file → `graphio.read_node_link_json()` → `(nx.Graph, ResourceTreeSet, resource_links)`
+2. `ResourceTreeSet` + `Registry` → `initialize_device.initialize_device_from_dict()` → `ROS2DeviceNode` instances
+3. Device nodes communicate via ROS2 topics/actions or direct Python calls (simple backend)
+4. Cloud sync via WebSocket (`app/ws_client.py`) and HTTP (`app/web/client.py`)
+
+### Test Data
+
+Example device graphs and experiment configs are in `unilabos/test/experiments/` (not `tests/`). Registry test fixtures in `unilabos/test/registry/`.
+
+## Code Conventions
+
+- Code comments and log messages in simplified Chinese
+- Python 3.11+, type hints expected
+- Pydantic models for data validation (`resource_tracker.py`)
+- Singleton pattern via `@singleton` decorator (`utils/decorator.py`)
+- Dynamic class loading via `utils/import_manager.py` — device classes resolved at runtime from registry YAML paths
+- CLI argument dashes auto-converted to underscores for consistency
+
+## Licensing
+
+- Framework code: GPL-3.0
+- Device drivers (`unilabos/devices/`): DP Technology Proprietary License — do not redistribute

CLAUDE.md

@@ -0,0 +1,4 @@
+
+Please follow the rules defined in:
+
+@AGENTS.md

docs/developer_guide/add_device.md

@@ -15,6 +15,9 @@ Python 类设备驱动在完成注册表后可以直接在 Uni-Lab 中使用，
 **示例：**
 
 ```python
+from unilabos.registry.decorators import device, topic_config
+
+@device(id="mock_gripper", category=["gripper"], description="Mock Gripper")
 class MockGripper:
     def __init__(self):
         self._position: float = 0.0
@@ -23,19 +26,23 @@ class MockGripper:
         self._status = "Idle"
 
     @property
+    @topic_config()  # 添加 @topic_config 才会定时广播
     def position(self) -> float:
         return self._position
 
     @property
+    @topic_config()
     def velocity(self) -> float:
         return self._velocity
 
     @property
+    @topic_config()
     def torque(self) -> float:
         return self._torque
 
-    # 会被自动识别的设备属性，接入 Uni-Lab 时会定时对外广播
+    # 使用 @topic_config 装饰的属性，接入 Uni-Lab 时会定时对外广播
     @property
+    @topic_config(period=2.0)  # 可自定义发布周期
     def status(self) -> str:
         return self._status
 
@@ -149,7 +156,7 @@ my_device: # 设备唯一标识符
 
 系统会自动分析您的 Python 驱动类并生成：
 
-- `status_types`：从 `@property` 装饰的方法自动识别状态属性
+- `status_types`：从 `@topic_config` 装饰的 `@property` 或方法自动识别状态属性
 - `action_value_mappings`：从类方法自动生成动作映射
 - `init_param_schema`：从 `__init__` 方法分析初始化参数
 - `schema`：前端显示用的属性类型定义
@@ -179,7 +186,9 @@ Uni-Lab 设备驱动是一个 Python 类，需要遵循以下结构：
 
 ```python
 from typing import Dict, Any
+from unilabos.registry.decorators import device, topic_config
 
+@device(id="my_device", category=["general"], description="My Device")
 class MyDevice:
     """设备类文档字符串
 
@@ -198,8 +207,9 @@ class MyDevice:
         # 初始化硬件连接
 
     @property
+    @topic_config()  # 必须添加 @topic_config 才会广播
     def status(self) -> str:
-        """设备状态（会自动广播）"""
+        """设备状态（通过 @topic_config 广播）"""
         return self._status
 
     def my_action(self, param: float) -> Dict[str, Any]:
@@ -217,34 +227,61 @@ class MyDevice:
 
 ## 状态属性 vs 动作方法
 
-### 状态属性（@property）
+### 状态属性（@property + @topic_config）
 
-状态属性会被自动识别并定期广播：
+状态属性需要同时使用 `@property` 和 `@topic_config` 装饰器才会被识别并定期广播：
 
 ```python
+from unilabos.registry.decorators import topic_config
+
 @property
+@topic_config()  # 必须添加，否则不会广播
 def temperature(self) -> float:
     """当前温度"""
     return self._read_temperature()
 
 @property
+@topic_config(period=2.0)  # 可自定义发布周期（秒）
 def status(self) -> str:
     """设备状态: idle, running, error"""
     return self._status
 
 @property
+@topic_config(name="ready")  # 可自定义发布名称
 def is_ready(self) -> bool:
     """设备是否就绪"""
     return self._status == "idle"
 ```
 
+也可以使用普通方法（非 @property）配合 `@topic_config`：
+
+```python
+@topic_config(period=10.0)
+def get_sensor_data(self) -> Dict[str, float]:
+    """获取传感器数据（get_ 前缀会自动去除，发布名为 sensor_data）"""
+    return {"temp": self._temp, "humidity": self._humidity}
+```
+
+**`@topic_config` 参数**:
+
+| 参数 | 类型 | 默认值 | 说明 |
+|------|------|--------|------|
+| `period` | float | 5.0 | 发布周期（秒） |
+| `print_publish` | bool | 节点默认 | 是否打印发布日志 |
+| `qos` | int | 10 | QoS 深度 |
+| `name` | str | None | 自定义发布名称 |
+
+**发布名称优先级**：`@topic_config(name=...)` > `get_` 前缀去除 > 方法名
+
 **特点**:
 
-- 使用`@property`装饰器
-- 只读，不能有参数
-- 自动添加到注册表的`status_types`
+- 必须使用 `@topic_config` 装饰器
+- 支持 `@property` 和普通方法
+- 添加到注册表的 `status_types`
 - 定期发布到 ROS2 topic
 
+> **⚠️ 重要：** 仅有 `@property` 装饰器而没有 `@topic_config` 的属性**不会**被广播。这是一个 Breaking Change。
+
 ### 动作方法
 
 动作方法是设备可以执行的操作：
@@ -497,6 +534,7 @@ class LiquidHandler:
         self._status = "idle"
 
     @property
+    @topic_config()
     def status(self) -> str:
         return self._status
 
@@ -886,7 +924,52 @@ class MyDevice:
 
 ## 最佳实践
 
-### 1. 类型注解
+### 1. 使用 `@device` 装饰器标识设备类
+
+```python
+from unilabos.registry.decorators import device
+
+@device(id="my_device", category=["heating"], description="My Heating Device", icon="heater.webp")
+class MyDevice:
+    ...
+```
+
+- `id`：设备唯一标识符，用于注册表匹配
+- `category`：分类列表，前端用于分组显示
+- `description`：设备描述
+- `icon`：图标文件名（可选）
+
+### 2. 使用 `@topic_config` 声明需要广播的状态
+
+```python
+from unilabos.registry.decorators import topic_config
+
+# ✓ @property + @topic_config → 会广播
+@property
+@topic_config(period=2.0)
+def temperature(self) -> float:
+    return self._temp
+
+# ✓ 普通方法 + @topic_config → 会广播（get_ 前缀自动去除）
+@topic_config(period=10.0)
+def get_sensor_data(self) -> Dict[str, float]:
+    return {"temp": self._temp}
+
+# ✓ 使用 name 参数自定义发布名称
+@property
+@topic_config(name="ready")
+def is_ready(self) -> bool:
+    return self._status == "idle"
+
+# ✗ 仅有 @property，没有 @topic_config → 不会广播
+@property
+def internal_state(self) -> str:
+    return self._state
+```
+
+> **注意：** 与 `@property` 连用时，`@topic_config` 必须放在 `@property` 下面。
+
+### 3. 类型注解
 
 ```python
 from typing import Dict, Any, Optional, List
@@ -901,7 +984,7 @@ def method(
     pass
 ```
 
-### 2. 文档字符串
+### 4. 文档字符串
 
 ```python
 def method(self, param: float) -> Dict[str, Any]:
@@ -923,7 +1006,7 @@ def method(self, param: float) -> Dict[str, Any]:
     pass
 ```
 
-### 3. 配置验证
+### 5. 配置验证
 
 ```python
 def __init__(self, config: Dict[str, Any]):
@@ -937,7 +1020,7 @@ def __init__(self, config: Dict[str, Any]):
     self.baudrate = config['baudrate']
 ```
 
-### 4. 资源清理
+### 6. 资源清理
 
 ```python
 def __del__(self):
@@ -946,7 +1029,7 @@ def __del__(self):
         self.connection.close()
 ```
 
-### 5. 设计前端友好的返回值
+### 7. 设计前端友好的返回值
 
 **记住：返回值会直接显示在 Web 界面**
 

docs/developer_guide/add_registry.md

@@ -422,18 +422,20 @@ placeholder_keys:
 
 ### status_types
 
-系统会扫描你的 Python 类，从状态方法（property 或 get\_方法）自动生成这部分：
+系统会扫描你的 Python 类，从带有 `@topic_config` 装饰器的 `@property` 或方法自动生成这部分：
 
 ```yaml
 status_types:
-  current_temperature: float # 从 get_current_temperature() 或 @property current_temperature
-  is_heating: bool # 从 get_is_heating() 或 @property is_heating
-  status: str # 从 get_status() 或 @property status
+  current_temperature: float # 从 @topic_config 装饰的 @property 或方法
+  is_heating: bool
+  status: str
 ```
 
 **注意事项**：
 
-- 系统会查找所有 `get_` 开头的方法和 `@property` 装饰的属性
+- 仅有带 `@topic_config` 装饰器的 `@property` 或方法才会被识别为状态属性
+- 没有 `@topic_config` 的 `@property` 不会生成 status_types，也不会广播
+- `get_` 前缀的方法名会自动去除前缀（如 `get_temperature` → `temperature`）
 - 类型会自动转成相应的类型（如 `str`、`float`、`bool`）
 - 如果类型是 `Any`、`None` 或未知的，默认使用 `String`
 
@@ -537,11 +539,13 @@ class AdvancedLiquidHandler:
         self._temperature = 25.0
 
     @property
+    @topic_config()
     def status(self) -> str:
         """设备状态"""
         return self._status
 
     @property
+    @topic_config()
     def temperature(self) -> float:
         """当前温度"""
         return self._temperature
@@ -809,21 +813,23 @@ my_temperature_controller:
 你的设备类需要符合以下要求：
 
 ```python
-from unilabos.common.device_base import DeviceBase
+from unilabos.registry.decorators import device, topic_config
 
-class MyDevice(DeviceBase):
+@device(id="my_device", category=["temperature"], description="My Device")
+class MyDevice:
     def __init__(self, config):
         """初始化，参数会自动分析到 init_param_schema.config"""
-        super().__init__(config)
         self.port = config.get('port', '/dev/ttyUSB0')
 
-    # 状态方法（会自动生成到 status_types）
+    # 状态方法（必须添加 @topic_config 才会生成到 status_types 并广播）
     @property
+    @topic_config()
     def status(self):
         """返回设备状态"""
         return "idle"
 
     @property
+    @topic_config()
     def temperature(self):
         """返回当前温度"""
         return 25.0
@@ -1039,7 +1045,34 @@ resource.type      # "resource"
 
 ### 代码规范
 
-1. **始终使用类型注解**
+1. **使用 `@device` 装饰器标识设备类**
+
+```python
+from unilabos.registry.decorators import device
+
+@device(id="my_device", category=["heating"], description="My Device")
+class MyDevice:
+    ...
+```
+
+2. **使用 `@topic_config` 声明广播属性**
+
+```python
+from unilabos.registry.decorators import topic_config
+
+# ✓ 需要广播的状态属性
+@property
+@topic_config(period=2.0)
+def temperature(self) -> float:
+    return self._temp
+
+# ✗ 仅有 @property 不会广播
+@property
+def internal_counter(self) -> int:
+    return self._counter
+```
+
+3. **始终使用类型注解**
 
 ```python
 # ✓ 好
@@ -1051,7 +1084,7 @@ def method(self, resource, device):
     pass
 ```
 
-2. **提供有意义的参数名**
+4. **提供有意义的参数名**
 
 ```python
 # ✓ 好 - 清晰的参数名
@@ -1063,7 +1096,7 @@ def transfer(self, r1: ResourceSlot, r2: ResourceSlot):
     pass
 ```
 
-3. **使用 Optional 表示可选参数**
+5. **使用 Optional 表示可选参数**
 
 ```python
 from typing import Optional
@@ -1076,7 +1109,7 @@ def method(
     pass
 ```
 
-4. **添加详细的文档字符串**
+6. **添加详细的文档字符串**
 
 ```python
 def method(
@@ -1096,13 +1129,13 @@ def method(
     pass
 ```
 
-5. **方法命名规范**
+7. **方法命名规范**
 
-   - 状态方法使用 `@property` 装饰器或 `get_` 前缀
+   - 状态方法使用 `@property` + `@topic_config` 装饰器，或普通方法 + `@topic_config`
    - 动作方法使用动词开头
    - 保持命名清晰、一致
 
-6. **完善的错误处理**
+8. **完善的错误处理**
    - 实现完善的错误处理
    - 添加日志记录
    - 提供有意义的错误信息

docs/developer_guide/networking_overview.md

@@ -221,10 +221,10 @@ Laboratory A    Laboratory B
 
 ```bash
 # 实验室A
-unilab --ak your_ak --sk your_sk --upload_registry --use_remote_resource
+unilab --ak your_ak --sk your_sk --upload_registry
 
 # 实验室B
-unilab --ak your_ak --sk your_sk --upload_registry --use_remote_resource
+unilab --ak your_ak --sk your_sk --upload_registry
 ```
 
 ---

docs/user_guide/launch.md

@@ -22,7 +22,6 @@ options:
   --is_slave            Run the backend as slave node (without host privileges).
   --slave_no_host       Skip waiting for host service in slave mode
   --upload_registry     Upload registry information when starting unilab
-  --use_remote_resource Use remote resources when starting unilab
   --config CONFIG       Configuration file path, supports .py format Python config files
   --port PORT           Port for web service information page
   --disable_browser     Disable opening information page on startup
@@ -85,7 +84,7 @@ Uni-Lab 的启动过程分为以下几个阶段：
 支持两种方式：
 
 - **本地文件**：使用 `-g` 指定图谱文件（支持 JSON 和 GraphML 格式）
-- **远程资源**：使用 `--use_remote_resource` 从云端获取
+- **远程资源**：不指定本地文件即可
 
 ### 7. 注册表构建
 
@@ -196,7 +195,7 @@ unilab --config path/to/your/config.py
 unilab --ak your_ak --sk your_sk -g path/to/graph.json --upload_registry
 
 # 使用远程资源启动
-unilab --ak your_ak --sk your_sk --use_remote_resource
+unilab --ak your_ak --sk your_sk
 
 # 更新注册表
 unilab --ak your_ak --sk your_sk --complete_registry

recipes/msgs/recipe.yaml

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

recipes/unilabos/recipe.yaml

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

setup.py

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

unilabos/__init__.py

@@ -1 +1 @@
-__version__ = "0.10.17"
+__version__ = "0.10.18"

unilabos/app/main.py

@@ -1,8 +1,10 @@
 import argparse
 import asyncio
 import os
+import platform
 import shutil
 import signal
+import subprocess
 import sys
 import threading
 import time
@@ -24,6 +26,84 @@ from unilabos.config.config import load_config, BasicConfig, HTTPConfig
 _restart_requested: bool = False
 _restart_reason: str = ""
 
+RESTART_EXIT_CODE = 42
+
+
+def _build_child_argv():
+    """Build sys.argv for child process, stripping supervisor-only arguments."""
+    result = []
+    skip_next = False
+    for arg in sys.argv:
+        if skip_next:
+            skip_next = False
+            continue
+        if arg in ("--restart_mode", "--restart-mode"):
+            continue
+        if arg in ("--auto_restart_count", "--auto-restart-count"):
+            skip_next = True
+            continue
+        if arg.startswith("--auto_restart_count=") or arg.startswith("--auto-restart-count="):
+            continue
+        result.append(arg)
+    return result
+
+
+def _run_as_supervisor(max_restarts: int):
+    """
+    Supervisor process that spawns and monitors child processes.
+
+    Similar to Uvicorn's --reload: the supervisor itself does no heavy work,
+    it only launches the real process as a child and restarts it when the child
+    exits with RESTART_EXIT_CODE.
+    """
+    child_argv = [sys.executable] + _build_child_argv()
+    restart_count = 0
+
+    print_status(
+        f"[Supervisor] Restart mode enabled (max restarts: {max_restarts}), "
+        f"child command: {' '.join(child_argv)}",
+        "info",
+    )
+
+    while True:
+        print_status(
+            f"[Supervisor] Launching process (restart {restart_count}/{max_restarts})...",
+            "info",
+        )
+
+        try:
+            process = subprocess.Popen(child_argv)
+            exit_code = process.wait()
+        except KeyboardInterrupt:
+            print_status("[Supervisor] Interrupted, terminating child process...", "info")
+            process.terminate()
+            try:
+                process.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+            sys.exit(1)
+
+        if exit_code == RESTART_EXIT_CODE:
+            restart_count += 1
+            if restart_count > max_restarts:
+                print_status(
+                    f"[Supervisor] Maximum restart count ({max_restarts}) reached, exiting",
+                    "warning",
+                )
+                sys.exit(1)
+            print_status(
+                f"[Supervisor] Child requested restart ({restart_count}/{max_restarts}), restarting in 2s...",
+                "info",
+            )
+            time.sleep(2)
+        else:
+            if exit_code != 0:
+                print_status(f"[Supervisor] Child exited with code {exit_code}", "warning")
+            else:
+                print_status("[Supervisor] Child exited normally", "info")
+            sys.exit(exit_code)
+
 
 def load_config_from_file(config_path):
     if config_path is None:
@@ -65,6 +145,13 @@ def parse_args():
         action="append",
         help="Path to the registry directory",
     )
+    parser.add_argument(
+        "--devices",
+        type=str,
+        default=None,
+        action="append",
+        help="Path to Python code directory for AST-based device/resource scanning",
+    )
     parser.add_argument(
         "--working_dir",
         type=str,
@@ -154,12 +241,6 @@ def parse_args():
         action="store_true",
         help="Skip environment dependency check on startup",
     )
-    parser.add_argument(
-        "--complete_registry",
-        action="store_true",
-        default=False,
-        help="Complete registry information",
-    )
     parser.add_argument(
         "--check_mode",
         action="store_true",
@@ -171,6 +252,30 @@ 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",
+    )
+    parser.add_argument(
+        "--extra_resource",
+        action="store_true",
+        default=False,
+        help="Load extra lab_ prefixed labware resources (529 auto-generated definitions from lab_resources.py)",
+    )
+    parser.add_argument(
+        "--restart_mode",
+        action="store_true",
+        default=False,
+        help="Enable supervisor mode: automatically restart the process when triggered via WebSocket",
+    )
+    parser.add_argument(
+        "--auto_restart_count",
+        type=int,
+        default=500,
+        help="Maximum number of automatic restarts in restart mode (default: 500)",
+    )
     # workflow upload subcommand
     workflow_parser = subparsers.add_parser(
         "workflow_upload",
@@ -204,6 +309,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
 
 
@@ -215,6 +326,11 @@ def main():
     args = parser.parse_args()
     args_dict = vars(args)
 
+    # Supervisor mode: spawn child processes and monitor for restart
+    if args_dict.get("restart_mode", False):
+        _run_as_supervisor(args_dict.get("auto_restart_count", 5))
+        return
+
     # 环境检查 - 检查并自动安装必需的包 (可选)
     skip_env_check = args_dict.get("skip_env_check", False)
     check_mode = args_dict.get("check_mode", False)
@@ -231,52 +347,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 +412,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,8 +458,14 @@ 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.extra_resource = args_dict.get("extra_resource", False)
+    if BasicConfig.extra_resource:
+        print_status("启用额外资源加载：将加载lab_开头的labware资源定义", "info")
     BasicConfig.communication_protocol = "websocket"
-    machine_name = os.popen("hostname").read().strip()
+    machine_name = platform.node()
     machine_name = "".join([c if c.isalnum() or c == "_" else "_" for c in machine_name])
     BasicConfig.machine_name = machine_name
     BasicConfig.vis_2d_enable = args_dict["2d_vis"]
@@ -356,22 +488,30 @@ def main():
     # 显示启动横幅
     print_unilab_banner(args_dict)
 
-    # 注册表 - check_mode 时强制启用 complete_registry
-    complete_registry = args_dict.get("complete_registry", False) or check_mode
-    lab_registry = build_registry(args_dict["registry_path"], complete_registry, BasicConfig.upload_registry)
+    # Step 0: AST 分析优先 + YAML 注册表加载
+    # check_mode 和 upload_registry 都会执行实际 import 验证
+    devices_dirs = args_dict.get("devices", None)
+    lab_registry = build_registry(
+        registry_paths=args_dict["registry_path"],
+        devices_dirs=devices_dirs,
+        upload_registry=BasicConfig.upload_registry,
+        check_mode=check_mode,
+    )
 
-    # Check mode: complete_registry 完成后直接退出，git diff 检测由 CI workflow 执行
+    # Check mode: 注册表验证完成后直接退出
     if check_mode:
-        print_status("Check mode: complete_registry 完成，退出", "info")
+        device_count = len(lab_registry.device_type_registry)
+        resource_count = len(lab_registry.resource_type_registry)
+        print_status(f"Check mode: 注册表验证完成 ({device_count} 设备, {resource_count} 资源)，退出", "info")
         os._exit(0)
 
+    # Step 1: 上传全部注册表到服务端，同步保存到 unilabos_data
     if BasicConfig.upload_registry:
-        # 设备注册到服务端 - 需要 ak 和 sk
         if BasicConfig.ak and BasicConfig.sk:
-            print_status("开始注册设备到服务端...", "info")
+            # print_status("开始注册设备到服务端...", "info")
             try:
                 register_devices_and_resources(lab_registry)
-                print_status("设备注册完成", "info")
+                # print_status("设备注册完成", "info")
             except Exception as e:
                 print_status(f"设备注册失败: {e}", "error")
         else:
@@ -456,7 +596,7 @@ def main():
             continue
 
     # 如果从远端获取了物料信息，则与本地物料进行同步
-    if request_startup_json and "nodes" in request_startup_json:
+    if file_path is not None and request_startup_json and "nodes" in request_startup_json:
         print_status("开始同步远端物料到本地...", "info")
         remote_tree_set = ResourceTreeSet.from_raw_dict_list(request_startup_json["nodes"])
         resource_tree_set.merge_remote_resources(remote_tree_set)
@@ -553,6 +693,10 @@ def main():
             open_browser=not args_dict["disable_browser"],
             port=BasicConfig.port,
         )
+        if restart_requested:
+            print_status("[Main] Restart requested, cleaning up...", "info")
+            cleanup_for_restart()
+            os._exit(RESTART_EXIT_CODE)
 
 
 if __name__ == "__main__":

unilabos/app/register.py

@@ -1,60 +1,83 @@
 import json
 import time
-from typing import Optional, Tuple, Dict, Any
+from typing import Any, Dict, Optional, Tuple
 
 from unilabos.utils.log import logger
 from unilabos.utils.type_check import TypeEncoder
 
+try:
+    import orjson
+
+    def _normalize_device(info: dict) -> dict:
+        """Serialize via orjson to strip non-JSON types (type objects etc.)."""
+        return orjson.loads(orjson.dumps(info, default=str))
+except ImportError:
+    def _normalize_device(info: dict) -> dict:
+        return json.loads(json.dumps(info, ensure_ascii=False, cls=TypeEncoder))
+
 
 def register_devices_and_resources(lab_registry, gather_only=False) -> Optional[Tuple[Dict[str, Any], Dict[str, Any]]]:
     """
     注册设备和资源到服务器（仅支持HTTP）
     """
 
-    # 注册资源信息 - 使用HTTP方式
     from unilabos.app.web.client import http_client
 
     logger.info("[UniLab Register] 开始注册设备和资源...")
 
-    # 注册设备信息
     devices_to_register = {}
     for device_info in lab_registry.obtain_registry_device_info():
-        devices_to_register[device_info["id"]] = json.loads(
-            json.dumps(device_info, ensure_ascii=False, cls=TypeEncoder)
-        )
-        logger.debug(f"[UniLab Register] 收集设备: {device_info['id']}")
+        devices_to_register[device_info["id"]] = _normalize_device(device_info)
+        logger.trace(f"[UniLab Register] 收集设备: {device_info['id']}")
 
     resources_to_register = {}
     for resource_info in lab_registry.obtain_registry_resource_info():
         resources_to_register[resource_info["id"]] = resource_info
-        logger.debug(f"[UniLab Register] 收集资源: {resource_info['id']}")
+        logger.trace(f"[UniLab Register] 收集资源: {resource_info['id']}")
 
     if gather_only:
         return devices_to_register, resources_to_register
-    # 注册设备
+
     if devices_to_register:
         try:
             start_time = time.time()
-            response = http_client.resource_registry({"resources": list(devices_to_register.values())})
+            response = http_client.resource_registry(
+                {"resources": list(devices_to_register.values())},
+                tag="device_registry",
+            )
             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")
+            res_data = response.json() if response.status_code == 200 else {}
+            skipped = res_data.get("data", {}).get("skipped", False)
+            if skipped:
+                logger.info(
+                    f"[UniLab Register] 设备注册跳过（内容未变化）"
+                    f" {len(devices_to_register)} 个 {cost_time:.3f}s"
+                )
+            elif response.status_code in [200, 201]:
+                logger.info(f"[UniLab Register] 成功注册 {len(devices_to_register)} 个设备 {cost_time:.3f}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:.3f}s")
         except Exception as e:
             logger.error(f"[UniLab Register] 设备注册异常: {e}")
 
-    # 注册资源
     if resources_to_register:
         try:
             start_time = time.time()
-            response = http_client.resource_registry({"resources": list(resources_to_register.values())})
+            response = http_client.resource_registry(
+                {"resources": list(resources_to_register.values())},
+                tag="resource_registry",
+            )
             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")
+            res_data = response.json() if response.status_code == 200 else {}
+            skipped = res_data.get("data", {}).get("skipped", False)
+            if skipped:
+                logger.info(
+                    f"[UniLab Register] 资源注册跳过（内容未变化）"
+                    f" {len(resources_to_register)} 个 {cost_time:.3f}s"
+                )
+            elif response.status_code in [200, 201]:
+                logger.info(f"[UniLab Register] 成功注册 {len(resources_to_register)} 个资源 {cost_time:.3f}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:.3f}s")
         except Exception as e:
             logger.error(f"[UniLab Register] 资源注册异常: {e}")
-
-    logger.info("[UniLab Register] 设备和资源注册完成.")

unilabos/app/web/api.py

@@ -1052,7 +1052,7 @@ async def handle_file_import(websocket: WebSocket, request_data: dict):
                                 "result": {},
                                 "schema": lab_registry._generate_unilab_json_command_schema(v["args"], k),
                                 "goal_default": {i["name"]: i["default"] for i in v["args"]},
-                                "handles": [],
+                                "handles": {},
                             }
                             # 不生成已配置action的动作
                             for k, v in enhanced_info["action_methods"].items()
@@ -1340,5 +1340,5 @@ def setup_api_routes(app):
     # 启动广播任务
     @app.on_event("startup")
     async def startup_event():
-        asyncio.create_task(broadcast_device_status())
-        asyncio.create_task(broadcast_status_page_data())
+        asyncio.create_task(broadcast_device_status(), name="web-api-startup-device")
+        asyncio.create_task(broadcast_status_page_data(), name="web-api-startup-status")

unilabos/app/web/client.py

@@ -3,11 +3,30 @@ HTTP客户端模块
 
 提供与远程服务器通信的客户端功能，只有host需要用
 """
-
+import gzip
 import json
 import os
 from typing import List, Dict, Any, Optional
 
+try:
+    import orjson as _json_fast
+
+    def _fast_dumps(obj, **kwargs) -> bytes:
+        return _json_fast.dumps(obj, option=_json_fast.OPT_NON_STR_KEYS, default=str)
+
+    def _fast_dumps_pretty(obj, **kwargs) -> bytes:
+        return _json_fast.dumps(
+            obj, option=_json_fast.OPT_NON_STR_KEYS | _json_fast.OPT_INDENT_2, default=str,
+        )
+except ImportError:
+    _json_fast = None  # type: ignore[assignment]
+
+    def _fast_dumps(obj, **kwargs) -> bytes:
+        return json.dumps(obj, ensure_ascii=False, default=str).encode("utf-8")
+
+    def _fast_dumps_pretty(obj, **kwargs) -> bytes:
+        return json.dumps(obj, indent=2, ensure_ascii=False, default=str).encode("utf-8")
+
 import requests
 from unilabos.resources.resource_tracker import ResourceTreeSet
 from unilabos.utils.log import info
@@ -280,22 +299,54 @@ class HTTPClient:
             )
         return response
 
-    def resource_registry(self, registry_data: Dict[str, Any] | List[Dict[str, Any]]) -> requests.Response:
+    def resource_registry(
+        self, registry_data: Dict[str, Any] | List[Dict[str, Any]], tag: str = "registry",
+    ) -> requests.Response:
         """
-        注册资源到服务器
+        注册资源到服务器，同步保存请求/响应到 unilabos_data
 
         Args:
             registry_data: 注册表数据，格式为 {resource_id: resource_info} / [{resource_info}]
+            tag: 保存文件的标签后缀 (如 "device_registry" / "resource_registry")
 
         Returns:
             Response: API响应对象
         """
+        # 序列化一次，同时用于保存和发送
+        json_bytes = _fast_dumps(registry_data)
+
+        # 保存请求数据到 unilabos_data
+        req_path = os.path.join(BasicConfig.working_dir, f"req_{tag}_upload.json")
+        try:
+            os.makedirs(BasicConfig.working_dir, exist_ok=True)
+            with open(req_path, "wb") as f:
+                f.write(_fast_dumps_pretty(registry_data))
+            logger.trace(f"注册表请求数据已保存: {req_path}")
+        except Exception as e:
+            logger.warning(f"保存注册表请求数据失败: {e}")
+
+        compressed_body = gzip.compress(json_bytes)
+        headers = {
+            "Authorization": f"Lab {self.auth}",
+            "Content-Type": "application/json",
+            "Content-Encoding": "gzip",
+        }
         response = requests.post(
             f"{self.remote_addr}/lab/resource",
-            json=registry_data,
-            headers={"Authorization": f"Lab {self.auth}"},
+            data=compressed_body,
+            headers=headers,
             timeout=30,
         )
+
+        # 保存响应数据到 unilabos_data
+        res_path = os.path.join(BasicConfig.working_dir, f"res_{tag}_upload.json")
+        try:
+            with open(res_path, "w", encoding="utf-8") as f:
+                f.write(f"{response.status_code}\n{response.text}")
+            logger.trace(f"注册表响应数据已保存: {res_path}")
+        except Exception as e:
+            logger.warning(f"保存注册表响应数据失败: {e}")
+
         if response.status_code not in [200, 201]:
             logger.error(f"注册资源失败: {response.status_code}, {response.text}")
         if response.status_code == 200:
@@ -343,9 +394,10 @@ class HTTPClient:
         edges: List[Dict[str, Any]],
         tags: Optional[List[str]] = None,
         published: bool = False,
+        description: str = "",
     ) -> Dict[str, Any]:
         """
-        导入工作流到服务器
+        导入工作流到服务器，如果 published 为 True，则额外发起发布请求
 
         Args:
             name: 工作流名称（顶层）
@@ -355,6 +407,7 @@ class HTTPClient:
             edges: 工作流边列表
             tags: 工作流标签列表，默认为空列表
             published: 是否发布工作流，默认为False
+            description: 工作流描述，发布时使用
 
         Returns:
             Dict: API响应数据，包含 code 和 data (uuid, name)
@@ -367,7 +420,6 @@ class HTTPClient:
                 "nodes": nodes,
                 "edges": edges,
                 "tags": tags if tags is not None else [],
-                "published": published,
             },
         }
         # 保存请求到文件
@@ -388,11 +440,51 @@ class HTTPClient:
             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/server.py

@@ -86,7 +86,7 @@ def setup_server() -> FastAPI:
     # 设置页面路由
     try:
         setup_web_pages(pages)
-        info("[Web] 已加载Web UI模块")
+        # info("[Web] 已加载Web UI模块")
     except ImportError as e:
         info(f"[Web] 未找到Web页面模块: {str(e)}")
     except Exception as e:
@@ -138,7 +138,7 @@ def start_server(host: str = "0.0.0.0", port: int = 8002, open_browser: bool = T
     server_thread = threading.Thread(target=server.run, daemon=True, name="uvicorn_server")
     server_thread.start()
 
-    info("[Web] Server started, monitoring for restart requests...")
+    # info("[Web] Server started, monitoring for restart requests...")
 
     # 监控重启标志
     import unilabos.app.main as main_module

unilabos/app/ws_client.py

@@ -26,6 +26,7 @@ from enum import Enum
 from typing_extensions import TypedDict
 
 from unilabos.app.model import JobAddReq
+from unilabos.resources.resource_tracker import ResourceDictType
 from unilabos.ros.nodes.presets.host_node import HostNode
 from unilabos.utils.type_check import serialize_result_info
 from unilabos.app.communication import BaseCommunicationClient
@@ -76,6 +77,7 @@ class JobInfo:
     start_time: float
     last_update_time: float = field(default_factory=time.time)
     ready_timeout: Optional[float] = None  # READY状态的超时时间
+    always_free: bool = False  # 是否为永久闲置动作(不受排队限制)
 
     def update_timestamp(self):
         """更新最后更新时间"""
@@ -127,6 +129,15 @@ class DeviceActionManager:
             # 总是将job添加到all_jobs中
             self.all_jobs[job_info.job_id] = job_info
 
+            # always_free的动作不受排队限制，直接设为READY
+            if job_info.always_free:
+                job_info.status = JobStatus.READY
+                job_info.update_timestamp()
+                job_info.set_ready_timeout(10)
+                job_log = format_job_log(job_info.job_id, job_info.task_id, job_info.device_id, job_info.action_name)
+                logger.trace(f"[DeviceActionManager] Job {job_log} always_free, start immediately")
+                return True
+
             # 检查是否有正在执行或准备执行的任务
             if device_key in self.active_jobs:
                 # 有正在执行或准备执行的任务，加入队列
@@ -176,11 +187,15 @@ class DeviceActionManager:
                 logger.error(f"[DeviceActionManager] Job {job_log} is not in READY status, current: {job_info.status}")
                 return False
 
-            # 检查设备上是否是这个job
-            if device_key not in self.active_jobs or self.active_jobs[device_key].job_id != job_id:
-                job_log = format_job_log(job_info.job_id, job_info.task_id, job_info.device_id, job_info.action_name)
-                logger.error(f"[DeviceActionManager] Job {job_log} is not the active job for {device_key}")
-                return False
+            # always_free的job不需要检查active_jobs
+            if not job_info.always_free:
+                # 检查设备上是否是这个job
+                if device_key not in self.active_jobs or self.active_jobs[device_key].job_id != job_id:
+                    job_log = format_job_log(
+                        job_info.job_id, job_info.task_id, job_info.device_id, job_info.action_name
+                    )
+                    logger.error(f"[DeviceActionManager] Job {job_log} is not the active job for {device_key}")
+                    return False
 
             # 开始执行任务，将状态从READY转换为STARTED
             job_info.status = JobStatus.STARTED
@@ -203,6 +218,13 @@ class DeviceActionManager:
             job_info = self.all_jobs[job_id]
             device_key = job_info.device_action_key
 
+            # always_free的job直接清理，不影响队列
+            if job_info.always_free:
+                job_info.status = JobStatus.ENDED
+                job_info.update_timestamp()
+                del self.all_jobs[job_id]
+                return None
+
             # 移除活跃任务
             if device_key in self.active_jobs and self.active_jobs[device_key].job_id == job_id:
                 del self.active_jobs[device_key]
@@ -234,9 +256,14 @@ class DeviceActionManager:
             return None
 
     def get_active_jobs(self) -> List[JobInfo]:
-        """获取所有正在执行的任务"""
+        """获取所有正在执行的任务(含active_jobs和always_free的STARTED job)"""
         with self.lock:
-            return list(self.active_jobs.values())
+            jobs = list(self.active_jobs.values())
+            # 补充 always_free 的 STARTED job(它们不在 active_jobs 中)
+            for job in self.all_jobs.values():
+                if job.always_free and job.status == JobStatus.STARTED and job not in jobs:
+                    jobs.append(job)
+            return jobs
 
     def get_queued_jobs(self) -> List[JobInfo]:
         """获取所有排队中的任务"""
@@ -261,6 +288,14 @@ class DeviceActionManager:
             job_info = self.all_jobs[job_id]
             device_key = job_info.device_action_key
 
+            # always_free的job直接清理
+            if job_info.always_free:
+                job_info.status = JobStatus.ENDED
+                del self.all_jobs[job_id]
+                job_log = format_job_log(job_info.job_id, job_info.task_id, job_info.device_id, job_info.action_name)
+                logger.trace(f"[DeviceActionManager] Always-free job {job_log} cancelled")
+                return True
+
             # 如果是正在执行的任务
             if device_key in self.active_jobs and self.active_jobs[device_key].job_id == job_id:
                 # 清理active job状态
@@ -334,13 +369,18 @@ class DeviceActionManager:
         timeout_jobs = []
 
         with self.lock:
-            # 统计READY状态的任务数量
-            ready_jobs_count = sum(1 for job in self.active_jobs.values() if job.status == JobStatus.READY)
+            # 收集所有需要检查的 READY 任务(active_jobs + always_free READY jobs)
+            ready_candidates = list(self.active_jobs.values())
+            for job in self.all_jobs.values():
+                if job.always_free and job.status == JobStatus.READY and job not in ready_candidates:
+                    ready_candidates.append(job)
+
+            ready_jobs_count = sum(1 for job in ready_candidates if job.status == JobStatus.READY)
             if ready_jobs_count > 0:
                 logger.trace(f"[DeviceActionManager] Checking {ready_jobs_count} READY jobs for timeout")  # type: ignore  # noqa: E501
 
             # 找到所有超时的READY任务（只检测，不处理）
-            for job_info in self.active_jobs.values():
+            for job_info in ready_candidates:
                 if job_info.is_ready_timeout():
                     timeout_jobs.append(job_info)
                     job_log = format_job_log(
@@ -369,6 +409,7 @@ class MessageProcessor:
         # 线程控制
         self.is_running = False
         self.thread = None
+        self._loop = None  # asyncio event loop引用，用于外部关闭websocket
         self.reconnect_count = 0
 
         logger.info(f"[MessageProcessor] Initialized for URL: {websocket_url}")
@@ -395,22 +436,31 @@ class MessageProcessor:
     def stop(self) -> None:
         """停止消息处理线程"""
         self.is_running = False
+        # 主动关闭websocket以快速中断消息接收循环
+        ws = self.websocket
+        loop = self._loop
+        if ws and loop and loop.is_running():
+            try:
+                asyncio.run_coroutine_threadsafe(ws.close(), loop)
+            except Exception:
+                pass
         if self.thread and self.thread.is_alive():
             self.thread.join(timeout=2)
         logger.info("[MessageProcessor] Stopped")
 
     def _run(self):
         """运行消息处理主循环"""
-        loop = asyncio.new_event_loop()
+        self._loop = asyncio.new_event_loop()
         try:
-            asyncio.set_event_loop(loop)
-            loop.run_until_complete(self._connection_handler())
+            asyncio.set_event_loop(self._loop)
+            self._loop.run_until_complete(self._connection_handler())
         except Exception as e:
             logger.error(f"[MessageProcessor] Thread error: {str(e)}")
             logger.error(traceback.format_exc())
         finally:
-            if loop:
-                loop.close()
+            if self._loop:
+                self._loop.close()
+            self._loop = None
 
     async def _connection_handler(self):
         """处理WebSocket连接和重连逻辑"""
@@ -427,8 +477,10 @@ class MessageProcessor:
                 async with websockets.connect(
                     self.websocket_url,
                     ssl=ssl_context,
+                    open_timeout=20,
                     ping_interval=WSConfig.ping_interval,
                     ping_timeout=10,
+                    close_timeout=5,
                     additional_headers={
                         "Authorization": f"Lab {BasicConfig.auth_secret()}",
                         "EdgeSession": f"{self.session_id}",
@@ -439,81 +491,94 @@ class MessageProcessor:
                     self.connected = True
                     self.reconnect_count = 0
 
-                    logger.trace(f"[MessageProcessor] Connected to {self.websocket_url}")
+                    logger.info(f"[MessageProcessor] 已连接到 {self.websocket_url}")
 
                     # 启动发送协程
-                    send_task = asyncio.create_task(self._send_handler())
+                    send_task = asyncio.create_task(self._send_handler(), name="websocket-send_task")
+
+                    # 每次连接（含重连）后重新向服务端注册，
+                    # 否则服务端不知道客户端已上线，不会推送消息。
+                    if self.websocket_client:
+                        self.websocket_client.publish_host_ready()
 
                     try:
                         # 接收消息循环
                         await self._message_handler()
                     finally:
+                        # 必须在 async with __aexit__ 之前停止 send_task，
+                        # 否则 send_task 会在关闭握手期间继续发送数据，
+                        # 干扰 websockets 库的内部清理，导致 task 泄漏。
+                        self.connected = False
                         send_task.cancel()
                         try:
                             await send_task
                         except asyncio.CancelledError:
                             pass
-                        self.connected = False
 
             except websockets.exceptions.ConnectionClosed:
-                logger.warning("[MessageProcessor] Connection closed")
-                self.connected = False
+                logger.warning("[MessageProcessor] 与服务端连接中断")
+            except TimeoutError:
+                logger.warning(
+                    f"[MessageProcessor] 与服务端连接通信超时 (已尝试 {self.reconnect_count + 1} 次)，请检查您的网络状况"
+                )
+            except websockets.exceptions.InvalidStatus as e:
+                logger.warning(
+                    f"[MessageProcessor] 收到服务端注册码 {e.response.status_code}, 上一进程可能还未退出"
+                )
             except Exception as e:
-                logger.error(f"[MessageProcessor] Connection error: {str(e)}")
                 logger.error(traceback.format_exc())
-                self.connected = False
+                logger.error(f"[MessageProcessor] 尝试重连时出错 {str(e)}")
             finally:
+                self.connected = False
                 self.websocket = None
 
             # 重连逻辑
-            if self.is_running and self.reconnect_count < WSConfig.max_reconnect_attempts:
+            if not self.is_running:
+                break
+            if self.reconnect_count < WSConfig.max_reconnect_attempts:
                 self.reconnect_count += 1
+                backoff = WSConfig.reconnect_interval
                 logger.info(
-                    f"[MessageProcessor] Reconnecting in {WSConfig.reconnect_interval}s "
-                    f"(attempt {self.reconnect_count}/{WSConfig.max_reconnect_attempts})"
+                    f"[MessageProcessor] 即将在 {backoff} 秒后重连 (已尝试 {self.reconnect_count}/{WSConfig.max_reconnect_attempts})"
                 )
-                await asyncio.sleep(WSConfig.reconnect_interval)
-            elif self.reconnect_count >= WSConfig.max_reconnect_attempts:
+                await asyncio.sleep(backoff)
+            else:
                 logger.error("[MessageProcessor] Max reconnection attempts reached")
                 break
-            else:
-                self.reconnect_count -= 1
 
     async def _message_handler(self):
-        """处理接收到的消息"""
+        """处理接收到的消息。
+
+        ConnectionClosed 不在此处捕获，让其向上传播到 _connection_handler，
+        以便 async with websockets.connect() 的 __aexit__ 能感知连接已断，
+        正确清理内部 task，避免 task 泄漏。
+        """
         if not self.websocket:
             logger.error("[MessageProcessor] WebSocket connection is None")
             return
 
-        try:
-            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)
+        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:
-                        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)
-                except json.JSONDecodeError:
-                    logger.error(f"[MessageProcessor] Invalid JSON received: {message}")
-                except Exception as e:
-                    logger.error(f"[MessageProcessor] Error processing message: {str(e)}")
-                    logger.error(traceback.format_exc())
-
-        except websockets.exceptions.ConnectionClosed:
-            logger.info("[MessageProcessor] Message handler stopped - connection closed")
-        except Exception as e:
-            logger.error(f"[MessageProcessor] Message handler error: {str(e)}")
-            logger.error(traceback.format_exc())
+                        await self._process_message(message_type, message_data)
+            except json.JSONDecodeError:
+                logger.error(f"[MessageProcessor] Invalid JSON received: {message}")
+            except Exception as e:
+                logger.error(f"[MessageProcessor] Error processing message: {str(e)}")
+                logger.error(traceback.format_exc())
 
     async def _send_handler(self):
         """处理发送队列中的消息"""
@@ -562,6 +627,7 @@ class MessageProcessor:
 
         except asyncio.CancelledError:
             logger.debug("[MessageProcessor] Send handler cancelled")
+            raise
         except Exception as e:
             logger.error(f"[MessageProcessor] Fatal error in send handler: {str(e)}")
             logger.error(traceback.format_exc())
@@ -593,6 +659,10 @@ class MessageProcessor:
             # 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 == "add_device":
+                await self._handle_device_manage(message_data, "add")
+            elif message_type == "remove_device":
+                await self._handle_device_manage(message_data, "remove")
             elif message_type == "request_restart":
                 await self._handle_request_restart(message_data)
             else:
@@ -608,6 +678,24 @@ class MessageProcessor:
         if host_node:
             host_node.handle_pong_response(pong_data)
 
+    def _check_action_always_free(self, device_id: str, action_name: str) -> bool:
+        """检查该action是否标记为always_free，通过HostNode统一的_action_value_mappings查找"""
+        try:
+            host_node = HostNode.get_instance(0)
+            if not host_node:
+                return False
+            # noinspection PyProtectedMember
+            action_mappings = host_node._action_value_mappings.get(device_id)
+            if not action_mappings:
+                return False
+            # 尝试直接匹配或 auto- 前缀匹配
+            for key in [action_name, f"auto-{action_name}"]:
+                if key in action_mappings:
+                    return action_mappings[key].get("always_free", False)
+            return False
+        except Exception:
+            return False
+
     async def _handle_query_action_state(self, data: Dict[str, Any]):
         """处理query_action_state消息"""
         device_id = data.get("device_id", "")
@@ -622,6 +710,9 @@ class MessageProcessor:
 
         device_action_key = f"/devices/{device_id}/{action_name}"
 
+        # 检查action是否为always_free
+        action_always_free = self._check_action_always_free(device_id, action_name)
+
         # 创建任务信息
         job_info = JobInfo(
             job_id=job_id,
@@ -631,6 +722,7 @@ class MessageProcessor:
             device_action_key=device_action_key,
             status=JobStatus.QUEUE,
             start_time=time.time(),
+            always_free=action_always_free,
         )
 
         # 添加到设备管理器
@@ -907,6 +999,37 @@ class MessageProcessor:
             )
             thread.start()
 
+    async def _handle_device_manage(self, device_list: list[ResourceDictType], action: str):
+        """Handle add_device / remove_device from LabGo server."""
+        if not device_list:
+            return
+
+        for item in device_list:
+            target_node_id = item.get("target_node_id", "host_node")
+
+            def _notify(target_id: str, act: str, cfg: ResourceDictType):
+                try:
+                    host_node = HostNode.get_instance(timeout=5)
+                    if not host_node:
+                        logger.error(f"[DeviceManage] HostNode not available for {act}_device")
+                        return
+                    success = host_node.notify_device_manage(target_id, act, cfg)
+                    if success:
+                        logger.info(f"[DeviceManage] {act}_device completed on {target_id}")
+                    else:
+                        logger.warning(f"[DeviceManage] {act}_device failed on {target_id}")
+                except Exception as e:
+                    logger.error(f"[DeviceManage] Error in {act}_device: {e}")
+                    logger.error(traceback.format_exc())
+
+            thread = threading.Thread(
+                target=_notify,
+                args=(target_node_id, action, item),
+                daemon=True,
+                name=f"DeviceManage-{action}-{item.get('id', '')}",
+            )
+            thread.start()
+
     async def _handle_request_restart(self, data: Dict[str, Any]):
         """
         处理重启请求
@@ -918,10 +1041,9 @@ class MessageProcessor:
         logger.info(f"[MessageProcessor] Received restart request, reason: {reason}, delay: {delay}s")
 
         # 发送确认消息
-        if self.websocket_client:
-            await self.websocket_client.send_message(
-                {"action": "restart_acknowledged", "data": {"reason": reason, "delay": delay}}
-            )
+        self.send_message(
+            {"action": "restart_acknowledged", "data": {"reason": reason, "delay": delay}}
+        )
 
         # 设置全局重启标志
         import unilabos.app.main as main_module
@@ -1023,6 +1145,7 @@ class QueueProcessor:
     def stop(self) -> None:
         """停止队列处理线程"""
         self.is_running = False
+        self.queue_update_event.set()  # 立即唤醒等待中的线程
         if self.thread and self.thread.is_alive():
             self.thread.join(timeout=2)
         logger.info("[QueueProcessor] Stopped")
@@ -1123,6 +1246,11 @@ class QueueProcessor:
         logger.debug(f"[QueueProcessor] Sending busy status for {len(queued_jobs)} queued jobs")
 
         for job_info in queued_jobs:
+            # 快照可能已过期：在遍历过程中 end_job() 可能已将此 job 移至 READY，
+            # 此时不应再发送 busy/need_more，否则会覆盖已发出的 free=True 通知
+            if job_info.status != JobStatus.QUEUE:
+                continue
+
             message = {
                 "action": "report_action_state",
                 "data": {
@@ -1271,8 +1399,8 @@ class WebSocketClient(BaseCommunicationClient):
                 message = {"action": "normal_exit", "data": {"session_id": session_id}}
                 self.message_processor.send_message(message)
                 logger.info(f"[WebSocketClient] Sent normal_exit message with session_id: {session_id}")
-                # 给一点时间让消息发送出去
-                time.sleep(1)
+                # send_handler 每100ms检查一次队列，等300ms足以让消息发出
+                time.sleep(0.3)
             except Exception as e:
                 logger.warning(f"[WebSocketClient] Failed to send normal_exit message: {str(e)}")
 

unilabos/config/config.py

@@ -23,6 +23,8 @@ class BasicConfig:
     disable_browser = False  # 禁止浏览器自动打开
     port = 8002  # 本地HTTP服务
     check_mode = False  # CI 检查模式，用于验证 registry 导入和文件一致性
+    test_mode = False  # 测试模式，所有动作不实际执行，返回模拟结果
+    extra_resource = False  # 是否加载lab_开头的额外资源
     # 'TRACE', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'
     log_level: Literal["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "DEBUG"
 
@@ -39,7 +41,7 @@ class BasicConfig:
 class WSConfig:
     reconnect_interval = 5  # 重连间隔（秒）
     max_reconnect_attempts = 999  # 最大重连次数
-    ping_interval = 30  # ping间隔（秒）
+    ping_interval = 20  # ping间隔（秒）
 
 
 # HTTP配置
@@ -145,5 +147,5 @@ def load_config(config_path=None):
             traceback.print_exc()
             exit(1)
     else:
-        config_path = os.path.join(os.path.dirname(__file__), "local_config.py")
+        config_path = os.path.join(os.path.dirname(__file__), "example_config.py")
         load_config(config_path)

unilabos/device_comms/universal_driver.py

@@ -1,4 +1,3 @@
-
 from abc import abstractmethod
 from functools import wraps
 import inspect

unilabos/devices/liquid_handling/liquid_handler_abstract.py

@@ -21,7 +21,7 @@ from pylabrobot.resources import (
     ResourceHolder,
     Lid,
     Trash,
-    Tip,
+    Tip, TubeRack,
 )
 from typing_extensions import TypedDict
 
@@ -696,10 +696,13 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
 
         如果 liquid_names 和 volumes 为空，但 plate 和 well_names 不为空，直接返回 plate 和 wells。
         """
-        assert issubclass(plate.__class__, Plate), "plate must be a Plate"
-        plate: Plate = cast(Plate, cast(Resource, plate))
+        assert issubclass(plate.__class__, Plate) or issubclass(plate.__class__, TubeRack) , f"plate must be a Plate, now: {type(plate)}"
+        plate: Union[Plate, TubeRack]
         # 根据 well_names 获取对应的 Well 对象
-        wells = [plate.get_well(name) for name in well_names]
+        if issubclass(plate.__class__, Plate):
+            wells = [plate.get_well(name) for name in well_names]
+        elif issubclass(plate.__class__, TubeRack):
+            wells = [plate.get_tube(name) for name in well_names]
         res_volumes = []
 
         # 如果 liquid_names 和 volumes 都为空，直接返回

unilabos/devices/liquid_handling/prcxi/prcxi.py

@@ -55,6 +55,7 @@ from unilabos.devices.liquid_handling.liquid_handler_abstract import (
     TransferLiquidReturn,
 )
 from unilabos.registry.placeholder_type import ResourceSlot
+from unilabos.resources.resource_tracker import ResourceTreeSet
 from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
 
 
@@ -90,20 +91,103 @@ class PRCXI9300Deck(Deck):
     该类定义了 PRCXI 9300 的工作台布局和槽位信息。
     """
 
-    def __init__(self, name: str, size_x: float, size_y: float, size_z: float, **kwargs):
-        super().__init__(name, size_x, size_y, size_z)
-        self.slots = [None] * 16  # PRCXI 9300/9320 最大有 16 个槽位
-        self.slot_locations = [Coordinate(0, 0, 0)] * 16
+    # T1-T16 默认位置 (4列×4行)
+    _DEFAULT_SITE_POSITIONS = [
+        (0, 0, 0), (138, 0, 0), (276, 0, 0), (414, 0, 0),         # T1-T4
+        (0, 96, 0), (138, 96, 0), (276, 96, 0), (414, 96, 0),     # T5-T8
+        (0, 192, 0), (138, 192, 0), (276, 192, 0), (414, 192, 0), # T9-T12
+        (0, 288, 0), (138, 288, 0), (276, 288, 0), (414, 288, 0), # T13-T16
+    ]
+    _DEFAULT_SITE_SIZE = {"width": 128.0, "height": 86, "depth": 0}
+    _DEFAULT_CONTENT_TYPE = ["plate", "tip_rack", "plates", "tip_racks", "tube_rack", "adaptor"]
+
+    def __init__(self, name: str, size_x: float, size_y: float, size_z: float,
+                 sites: Optional[List[Dict[str, Any]]] = None, **kwargs):
+        super().__init__(size_x, size_y, size_z, name)
+        if sites is not None:
+            self.sites: List[Dict[str, Any]] = [dict(s) for s in sites]
+        else:
+            self.sites = []
+            for i, (x, y, z) in enumerate(self._DEFAULT_SITE_POSITIONS):
+                self.sites.append({
+                    "label": f"T{i + 1}",
+                    "visible": True,
+                    "position": {"x": x, "y": y, "z": z},
+                    "size": dict(self._DEFAULT_SITE_SIZE),
+                    "content_type": list(self._DEFAULT_CONTENT_TYPE),
+                })
+        # _ordering: label -> None, 用于外部通过 list(keys()).index(site) 将 Tn 转换为 spot index
+        self._ordering = collections.OrderedDict(
+            (site["label"], None) for site in self.sites
+        )
+
+    def _get_site_location(self, idx: int) -> Coordinate:
+        pos = self.sites[idx]["position"]
+        return Coordinate(pos["x"], pos["y"], pos["z"])
+
+    def _get_site_resource(self, idx: int) -> Optional[Resource]:
+        site_loc = self._get_site_location(idx)
+        for child in self.children:
+            if child.location == site_loc:
+                return child
+        return None
+
+    def assign_child_resource(
+        self,
+        resource: Resource,
+        location: Optional[Coordinate] = None,
+        reassign: bool = True,
+        spot: Optional[int] = None,
+    ):
+        idx = spot
+        if spot is not None:
+            idx = spot
+        else:
+            for i, site in enumerate(self.sites):
+                site_loc = self._get_site_location(i)
+                if site.get("label") == resource.name:
+                    idx = i
+                    break
+                if location is not None and site_loc == location:
+                    idx = i
+                    break
+
+        if idx is None:
+            for i in range(len(self.sites)):
+                if self._get_site_resource(i) is None:
+                    idx = i
+                    break
+
+        if idx is None:
+            raise ValueError(f"No available site on deck '{self.name}' for resource '{resource.name}'")
+
+        if not reassign and self._get_site_resource(idx) is not None:
+            raise ValueError(f"Site {idx} ('{self.sites[idx]['label']}') is already occupied")
+
+        loc = self._get_site_location(idx)
+        super().assign_child_resource(resource, location=loc, reassign=reassign)
 
     def assign_child_at_slot(self, resource: Resource, slot: int, reassign: bool = False) -> None:
-        if self.slots[slot - 1] is not None and not reassign:
-            raise ValueError(f"Spot {slot} is already occupied")
+        self.assign_child_resource(resource, spot=slot - 1, reassign=reassign)
 
-        self.slots[slot - 1] = resource
-        super().assign_child_resource(resource, location=self.slot_locations[slot - 1])
+    def serialize(self) -> dict:
+        data = super().serialize()
+        sites_out = []
+        for i, site in enumerate(self.sites):
+            occupied = self._get_site_resource(i)
+            sites_out.append({
+                "label": site["label"],
+                "visible": site.get("visible", True),
+                "occupied_by": occupied.name if occupied is not None else None,
+                "position": site["position"],
+                "size": site["size"],
+                "content_type": site["content_type"],
+            })
+        data["sites"] = sites_out
+        return data
 
 
-class PRCXI9300Container(Plate):
+class PRCXI9300Container(Container):
     """PRCXI 9300 的专用 Container 类，继承自 Plate，用于槽位定位和未知模块。
 
     该类定义了 PRCXI 9300 的工作台布局和槽位信息。
@@ -116,11 +200,10 @@ class PRCXI9300Container(Plate):
         size_y: float,
         size_z: float,
         category: str,
-        ordering: collections.OrderedDict,
         model: Optional[str] = None,
         **kwargs,
     ):
-        super().__init__(name, size_x, size_y, size_z, category=category, ordering=ordering, model=model)
+        super().__init__(name, size_x, size_y, size_z, category=category, model=model)
         self._unilabos_state = {}
 
     def load_state(self, state: Dict[str, Any]) -> None:
@@ -248,14 +331,15 @@ class PRCXI9300TipRack(TipRack):
         if ordered_items is not None:
             items = ordered_items
         elif ordering is not None:
-            # 检查 ordering 中的值是否是字符串（从 JSON 反序列化时的情况）
-            # 如果是字符串，说明这是位置名称，需要让 TipRack 自己创建 Tip 对象
-            # 我们只传递位置信息（键），不传递值，使用 ordering 参数
-            if ordering and isinstance(next(iter(ordering.values()), None), str):
-                # ordering 的值是字符串，只使用键（位置信息）创建新的 OrderedDict
+            # 检查 ordering 中的值类型来决定如何处理：
+            # - 字符串值（从 JSON 反序列化）: 只用键创建 ordering_param
+            # - None 值（从第二次往返序列化）: 同样只用键创建 ordering_param
+            # - 对象值（已经是实际的 Resource 对象）: 直接作为 ordered_items 使用
+            first_val = next(iter(ordering.values()), None) if ordering else None
+            if not ordering or first_val is None or isinstance(first_val, str):
+                # ordering 的值是字符串或 None，只使用键（位置信息）创建新的 OrderedDict
                 # 传递 ordering 参数而不是 ordered_items，让 TipRack 自己创建 Tip 对象
                 items = None
-                # 使用 ordering 参数，只包含位置信息（键）
                 ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
             else:
                 # ordering 的值已经是对象，可以直接使用
@@ -397,14 +481,15 @@ class PRCXI9300TubeRack(TubeRack):
             items_to_pass = ordered_items
             ordering_param = None
         elif ordering is not None:
-            # 检查 ordering 中的值是否是字符串（从 JSON 反序列化时的情况）
-            # 如果是字符串，说明这是位置名称，需要让 TubeRack 自己创建 Tube 对象
-            # 我们只传递位置信息（键），不传递值，使用 ordering 参数
-            if ordering and isinstance(next(iter(ordering.values()), None), str):
-                # ordering 的值是字符串，只使用键（位置信息）创建新的 OrderedDict
+            # 检查 ordering 中的值类型来决定如何处理：
+            # - 字符串值（从 JSON 反序列化）: 只用键创建 ordering_param
+            # - None 值（从第二次往返序列化）: 同样只用键创建 ordering_param
+            # - 对象值（已经是实际的 Resource 对象）: 直接作为 ordered_items 使用
+            first_val = next(iter(ordering.values()), None) if ordering else None
+            if not ordering or first_val is None or isinstance(first_val, str):
+                # ordering 的值是字符串或 None，只使用键（位置信息）创建新的 OrderedDict
                 # 传递 ordering 参数而不是 ordered_items，让 TubeRack 自己创建 Tube 对象
                 items_to_pass = None
-                # 使用 ordering 参数，只包含位置信息（键）
                 ordering_param = collections.OrderedDict((k, None) for k in ordering.keys())
             else:
                 # ordering 的值已经是对象，可以直接使用
@@ -549,7 +634,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
 
     def __init__(
         self,
-        deck: Deck,
+        deck: PRCXI9300Deck,
         host: str,
         port: int,
         timeout: float,
@@ -563,16 +648,16 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
         is_9320=False,
     ):
         tablets_info = []
-        count = 0
-        for child in deck.children:
-            if child.children:
-                if "Material" in child.children[0]._unilabos_state:
-                    number = int(child.name.replace("T", ""))
-                    tablets_info.append(
-                        WorkTablets(
-                            Number=number, Code=f"T{number}", Material=child.children[0]._unilabos_state["Material"]
-                        )
+        for site_id in range(len(deck.sites)):
+            child = deck._get_site_resource(site_id)
+            # 如果放其他类型的物料，是不可以的
+            if hasattr(child, "_unilabos_state") and "Material" in child._unilabos_state:
+                number = site_id + 1
+                tablets_info.append(
+                    WorkTablets(
+                        Number=number, Code=f"T{number}", Material=child._unilabos_state["Material"]
                     )
+                )
         if is_9320:
             print("当前设备是9320")
         # 始终初始化 step_mode 属性

unilabos/devices/ros_dev/liquid_handler_joint_publisher.py

@@ -19,10 +19,11 @@ from rclpy.node import Node
 import re
 
 class LiquidHandlerJointPublisher(BaseROS2DeviceNode):
-    def __init__(self,resources_config:list, resource_tracker, rate=50, device_id:str = "lh_joint_publisher", **kwargs):
+    def __init__(self,resources_config:list, resource_tracker, rate=50, device_id:str = "lh_joint_publisher", registry_name: str = "lh_joint_publisher", **kwargs):
         super().__init__(
             driver_instance=self,
             device_id=device_id,
+            registry_name=registry_name,
             status_types={},
             action_value_mappings={},
             hardware_interface={},

unilabos/devices/virtual/workbench.py

@@ -1,15 +1,15 @@
 """
 Virtual Workbench Device - 模拟工作台设备
-包含：
+包含:
 - 1个机械臂 (每次操作3s, 独占锁)
 - 3个加热台 (每次加热10s, 可并行)
 
-工作流程：
-1. A1-A5 物料同时启动，竞争机械臂
+工作流程:
+1. A1-A5 物料同时启动, 竞争机械臂
 2. 机械臂将物料移动到空闲加热台
-3. 加热完成后，机械臂将物料移动到C1-C5
+3. 加热完成后, 机械臂将物料移动到C1-C5
 
-注意：调用来自线程池，使用 threading.Lock 进行同步
+注意: 调用来自线程池, 使用 threading.Lock 进行同步
 """
 
 import logging
@@ -21,9 +21,11 @@ from threading import Lock, RLock
 
 from typing_extensions import TypedDict
 
+from unilabos.registry.decorators import (
+    device, action, ActionInputHandle, ActionOutputHandle, DataSource, topic_config, not_action
+)
 from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
-from unilabos.utils.decorator import not_action
-from unilabos.resources.resource_tracker import SampleUUIDsType, LabSample, RETURN_UNILABOS_SAMPLES
+from unilabos.resources.resource_tracker import SampleUUIDsType, LabSample
 
 
 # ============ TypedDict 返回类型定义 ============
@@ -57,6 +59,8 @@ class MoveToOutputResult(TypedDict):
     success: bool
     station_id: int
     material_id: str
+    output_position: str
+    message: str
     unilabos_samples: List[LabSample]
 
 
@@ -81,9 +85,9 @@ class HeatingStationState(Enum):
     """加热台状态枚举"""
 
     IDLE = "idle"  # 空闲
-    OCCUPIED = "occupied"  # 已放置物料，等待加热
+    OCCUPIED = "occupied"  # 已放置物料, 等待加热
     HEATING = "heating"  # 加热中
-    COMPLETED = "completed"  # 加热完成，等待取走
+    COMPLETED = "completed"  # 加热完成, 等待取走
 
 
 class ArmState(Enum):
@@ -105,26 +109,31 @@ class HeatingStation:
     heating_progress: float = 0.0
 
 
+@device(
+    id="virtual_workbench",
+    category=["virtual_device"],
+    description="Virtual Workbench with 1 robotic arm and 3 heating stations for concurrent material processing",
+)
 class VirtualWorkbench:
     """
     Virtual Workbench Device - 虚拟工作台设备
 
     模拟一个包含1个机械臂和3个加热台的工作站
-    - 机械臂操作耗时3秒，同一时间只能执行一个操作
-    - 加热台加热耗时10秒，3个加热台可并行工作
+    - 机械臂操作耗时3秒, 同一时间只能执行一个操作
+    - 加热台加热耗时10秒, 3个加热台可并行工作
 
     工作流:
-    1. 物料A1-A5并发启动（线程池），竞争机械臂使用权
-    2. 获取机械臂后，查找空闲加热台
-    3. 机械臂将物料放入加热台，开始加热
-    4. 加热完成后，机械臂将物料移动到目标位置Cn
+    1. 物料A1-A5并发启动(线程池), 竞争机械臂使用权
+    2. 获取机械臂后, 查找空闲加热台
+    3. 机械臂将物料放入加热台, 开始加热
+    4. 加热完成后, 机械臂将物料移动到目标位置Cn
     """
 
     _ros_node: BaseROS2DeviceNode
 
     # 配置常量
-    ARM_OPERATION_TIME: float = 3.0  # 机械臂操作时间(秒)
-    HEATING_TIME: float = 10.0  # 加热时间(秒)
+    ARM_OPERATION_TIME: float = 2  # 机械臂操作时间(秒)
+    HEATING_TIME: float = 60.0  # 加热时间(秒)
     NUM_HEATING_STATIONS: int = 3  # 加热台数量
 
     def __init__(self, device_id: Optional[str] = None, config: Optional[Dict[str, Any]] = None, **kwargs):
@@ -141,23 +150,23 @@ class VirtualWorkbench:
         self.data: Dict[str, Any] = {}
 
         # 从config中获取可配置参数
-        self.ARM_OPERATION_TIME = float(self.config.get("arm_operation_time", 3.0))
-        self.HEATING_TIME = float(self.config.get("heating_time", 10.0))
-        self.NUM_HEATING_STATIONS = int(self.config.get("num_heating_stations", 3))
+        self.ARM_OPERATION_TIME = float(self.config.get("arm_operation_time", self.ARM_OPERATION_TIME))
+        self.HEATING_TIME = float(self.config.get("heating_time", self.HEATING_TIME))
+        self.NUM_HEATING_STATIONS = int(self.config.get("num_heating_stations", self.NUM_HEATING_STATIONS))
 
-        # 机械臂状态和锁 (使用threading.Lock)
+        # 机械臂状态和锁
         self._arm_lock = Lock()
         self._arm_state = ArmState.IDLE
         self._arm_current_task: Optional[str] = None
 
-        # 加热台状态 (station_id -> HeatingStation) - 立即初始化，不依赖initialize()
+        # 加热台状态
         self._heating_stations: Dict[int, HeatingStation] = {
             i: HeatingStation(station_id=i) for i in range(1, self.NUM_HEATING_STATIONS + 1)
         }
-        self._stations_lock = RLock()  # 可重入锁，保护加热台状态
+        self._stations_lock = RLock()
 
         # 任务追踪
-        self._active_tasks: Dict[str, Dict[str, Any]] = {}  # material_id -> task_info
+        self._active_tasks: Dict[str, Dict[str, Any]] = {}
         self._tasks_lock = Lock()
 
         # 处理其他kwargs参数
@@ -183,7 +192,6 @@ class VirtualWorkbench:
         """初始化虚拟工作台"""
         self.logger.info(f"初始化虚拟工作台 {self.device_id}")
 
-        # 重置加热台状态 (已在__init__中创建，这里重置为初始状态)
         with self._stations_lock:
             for station in self._heating_stations.values():
                 station.state = HeatingStationState.IDLE
@@ -191,7 +199,6 @@ class VirtualWorkbench:
                 station.material_number = None
                 station.heating_progress = 0.0
 
-        # 初始化状态
         self.data.update(
             {
                 "status": "Ready",
@@ -257,11 +264,7 @@ class VirtualWorkbench:
             self.data["message"] = message
 
     def _find_available_heating_station(self) -> Optional[int]:
-        """查找空闲的加热台
-
-        Returns:
-            空闲加热台ID，如果没有则返回None
-        """
+        """查找空闲的加热台"""
         with self._stations_lock:
             for station_id, station in self._heating_stations.items():
                 if station.state == HeatingStationState.IDLE:
@@ -269,23 +272,12 @@ class VirtualWorkbench:
         return None
 
     def _acquire_arm(self, task_description: str) -> bool:
-        """获取机械臂使用权（阻塞直到获取）
-
-        Args:
-            task_description: 任务描述，用于日志
-
-        Returns:
-            是否成功获取
-        """
+        """获取机械臂使用权(阻塞直到获取)"""
         self.logger.info(f"[{task_description}] 等待获取机械臂...")
-
-        # 阻塞等待获取锁
         self._arm_lock.acquire()
-
         self._arm_state = ArmState.BUSY
         self._arm_current_task = task_description
         self._update_data_status(f"机械臂执行: {task_description}")
-
         self.logger.info(f"[{task_description}] 成功获取机械臂使用权")
         return True
 
@@ -298,6 +290,22 @@ class VirtualWorkbench:
         self._update_data_status(f"机械臂已释放 (完成: {task})")
         self.logger.info(f"机械臂已释放 (完成: {task})")
 
+    @action(
+        auto_prefix=True,
+        description="批量准备物料 - 虚拟起始节点, 生成A1-A5物料, 输出5个handle供后续节点使用",
+        handles=[
+            ActionOutputHandle(key="channel_1", data_type="workbench_material",
+                         label="实验1", data_key="material_1", data_source=DataSource.EXECUTOR),
+            ActionOutputHandle(key="channel_2", data_type="workbench_material",
+                         label="实验2", data_key="material_2", data_source=DataSource.EXECUTOR),
+            ActionOutputHandle(key="channel_3", data_type="workbench_material",
+                         label="实验3", data_key="material_3", data_source=DataSource.EXECUTOR),
+            ActionOutputHandle(key="channel_4", data_type="workbench_material",
+                         label="实验4", data_key="material_4", data_source=DataSource.EXECUTOR),
+            ActionOutputHandle(key="channel_5", data_type="workbench_material",
+                         label="实验5", data_key="material_5", data_source=DataSource.EXECUTOR),
+        ],
+    )
     def prepare_materials(
         self,
         sample_uuids: SampleUUIDsType,
@@ -306,19 +314,14 @@ class VirtualWorkbench:
         """
         批量准备物料 - 虚拟起始节点
 
-        作为工作流的起始节点，生成指定数量的物料编号供后续节点使用。
-        输出5个handle (material_1 ~ material_5)，分别对应实验1~5。
-
-        Args:
-            count: 待生成的物料数量，默认5 (生成 A1-A5)
-
-        Returns:
-            PrepareMaterialsResult: 包含 material_1 ~ material_5 用于传递给 move_to_heating_station
+        作为工作流的起始节点, 生成指定数量的物料编号供后续节点使用。
+        输出5个handle (material_1 ~ material_5), 分别对应实验1~5。
         """
-        # 生成物料列表 A1 - A{count}
         materials = [i for i in range(1, count + 1)]
 
-        self.logger.info(f"[准备物料] 生成 {count} 个物料: " f"A1-A{count} -> material_1~material_{count}")
+        self.logger.info(
+            f"[准备物料] 生成 {count} 个物料: A1-A{count} -> material_1~material_{count}"
+        )
 
         return {
             "success": True,
@@ -329,9 +332,28 @@ class VirtualWorkbench:
             "material_4": materials[3] if len(materials) > 3 else 0,
             "material_5": materials[4] if len(materials) > 4 else 0,
             "message": f"已准备 {count} 个物料: A1-A{count}",
-            "unilabos_samples": [LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for sample_uuid, content in sample_uuids.items()]
+            "unilabos_samples": [
+                LabSample(
+                    sample_uuid=sample_uuid,
+                    oss_path="",
+                    extra={"material_uuid": content} if isinstance(content, str) else (content.serialize() if content else {}),
+                )
+                for sample_uuid, content in sample_uuids.items()
+            ],
         }
 
+    @action(
+        auto_prefix=True,
+        description="将物料从An位置移动到空闲加热台, 返回分配的加热台ID",
+        handles=[
+            ActionInputHandle(key="material_input", data_type="workbench_material",
+                        label="物料编号", data_key="material_number", data_source=DataSource.HANDLE),
+            ActionOutputHandle(key="heating_station_output", data_type="workbench_station",
+                         label="加热台ID", data_key="station_id", data_source=DataSource.EXECUTOR),
+            ActionOutputHandle(key="material_number_output", data_type="workbench_material",
+                         label="物料编号", data_key="material_number", data_source=DataSource.EXECUTOR),
+        ],
+    )
     def move_to_heating_station(
         self,
         sample_uuids: SampleUUIDsType,
@@ -340,20 +362,12 @@ class VirtualWorkbench:
         """
         将物料从An位置移动到加热台
 
-        多线程并发调用时，会竞争机械臂使用权，并自动查找空闲加热台
-
-        Args:
-            material_number: 物料编号 (1-5)
-
-        Returns:
-            MoveToHeatingStationResult: 包含 station_id, material_number 等用于传递给下一个节点
+        多线程并发调用时, 会竞争机械臂使用权, 并自动查找空闲加热台
         """
-        # 根据物料编号生成物料ID
         material_id = f"A{material_number}"
         task_desc = f"移动{material_id}到加热台"
         self.logger.info(f"[任务] {task_desc} - 开始执行")
 
-        # 记录任务
         with self._tasks_lock:
             self._active_tasks[material_id] = {
                 "status": "waiting_for_arm",
@@ -361,33 +375,27 @@ class VirtualWorkbench:
             }
 
         try:
-            # 步骤1: 等待获取机械臂使用权(竞争)
             with self._tasks_lock:
                 self._active_tasks[material_id]["status"] = "waiting_for_arm"
             self._acquire_arm(task_desc)
 
-            # 步骤2: 查找空闲加热台
             with self._tasks_lock:
                 self._active_tasks[material_id]["status"] = "finding_station"
             station_id = None
 
-            # 循环等待直到找到空闲加热台
             while station_id is None:
                 station_id = self._find_available_heating_station()
                 if station_id is None:
-                    self.logger.info(f"[{material_id}] 没有空闲加热台，等待中...")
-                    # 释放机械臂，等待后重试
+                    self.logger.info(f"[{material_id}] 没有空闲加热台, 等待中...")
                     self._release_arm()
                     time.sleep(0.5)
                     self._acquire_arm(task_desc)
 
-            # 步骤3: 占用加热台 - 立即标记为OCCUPIED，防止其他任务选择同一加热台
             with self._stations_lock:
                 self._heating_stations[station_id].state = HeatingStationState.OCCUPIED
                 self._heating_stations[station_id].current_material = material_id
                 self._heating_stations[station_id].material_number = material_number
 
-            # 步骤4: 模拟机械臂移动操作 (3秒)
             with self._tasks_lock:
                 self._active_tasks[material_id]["status"] = "arm_moving"
                 self._active_tasks[material_id]["assigned_station"] = station_id
@@ -395,11 +403,11 @@ class VirtualWorkbench:
 
             time.sleep(self.ARM_OPERATION_TIME)
 
-            # 步骤5: 放入加热台完成
             self._update_data_status(f"{material_id}已放入加热台{station_id}")
-            self.logger.info(f"[{material_id}] 已放入加热台{station_id} (用时{self.ARM_OPERATION_TIME}s)")
+            self.logger.info(
+                f"[{material_id}] 已放入加热台{station_id} (用时{self.ARM_OPERATION_TIME}s)"
+            )
 
-            # 释放机械臂
             self._release_arm()
 
             with self._tasks_lock:
@@ -412,8 +420,16 @@ class VirtualWorkbench:
                 "material_number": material_number,
                 "message": f"{material_id}已成功移动到加热台{station_id}",
                 "unilabos_samples": [
-                    LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                    sample_uuid, content in sample_uuids.items()]
+                    LabSample(
+                        sample_uuid=sample_uuid,
+                        oss_path="",
+                        extra=(
+                            {"material_uuid": content}
+                            if isinstance(content, str) else (content.serialize() if content else {})
+                        ),
+                    )
+                    for sample_uuid, content in sample_uuids.items()
+                ],
             }
 
         except Exception as e:
@@ -427,10 +443,33 @@ class VirtualWorkbench:
                 "material_number": material_number,
                 "message": f"移动失败: {str(e)}",
                 "unilabos_samples": [
-                    LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                    sample_uuid, content in sample_uuids.items()]
+                    LabSample(
+                        sample_uuid=sample_uuid,
+                        oss_path="",
+                        extra=(
+                            {"material_uuid": content}
+                            if isinstance(content, str) else (content.serialize() if content else {})
+                        ),
+                    )
+                    for sample_uuid, content in sample_uuids.items()
+                ],
             }
 
+    @action(
+        auto_prefix=True,
+        always_free=True,
+        description="启动指定加热台的加热程序",
+        handles=[
+            ActionInputHandle(key="station_id_input", data_type="workbench_station",
+                        label="加热台ID", data_key="station_id", data_source=DataSource.HANDLE),
+            ActionInputHandle(key="material_number_input", data_type="workbench_material",
+                        label="物料编号", data_key="material_number", data_source=DataSource.HANDLE),
+            ActionOutputHandle(key="heating_done_station", data_type="workbench_station",
+                         label="加热完成-加热台ID", data_key="station_id", data_source=DataSource.EXECUTOR),
+            ActionOutputHandle(key="heating_done_material", data_type="workbench_material",
+                         label="加热完成-物料编号", data_key="material_number", data_source=DataSource.EXECUTOR),
+        ],
+    )
     def start_heating(
         self,
         sample_uuids: SampleUUIDsType,
@@ -439,13 +478,6 @@ class VirtualWorkbench:
     ) -> StartHeatingResult:
         """
         启动指定加热台的加热程序
-
-        Args:
-            station_id: 加热台ID (1-3)，从 move_to_heating_station 的 handle 传入
-            material_number: 物料编号，从 move_to_heating_station 的 handle 传入
-
-        Returns:
-            StartHeatingResult: 包含 station_id, material_number 等用于传递给下一个节点
         """
         self.logger.info(f"[加热台{station_id}] 开始加热")
 
@@ -457,8 +489,16 @@ class VirtualWorkbench:
                 "material_number": material_number,
                 "message": f"无效的加热台ID: {station_id}",
                 "unilabos_samples": [
-                    LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                    sample_uuid, content in sample_uuids.items()]
+                    LabSample(
+                        sample_uuid=sample_uuid,
+                        oss_path="",
+                        extra=(
+                            {"material_uuid": content}
+                            if isinstance(content, str) else (content.serialize() if content else {})
+                        ),
+                    )
+                    for sample_uuid, content in sample_uuids.items()
+                ],
             }
 
         with self._stations_lock:
@@ -472,8 +512,16 @@ class VirtualWorkbench:
                     "material_number": material_number,
                     "message": f"加热台{station_id}上没有物料",
                     "unilabos_samples": [
-                        LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                        sample_uuid, content in sample_uuids.items()]
+                        LabSample(
+                            sample_uuid=sample_uuid,
+                            oss_path="",
+                            extra=(
+                                {"material_uuid": content}
+                                if isinstance(content, str) else (content.serialize() if content else {})
+                            ),
+                        )
+                        for sample_uuid, content in sample_uuids.items()
+                    ],
                 }
 
             if station.state == HeatingStationState.HEATING:
@@ -484,13 +532,20 @@ class VirtualWorkbench:
                     "material_number": material_number,
                     "message": f"加热台{station_id}已经在加热中",
                     "unilabos_samples": [
-                        LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                        sample_uuid, content in sample_uuids.items()]
+                        LabSample(
+                            sample_uuid=sample_uuid,
+                            oss_path="",
+                            extra=(
+                                {"material_uuid": content}
+                                if isinstance(content, str) else (content.serialize() if content else {})
+                            ),
+                        )
+                        for sample_uuid, content in sample_uuids.items()
+                    ],
                 }
 
             material_id = station.current_material
 
-            # 开始加热
             station.state = HeatingStationState.HEATING
             station.heating_start_time = time.time()
             station.heating_progress = 0.0
@@ -501,10 +556,19 @@ class VirtualWorkbench:
 
         self._update_data_status(f"加热台{station_id}开始加热{material_id}")
 
-        # 模拟加热过程 (10秒)
+        with self._stations_lock:
+            heating_list = [
+                f"加热台{sid}:{s.current_material}"
+                for sid, s in self._heating_stations.items()
+                if s.state == HeatingStationState.HEATING and s.current_material
+            ]
+        self.logger.info(f"[并行加热] 当前同时加热中: {', '.join(heating_list)}")
+
         start_time = time.time()
+        last_countdown_log = start_time
         while True:
             elapsed = time.time() - start_time
+            remaining = max(0.0, self.HEATING_TIME - elapsed)
             progress = min(100.0, (elapsed / self.HEATING_TIME) * 100)
 
             with self._stations_lock:
@@ -512,12 +576,15 @@ class VirtualWorkbench:
 
             self._update_data_status(f"加热台{station_id}加热中: {progress:.1f}%")
 
+            if time.time() - last_countdown_log >= 5.0:
+                self.logger.info(f"[加热台{station_id}] {material_id} 剩余 {remaining:.1f}s")
+                last_countdown_log = time.time()
+
             if elapsed >= self.HEATING_TIME:
                 break
 
             time.sleep(1.0)
 
-        # 加热完成
         with self._stations_lock:
             self._heating_stations[station_id].state = HeatingStationState.COMPLETED
             self._heating_stations[station_id].heating_progress = 100.0
@@ -536,10 +603,28 @@ class VirtualWorkbench:
             "material_number": material_number,
             "message": f"加热台{station_id}加热完成",
             "unilabos_samples": [
-                LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                sample_uuid, content in sample_uuids.items()]
+                LabSample(
+                    sample_uuid=sample_uuid,
+                    oss_path="",
+                    extra=(
+                        {"material_uuid": content}
+                        if isinstance(content, str) else (content.serialize() if content else {})
+                    ),
+                )
+                for sample_uuid, content in sample_uuids.items()
+            ],
         }
 
+    @action(
+        auto_prefix=True,
+        description="将物料从加热台移动到输出位置Cn",
+        handles=[
+            ActionInputHandle(key="output_station_input", data_type="workbench_station",
+                        label="加热台ID", data_key="station_id", data_source=DataSource.HANDLE),
+            ActionInputHandle(key="output_material_input", data_type="workbench_material",
+                        label="物料编号", data_key="material_number", data_source=DataSource.HANDLE),
+        ],
+    )
     def move_to_output(
         self,
         sample_uuids: SampleUUIDsType,
@@ -548,15 +633,8 @@ class VirtualWorkbench:
     ) -> MoveToOutputResult:
         """
         将物料从加热台移动到输出位置Cn
-
-        Args:
-            station_id: 加热台ID (1-3)，从 start_heating 的 handle 传入
-            material_number: 物料编号，从 start_heating 的 handle 传入，用于确定输出位置 Cn
-
-        Returns:
-            MoveToOutputResult: 包含执行结果
         """
-        output_number = material_number  # 物料编号决定输出位置
+        output_number = material_number
 
         if station_id not in self._heating_stations:
             return {
@@ -566,8 +644,16 @@ class VirtualWorkbench:
                 "output_position": f"C{output_number}",
                 "message": f"无效的加热台ID: {station_id}",
                 "unilabos_samples": [
-                    LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                    sample_uuid, content in sample_uuids.items()]
+                    LabSample(
+                        sample_uuid=sample_uuid,
+                        oss_path="",
+                        extra=(
+                            {"material_uuid": content}
+                            if isinstance(content, str) else (content.serialize() if content else {})
+                        ),
+                    )
+                    for sample_uuid, content in sample_uuids.items()
+                ],
             }
 
         with self._stations_lock:
@@ -582,8 +668,16 @@ class VirtualWorkbench:
                     "output_position": f"C{output_number}",
                     "message": f"加热台{station_id}上没有物料",
                     "unilabos_samples": [
-                        LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                        sample_uuid, content in sample_uuids.items()]
+                        LabSample(
+                            sample_uuid=sample_uuid,
+                            oss_path="",
+                            extra=(
+                                {"material_uuid": content}
+                                if isinstance(content, str) else (content.serialize() if content else {})
+                            ),
+                        )
+                        for sample_uuid, content in sample_uuids.items()
+                    ],
                 }
 
             if station.state != HeatingStationState.COMPLETED:
@@ -594,8 +688,16 @@ class VirtualWorkbench:
                     "output_position": f"C{output_number}",
                     "message": f"加热台{station_id}尚未完成加热 (当前状态: {station.state.value})",
                     "unilabos_samples": [
-                        LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                        sample_uuid, content in sample_uuids.items()]
+                        LabSample(
+                            sample_uuid=sample_uuid,
+                            oss_path="",
+                            extra=(
+                                {"material_uuid": content}
+                                if isinstance(content, str) else (content.serialize() if content else {})
+                            ),
+                        )
+                        for sample_uuid, content in sample_uuids.items()
+                    ],
                 }
 
         output_position = f"C{output_number}"
@@ -607,18 +709,17 @@ class VirtualWorkbench:
                 if material_id in self._active_tasks:
                     self._active_tasks[material_id]["status"] = "waiting_for_arm_output"
 
-            # 获取机械臂
             self._acquire_arm(task_desc)
 
             with self._tasks_lock:
                 if material_id in self._active_tasks:
                     self._active_tasks[material_id]["status"] = "arm_moving_to_output"
 
-            # 模拟机械臂操作 (3秒)
-            self.logger.info(f"[{material_id}] 机械臂正在从加热台{station_id}取出并移动到{output_position}...")
+            self.logger.info(
+                f"[{material_id}] 机械臂正在从加热台{station_id}取出并移动到{output_position}..."
+            )
             time.sleep(self.ARM_OPERATION_TIME)
 
-            # 清空加热台
             with self._stations_lock:
                 self._heating_stations[station_id].state = HeatingStationState.IDLE
                 self._heating_stations[station_id].current_material = None
@@ -626,17 +727,17 @@ class VirtualWorkbench:
                 self._heating_stations[station_id].heating_progress = 0.0
                 self._heating_stations[station_id].heating_start_time = None
 
-            # 释放机械臂
             self._release_arm()
 
-            # 任务完成
             with self._tasks_lock:
                 if material_id in self._active_tasks:
                     self._active_tasks[material_id]["status"] = "completed"
                     self._active_tasks[material_id]["end_time"] = time.time()
 
             self._update_data_status(f"{material_id}已移动到{output_position}")
-            self.logger.info(f"[{material_id}] 已成功移动到{output_position} (用时{self.ARM_OPERATION_TIME}s)")
+            self.logger.info(
+                f"[{material_id}] 已成功移动到{output_position} (用时{self.ARM_OPERATION_TIME}s)"
+            )
 
             return {
                 "success": True,
@@ -645,8 +746,17 @@ class VirtualWorkbench:
                 "output_position": output_position,
                 "message": f"{material_id}已成功移动到{output_position}",
                 "unilabos_samples": [
-                    LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                    sample_uuid, content in sample_uuids.items()]
+                    LabSample(
+                        sample_uuid=sample_uuid,
+                        oss_path="",
+                        extra=(
+                            {"material_uuid": content}
+                            if isinstance(content, str)
+                            else (content.serialize() if content is not None else {})
+                        ),
+                    )
+                    for sample_uuid, content in sample_uuids.items()
+                ],
             }
 
         except Exception as e:
@@ -660,83 +770,105 @@ class VirtualWorkbench:
                 "output_position": output_position,
                 "message": f"移动失败: {str(e)}",
                 "unilabos_samples": [
-                    LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for
-                    sample_uuid, content in sample_uuids.items()]
+                    LabSample(
+                        sample_uuid=sample_uuid,
+                        oss_path="",
+                        extra=(
+                            {"material_uuid": content}
+                            if isinstance(content, str) else (content.serialize() if content else {})
+                        ),
+                    )
+                    for sample_uuid, content in sample_uuids.items()
+                ],
             }
 
     # ============ 状态属性 ============
 
     @property
+    @topic_config()
     def status(self) -> str:
         return self.data.get("status", "Unknown")
 
     @property
+    @topic_config()
     def arm_state(self) -> str:
         return self._arm_state.value
 
     @property
+    @topic_config()
     def arm_current_task(self) -> str:
         return self._arm_current_task or ""
 
     @property
+    @topic_config()
     def heating_station_1_state(self) -> str:
         with self._stations_lock:
             station = self._heating_stations.get(1)
             return station.state.value if station else "unknown"
 
     @property
+    @topic_config()
     def heating_station_1_material(self) -> str:
         with self._stations_lock:
             station = self._heating_stations.get(1)
             return station.current_material or "" if station else ""
 
     @property
+    @topic_config()
     def heating_station_1_progress(self) -> float:
         with self._stations_lock:
             station = self._heating_stations.get(1)
             return station.heating_progress if station else 0.0
 
     @property
+    @topic_config()
     def heating_station_2_state(self) -> str:
         with self._stations_lock:
             station = self._heating_stations.get(2)
             return station.state.value if station else "unknown"
 
     @property
+    @topic_config()
     def heating_station_2_material(self) -> str:
         with self._stations_lock:
             station = self._heating_stations.get(2)
             return station.current_material or "" if station else ""
 
     @property
+    @topic_config()
     def heating_station_2_progress(self) -> float:
         with self._stations_lock:
             station = self._heating_stations.get(2)
             return station.heating_progress if station else 0.0
 
     @property
+    @topic_config()
     def heating_station_3_state(self) -> str:
         with self._stations_lock:
             station = self._heating_stations.get(3)
             return station.state.value if station else "unknown"
 
     @property
+    @topic_config()
     def heating_station_3_material(self) -> str:
         with self._stations_lock:
             station = self._heating_stations.get(3)
             return station.current_material or "" if station else ""
 
     @property
+    @topic_config()
     def heating_station_3_progress(self) -> float:
         with self._stations_lock:
             station = self._heating_stations.get(3)
             return station.heating_progress if station else 0.0
 
     @property
+    @topic_config()
     def active_tasks_count(self) -> int:
         with self._tasks_lock:
             return len(self._active_tasks)
 
     @property
+    @topic_config()
     def message(self) -> str:
         return self.data.get("message", "")

unilabos/registry/decorators.py

@@ -0,0 +1,614 @@
+"""
+装饰器注册表系统
+
+通过 @device, @action, @resource 装饰器替代 YAML 配置文件来定义设备/动作/资源注册表信息。
+
+Usage:
+    from unilabos.registry.decorators import (
+        device, action, resource,
+        InputHandle, OutputHandle,
+        ActionInputHandle, ActionOutputHandle,
+        HardwareInterface, Side, DataSource,
+    )
+
+    @device(
+        id="solenoid_valve.mock",
+        category=["pump_and_valve"],
+        description="模拟电磁阀设备",
+        handles=[
+            InputHandle(key="in", data_type="fluid", label="in", side=Side.NORTH),
+            OutputHandle(key="out", data_type="fluid", label="out", side=Side.SOUTH),
+        ],
+        hardware_interface=HardwareInterface(
+            name="hardware_interface",
+            read="send_command",
+            write="send_command",
+        ),
+    )
+    class SolenoidValveMock:
+        @action(action_type=EmptyIn)
+        def close(self):
+            ...
+
+        @action(
+            handles=[
+                ActionInputHandle(key="in", data_type="fluid", label="in"),
+                ActionOutputHandle(key="out", data_type="fluid", label="out"),
+            ],
+        )
+        def set_valve_position(self, position):
+            ...
+
+        # 无 @action 装饰器 => auto- 前缀动作
+        def is_open(self):
+            ...
+"""
+
+from enum import Enum
+from functools import wraps
+from typing import Any, Callable, Dict, List, Optional, TypeVar
+
+from pydantic import BaseModel, ConfigDict, Field
+
+F = TypeVar("F", bound=Callable[..., Any])
+
+# ---------------------------------------------------------------------------
+# 枚举
+# ---------------------------------------------------------------------------
+
+
+class Side(str, Enum):
+    """UI 上 Handle 的显示位置"""
+
+    NORTH = "NORTH"
+    SOUTH = "SOUTH"
+    EAST = "EAST"
+    WEST = "WEST"
+
+
+class DataSource(str, Enum):
+    """Handle 的数据来源"""
+
+    HANDLE = "handle"  # 从上游 handle 获取数据 (用于 InputHandle)
+    EXECUTOR = "executor"  # 从执行器输出数据 (用于 OutputHandle)
+
+
+# ---------------------------------------------------------------------------
+# Device / Resource Handle (设备/资源级别端口, 序列化时包含 io_type)
+# ---------------------------------------------------------------------------
+
+
+class _DeviceHandleBase(BaseModel):
+    """设备/资源端口基类 (内部使用)"""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    key: str = Field(serialization_alias="handler_key")
+    data_type: str
+    label: str
+    side: Optional[Side] = None
+    data_key: Optional[str] = None
+    data_source: Optional[str] = None
+    description: Optional[str] = None
+
+    # 子类覆盖
+    io_type: str = ""
+
+    def to_registry_dict(self) -> Dict[str, Any]:
+        return self.model_dump(by_alias=True, exclude_none=True)
+
+
+class InputHandle(_DeviceHandleBase):
+    """
+    输入端口 (io_type="target"), 用于 @device / @resource handles
+
+    Example:
+        InputHandle(key="in", data_type="fluid", label="in", side=Side.NORTH)
+    """
+
+    io_type: str = "target"
+
+
+class OutputHandle(_DeviceHandleBase):
+    """
+    输出端口 (io_type="source"), 用于 @device / @resource handles
+
+    Example:
+        OutputHandle(key="out", data_type="fluid", label="out", side=Side.SOUTH)
+    """
+
+    io_type: str = "source"
+
+
+# ---------------------------------------------------------------------------
+# Action Handle (动作级别端口, 序列化时不含 io_type, 按类型自动分组)
+# ---------------------------------------------------------------------------
+
+
+class _ActionHandleBase(BaseModel):
+    """动作端口基类 (内部使用)"""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    key: str = Field(serialization_alias="handler_key")
+    data_type: str
+    label: str
+    side: Optional[Side] = None
+    data_key: Optional[str] = None
+    data_source: Optional[str] = None
+    description: Optional[str] = None
+    io_type: Optional[str] = None  # source/sink (dataflow) or target/source (device-style)
+
+    def to_registry_dict(self) -> Dict[str, Any]:
+        return self.model_dump(by_alias=True, exclude_none=True)
+
+
+class ActionInputHandle(_ActionHandleBase):
+    """
+    动作输入端口, 用于 @action handles, 序列化后归入 "input" 组
+
+    Example:
+        ActionInputHandle(
+            key="material_input", data_type="workbench_material",
+            label="物料编号", data_key="material_number", data_source="handle",
+        )
+    """
+
+    pass
+
+
+class ActionOutputHandle(_ActionHandleBase):
+    """
+    动作输出端口, 用于 @action handles, 序列化后归入 "output" 组
+
+    Example:
+        ActionOutputHandle(
+            key="station_output", data_type="workbench_station",
+            label="加热台ID", data_key="station_id", data_source="executor",
+        )
+    """
+
+    pass
+
+
+# ---------------------------------------------------------------------------
+# HardwareInterface
+# ---------------------------------------------------------------------------
+
+
+class HardwareInterface(BaseModel):
+    """
+    硬件通信接口定义
+
+    描述设备与底层硬件通信的方式 (串口、Modbus 等)。
+
+    Example:
+        HardwareInterface(name="hardware_interface", read="send_command", write="send_command")
+    """
+
+    name: str
+    read: Optional[str] = None
+    write: Optional[str] = None
+    extra_info: Optional[List[str]] = None
+
+
+# ---------------------------------------------------------------------------
+# 全局注册表 -- 记录所有被装饰器标记的类/函数
+# ---------------------------------------------------------------------------
+_registered_devices: Dict[str, type] = {}  # device_id -> class
+_registered_resources: Dict[str, Any] = {}  # resource_id -> class or function
+
+
+def _device_handles_to_list(
+    handles: Optional[List[_DeviceHandleBase]],
+) -> List[Dict[str, Any]]:
+    """将设备/资源 Handle 列表序列化为字典列表 (含 io_type)"""
+    if handles is None:
+        return []
+    return [h.to_registry_dict() for h in handles]
+
+
+def _action_handles_to_dict(
+    handles: Optional[List[_ActionHandleBase]],
+) -> Dict[str, Any]:
+    """
+    将动作 Handle 列表序列化为 {"input": [...], "output": [...]} 格式。
+
+    ActionInputHandle => "input", ActionOutputHandle => "output"
+    """
+    if handles is None:
+        return {}
+    input_list = [h.to_registry_dict() for h in handles if isinstance(h, ActionInputHandle)]
+    output_list = [h.to_registry_dict() for h in handles if isinstance(h, ActionOutputHandle)]
+    result: Dict[str, Any] = {}
+    if input_list:
+        result["input"] = input_list
+    if output_list:
+        result["output"] = output_list
+    return result
+
+
+# ---------------------------------------------------------------------------
+# @device 类装饰器
+# ---------------------------------------------------------------------------
+
+
+# noinspection PyShadowingBuiltins
+def device(
+    id: Optional[str] = None,
+    ids: Optional[List[str]] = None,
+    id_meta: Optional[Dict[str, Dict[str, Any]]] = None,
+    category: Optional[List[str]] = None,
+    description: str = "",
+    display_name: str = "",
+    icon: str = "",
+    version: str = "1.0.0",
+    handles: Optional[List[_DeviceHandleBase]] = None,
+    model: Optional[Dict[str, Any]] = None,
+    device_type: str = "python",
+    hardware_interface: Optional[HardwareInterface] = None,
+):
+    """
+    设备类装饰器
+
+    将类标记为一个 UniLab-OS 设备，并附加注册表元数据。
+
+    支持两种模式:
+      1. 单设备: id="xxx", category=[...]
+      2. 多设备: ids=["id1","id2"], id_meta={"id1":{handles:[...]}, "id2":{...}}
+
+    Args:
+        id: 单设备时的注册表唯一标识
+        ids: 多设备时的 id 列表，与 id_meta 配合使用
+        id_meta: 每个 device_id 的覆盖元数据 (handles/description/icon/model)
+        category: 设备分类标签列表 (必填)
+        description: 设备描述
+        display_name: 人类可读的设备显示名称，缺失时默认使用 id
+        icon: 图标路径
+        version: 版本号
+        handles: 设备端口列表 (单设备或 id_meta 未覆盖时使用)
+        model: 可选的 3D 模型配置
+        device_type: 设备实现类型 ("python" / "ros2")
+        hardware_interface: 硬件通信接口 (HardwareInterface)
+    """
+    # Resolve device ids
+    if ids is not None:
+        device_ids = list(ids)
+        if not device_ids:
+            raise ValueError("@device ids 不能为空")
+        id_meta = id_meta or {}
+    elif id is not None:
+        device_ids = [id]
+        id_meta = {}
+    else:
+        raise ValueError("@device 必须提供 id 或 ids")
+
+    if category is None:
+        raise ValueError("@device category 必填")
+
+    base_meta = {
+        "category": category,
+        "description": description,
+        "display_name": display_name,
+        "icon": icon,
+        "version": version,
+        "handles": _device_handles_to_list(handles),
+        "model": model,
+        "device_type": device_type,
+        "hardware_interface": (hardware_interface.model_dump(exclude_none=True) if hardware_interface else None),
+    }
+
+    def decorator(cls):
+        cls._device_registry_meta = base_meta
+        cls._device_registry_id_meta = id_meta
+        cls._device_registry_ids = device_ids
+
+        for did in device_ids:
+            if did in _registered_devices:
+                raise ValueError(f"@device id 重复: '{did}' 已被 {_registered_devices[did]} 注册")
+            _registered_devices[did] = cls
+
+        return cls
+
+    return decorator
+
+
+# ---------------------------------------------------------------------------
+# @action 方法装饰器
+# ---------------------------------------------------------------------------
+
+# 区分 "用户没传 action_type" 和 "用户传了 None"
+_ACTION_TYPE_UNSET = object()
+
+
+# noinspection PyShadowingNames
+def action(
+    action_type: Any = _ACTION_TYPE_UNSET,
+    goal: Optional[Dict[str, str]] = None,
+    feedback: Optional[Dict[str, str]] = None,
+    result: Optional[Dict[str, str]] = None,
+    handles: Optional[List[_ActionHandleBase]] = None,
+    goal_default: Optional[Dict[str, Any]] = None,
+    placeholder_keys: Optional[Dict[str, str]] = None,
+    always_free: bool = False,
+    is_protocol: bool = False,
+    description: str = "",
+    auto_prefix: bool = False,
+    parent: bool = False,
+):
+    """
+    动作方法装饰器
+
+    标记方法为注册表动作。有三种用法:
+      1. @action(action_type=EmptyIn, ...)  -- 非 auto, 使用指定 ROS Action 类型
+      2. @action()                          -- 非 auto, UniLabJsonCommand (从方法签名生成 schema)
+      3. 不加 @action                       -- auto- 前缀, UniLabJsonCommand
+
+    Protocol 用法:
+      @action(action_type=Add, is_protocol=True)
+      def AddProtocol(self): ...
+      标记该动作为高级协议 (protocol)，运行时通过 ROS Action 路由到
+      protocol generator 执行。action_type 指向 unilabos_msgs 的 Action 类型。
+
+    Args:
+        action_type: ROS Action 消息类型 (如 EmptyIn, SendCmd, HeatChill).
+                     不传/默认 = UniLabJsonCommand (非 auto).
+        goal: Goal 字段映射 (ROS字段名 -> 设备参数名).
+              protocol 模式下可留空，系统自动生成 identity 映射.
+        feedback: Feedback 字段映射
+        result: Result 字段映射
+        handles: 动作端口列表 (ActionInputHandle / ActionOutputHandle)
+        goal_default: Goal 字段默认值映射 (字段名 -> 默认值), 与自动生成的 goal_default 合并
+        placeholder_keys: 参数占位符配置
+        always_free: 是否为永久闲置动作 (不受排队限制)
+        is_protocol: 是否为工作站协议 (protocol)。True 时运行时走 protocol generator 路径。
+        description: 动作描述
+        auto_prefix: 若为 True，动作名使用 auto-{method_name} 形式（与无 @action 时一致）
+        parent: 若为 True，当方法参数为空 (*args, **kwargs) 时，通过 MRO 从父类获取真实方法参数
+    """
+
+    def decorator(func: F) -> F:
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            return func(*args, **kwargs)
+
+        # action_type 为哨兵值 => 用户没传, 视为 None (UniLabJsonCommand)
+        resolved_type = None if action_type is _ACTION_TYPE_UNSET else action_type
+
+        meta = {
+            "action_type": resolved_type,
+            "goal": goal or {},
+            "feedback": feedback or {},
+            "result": result or {},
+            "handles": _action_handles_to_dict(handles),
+            "goal_default": goal_default or {},
+            "placeholder_keys": placeholder_keys or {},
+            "always_free": always_free,
+            "is_protocol": is_protocol,
+            "description": description,
+            "auto_prefix": auto_prefix,
+            "parent": parent,
+        }
+        wrapper._action_registry_meta = meta  # type: ignore[attr-defined]
+
+        # 设置 _is_always_free 保持与旧 @always_free 装饰器兼容
+        if always_free:
+            wrapper._is_always_free = True  # type: ignore[attr-defined]
+
+        return wrapper  # type: ignore[return-value]
+
+    return decorator
+
+
+def get_action_meta(func) -> Optional[Dict[str, Any]]:
+    """获取方法上的 @action 装饰器元数据"""
+    return getattr(func, "_action_registry_meta", None)
+
+
+def has_action_decorator(func) -> bool:
+    """检查函数是否带有 @action 装饰器"""
+    return hasattr(func, "_action_registry_meta")
+
+
+# ---------------------------------------------------------------------------
+# @resource 类/函数装饰器
+# ---------------------------------------------------------------------------
+
+
+def resource(
+    id: str,
+    category: List[str],
+    description: str = "",
+    icon: str = "",
+    version: str = "1.0.0",
+    handles: Optional[List[_DeviceHandleBase]] = None,
+    model: Optional[Dict[str, Any]] = None,
+    class_type: str = "pylabrobot",
+):
+    """
+    资源类/函数装饰器
+
+    将类或工厂函数标记为一个 UniLab-OS 资源，附加注册表元数据。
+
+    Args:
+        id: 注册表唯一标识 (必填, 不可重复)
+        category: 资源分类标签列表 (必填)
+        description: 资源描述
+        icon: 图标路径
+        version: 版本号
+        handles: 端口列表 (InputHandle / OutputHandle)
+        model: 可选的 3D 模型配置
+        class_type: 资源实现类型 ("python" / "pylabrobot" / "unilabos")
+    """
+
+    def decorator(obj):
+        meta = {
+            "resource_id": id,
+            "category": category,
+            "description": description,
+            "icon": icon,
+            "version": version,
+            "handles": _device_handles_to_list(handles),
+            "model": model,
+            "class_type": class_type,
+        }
+        obj._resource_registry_meta = meta
+
+        if id in _registered_resources:
+            raise ValueError(f"@resource id 重复: '{id}' 已被 {_registered_resources[id]} 注册")
+        _registered_resources[id] = obj
+
+        return obj
+
+    return decorator
+
+
+def get_device_meta(cls, device_id: Optional[str] = None) -> Optional[Dict[str, Any]]:
+    """
+    获取类上的 @device 装饰器元数据。
+
+    当 device_id 存在且类使用 ids+id_meta 时，返回合并后的 meta
+    (base_meta 与 id_meta[device_id] 深度合并)。
+    """
+    base = getattr(cls, "_device_registry_meta", None)
+    if base is None:
+        return None
+    id_meta = getattr(cls, "_device_registry_id_meta", None) or {}
+    if device_id is None or device_id not in id_meta:
+        result = dict(base)
+        ids = getattr(cls, "_device_registry_ids", None)
+        result["device_id"] = device_id if device_id is not None else (ids[0] if ids else None)
+        return result
+
+    overrides = id_meta[device_id]
+    result = dict(base)
+    result["device_id"] = device_id
+    for key in ["handles", "description", "icon", "model"]:
+        if key in overrides:
+            val = overrides[key]
+            if key == "handles" and isinstance(val, list):
+                # handles 必须是 Handle 对象列表
+                result[key] = [h.to_registry_dict() for h in val]
+            else:
+                result[key] = val
+    return result
+
+
+def get_resource_meta(obj) -> Optional[Dict[str, Any]]:
+    """获取对象上的 @resource 装饰器元数据"""
+    return getattr(obj, "_resource_registry_meta", None)
+
+
+def get_all_registered_devices() -> Dict[str, type]:
+    """获取所有已注册的设备类"""
+    return _registered_devices.copy()
+
+
+def get_all_registered_resources() -> Dict[str, Any]:
+    """获取所有已注册的资源"""
+    return _registered_resources.copy()
+
+
+def clear_registry():
+    """清空全局注册表 (用于测试)"""
+    _registered_devices.clear()
+    _registered_resources.clear()
+
+
+# ---------------------------------------------------------------------------
+# topic_config / not_action / always_free 装饰器
+# ---------------------------------------------------------------------------
+
+
+def topic_config(
+    period: Optional[float] = None,
+    print_publish: Optional[bool] = None,
+    qos: Optional[int] = None,
+    name: Optional[str] = None,
+) -> Callable[[F], F]:
+    """
+    Topic发布配置装饰器
+
+    用于装饰 get_{attr_name} 方法或 @property，控制对应属性的ROS topic发布行为。
+
+    Args:
+        period: 发布周期（秒）。None 表示使用默认值 5.0
+        print_publish: 是否打印发布日志。None 表示使用节点默认配置
+        qos: QoS深度配置。None 表示使用默认值 10
+        name: 自定义发布名称。None 表示使用方法名（去掉 get_ 前缀）
+
+    Note:
+        与 @property 连用时，@topic_config 必须放在 @property 下面，
+        这样装饰器执行顺序为：先 topic_config 添加配置，再 property 包装。
+    """
+
+    def decorator(func: F) -> F:
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            return func(*args, **kwargs)
+
+        wrapper._topic_period = period  # type: ignore[attr-defined]
+        wrapper._topic_print_publish = print_publish  # type: ignore[attr-defined]
+        wrapper._topic_qos = qos  # type: ignore[attr-defined]
+        wrapper._topic_name = name  # type: ignore[attr-defined]
+        wrapper._has_topic_config = True  # type: ignore[attr-defined]
+
+        return wrapper  # type: ignore[return-value]
+
+    return decorator
+
+
+def get_topic_config(func) -> dict:
+    """获取函数上的 topic 配置 (period, print_publish, qos, name)"""
+    if hasattr(func, "_has_topic_config") and getattr(func, "_has_topic_config", False):
+        return {
+            "period": getattr(func, "_topic_period", None),
+            "print_publish": getattr(func, "_topic_print_publish", None),
+            "qos": getattr(func, "_topic_qos", None),
+            "name": getattr(func, "_topic_name", None),
+        }
+    return {}
+
+
+def always_free(func: F) -> F:
+    """
+    标记动作为永久闲置(不受busy队列限制)的装饰器
+
+    被此装饰器标记的 action 方法，在执行时不会受到设备级别的排队限制，
+    任何时候请求都可以立即执行。适用于查询类、状态读取类等轻量级操作。
+    """
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        return func(*args, **kwargs)
+
+    wrapper._is_always_free = True  # type: ignore[attr-defined]
+
+    return wrapper  # type: ignore[return-value]
+
+
+def is_always_free(func) -> bool:
+    """检查函数是否被标记为永久闲置"""
+    return getattr(func, "_is_always_free", False)
+
+
+def not_action(func: F) -> F:
+    """
+    标记方法为非动作的装饰器
+
+    用于装饰 driver 类中的方法，使其在注册表扫描时不被识别为动作。
+    适用于辅助方法、内部工具方法等不应暴露为设备动作的公共方法。
+    """
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        return func(*args, **kwargs)
+
+    wrapper._is_not_action = True  # type: ignore[attr-defined]
+
+    return wrapper  # type: ignore[return-value]
+
+
+def is_not_action(func) -> bool:
+    """检查函数是否被标记为非动作"""
+    return getattr(func, "_is_not_action", False)

unilabos/registry/device_comms/communication_devices.yaml

@@ -96,10 +96,13 @@ serial:
           type: string
         port:
           type: string
+        registry_name:
+          type: string
         resource_tracker:
           type: object
       required:
       - device_id
+      - registry_name
       - port
       type: object
     data:

unilabos/registry/devices/camera.yaml

@@ -67,6 +67,9 @@ camera:
         period:
           default: 0.1
           type: number
+        registry_name:
+          default: ''
+          type: string
         resource_tracker:
           type: object
       required: []

unilabos/registry/devices/virtual_device.yaml

@@ -6090,6 +6090,7 @@ virtual_workbench:
           type: object
         type: UniLabJsonCommand
       auto-start_heating:
+        always_free: true
         feedback: {}
         goal: {}
         goal_default:

unilabos/registry/utils.py

@@ -0,0 +1,699 @@
+"""
+注册表工具函数
+
+从 registry.py 中提取的纯工具函数，包括：
+- docstring 解析
+- 类型字符串 → JSON Schema 转换
+- AST 类型节点解析
+- TypedDict / Slot / Handle 等辅助检测
+"""
+
+import inspect
+import logging
+import re
+import typing
+from typing import Any, Dict, List, Optional, Tuple, Union
+
+from msgcenterpy.instances.typed_dict_instance import TypedDictMessageInstance
+
+from unilabos.utils.cls_creator import import_class
+
+_logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# 异常
+# ---------------------------------------------------------------------------
+
+
+class ROSMsgNotFound(Exception):
+    pass
+
+
+# ---------------------------------------------------------------------------
+# Docstring 解析 (Google-style)
+# ---------------------------------------------------------------------------
+
+_SECTION_RE = re.compile(r"^(\w[\w\s]*):\s*$")
+
+
+def parse_docstring(docstring: Optional[str]) -> Dict[str, Any]:
+    """
+    解析 Google-style docstring，提取描述和参数说明。
+
+    Returns:
+        {"description": "短描述", "params": {"param1": "参数1描述", ...}}
+    """
+    result: Dict[str, Any] = {"description": "", "params": {}}
+    if not docstring:
+        return result
+
+    lines = docstring.strip().splitlines()
+    if not lines:
+        return result
+
+    result["description"] = lines[0].strip()
+
+    in_args = False
+    current_param: Optional[str] = None
+    current_desc_parts: list = []
+
+    for line in lines[1:]:
+        stripped = line.strip()
+        section_match = _SECTION_RE.match(stripped)
+        if section_match:
+            if current_param is not None:
+                result["params"][current_param] = "\n".join(current_desc_parts).strip()
+                current_param = None
+                current_desc_parts = []
+            section_name = section_match.group(1).lower()
+            in_args = section_name in ("args", "arguments", "parameters", "params")
+            continue
+
+        if not in_args:
+            continue
+
+        if ":" in stripped and not stripped.startswith(" "):
+            if current_param is not None:
+                result["params"][current_param] = "\n".join(current_desc_parts).strip()
+            param_part, _, desc_part = stripped.partition(":")
+            param_name = param_part.strip().split("(")[0].strip()
+            current_param = param_name
+            current_desc_parts = [desc_part.strip()]
+        elif current_param is not None:
+            aline = line
+            if aline.startswith("    "):
+                aline = aline[4:]
+            elif aline.startswith("\t"):
+                aline = aline[1:]
+            current_desc_parts.append(aline.strip())
+
+    if current_param is not None:
+        result["params"][current_param] = "\n".join(current_desc_parts).strip()
+
+    return result
+
+
+# ---------------------------------------------------------------------------
+# 类型常量
+# ---------------------------------------------------------------------------
+
+SIMPLE_TYPE_MAP = {
+    "str": "string",
+    "string": "string",
+    "int": "integer",
+    "integer": "integer",
+    "float": "number",
+    "number": "number",
+    "bool": "boolean",
+    "boolean": "boolean",
+    "list": "array",
+    "array": "array",
+    "dict": "object",
+    "object": "object",
+}
+
+ARRAY_TYPES = {"list", "List", "tuple", "Tuple", "set", "Set", "Sequence", "Iterable"}
+OBJECT_TYPES = {"dict", "Dict", "Mapping"}
+WRAPPER_TYPES = {"Optional"}
+SLOT_TYPES = {"ResourceSlot", "DeviceSlot"}
+
+
+# ---------------------------------------------------------------------------
+# 简单类型映射
+# ---------------------------------------------------------------------------
+
+
+def get_json_schema_type(type_str: str) -> str:
+    """简单类型名 -> JSON Schema type"""
+    return SIMPLE_TYPE_MAP.get(type_str.lower(), "string")
+
+
+# ---------------------------------------------------------------------------
+# AST 类型解析
+# ---------------------------------------------------------------------------
+
+
+def parse_type_node(type_str: str):
+    """将类型注解字符串解析为 AST 节点，失败返回 None。"""
+    import ast as _ast
+
+    try:
+        return _ast.parse(type_str.strip(), mode="eval").body
+    except Exception:
+        return None
+
+
+def _collect_bitor(node, out: list):
+    """递归收集 X | Y | Z 的所有分支。"""
+    import ast as _ast
+
+    if isinstance(node, _ast.BinOp) and isinstance(node.op, _ast.BitOr):
+        _collect_bitor(node.left, out)
+        _collect_bitor(node.right, out)
+    else:
+        out.append(node)
+
+
+def type_node_to_schema(
+    node,
+    import_map: Optional[Dict[str, str]] = None,
+) -> Dict[str, Any]:
+    """将 AST 类型注解节点递归转换为 JSON Schema dict。
+
+    当提供 import_map 时，对于未知类名会尝试通过 import_map 解析模块路径，
+    然后 import 真实类型对象来生成 schema (支持 TypedDict 等)。
+
+    映射规则:
+    - Optional[X]                    → X 的 schema (剥掉 Optional)
+    - Union[X, Y]                    → {"anyOf": [X_schema, Y_schema]}
+    - List[X] / Tuple[X] / Set[X]   → {"type": "array", "items": X_schema}
+    - Dict[K, V]                     → {"type": "object", "additionalProperties": V_schema}
+    - Literal["a", "b"]             → {"type": "string", "enum": ["a", "b"]}
+    - TypedDict (via import_map)     → {"type": "object", "properties": {...}}
+    - 基本类型 str/int/...          → {"type": "string"/"integer"/...}
+    """
+    import ast as _ast
+
+    # --- Name 节点: str / int / dict / ResourceSlot / 自定义类 ---
+    if isinstance(node, _ast.Name):
+        name = node.id
+        if name in SLOT_TYPES:
+            return {"$slot": name}
+        json_type = SIMPLE_TYPE_MAP.get(name.lower())
+        if json_type:
+            return {"type": json_type}
+        # 尝试通过 import_map 解析并 import 真实类型
+        if import_map and name in import_map:
+            type_obj = resolve_type_object(import_map[name])
+            if type_obj is not None:
+                return type_to_schema(type_obj)
+        # 未知类名 → 无法转 schema 的自定义类型默认当 object
+        return {"type": "object"}
+
+    if isinstance(node, _ast.Constant):
+        if isinstance(node.value, str):
+            return {"type": SIMPLE_TYPE_MAP.get(node.value.lower(), "string")}
+        return {"type": "string"}
+
+    # --- Subscript 节点: List[X], Dict[K,V], Optional[X], Literal[...] 等 ---
+    if isinstance(node, _ast.Subscript):
+        base_name = node.value.id if isinstance(node.value, _ast.Name) else ""
+
+        # Optional[X] → 剥掉
+        if base_name in WRAPPER_TYPES:
+            return type_node_to_schema(node.slice, import_map)
+
+        # Union[X, None] → 剥掉 None; Union[X, Y] → anyOf
+        if base_name == "Union":
+            elts = node.slice.elts if isinstance(node.slice, _ast.Tuple) else [node.slice]
+            non_none = [
+                e
+                for e in elts
+                if not (isinstance(e, _ast.Constant) and e.value is None)
+                and not (isinstance(e, _ast.Name) and e.id == "None")
+            ]
+            if len(non_none) == 1:
+                return type_node_to_schema(non_none[0], import_map)
+            if len(non_none) > 1:
+                return {"anyOf": [type_node_to_schema(e, import_map) for e in non_none]}
+            return {"type": "string"}
+
+        # Literal["a", "b", 1] → enum
+        if base_name == "Literal":
+            elts = node.slice.elts if isinstance(node.slice, _ast.Tuple) else [node.slice]
+            values = []
+            for e in elts:
+                if isinstance(e, _ast.Constant):
+                    values.append(e.value)
+                elif isinstance(e, _ast.Name):
+                    values.append(e.id)
+            if values:
+                return {"type": "string", "enum": values}
+            return {"type": "string"}
+
+        # List / Tuple / Set → array
+        if base_name in ARRAY_TYPES:
+            if isinstance(node.slice, _ast.Tuple) and node.slice.elts:
+                inner_node = node.slice.elts[0]
+            else:
+                inner_node = node.slice
+            return {"type": "array", "items": type_node_to_schema(inner_node, import_map)}
+
+        # Dict → object
+        if base_name in OBJECT_TYPES:
+            schema: Dict[str, Any] = {"type": "object"}
+            if isinstance(node.slice, _ast.Tuple) and len(node.slice.elts) >= 2:
+                val_node = node.slice.elts[1]
+                # Dict[str, Any] → 不加 additionalProperties (Any 等同于无约束)
+                is_any = (isinstance(val_node, _ast.Name) and val_node.id == "Any") or (
+                    isinstance(val_node, _ast.Constant) and val_node.value is None
+                )
+                if not is_any:
+                    val_schema = type_node_to_schema(val_node, import_map)
+                    schema["additionalProperties"] = val_schema
+            return schema
+
+    # --- BinOp: X | Y (Python 3.10+) → 当 Union 处理 ---
+    if isinstance(node, _ast.BinOp) and isinstance(node.op, _ast.BitOr):
+        parts: list = []
+        _collect_bitor(node, parts)
+        non_none = [
+            p
+            for p in parts
+            if not (isinstance(p, _ast.Constant) and p.value is None)
+            and not (isinstance(p, _ast.Name) and p.id == "None")
+        ]
+        if len(non_none) == 1:
+            return type_node_to_schema(non_none[0], import_map)
+        if len(non_none) > 1:
+            return {"anyOf": [type_node_to_schema(p, import_map) for p in non_none]}
+        return {"type": "string"}
+
+    return {"type": "string"}
+
+
+# ---------------------------------------------------------------------------
+# 真实类型对象解析 (import-based)
+# ---------------------------------------------------------------------------
+
+
+def resolve_type_object(type_ref: str) -> Optional[Any]:
+    """通过 'module.path:ClassName' 格式的引用 import 并返回真实类型对象。
+
+    对于 typing 内置名 (str, int, List 等) 直接返回 None (由 AST 路径处理)。
+    import 失败时静默返回 None。
+    """
+    if ":" not in type_ref:
+        return None
+    try:
+        return import_class(type_ref)
+    except Exception:
+        return None
+
+
+def is_typed_dict_class(obj: Any) -> bool:
+    """检查对象是否是 TypedDict 类。"""
+    if obj is None:
+        return False
+    try:
+        from typing_extensions import is_typeddict
+
+        return is_typeddict(obj)
+    except ImportError:
+        if isinstance(obj, type):
+            return hasattr(obj, "__required_keys__") and hasattr(obj, "__optional_keys__")
+        return False
+
+
+def type_to_schema(tp: Any) -> Dict[str, Any]:
+    """将真实 typing 对象递归转换为 JSON Schema dict。
+
+    支持:
+    - 基本类型: str, int, float, bool → {"type": "string"/"integer"/...}
+    - typing 泛型: List[X], Dict[K,V], Optional[X], Union[X,Y], Literal[...]
+    - TypedDict → {"type": "object", "properties": {...}, "required": [...]}
+    - 自定义类 (ResourceSlot 等) → {"$slot": "..."} 或 {"type": "string"}
+    """
+    origin = getattr(tp, "__origin__", None)
+    args = getattr(tp, "__args__", None)
+
+    # --- None / NoneType ---
+    if tp is type(None):
+        return {"type": "null"}
+
+    # --- 基本类型 ---
+    if tp is str:
+        return {"type": "string"}
+    if tp is int:
+        return {"type": "integer"}
+    if tp is float:
+        return {"type": "number"}
+    if tp is bool:
+        return {"type": "boolean"}
+
+    # --- TypedDict ---
+    if is_typed_dict_class(tp):
+        try:
+            return TypedDictMessageInstance.get_json_schema_from_typed_dict(tp)
+        except Exception:
+            return {"type": "object"}
+
+    # --- Literal ---
+    if origin is typing.Literal:
+        values = list(args) if args else []
+        return {"type": "string", "enum": values}
+
+    # --- Optional / Union ---
+    if origin is typing.Union:
+        non_none = [a for a in (args or ()) if a is not type(None)]
+        if len(non_none) == 1:
+            return type_to_schema(non_none[0])
+        if len(non_none) > 1:
+            return {"anyOf": [type_to_schema(a) for a in non_none]}
+        return {"type": "string"}
+
+    # --- List / Sequence / Set / Tuple / Iterable ---
+    if origin in (list, tuple, set, frozenset) or (
+        origin is not None
+        and getattr(origin, "__name__", "") in ("Sequence", "Iterable", "Iterator", "MutableSequence")
+    ):
+        if args:
+            return {"type": "array", "items": type_to_schema(args[0])}
+        return {"type": "array"}
+
+    # --- Dict / Mapping ---
+    if origin in (dict,) or (origin is not None and getattr(origin, "__name__", "") in ("Mapping", "MutableMapping")):
+        schema: Dict[str, Any] = {"type": "object"}
+        if args and len(args) >= 2:
+            schema["additionalProperties"] = type_to_schema(args[1])
+        return schema
+
+    # --- Slot 类型 ---
+    if isinstance(tp, type):
+        name = tp.__name__
+        if name in SLOT_TYPES:
+            return {"$slot": name}
+
+    # --- 其他未知类型 fallback ---
+    if isinstance(tp, type):
+        return {"type": "object"}
+    return {"type": "string"}
+
+
+# ---------------------------------------------------------------------------
+# Slot / Placeholder 检测
+# ---------------------------------------------------------------------------
+
+
+def detect_slot_type(ptype) -> Tuple[Optional[str], bool]:
+    """检测参数类型是否为 ResourceSlot / DeviceSlot。
+
+    兼容多种格式:
+    - runtime: "unilabos.registry.placeholder_type:ResourceSlot"
+    - runtime tuple: ("list", "unilabos.registry.placeholder_type:ResourceSlot")
+    - AST 裸名: "ResourceSlot", "List[ResourceSlot]", "Optional[ResourceSlot]"
+
+    Returns: (slot_name | None, is_list)
+    """
+    ptype_str = str(ptype)
+
+    # 快速路径: 字符串里根本没有 Slot
+    if "ResourceSlot" not in ptype_str and "DeviceSlot" not in ptype_str:
+        return (None, False)
+
+    # runtime 格式: 完整模块路径
+    if isinstance(ptype, str):
+        if ptype.endswith(":ResourceSlot") or ptype == "ResourceSlot":
+            return ("ResourceSlot", False)
+        if ptype.endswith(":DeviceSlot") or ptype == "DeviceSlot":
+            return ("DeviceSlot", False)
+        # AST 复杂格式: List[ResourceSlot], Optional[ResourceSlot] 等
+        if "[" in ptype:
+            node = parse_type_node(ptype)
+            if node is not None:
+                schema = type_node_to_schema(node)
+                # 直接是 slot
+                if "$slot" in schema:
+                    return (schema["$slot"], False)
+                # array 包裹 slot: {"type": "array", "items": {"$slot": "..."}}
+                items = schema.get("items", {})
+                if isinstance(items, dict) and "$slot" in items:
+                    return (items["$slot"], True)
+        return (None, False)
+
+    # runtime tuple 格式
+    if isinstance(ptype, tuple) and len(ptype) == 2:
+        inner_str = str(ptype[1])
+        if "ResourceSlot" in inner_str:
+            return ("ResourceSlot", True)
+        if "DeviceSlot" in inner_str:
+            return ("DeviceSlot", True)
+
+    return (None, False)
+
+
+def detect_placeholder_keys(params: list) -> Dict[str, str]:
+    """Detect parameters that reference ResourceSlot or DeviceSlot."""
+    result: Dict[str, str] = {}
+    for p in params:
+        ptype = p.get("type", "")
+        if "ResourceSlot" in str(ptype):
+            result[p["name"]] = "unilabos_resources"
+        elif "DeviceSlot" in str(ptype):
+            result[p["name"]] = "unilabos_devices"
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Handle 规范化
+# ---------------------------------------------------------------------------
+
+
+def normalize_ast_handles(handles_raw: Any) -> List[Dict[str, Any]]:
+    """Convert AST-parsed handle structures to the standard registry format."""
+    if not handles_raw:
+        return []
+
+    # handle_type → io_type 映射 (AST 内部类名 → YAML 标准字段值)
+    _HANDLE_TYPE_TO_IO_TYPE = {
+        "input": "target",
+        "output": "source",
+        "action_input": "action_target",
+        "action_output": "action_source",
+    }
+
+    result: List[Dict[str, Any]] = []
+    for h in handles_raw:
+        if isinstance(h, dict):
+            call = h.get("_call", "")
+            if "InputHandle" in call:
+                handle_type = "input"
+            elif "OutputHandle" in call:
+                handle_type = "output"
+            elif "ActionInputHandle" in call:
+                handle_type = "action_input"
+            elif "ActionOutputHandle" in call:
+                handle_type = "action_output"
+            else:
+                handle_type = h.get("handle_type", "unknown")
+
+            io_type = _HANDLE_TYPE_TO_IO_TYPE.get(handle_type, handle_type)
+
+            entry: Dict[str, Any] = {
+                "handler_key": h.get("key", ""),
+                "data_type": h.get("data_type", ""),
+                "io_type": io_type,
+            }
+            side = h.get("side")
+            if side:
+                if isinstance(side, str) and "." in side:
+                    val = side.rsplit(".", 1)[-1]
+                    side = val.lower() if val in ("LEFT", "RIGHT", "TOP", "BOTTOM") else val
+                entry["side"] = side
+            label = h.get("label")
+            if label:
+                entry["label"] = label
+            data_key = h.get("data_key")
+            if data_key:
+                entry["data_key"] = data_key
+            data_source = h.get("data_source")
+            if data_source:
+                if isinstance(data_source, str) and "." in data_source:
+                    val = data_source.rsplit(".", 1)[-1]
+                    data_source = val.lower() if val in ("HANDLE", "EXECUTOR") else val
+                entry["data_source"] = data_source
+            description = h.get("description")
+            if description:
+                entry["description"] = description
+
+            result.append(entry)
+    return result
+
+
+def normalize_ast_action_handles(handles_raw: Any) -> Dict[str, Any]:
+    """Convert AST-parsed action handle list to {"input": [...], "output": [...]}.
+
+    Mirrors the runtime behavior of decorators._action_handles_to_dict:
+      - ActionInputHandle  => grouped under "input"
+      - ActionOutputHandle => grouped under "output"
+    Field mapping: key -> handler_key (matches Pydantic serialization_alias).
+    """
+    if not handles_raw or not isinstance(handles_raw, list):
+        return {}
+
+    input_list: List[Dict[str, Any]] = []
+    output_list: List[Dict[str, Any]] = []
+
+    for h in handles_raw:
+        if not isinstance(h, dict):
+            continue
+        call = h.get("_call", "")
+        is_input = "ActionInputHandle" in call or "InputHandle" in call
+        is_output = "ActionOutputHandle" in call or "OutputHandle" in call
+
+        entry: Dict[str, Any] = {
+            "handler_key": h.get("key", ""),
+            "data_type": h.get("data_type", ""),
+            "label": h.get("label", ""),
+        }
+        for opt_key in ("side", "data_key", "data_source", "description", "io_type"):
+            val = h.get(opt_key)
+            if val is not None:
+                # Only resolve enum-style refs (e.g. DataSource.HANDLE -> handle) for data_source/side
+                # data_key values like "wells.@flatten", "@this.0@@@plate" must be preserved as-is
+                if (
+                    isinstance(val, str)
+                    and "." in val
+                    and opt_key not in ("io_type", "data_key")
+                ):
+                    val = val.rsplit(".", 1)[-1].lower()
+                entry[opt_key] = val
+
+        # io_type: only add when explicitly set; do not default output to "sink" (YAML convention omits it)
+        if "io_type" not in entry and is_input:
+            entry["io_type"] = "source"
+
+        if is_input:
+            input_list.append(entry)
+        elif is_output:
+            output_list.append(entry)
+
+    result: Dict[str, Any] = {}
+    if input_list:
+        result["input"] = input_list
+    # Always include output (empty list when no outputs) to match YAML
+    result["output"] = output_list
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Schema 辅助
+# ---------------------------------------------------------------------------
+
+
+def wrap_action_schema(
+    goal_schema: Dict[str, Any],
+    action_name: str,
+    description: str = "",
+    result_schema: Optional[Dict[str, Any]] = None,
+    feedback_schema: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """
+    将 goal 参数 schema 包装为标准的 action schema 格式:
+    { "properties": { "goal": ..., "feedback": ..., "result": ... }, ... }
+    """
+    # 去掉 auto- 前缀用于 title/description，与 YAML 路径保持一致
+    display_name = action_name.removeprefix("auto-")
+    return {
+        "title": f"{display_name}参数",
+        "description": description or f"{display_name}的参数schema",
+        "type": "object",
+        "properties": {
+            "goal": goal_schema,
+            "feedback": feedback_schema or {},
+            "result": result_schema or {},
+        },
+        "required": ["goal"],
+    }
+
+
+def preserve_field_descriptions(new_schema: Dict[str, Any], prev_schema: Dict[str, Any]):
+    """保留之前 schema 中的 field descriptions"""
+    if not prev_schema or not new_schema:
+        return
+    prev_props = prev_schema.get("properties", {})
+    new_props = new_schema.get("properties", {})
+    for field_name, prev_field in prev_props.items():
+        if field_name in new_props and "title" in prev_field:
+            new_props[field_name].setdefault("title", prev_field["title"])
+
+
+# ---------------------------------------------------------------------------
+# 深度对比
+# ---------------------------------------------------------------------------
+
+
+def _short(val, limit=120):
+    """截断过长的值用于日志显示。"""
+    s = repr(val)
+    return s if len(s) <= limit else s[:limit] + "..."
+
+
+def deep_diff(old, new, path="", max_depth=10) -> list:
+    """递归对比两个对象，返回所有差异的描述列表。"""
+    diffs = []
+    if max_depth <= 0:
+        if old != new:
+            diffs.append(f"{path}: (达到最大深度) OLD≠NEW")
+        return diffs
+
+    if type(old) != type(new):
+        diffs.append(f"{path}: 类型不同 OLD={type(old).__name__}({_short(old)}) NEW={type(new).__name__}({_short(new)})")
+        return diffs
+
+    if isinstance(old, dict):
+        old_keys = set(old.keys())
+        new_keys = set(new.keys())
+        for k in sorted(new_keys - old_keys):
+            diffs.append(f"{path}.{k}: 新增字段 (AST有, YAML无) = {_short(new[k])}")
+        for k in sorted(old_keys - new_keys):
+            diffs.append(f"{path}.{k}: 缺失字段 (YAML有, AST无) = {_short(old[k])}")
+        for k in sorted(old_keys & new_keys):
+            diffs.extend(deep_diff(old[k], new[k], f"{path}.{k}", max_depth - 1))
+    elif isinstance(old, (list, tuple)):
+        if len(old) != len(new):
+            diffs.append(f"{path}: 列表长度不同 OLD={len(old)} NEW={len(new)}")
+        for i in range(min(len(old), len(new))):
+            diffs.extend(deep_diff(old[i], new[i], f"{path}[{i}]", max_depth - 1))
+        if len(new) > len(old):
+            for i in range(len(old), len(new)):
+                diffs.append(f"{path}[{i}]: 新增元素 = {_short(new[i])}")
+        elif len(old) > len(new):
+            for i in range(len(new), len(old)):
+                diffs.append(f"{path}[{i}]: 缺失元素 = {_short(old[i])}")
+    else:
+        if old != new:
+            diffs.append(f"{path}: OLD={_short(old)} NEW={_short(new)}")
+    return diffs
+
+
+# ---------------------------------------------------------------------------
+# MRO 方法参数解析
+# ---------------------------------------------------------------------------
+
+
+def resolve_method_params_via_import(module_str: str, method_name: str) -> Dict[str, str]:
+    """当 AST 方法参数为空 (如 *args, **kwargs) 时, import class 并通过 MRO 获取真实方法参数.
+
+    返回 identity mapping {param_name: param_name}.
+    """
+    if not module_str or ":" not in module_str:
+        return {}
+    try:
+        cls = import_class(module_str)
+    except Exception as e:
+        _logger.debug(f"[AST] resolve_method_params_via_import: import_class('{module_str}') failed: {e}")
+        return {}
+
+    try:
+        for base_cls in cls.__mro__:
+            if method_name not in base_cls.__dict__:
+                continue
+            method = base_cls.__dict__[method_name]
+            actual = getattr(method, "__wrapped__", method)
+            if isinstance(actual, (staticmethod, classmethod)):
+                actual = actual.__func__
+            if not callable(actual):
+                continue
+            sig = inspect.signature(actual, follow_wrapped=True)
+            params = [
+                p.name for p in sig.parameters.values()
+                if p.name not in ("self", "cls")
+                and p.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
+            ]
+            if params:
+                return {p: p for p in params}
+    except Exception as e:
+        _logger.debug(f"[AST] resolve_method_params_via_import: MRO walk for '{method_name}' failed: {e}")
+    return {}

unilabos/resources/container.py

@@ -1,10 +1,6 @@
-import json
 from typing import Dict, Any
 
 from pylabrobot.resources import Container
-from unilabos_msgs.msg import Resource
-
-from unilabos.ros.msgs.message_converter import convert_from_ros_msg
 
 
 class RegularContainer(Container):
@@ -16,12 +12,14 @@ class RegularContainer(Container):
             kwargs["size_y"] = 0
         if "size_z" not in kwargs:
             kwargs["size_z"] = 0
+        if "category" not in kwargs:
+            kwargs["category"] = "container"
+
         self.kwargs = kwargs
-        self.state = {}
-        super().__init__(*args, category="container", **kwargs)
+        super().__init__(*args, **kwargs)
 
     def load_state(self, state: Dict[str, Any]):
-        self.state = state
+        super().load_state(state)
 
 
 def get_regular_container(name="container"):
@@ -29,7 +27,6 @@ def get_regular_container(name="container"):
     r.category = "container"
     return r
 
-#
 # class RegularContainer(object):
 #     # 第一个参数必须是id传入
 #     # noinspection PyShadowingBuiltins
@@ -89,4 +86,4 @@ def get_regular_container(name="container"):
 #         return to_dict
 #
 #     def __str__(self):
-#         return f"{self.id}"
+#         return f"{self.id}"

unilabos/resources/graphio.py

@@ -76,7 +76,7 @@ def canonicalize_nodes_data(
             if sample_id:
                 logger.error(f"{node}的sample_id参数已弃用，sample_id: {sample_id}")
         for k in list(node.keys()):
-            if k not in ["id", "uuid", "name", "description", "schema", "model", "icon", "parent_uuid", "parent", "type", "class", "position", "config", "data", "children", "pose"]:
+            if k not in ["id", "uuid", "name", "description", "schema", "model", "icon", "parent_uuid", "parent", "type", "class", "position", "config", "data", "children", "pose", "extra", "machine_name"]:
                 v = node.pop(k)
                 node["config"][k] = v
     if outer_host_node_id is not None:
@@ -288,6 +288,15 @@ def read_node_link_json(
     physical_setup_graph = nx.node_link_graph(graph_data, edges="links", multigraph=False)
     handle_communications(physical_setup_graph)
 
+    # Stamp machine_name on device trees only (resources are cloud-managed)
+    local_machine = BasicConfig.machine_name or "本地"
+    for tree in resource_tree_set.trees:
+        if tree.root_node.res_content.type != "device":
+            continue
+        for node in tree.get_all_nodes():
+            if not node.res_content.machine_name:
+                node.res_content.machine_name = local_machine
+
     return physical_setup_graph, resource_tree_set, standardized_links
 
 
@@ -372,6 +381,15 @@ def read_graphml(graphml_file: str) -> tuple[nx.Graph, ResourceTreeSet, List[Dic
     physical_setup_graph = nx.node_link_graph(graph_data, link="links", multigraph=False)
     handle_communications(physical_setup_graph)
 
+    # Stamp machine_name on device trees only (resources are cloud-managed)
+    local_machine = BasicConfig.machine_name or "本地"
+    for tree in resource_tree_set.trees:
+        if tree.root_node.res_content.type != "device":
+            continue
+        for node in tree.get_all_nodes():
+            if not node.res_content.machine_name:
+                node.res_content.machine_name = local_machine
+
     return physical_setup_graph, resource_tree_set, standardized_links
 
 

unilabos/resources/resource_tracker.py

@@ -16,6 +16,7 @@ if TYPE_CHECKING:
 
 
 EXTRA_CLASS = "unilabos_resource_class"
+FRONTEND_POSE_EXTRA = "unilabos_frontend_pose_extra"
 EXTRA_SAMPLE_UUID = "sample_uuid"
 EXTRA_UNILABOS_SAMPLE_UUID = "unilabos_sample_uuid"
 
@@ -38,24 +39,52 @@ class LabSample(TypedDict):
     extra: Dict[str, Any]
 
 
+class ResourceDictPositionSizeType(TypedDict):
+    depth: float
+    width: float
+    height: float
+
+
 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
 
 
+class ResourceDictPositionScaleType(TypedDict):
+    x: float
+    y: float
+    z: float
+
+
 class ResourceDictPositionScale(BaseModel):
     x: float = Field(description="x scale", default=0.0)
     y: float = Field(description="y scale", default=0.0)
     z: float = Field(description="z scale", default=0.0)
 
 
+class ResourceDictPositionObjectType(TypedDict):
+    x: float
+    y: float
+    z: float
+
+
 class ResourceDictPositionObject(BaseModel):
     x: float = Field(description="X coordinate", default=0.0)
     y: float = Field(description="Y coordinate", default=0.0)
     z: float = Field(description="Z coordinate", default=0.0)
 
 
+class ResourceDictPositionType(TypedDict):
+    size: ResourceDictPositionSizeType
+    scale: ResourceDictPositionScaleType
+    layout: Literal["2d", "x-y", "z-y", "x-z"]
+    position: ResourceDictPositionObjectType
+    position3d: ResourceDictPositionObjectType
+    rotation: ResourceDictPositionObjectType
+    cross_section_type: Literal["rectangle", "circle", "rounded_rectangle"]
+
+
 class ResourceDictPosition(BaseModel):
     size: ResourceDictPositionSize = Field(description="Resource size", default_factory=ResourceDictPositionSize)
     scale: ResourceDictPositionScale = Field(description="Resource scale", default_factory=ResourceDictPositionScale)
@@ -72,6 +101,26 @@ class ResourceDictPosition(BaseModel):
     cross_section_type: Literal["rectangle", "circle", "rounded_rectangle"] = Field(
         description="Cross section type", default="rectangle"
     )
+    extra: Optional[Dict[str, Any]] = Field(description="Extra data", default=None)
+
+
+class ResourceDictType(TypedDict):
+    id: str
+    uuid: str
+    name: str
+    description: str
+    resource_schema: Dict[str, Any]
+    model: Dict[str, Any]
+    icon: str
+    parent_uuid: Optional[str]
+    parent: Optional["ResourceDictType"]
+    type: Union[Literal["device"], str]
+    klass: str
+    pose: ResourceDictPositionType
+    config: Dict[str, Any]
+    data: Dict[str, Any]
+    extra: Dict[str, Any]
+    machine_name: str
 
 
 # 统一的资源字典模型，parent 自动序列化为 parent_uuid，children 不序列化
@@ -93,6 +142,7 @@ class ResourceDict(BaseModel):
     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")
+    machine_name: str = Field(description="Machine this resource belongs to", default="")
 
     @field_serializer("parent_uuid")
     def _serialize_parent(self, parent_uuid: Optional["ResourceDict"]):
@@ -148,22 +198,30 @@ class ResourceDictInstance(object):
         self.typ = "dict"
 
     @classmethod
-    def get_resource_instance_from_dict(cls, content: Dict[str, Any]) -> "ResourceDictInstance":
+    def get_resource_instance_from_dict(cls, content: ResourceDictType) -> "ResourceDictInstance":
         """从字典创建资源实例"""
         if "id" not in content:
             content["id"] = content["name"]
         if "uuid" not in content:
             content["uuid"] = str(uuid.uuid4())
         if "description" in content and content["description"] is None:
+            # noinspection PyTypedDict
             del content["description"]
         if "model" in content and content["model"] is None:
+            # noinspection PyTypedDict
             del content["model"]
+        # noinspection PyTypedDict
         if "schema" in content and content["schema"] is None:
+            # noinspection PyTypedDict
             del content["schema"]
+        # noinspection PyTypedDict
         if "x" in content.get("position", {}):
             # 说明是老版本的position格式，转换成新的
+            # noinspection PyTypedDict
             content["position"] = {"position": content["position"]}
+        # noinspection PyTypedDict
         if not content.get("class"):
+            # noinspection PyTypedDict
             content["class"] = ""
         if not content.get("config"):  # todo: 后续从后端保证字段非空
             content["config"] = {}
@@ -174,16 +232,18 @@ class ResourceDictInstance(object):
         if "position" in content:
             pose = content.get("pose", {})
             if "position" not in pose:
+                # noinspection PyTypedDict
                 if "position" in content["position"]:
+                    # noinspection PyTypedDict
                     pose["position"] = content["position"]["position"]
                 else:
-                    pose["position"] = {"x": 0, "y": 0, "z": 0}
+                    pose["position"] = ResourceDictPositionObjectType(x=0, y=0, z=0)
             if "size" not in pose:
-                pose["size"] = {
-                    "width": content["config"].get("size_x", 0),
-                    "height": content["config"].get("size_y", 0),
-                    "depth": content["config"].get("size_z", 0),
-                }
+                pose["size"] = ResourceDictPositionSizeType(
+                    width= content["config"].get("size_x", 0),
+                    height= content["config"].get("size_y", 0),
+                    depth= content["config"].get("size_z", 0),
+                )
             content["pose"] = pose
         try:
             res_dict = ResourceDict.model_validate(content)
@@ -351,7 +411,7 @@ class ResourceTreeSet(object):
             )
 
     @classmethod
-    def from_plr_resources(cls, resources: List["PLRResource"], known_newly_created=False) -> "ResourceTreeSet":
+    def from_plr_resources(cls, resources: List["PLRResource"], known_newly_created=False, old_size=False) -> "ResourceTreeSet":
         """
         从plr资源创建ResourceTreeSet
         """
@@ -365,13 +425,29 @@ class ResourceTreeSet(object):
                 "tip_spot": "tip_spot",
                 "tube": "tube",
                 "bottle_carrier": "bottle_carrier",
+                "material_hole": "material_hole",
+                "container": "container",
+                "material_plate": "material_plate",
+                "electrode_sheet": "electrode_sheet",
+                "warehouse": "warehouse",
+                "magazine_holder": "magazine_holder",
+                "resource_group": "resource_group",
+                "trash": "trash",
+                "plate_adapter": "plate_adapter",
+                "consumable": "consumable",
+                "tool": "tool",
+                "condenser": "condenser",
+                "crucible": "crucible",
+                "reagent_bottle": "reagent_bottle",
+                "flask": "flask",
+                "beaker": "beaker",
             }
             if source in replace_info:
                 return replace_info[source]
             elif source is None:
                 return ""
             else:
-                print("转换pylabrobot的时候，出现未知类型", source)
+                logger.trace(f"转换pylabrobot的时候，出现未知类型 {source}")
                 return source
 
         def build_uuid_mapping(res: "PLRResource", uuid_list: list, parent_uuid: Optional[str] = None):
@@ -408,6 +484,7 @@ class ResourceTreeSet(object):
                 "position3d": raw_pos,
                 "rotation": d["rotation"],
                 "cross_section_type": d.get("cross_section_type", "rectangle"),
+                "extra": extra.get(FRONTEND_POSE_EXTRA)
             }
 
             # 先构建当前节点的字典（不包含children）
@@ -425,7 +502,7 @@ class ResourceTreeSet(object):
                     k: v
                     for k, v in d.items()
                     if k
-                    not in [
+                    not in ([
                         "name",
                         "children",
                         "parent_name",
@@ -436,7 +513,15 @@ class ResourceTreeSet(object):
                         "size_z",
                         "cross_section_type",
                         "bottom_type",
-                    ]
+                    ] if not old_size else [
+                        "name",
+                        "children",
+                        "parent_name",
+                        "location",
+                        "rotation",
+                        "cross_section_type",
+                        "bottom_type",
+                    ])
                 },
                 "data": states[d["name"]],
                 "extra": extra,
@@ -493,6 +578,7 @@ class ResourceTreeSet(object):
             name_to_uuid[node.res_content.name] = node.res_content.uuid
             all_states[node.res_content.name] = node.res_content.data
             name_to_extra[node.res_content.name] = node.res_content.extra
+            name_to_extra[node.res_content.name][FRONTEND_POSE_EXTRA] = node.res_content.pose.extra
             name_to_extra[node.res_content.name][EXTRA_CLASS] = node.res_content.klass
             for child in node.children:
                 collect_node_data(child, name_to_uuid, all_states, name_to_extra)
@@ -561,7 +647,7 @@ class ResourceTreeSet(object):
                 plr_resources.append(plr_resource)
 
             except Exception as e:
-                logger.error(f"转换 PLR 资源失败: {e}")
+                logger.error(f"转换 PLR 资源失败: {e} {str(plr_dict)[:1000]}")
                 import traceback
 
                 logger.error(f"堆栈: {traceback.format_exc()}")
@@ -734,7 +820,8 @@ class ResourceTreeSet(object):
             if remote_root_type == "device":
                 # 情况1: 一级是 device
                 if remote_root_id not in local_device_map:
-                    logger.warning(f"Device '{remote_root_id}' 在本地不存在，跳过该 device 下的物料同步")
+                    if remote_root_id != "host_node":
+                        logger.warning(f"Device '{remote_root_id}' 在本地不存在，跳过该 device 下的物料同步")
                     continue
 
                 local_device = local_device_map[remote_root_id]
@@ -781,14 +868,27 @@ class ResourceTreeSet(object):
                                 f"从远端同步了 {added_count} 个物料子树"
                             )
                     else:
-                        # 情况2: 二级是物料（不是 device）
-                        if remote_child_name not in local_children_map:
-                            # 引入整个子树
-                            remote_child.res_content.parent = local_device.res_content
-                            local_device.children.append(remote_child)
-                            logger.info(f"Device '{remote_root_id}': 从远端同步物料子树 '{remote_child_name}'")
-                        else:
-                            logger.info(f"物料 '{remote_root_id}/{remote_child_name}' 已存在，跳过")
+                        # 二级物料已存在，比较三级子节点是否缺失
+                        local_material = local_children_map[remote_child_name]
+                        local_material_children_map = {child.res_content.name: child for child in
+                                                       local_material.children}
+                        added_count = 0
+                        for remote_sub in remote_child.children:
+                            remote_sub_name = remote_sub.res_content.name
+                            if remote_sub_name not in local_material_children_map:
+                                remote_sub.res_content.parent = local_material.res_content
+                                local_material.children.append(remote_sub)
+                                added_count += 1
+                            else:
+                                logger.info(
+                                    f"物料 '{remote_root_id}/{remote_child_name}/{remote_sub_name}' "
+                                    f"已存在，跳过"
+                                )
+                        if added_count > 0:
+                            logger.info(
+                                f"物料 '{remote_root_id}/{remote_child_name}': "
+                                f"从远端同步了 {added_count} 个子物料"
+                            )
             else:
                 # 情况1: 一级节点是物料（不是 device）
                 # 检查是否已存在
@@ -811,7 +911,7 @@ class ResourceTreeSet(object):
 
         return self
 
-    def dump(self) -> List[List[Dict[str, Any]]]:
+    def dump(self, old_position=False) -> List[List[Dict[str, Any]]]:
         """
         将 ResourceTreeSet 序列化为嵌套列表格式
 
@@ -827,6 +927,10 @@ class ResourceTreeSet(object):
             # 获取树的所有节点并序列化
             tree_nodes = [node.res_content.model_dump(by_alias=True) for node in tree.get_all_nodes()]
             result.append(tree_nodes)
+        if old_position:
+            for r in result:
+                for rr in r:
+                    rr["position"] = rr["pose"]["position"]
         return result
 
     @classmethod

unilabos/ros/device_node_wrapper.py

@@ -44,8 +44,7 @@ def ros2_device_node(
     # 从属性中自动发现可发布状态
     if status_types is None:
         status_types = {}
-    if device_config is None:
-        raise ValueError("device_config cannot be None")
+    assert device_config is not None, "device_config cannot be None"
     if action_value_mappings is None:
         action_value_mappings = {}
     if hardware_interface is None:

unilabos/ros/msgs/message_converter.py

@@ -11,6 +11,7 @@ from io import StringIO
 from typing import Iterable, Any, Dict, Type, TypeVar, Union
 
 import yaml
+from msgcenterpy.instances.ros2_instance import ROS2MessageInstance
 from pydantic import BaseModel
 from dataclasses import asdict, is_dataclass
 
@@ -727,46 +728,9 @@ def ros_message_to_json_schema(msg_class: Any, field_name: str) -> Dict[str, Any
     Returns:
         对应的 JSON Schema 定义
     """
-    schema = {"type": "object", "properties": {}, "required": []}
-
-    # 优先使用字段名作为标题，否则使用类名
+    schema = ROS2MessageInstance(msg_class()).get_json_schema()
     schema["title"] = field_name
-
-    # 获取消息的字段和字段类型
-    try:
-        for ind, slot_info in enumerate(msg_class._fields_and_field_types.items()):
-            slot_name, slot_type = slot_info
-            type_info = msg_class.SLOT_TYPES[ind]
-            field_schema = ros_field_type_to_json_schema(type_info, slot_name)
-            schema["properties"][slot_name] = field_schema
-            schema["required"].append(slot_name)
-        # if hasattr(msg_class, 'get_fields_and_field_types'):
-        #     fields_and_types = msg_class.get_fields_and_field_types()
-        #
-        #     for field_name, field_type in fields_and_types.items():
-        #         # 将 ROS 字段类型转换为 JSON Schema
-        #         field_schema = ros_field_type_to_json_schema(field_type)
-        #
-        #         schema['properties'][field_name] = field_schema
-        #         schema['required'].append(field_name)
-        # elif hasattr(msg_class, '__slots__') and hasattr(msg_class, '_fields_and_field_types'):
-        #     # 直接从实例属性获取
-        #     for field_name in msg_class.__slots__:
-        #         # 移除前导下划线（如果有）
-        #         clean_name = field_name[1:] if field_name.startswith('_') else field_name
-        #
-        #         # 从 _fields_and_field_types 获取类型
-        #         if clean_name in msg_class._fields_and_field_types:
-        #             field_type = msg_class._fields_and_field_types[clean_name]
-        #             field_schema = ros_field_type_to_json_schema(field_type)
-        #
-        #             schema['properties'][clean_name] = field_schema
-        #             schema['required'].append(clean_name)
-    except Exception as e:
-        # 如果获取字段类型失败，添加错误信息
-        schema["description"] = f"解析消息字段时出错: {str(e)}"
-        logger.error(f"解析 {msg_class.__name__} 消息字段失败: {str(e)}")
-
+    schema.pop("description")
     return schema
 
 

unilabos/ros/nodes/base_device_node.py

@@ -34,7 +34,8 @@ 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.registry.decorators import get_topic_config
+from unilabos.utils.decorator import get_all_subscriptions
 
 from unilabos.resources.container import RegularContainer
 from unilabos.resources.graphio import (
@@ -57,6 +58,7 @@ from unilabos_msgs.msg import Resource  # type: ignore
 
 from unilabos.resources.resource_tracker import (
     DeviceNodeResourceTracker,
+    ResourceDictType,
     ResourceTreeSet,
     ResourceTreeInstance,
     ResourceDictInstance,
@@ -146,7 +148,7 @@ def init_wrapper(
     device_id: str,
     device_uuid: str,
     driver_class: type[T],
-    device_config: ResourceTreeInstance,
+    device_config: ResourceDictInstance,
     status_types: Dict[str, Any],
     action_value_mappings: Dict[str, Any],
     hardware_interface: Dict[str, Any],
@@ -194,9 +196,9 @@ class PropertyPublisher:
         self._value = None
         try:
             self.publisher_ = node.create_publisher(msg_type, f"{name}", qos)
-        except AttributeError as ex:
+        except Exception as e:
             self.node.lab_logger().error(
-                f"创建发布者 {name} 失败，可能由于注册表有误，类型: {msg_type}，错误: {ex}\n{traceback.format_exc()}"
+                f"StatusError, DeviceId: {self.node.device_id} 创建发布者 {name} 失败，可能由于注册表有误，类型: {msg_type}，错误: {e}"
             )
         self.timer = node.create_timer(self.timer_period, self.publish_property)
         self.__loop = ROS2DeviceNode.get_asyncio_loop()
@@ -279,6 +281,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         self,
         driver_instance: T,
         device_id: str,
+        registry_name: str,
         device_uuid: str,
         status_types: Dict[str, Any],
         action_value_mappings: Dict[str, Any],
@@ -300,6 +303,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         """
         self.driver_instance = driver_instance
         self.device_id = device_id
+        self.registry_name = registry_name
         self.uuid = device_uuid
         self.publish_high_frequency = False
         self.callback_group = ReentrantCallbackGroup()
@@ -416,7 +420,9 @@ class BaseROS2DeviceNode(Node, Generic[T]):
             if len(rts.root_nodes) == 1 and isinstance(rts_plr_instances[0], RegularContainer):
                 # noinspection PyTypeChecker
                 container_instance: RegularContainer = rts_plr_instances[0]
-                found_resources = self.resource_tracker.figure_resource({"name": container_instance.name}, try_mode=True)
+                found_resources = self.resource_tracker.figure_resource(
+                    {"name": container_instance.name}, try_mode=True
+                )
                 if not len(found_resources):
                     self.resource_tracker.add_resource(container_instance)
                     logger.info(f"添加物料{container_instance.name}到资源跟踪器")
@@ -456,7 +462,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
             }
             res.response = json.dumps(final_response)
             # 如果driver自己就有assign的方法，那就使用driver自己的assign方法
-            if hasattr(self.driver_instance, "create_resource"):
+            if hasattr(self.driver_instance, "create_resource") and self.node_name != "host_node":
                 create_resource_func = getattr(self.driver_instance, "create_resource")
                 try:
                     ret = create_resource_func(
@@ -565,9 +571,11 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                 future.add_done_callback(done_cb)
             except ImportError:
                 self.lab_logger().error("Host请求添加物料时，本环境并不存在pylabrobot")
+                res.response = get_result_info_str(traceback.format_exc(), False, {})
             except Exception as e:
                 self.lab_logger().error("Host请求添加物料时出错")
                 self.lab_logger().error(traceback.format_exc())
+                res.response = get_result_info_str(traceback.format_exc(), False, {})
             return res
 
         # noinspection PyTypeChecker
@@ -590,6 +598,12 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                 self.s2c_resource_tree,  # type: ignore
                 callback_group=self.callback_group,
             ),
+            "s2c_device_manage": self.create_service(
+                SerialCommand,
+                f"/srv{self.namespace}/s2c_device_manage",
+                self.s2c_device_manage,  # type: ignore
+                callback_group=self.callback_group,
+            ),
         }
 
         # 向全局在线设备注册表添加设备信息
@@ -911,8 +925,24 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                         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]
+                        site_index = None
+                        try:
+                            # sites 可能是 Resource 列表或 dict 列表 (如 PRCXI9300Deck)
+                            # 只有itemized_carrier在使用，准备弃用
+                            site_index = sites.index(original_instance)
+                        except ValueError:
+                            # dict 类型的 sites: 通过name匹配
+                            for idx, site in enumerate(sites):
+                                if original_instance.name == site["occupied_by"]:
+                                    site_index = idx
+                                    break
+                                elif (original_instance.location.x == site["position"]["x"] and original_instance.location.y == site["position"]["y"] and original_instance.location.z == site["position"]["z"]):
+                                    site_index = idx
+                                    break
+                        if site_index is None:
+                            site_name = None
+                        else:
+                            site_name = site_names[site_index]
                         if site_name != target_site:
                             parent = self.transfer_to_new_resource(original_instance, tree, additional_add_params)
                             if parent is not None:
@@ -920,6 +950,14 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                                 parent_appended = True
 
                 # 加载状态
+                # noinspection PyProtectedMember
+                original_instance._size_x = plr_resource._size_x
+                # noinspection PyProtectedMember
+                original_instance._size_y = plr_resource._size_y
+                # noinspection PyProtectedMember
+                original_instance._size_z = plr_resource._size_z
+                # noinspection PyProtectedMember
+                original_instance._local_size_z = plr_resource._local_size_z
                 original_instance.location = plr_resource.location
                 original_instance.rotation = plr_resource.rotation
                 original_instance.barcode = plr_resource.barcode
@@ -980,7 +1018,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                         ].call_async(
                             r
                         )  # type: ignore
-                        self.lab_logger().info(f"确认资源云端 Add 结果: {response.response}")
+                        self.lab_logger().trace(f"确认资源云端 Add 结果: {response.response}")
                         results.append(result)
                     elif action == "update":
                         if tree_set is None:
@@ -1006,7 +1044,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                             ].call_async(
                                 r
                             )  # type: ignore
-                            self.lab_logger().info(f"确认资源云端 Update 结果: {response.response}")
+                            self.lab_logger().trace(f"确认资源云端 Update 结果: {response.response}")
                         results.append(result)
                     elif action == "remove":
                         result = _handle_remove(resources_uuid)
@@ -1034,6 +1072,48 @@ class BaseROS2DeviceNode(Node, Generic[T]):
 
         return res
 
+    async def s2c_device_manage(self, req: SerialCommand_Request, res: SerialCommand_Response):
+        """Handle add/remove device requests from HostNode via SerialCommand."""
+        try:
+            cmd = json.loads(req.command)
+            action = cmd.get("action", "")
+            data = cmd.get("data", {})
+            device_id = data.get("device_id", "")
+
+            if not device_id:
+                res.response = json.dumps({"success": False, "error": "device_id required"})
+                return res
+
+            if action == "add":
+                result = self.create_device(device_id, data)
+            elif action == "remove":
+                result = self.destroy_device(device_id)
+            else:
+                result = {"success": False, "error": f"Unknown action: {action}"}
+
+            res.response = json.dumps(result, ensure_ascii=False)
+
+        except NotImplementedError as e:
+            self.lab_logger().warning(f"[DeviceManage] {e}")
+            res.response = json.dumps({"success": False, "error": str(e)})
+        except Exception as e:
+            self.lab_logger().error(f"[DeviceManage] Error: {e}")
+            res.response = json.dumps({"success": False, "error": str(e)})
+
+        return res
+
+    def create_device(self, device_id: str, config: "ResourceDictType") -> dict:
+        """Create a sub-device dynamically. Override in HostNode / WorkstationNode."""
+        raise NotImplementedError(
+            f"{self.__class__.__name__} does not support dynamic device creation"
+        )
+
+    def destroy_device(self, device_id: str) -> dict:
+        """Destroy a sub-device dynamically. Override in HostNode / WorkstationNode."""
+        raise NotImplementedError(
+            f"{self.__class__.__name__} does not support dynamic device removal"
+        )
+
     async def transfer_resource_to_another(
         self,
         plr_resources: List["ResourcePLR"],
@@ -1152,6 +1232,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                     "machine_name": BasicConfig.machine_name,
                     "type": "slave",
                     "edge_device_id": self.device_id,
+                    "registry_name": self.registry_name,
                 }
             },
             ensure_ascii=False,
@@ -1175,22 +1256,40 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         return self._lab_logger
 
     def create_ros_publisher(self, attr_name, msg_type, initial_period=5.0):
-        """创建ROS发布者"""
-        # 检测装饰器配置（支持 get_{attr_name} 方法和 @property）
+        """创建ROS发布者，仅当方法/属性有 @topic_config 装饰器时才创建。"""
+        # 检测 @topic_config 装饰器配置
         topic_config = {}
+        driver_class = type(self.driver_instance)
 
-        # 优先检测 get_{attr_name} 方法
-        if hasattr(self.driver_instance, f"get_{attr_name}"):
-            getter_method = getattr(self.driver_instance, f"get_{attr_name}")
-            topic_config = get_topic_config(getter_method)
+        # 区分 @property 和普通方法两种情况
+        is_prop = hasattr(driver_class, attr_name) and isinstance(
+            getattr(driver_class, attr_name), property
+        )
 
-        # 如果没有配置，检测 @property 装饰的属性
+        if is_prop:
+            # @property: 检测 fget 上的 @topic_config
+            class_attr = getattr(driver_class, attr_name)
+            if class_attr.fget is not None:
+                topic_config = get_topic_config(class_attr.fget)
+        else:
+            # 普通方法: 直接检测 attr_name 方法上的 @topic_config
+            if hasattr(self.driver_instance, attr_name):
+                method = getattr(self.driver_instance, attr_name)
+                if callable(method):
+                    topic_config = get_topic_config(method)
+
+        # 没有 @topic_config 装饰器则跳过发布
         if not topic_config:
-            driver_class = type(self.driver_instance)
-            if hasattr(driver_class, attr_name):
-                class_attr = getattr(driver_class, attr_name)
-                if isinstance(class_attr, property) and class_attr.fget is not None:
-                    topic_config = get_topic_config(class_attr.fget)
+            return
+
+        # 发布名称优先级: @topic_config(name=...) > get_ 前缀去除 > attr_name
+        cfg_name = topic_config.get("name")
+        if cfg_name:
+            publish_name = cfg_name
+        elif attr_name.startswith("get_"):
+            publish_name = attr_name[4:]
+        else:
+            publish_name = attr_name
 
         # 使用装饰器配置或默认值
         cfg_period = topic_config.get("period")
@@ -1203,10 +1302,10 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         # 获取属性值的方法
         def get_device_attr():
             try:
-                if hasattr(self.driver_instance, f"get_{attr_name}"):
-                    return getattr(self.driver_instance, f"get_{attr_name}")()
-                else:
+                if is_prop:
                     return getattr(self.driver_instance, attr_name)
+                else:
+                    return getattr(self.driver_instance, attr_name)()
             except AttributeError as ex:
                 if ex.args[0].startswith(f"AttributeError: '{self.driver_instance.__class__.__name__}' object"):
                     self.lab_logger().error(
@@ -1218,8 +1317,8 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                     )
                     self.lab_logger().error(traceback.format_exc())
 
-        self._property_publishers[attr_name] = PropertyPublisher(
-            self, attr_name, get_device_attr, msg_type, period, print_publish, qos
+        self._property_publishers[publish_name] = PropertyPublisher(
+            self, publish_name, get_device_attr, msg_type, period, print_publish, qos
         )
 
     def create_ros_action_server(self, action_name, action_value_mapping):
@@ -1227,14 +1326,17 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         action_type = action_value_mapping["type"]
         str_action_type = str(action_type)[8:-2]
 
-        self._action_servers[action_name] = ActionServer(
-            self,
-            action_type,
-            action_name,
-            execute_callback=self._create_execute_callback(action_name, action_value_mapping),
-            callback_group=self.callback_group,
-        )
-
+        try:
+            self._action_servers[action_name] = ActionServer(
+                self,
+                action_type,
+                action_name,
+                execute_callback=self._create_execute_callback(action_name, action_value_mapping),
+                callback_group=self.callback_group,
+            )
+        except Exception as e:
+            self.lab_logger().error(f"创建ActionServer失败，Device: {self.device_id}, Action Name: {action_name}, Action Type: {action_type}, Error: {e}")
+            return
         self.lab_logger().trace(f"发布动作: {action_name}, 类型: {str_action_type}")
 
     def _setup_decorated_subscribers(self):
@@ -1626,9 +1728,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                             else:
                                 resolved_sample_uuids[sample_uuid] = material_uuid
                         function_args[PARAM_SAMPLE_UUIDS] = resolved_sample_uuids
-                        self.lab_logger().debug(
-                            f"[JsonCommand] 注入 {PARAM_SAMPLE_UUIDS}: {resolved_sample_uuids}"
-                        )
+                        self.lab_logger().debug(f"[JsonCommand] 注入 {PARAM_SAMPLE_UUIDS}: {resolved_sample_uuids}")
                     continue
 
                 # 处理单个 ResourceSlot
@@ -2005,6 +2105,7 @@ class ROS2DeviceNode:
 
         if driver_is_ros:
             driver_params["device_id"] = device_id
+            driver_params["registry_name"] = device_config.res_content.klass
             driver_params["resource_tracker"] = self.resource_tracker
         self._driver_instance = self._driver_creator.create_instance(driver_params)
         if self._driver_instance is None:
@@ -2022,6 +2123,7 @@ class ROS2DeviceNode:
                 children=children,
                 driver_instance=self._driver_instance,  # type: ignore
                 device_id=device_id,
+                registry_name=device_config.res_content.klass,
                 device_uuid=device_uuid,
                 status_types=status_types,
                 action_value_mappings=action_value_mappings,
@@ -2033,6 +2135,7 @@ class ROS2DeviceNode:
             self._ros_node = BaseROS2DeviceNode(
                 driver_instance=self._driver_instance,
                 device_id=device_id,
+                registry_name=device_config.res_content.klass,
                 device_uuid=device_uuid,
                 status_types=status_types,
                 action_value_mappings=action_value_mappings,
@@ -2041,6 +2144,7 @@ class ROS2DeviceNode:
                 resource_tracker=self.resource_tracker,
             )
         self._ros_node: BaseROS2DeviceNode
+        # 将注册表类型名传递给BaseROS2DeviceNode,用于slave上报
         self._ros_node.lab_logger().info(f"初始化完成 {self._ros_node.uuid} {self.driver_is_ros}")
         self.driver_instance._ros_node = self._ros_node  # type: ignore
         self.driver_instance._execute_driver_command = self._ros_node._execute_driver_command  # type: ignore

unilabos/ros/nodes/presets/camera.py

@@ -4,14 +4,22 @@ import cv2
 from sensor_msgs.msg import Image
 from cv_bridge import CvBridge
 from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, DeviceNodeResourceTracker
+from unilabos.registry.decorators import device
 
+
+@device(
+    id="camera",
+    category=["camera"],
+    description="""VideoPublisher摄像头设备节点，用于实时视频采集和流媒体发布。该设备通过OpenCV连接本地摄像头（如USB摄像头、内置摄像头等），定时采集视频帧并将其转换为ROS2的sensor_msgs/Image消息格式发布到视频话题。主要用于实验室自动化系统中的视觉监控、图像分析、实时观察等应用场景。支持可配置的摄像头索引、发布频率等参数。""",
+)
 class VideoPublisher(BaseROS2DeviceNode):
-    def __init__(self, device_id='video_publisher', device_uuid='', camera_index=0, period: float = 0.1, resource_tracker: DeviceNodeResourceTracker = None):
+    def __init__(self, device_id='video_publisher', registry_name="", device_uuid='', camera_index=0, period: float = 0.1, resource_tracker: DeviceNodeResourceTracker = None):
         # 初始化BaseROS2DeviceNode，使用自身作为driver_instance
         BaseROS2DeviceNode.__init__(
             self,
             driver_instance=self,
             device_id=device_id,
+            registry_name=registry_name,
             device_uuid=device_uuid,
             status_types={},
             action_value_mappings={},

unilabos/ros/nodes/presets/controller_node.py

@@ -10,6 +10,7 @@ class ControllerNode(BaseROS2DeviceNode):
     def __init__(
         self,
         device_id: str,
+        registry_name: str,
         controller_func: Callable,
         update_rate: float,
         inputs: Dict[str, Dict[str, type | str]],
@@ -51,6 +52,7 @@ class ControllerNode(BaseROS2DeviceNode):
             self,
             driver_instance=self,
             device_id=device_id,
+            registry_name=registry_name,
             status_types=status_types,
             action_value_mappings=action_value_mappings,
             hardware_interface=hardware_interface,

unilabos/ros/nodes/presets/host_node.py

@@ -12,6 +12,7 @@ from geometry_msgs.msg import Point
 from rclpy.action import ActionClient, get_action_server_names_and_types_by_node
 from rclpy.service import Service
 from typing_extensions import TypedDict
+from unilabos_msgs.action import EmptyIn, StrSingleInput, ResourceCreateFromOuterEasy, ResourceCreateFromOuter
 from unilabos_msgs.msg import Resource  # type: ignore
 from unilabos_msgs.srv import (
     ResourceAdd,
@@ -23,6 +24,7 @@ from unilabos_msgs.srv import (
 from unilabos_msgs.srv._serial_command import SerialCommand_Request, SerialCommand_Response
 from unique_identifier_msgs.msg import UUID
 
+from unilabos.registry.decorators import device
 from unilabos.registry.placeholder_type import ResourceSlot, DeviceSlot
 from unilabos.registry.registry import lab_registry
 from unilabos.resources.container import RegularContainer
@@ -30,12 +32,13 @@ from unilabos.resources.graphio import initialize_resource
 from unilabos.resources.registry import add_schema
 from unilabos.resources.resource_tracker import (
     ResourceDict,
+    ResourceDictType,
     ResourceDictInstance,
     ResourceTreeSet,
     ResourceTreeInstance,
     RETURN_UNILABOS_SAMPLES,
     JSON_UNILABOS_PARAM,
-    PARAM_SAMPLE_UUIDS,
+    PARAM_SAMPLE_UUIDS, SampleUUIDsType, LabSample,
 )
 from unilabos.ros.initialize_device import initialize_device_from_dict
 from unilabos.ros.msgs.message_converter import (
@@ -51,6 +54,7 @@ from unilabos.utils import logger
 from unilabos.utils.exception import DeviceClassInvalid
 from unilabos.utils.log import warning
 from unilabos.utils.type_check import serialize_result_info
+from unilabos.config.config import BasicConfig
 
 if TYPE_CHECKING:
     from unilabos.app.ws_client import QueueItem
@@ -63,7 +67,14 @@ class DeviceActionStatus:
 
 class TestResourceReturn(TypedDict):
     resources: List[List[ResourceDict]]
-    devices: List[DeviceSlot]
+    devices: List[Dict[str, Any]]
+    # unilabos_samples: List[LabSample]
+
+
+class CreateResourceReturn(TypedDict):
+    created_resource_tree: List[List[ResourceDict]]
+    liquid_input_resource_tree: List[Dict[str, Any]]
+    # unilabos_samples: List[LabSample]
 
 
 class TestLatencyReturn(TypedDict):
@@ -78,6 +89,7 @@ class TestLatencyReturn(TypedDict):
     status: str
 
 
+@device(id="host_node", category=[], description="Host Node", icon="icon_device.webp")
 class HostNode(BaseROS2DeviceNode):
     """
     主机节点类，负责管理设备、资源和控制器
@@ -248,6 +260,7 @@ class HostNode(BaseROS2DeviceNode):
             self,
             driver_instance=self,
             device_id=device_id,
+            registry_name="host_node",
             device_uuid=host_node_dict["uuid"],
             status_types={},
             action_value_mappings=lab_registry.device_type_registry["host_node"]["class"]["action_value_mappings"],
@@ -265,44 +278,43 @@ class HostNode(BaseROS2DeviceNode):
         self._action_clients: Dict[str, ActionClient] = {  # 为了方便了解实际的数据类型，host的默认写好
             "/devices/host_node/create_resource": ActionClient(
                 self,
-                lab_registry.ResourceCreateFromOuterEasy,
+                ResourceCreateFromOuterEasy,
                 "/devices/host_node/create_resource",
                 callback_group=self.callback_group,
             ),
             "/devices/host_node/create_resource_detailed": ActionClient(
                 self,
-                lab_registry.ResourceCreateFromOuter,
+                ResourceCreateFromOuter,
                 "/devices/host_node/create_resource_detailed",
                 callback_group=self.callback_group,
             ),
             "/devices/host_node/test_latency": ActionClient(
                 self,
-                lab_registry.EmptyIn,
+                EmptyIn,
                 "/devices/host_node/test_latency",
                 callback_group=self.callback_group,
             ),
             "/devices/host_node/test_resource": ActionClient(
                 self,
-                lab_registry.EmptyIn,
+                EmptyIn,
                 "/devices/host_node/test_resource",
                 callback_group=self.callback_group,
             ),
             "/devices/host_node/_execute_driver_command": ActionClient(
                 self,
-                lab_registry.StrSingleInput,
+                StrSingleInput,
                 "/devices/host_node/_execute_driver_command",
                 callback_group=self.callback_group,
             ),
             "/devices/host_node/_execute_driver_command_async": ActionClient(
                 self,
-                lab_registry.StrSingleInput,
+                StrSingleInput,
                 "/devices/host_node/_execute_driver_command_async",
                 callback_group=self.callback_group,
             ),
         }  # 用来存储多个ActionClient实例
-        self._action_value_mappings: Dict[str, Dict] = (
-            {}
-        )  # 用来存储多个ActionClient的type, goal, feedback, result的变量名映射关系
+        self._action_value_mappings: Dict[str, Dict] = {}  # device_id -> action_value_mappings(本地+远程设备统一存储)
+        self._slave_registry_configs: Dict[str, Dict] = {}  # registry_name -> registry_config(含action_value_mappings)
         self._goals: Dict[str, Any] = {}  # 用来存储多个目标的状态
         self._online_devices: Set[str] = {f"{self.namespace}/{device_id}"}  # 用于跟踪在线设备
         self._last_discovery_time = 0.0  # 上次设备发现的时间
@@ -319,10 +331,18 @@ class HostNode(BaseROS2DeviceNode):
         self._discover_devices()
 
         # 初始化所有本机设备节点，多一次过滤，防止重复初始化
+        local_machine = BasicConfig.machine_name
         for device_config in devices_config.root_nodes:
             device_id = device_config.res_content.id
             if device_config.res_content.type != "device":
                 continue
+            dev_machine = device_config.res_content.machine_name
+            if dev_machine and local_machine and dev_machine != local_machine:
+                self.lab_logger().info(
+                    f"[Host Node] Device {device_id} belongs to machine '{dev_machine}', "
+                    f"local is '{local_machine}', skipping initialization."
+                )
+                continue
             if device_id not in self.devices_names:
                 self.initialize_device(device_id, device_config)
             else:
@@ -552,7 +572,7 @@ class HostNode(BaseROS2DeviceNode):
         liquid_type: list[str] = [],
         liquid_volume: list[int] = [],
         slot_on_deck: str = "",
-    ):
+    ) -> CreateResourceReturn:
         # 暂不支持多对同名父子同时存在
         res_creation_input = {
             "id": res_id.split("/")[-1],
@@ -605,6 +625,8 @@ class HostNode(BaseROS2DeviceNode):
             assert len(response) == 1, "Create Resource应当只返回一个结果"
             for i in response:
                 res = json.loads(i)
+                if "suc" in res:
+                    raise ValueError(res.get("error"))
                 return res
         except Exception as ex:
             pass
@@ -636,6 +658,8 @@ class HostNode(BaseROS2DeviceNode):
         self.device_machine_names[device_id] = "本地"
         self.devices_instances[device_id] = d
         # noinspection PyProtectedMember
+        self._action_value_mappings[device_id] = d._ros_node._action_value_mappings
+        # noinspection PyProtectedMember
         for action_name, action_value_mapping in d._ros_node._action_value_mappings.items():
             if action_name.startswith("auto-") or str(action_value_mapping.get("type", "")).startswith(
                 "UniLabJsonCommand"
@@ -644,7 +668,12 @@ class HostNode(BaseROS2DeviceNode):
             action_id = f"/devices/{device_id}/{action_name}"
             if action_id not in self._action_clients:
                 action_type = action_value_mapping["type"]
-                self._action_clients[action_id] = ActionClient(self, action_type, action_id)
+                try:
+                    self._action_clients[action_id] = ActionClient(self, action_type, action_id)
+                except Exception as e:
+                    self.lab_logger().error(
+                        f"创建ActionClient失败，Device: {device_id}, Action Name: {action_name}, Action Type: {action_type}, Error: {e}")
+                    continue
                 self.lab_logger().trace(
                     f"[Host Node] Created ActionClient (Local): {action_id}"
                 )  # 子设备再创建用的是Discover发现的
@@ -772,6 +801,17 @@ class HostNode(BaseROS2DeviceNode):
         u = uuid.UUID(item.job_id)
         device_id = item.device_id
         action_name = item.action_name
+
+        if BasicConfig.test_mode:
+            action_id = f"/devices/{device_id}/{action_name}"
+            self.lab_logger().info(
+                f"[TEST MODE] 模拟执行: {action_id} (job={item.job_id[:8]}), 参数: {str(action_kwargs)[:500]}"
+            )
+            # 根据注册表 handles 构建模拟返回值
+            mock_return = self._build_test_mode_return(device_id, action_name, action_kwargs)
+            self._handle_test_mode_result(item, action_id, mock_return)
+            return
+
         if action_type.startswith("UniLabJsonCommand"):
             if action_name.startswith("auto-"):
                 action_name = action_name[5:]
@@ -809,6 +849,51 @@ class HostNode(BaseROS2DeviceNode):
         )
         future.add_done_callback(lambda f: self.goal_response_callback(item, action_id, f))
 
+    def _build_test_mode_return(
+        self, device_id: str, action_name: str, action_kwargs: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """
+        根据注册表 handles 的 output 定义构建测试模式的模拟返回值
+
+        根据 data_key 中 @flatten 的层数决定嵌套数组层数，叶子值为空字典。
+        例如: "vessel" → {}, "plate.@flatten" → [{}], "a.@flatten.@flatten" → [[{}]]
+        """
+        mock_return: Dict[str, Any] = {"test_mode": True, "action_name": action_name}
+        action_mappings = self._action_value_mappings.get(device_id, {})
+        action_mapping = action_mappings.get(action_name, {})
+        handles = action_mapping.get("handles", {})
+        if isinstance(handles, dict):
+            for output_handle in handles.get("output", []):
+                data_key = output_handle.get("data_key", "")
+                handler_key = output_handle.get("handler_key", "")
+                # 根据 @flatten 层数构建嵌套数组，叶子为空字典
+                flatten_count = data_key.count("@flatten")
+                value: Any = {}
+                for _ in range(flatten_count):
+                    value = [value]
+                mock_return[handler_key] = value
+        return mock_return
+
+    def _handle_test_mode_result(
+        self, item: "QueueItem", action_id: str, mock_return: Dict[str, Any]
+    ) -> None:
+        """
+        测试模式下直接构建结果并走正常的结果回调流程（跳过 ROS）
+        """
+        job_id = item.job_id
+        status = "success"
+        return_info = serialize_result_info("", True, mock_return)
+
+        self.lab_logger().info(f"[TEST MODE] Result for {action_id} ({job_id[:8]}): {status}")
+
+        from unilabos.app.web.controller import store_job_result
+        store_job_result(job_id, status, return_info, mock_return)
+
+        # 发布状态到桥接器
+        for bridge in self.bridges:
+            if hasattr(bridge, "publish_job_status"):
+                bridge.publish_job_status(mock_return, item, status, return_info)
+
     def goal_response_callback(self, item: "QueueItem", action_id: str, future) -> None:
         """目标响应回调"""
         goal_handle = future.result()
@@ -1133,7 +1218,7 @@ class HostNode(BaseROS2DeviceNode):
                 self.lab_logger().info(f"[Host Node-Resource] UUID映射: {len(uuid_mapping)} 个节点")
             # 还需要加入到资源图中，暂不实现，考虑资源图新的获取方式
             response.response = json.dumps(uuid_mapping)
-            self.lab_logger().info(f"[Host Node-Resource] Resource tree add completed, success: {success}")
+            self.lab_logger().info(f"[Host Node-Resource] Resource tree update completed, success: {success}")
 
     async def _resource_tree_update_callback(self, request: SerialCommand_Request, response: SerialCommand_Response):
         """
@@ -1168,6 +1253,10 @@ class HostNode(BaseROS2DeviceNode):
     def _node_info_update_callback(self, request, response):
         """
         更新节点信息回调
+
+        处理两种消息:
+        1. 首次上报(main_slave_run): 带 devices_config + registry_config,存储 action_value_mappings
+        2. 设备重注册(SYNC_SLAVE_NODE_INFO): 带 edge_device_id + registry_name,用 registry_name 索引已存储的 mappings
         """
         self.lab_logger().trace(f"[Host Node] Node info update request received: {request}")
         try:
@@ -1179,12 +1268,65 @@ class HostNode(BaseROS2DeviceNode):
                 info = info["SYNC_SLAVE_NODE_INFO"]
                 machine_name = info["machine_name"]
                 edge_device_id = info["edge_device_id"]
+                registry_name = info.get("registry_name", "")
                 self.device_machine_names[edge_device_id] = machine_name
+
+                # 用 registry_name 索引已存储的 registry_config,获取 action_value_mappings
+                if registry_name and registry_name in self._slave_registry_configs:
+                    action_mappings = (
+                        self._slave_registry_configs[registry_name].get("class", {}).get("action_value_mappings", {})
+                    )
+                    if action_mappings:
+                        self._action_value_mappings[edge_device_id] = action_mappings
+                        self.lab_logger().info(
+                            f"[Host Node] Loaded {len(action_mappings)} action mappings "
+                            f"for remote device {edge_device_id} (registry: {registry_name})"
+                        )
             else:
                 devices_config = info.pop("devices_config")
                 registry_config = info.pop("registry_config")
                 if registry_config:
                     http_client.resource_registry({"resources": registry_config})
+
+                    # 存储 slave 的 registry_config,用于后续 SYNC_SLAVE_NODE_INFO 索引
+                    for reg_name, reg_data in registry_config.items():
+                        if isinstance(reg_data, dict) and "class" in reg_data:
+                            self._slave_registry_configs[reg_name] = reg_data
+
+                # 解析 devices_config,建立 device_id -> action_value_mappings 映射
+                if devices_config:
+                    machine_name = info["machine_name"]
+                    # Stamp machine_name on each device dict before parsing
+                    for device_tree in devices_config:
+                        for device_dict in device_tree:
+                            device_dict["machine_name"] = machine_name
+                            device_id = device_dict.get("id", "")
+                            class_name = device_dict.get("class", "")
+                            if device_id and class_name and class_name in self._slave_registry_configs:
+                                action_mappings = (
+                                    self._slave_registry_configs[class_name]
+                                    .get("class", {})
+                                    .get("action_value_mappings", {})
+                                )
+                                if action_mappings:
+                                    self._action_value_mappings[device_id] = action_mappings
+                                    self.lab_logger().info(
+                                        f"[Host Node] Stored {len(action_mappings)} action mappings "
+                                        f"for remote device {device_id} (class: {class_name})"
+                                    )
+
+                    # Merge slave devices_config into self.devices_config tree
+                    try:
+                        slave_tree_set = ResourceTreeSet.load(devices_config)  # slave一定是根节点的tree
+                        for tree in slave_tree_set.trees:
+                            self.devices_config.trees.append(tree)
+                        self.lab_logger().info(
+                            f"[Host Node] Merged {len(slave_tree_set.trees)} slave device trees "
+                            f"(machine: {machine_name}) into devices_config"
+                        )
+                    except Exception as e:
+                        self.lab_logger().error(f"[Host Node] Failed to merge slave devices_config: {e}")
+
             self.lab_logger().debug(f"[Host Node] Node info update: {info}")
             response.response = "OK"
         except Exception as e:
@@ -1481,6 +1623,7 @@ class HostNode(BaseROS2DeviceNode):
 
     def test_resource(
         self,
+        sample_uuids: SampleUUIDsType,
         resource: ResourceSlot = None,
         resources: List[ResourceSlot] = None,
         device: DeviceSlot = None,
@@ -1495,6 +1638,7 @@ class HostNode(BaseROS2DeviceNode):
         return {
             "resources": ResourceTreeSet.from_plr_resources([resource, *resources], known_newly_created=True).dump(),
             "devices": [device, *devices],
+            "unilabos_samples": [LabSample(sample_uuid=sample_uuid, oss_path="", extra={"material_uuid": content} if isinstance(content, str) else content.serialize()) for sample_uuid, content in sample_uuids.items()]
         }
 
     def handle_pong_response(self, pong_data: dict):
@@ -1591,3 +1735,177 @@ class HostNode(BaseROS2DeviceNode):
             self.lab_logger().error(f"[Host Node-Resource] Error notifying resource tree update: {str(e)}")
             self.lab_logger().error(traceback.format_exc())
             return False
+
+    # ------------------------------------------------------------------
+    # Device lifecycle (add / remove) — pure forwarder
+    # ------------------------------------------------------------------
+
+    def notify_device_manage(self, target_node_id: str, action: str, config: ResourceDictType) -> bool:
+        """Forward an add/remove device command to the target node via ROS2 SerialCommand.
+
+        The HostNode does NOT interpret the command; it simply resolves the
+        target namespace and forwards the request to ``s2c_device_manage``.
+
+        If *target_node_id* equals the HostNode's own device_id (i.e. the
+        command targets the host itself), we call our local ``create_device``
+        / ``destroy_device`` directly instead of going through ROS2.
+        """
+        try:
+            # If the target is the host itself, handle locally
+            device_id = config["id"]
+            if target_node_id == self.device_id:
+                if action == "add":
+                    return self.create_device(device_id, config).get("success", False)
+                elif action == "remove":
+                    return self.destroy_device(device_id).get("success", False)
+
+            if target_node_id not in self.devices_names:
+                self.lab_logger().error(
+                    f"[Host Node-DeviceMgr] Target {target_node_id} not found in devices_names"
+                )
+                return False
+
+            namespace = self.devices_names[target_node_id]
+            device_key = f"{namespace}/{target_node_id}"
+            if device_key not in self._online_devices:
+                self.lab_logger().error(f"[Host Node-DeviceMgr] Target {device_key} is offline")
+                return False
+
+            srv_address = f"/srv{namespace}/s2c_device_manage"
+            self.lab_logger().info(
+                f"[Host Node-DeviceMgr] Forwarding {action}_device to {target_node_id} ({srv_address})"
+            )
+
+            sclient = self.create_client(SerialCommand, srv_address)
+            if not sclient.wait_for_service(timeout_sec=5.0):
+                self.lab_logger().error(f"[Host Node-DeviceMgr] Service {srv_address} not available")
+                return False
+
+            request = SerialCommand.Request()
+            request.command = json.dumps({"action": action, "data": config}, ensure_ascii=False)
+
+            future = sclient.call_async(request)
+            timeout = 30.0
+            start_time = time.time()
+            while not future.done():
+                if time.time() - start_time > timeout:
+                    self.lab_logger().error(
+                        f"[Host Node-DeviceMgr] Timeout waiting for {action}_device on {target_node_id}"
+                    )
+                    return False
+                time.sleep(0.05)
+
+            response = future.result()
+            self.lab_logger().info(
+                f"[Host Node-DeviceMgr] {action}_device on {target_node_id} completed"
+            )
+            return True
+
+        except Exception as e:
+            self.lab_logger().error(f"[Host Node-DeviceMgr] Error: {e}")
+            self.lab_logger().error(traceback.format_exc())
+            return False
+
+    def create_device(self, device_id: str, config: ResourceDictType) -> dict:
+        """Dynamically create a root-level device on the host."""
+        if not device_id:
+            return {"success": False, "error": "device_id required"}
+
+        if device_id in self.devices_names:
+            return {"success": False, "error": f"Device {device_id} already exists"}
+
+        try:
+            config.setdefault("id", device_id)
+            config.setdefault("type", "device")
+            config.setdefault("machine_name", BasicConfig.machine_name or "本地")
+            res_dict = ResourceDictInstance.get_resource_instance_from_dict(config)
+
+            self.initialize_device(device_id, res_dict)
+
+            if device_id not in self.devices_names:
+                return {"success": False, "error": f"initialize_device failed for {device_id}"}
+
+            # Add to config tree (devices_config)
+            tree = ResourceTreeInstance(res_dict)
+            self.devices_config.trees.append(tree)
+
+            # Add to resource tracker so s2c_resource_tree can find it
+            try:
+                for plr_resource in ResourceTreeSet([tree]).to_plr_resources():
+                    self._resource_tracker.add_resource(plr_resource)
+            except Exception as ex:
+                self.lab_logger().warning(f"[Host Node-DeviceMgr] PLR resource registration skipped: {ex}")
+
+            self.lab_logger().info(f"[Host Node-DeviceMgr] Device {device_id} created successfully")
+            return {"success": True, "device_id": device_id}
+
+        except Exception as e:
+            self.lab_logger().error(f"[Host Node-DeviceMgr] Failed to create {device_id}: {e}")
+            self.lab_logger().error(traceback.format_exc())
+            return {"success": False, "error": str(e)}
+
+    def destroy_device(self, device_id: str) -> dict:
+        """Remove a root-level device from the host."""
+        if not device_id:
+            return {"success": False, "error": "device_id required"}
+
+        if device_id not in self.devices_names:
+            return {"success": False, "error": f"Device {device_id} not found"}
+
+        if device_id == self.device_id:
+            return {"success": False, "error": "Cannot destroy host_node itself"}
+
+        try:
+            namespace = self.devices_names[device_id]
+            device_key = f"{namespace}/{device_id}"
+
+            # Remove action clients
+            action_prefix = f"/devices/{device_id}/"
+            to_remove = [k for k in self._action_clients if k.startswith(action_prefix)]
+            for k in to_remove:
+                try:
+                    self._action_clients[k].destroy()
+                except Exception:
+                    pass
+                del self._action_clients[k]
+
+            # Remove from config tree (devices_config)
+            self.devices_config.trees = [
+                t for t in self.devices_config.trees
+                if t.root_node.res_content.id != device_id
+            ]
+
+            # Remove from resource tracker
+            try:
+                tracked = self._resource_tracker.uuid_to_resources.copy()
+                for uid, res in tracked.items():
+                    res_id = res.get("id") if isinstance(res, dict) else getattr(res, "name", None)
+                    if res_id == device_id:
+                        self._resource_tracker.remove_resource(res)
+            except Exception as ex:
+                self.lab_logger().warning(f"[Host Node-DeviceMgr] Resource tracker cleanup: {ex}")
+
+            # Clean internal state
+            self._online_devices.discard(device_key)
+            self.devices_names.pop(device_id, None)
+            self.device_machine_names.pop(device_id, None)
+            self._action_value_mappings.pop(device_id, None)
+
+            # Destroy the ROS2 node of the device
+            instance = self.devices_instances.pop(device_id, None)
+            if instance is not None:
+                try:
+                    # noinspection PyProtectedMember
+                    ros_node = getattr(instance, "_ros_node", None)
+                    if ros_node is not None:
+                        ros_node.destroy_node()
+                except Exception as e:
+                    self.lab_logger().warning(f"[Host Node-DeviceMgr] Error destroying ROS node for {device_id}: {e}")
+
+            self.lab_logger().info(f"[Host Node-DeviceMgr] Device {device_id} destroyed")
+            return {"success": True, "device_id": device_id}
+
+        except Exception as e:
+            self.lab_logger().error(f"[Host Node-DeviceMgr] Failed to destroy {device_id}: {e}")
+            self.lab_logger().error(traceback.format_exc())
+            return {"success": False, "error": str(e)}

unilabos/ros/nodes/presets/joint_republisher.py

@@ -7,10 +7,11 @@ from rclpy.callback_groups import ReentrantCallbackGroup
 from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode
 
 class JointRepublisher(BaseROS2DeviceNode):
-    def __init__(self,device_id,resource_tracker, **kwargs):
+    def __init__(self,device_id, registry_name, resource_tracker, **kwargs):
         super().__init__(
             driver_instance=self,
             device_id=device_id,
+            registry_name=registry_name,
             status_types={},
             action_value_mappings={},
             hardware_interface={},

unilabos/ros/nodes/presets/resource_mesh_manager.py

@@ -26,7 +26,7 @@ from unilabos.resources.graphio import initialize_resources
 from unilabos.registry.registry import lab_registry
 
 class ResourceMeshManager(BaseROS2DeviceNode):
-    def __init__(self, resource_model: dict, resource_config: list,resource_tracker, device_id: str = "resource_mesh_manager", rate=50, **kwargs):
+    def __init__(self, resource_model: dict, resource_config: list,resource_tracker, device_id: str = "resource_mesh_manager", registry_name: str = "", rate=50, **kwargs):
         """初始化资源网格管理器节点
         
         Args:
@@ -37,6 +37,7 @@ class ResourceMeshManager(BaseROS2DeviceNode):
         super().__init__(
             driver_instance=self,
             device_id=device_id,
+            registry_name=registry_name,
             status_types={},
             action_value_mappings={},
             hardware_interface={},

unilabos/ros/nodes/presets/serial_node.py

@@ -7,7 +7,7 @@ from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, DeviceNodeRe
 
 
 class ROS2SerialNode(BaseROS2DeviceNode):
-    def __init__(self, device_id, port: str, baudrate: int = 9600, resource_tracker: DeviceNodeResourceTracker=None):
+    def __init__(self, device_id, registry_name, port: str, baudrate: int = 9600, resource_tracker: DeviceNodeResourceTracker=None):
         # 保存属性，以便在调用父类初始化前使用
         self.port = port
         self.baudrate = baudrate
@@ -28,6 +28,7 @@ class ROS2SerialNode(BaseROS2DeviceNode):
         BaseROS2DeviceNode.__init__(
             self,
             driver_instance=self,
+            registry_name=registry_name,
             device_id=device_id,
             status_types={},
             action_value_mappings={},

unilabos/ros/nodes/presets/workstation.py

@@ -20,7 +20,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.resources.resource_tracker import ResourceDictType, ResourceTreeSet, ResourceDictInstance
 from unilabos.utils.type_check import get_result_info_str
 
 if TYPE_CHECKING:
@@ -47,6 +47,7 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
         *,
         driver_instance: "WorkstationBase",
         device_id: str,
+        registry_name: str,
         device_uuid: str,
         status_types: Dict[str, Any],
         action_value_mappings: Dict[str, Any],
@@ -62,6 +63,7 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
         super().__init__(
             driver_instance=driver_instance,
             device_id=device_id,
+            registry_name=registry_name,
             device_uuid=device_uuid,
             status_types=status_types,
             action_value_mappings={**action_value_mappings, **self.protocol_action_mappings},
@@ -175,6 +177,103 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
                     self.lab_logger().trace(f"为子设备 {device_id} 创建动作客户端: {action_name}")
         return d
 
+    def create_device(self, device_id: str, config: ResourceDictType) -> dict:
+        """Dynamically add a sub-device to this workstation."""
+        if not device_id:
+            return {"success": False, "error": "device_id required"}
+
+        if device_id in self.sub_devices:
+            return {"success": False, "error": f"Sub-device {device_id} already exists"}
+
+        try:
+            from unilabos.config.config import BasicConfig
+            config.setdefault("id", device_id)
+            config.setdefault("type", "device")
+            config.setdefault("machine_name", BasicConfig.machine_name or "本地")
+            res_dict = ResourceDictInstance.get_resource_instance_from_dict(config)
+
+            d = self.initialize_device(device_id, res_dict)
+            if d is None:
+                return {"success": False, "error": f"initialize_device returned None for {device_id}"}
+
+            # Add to children config list
+            self.children.append(res_dict)
+
+            # Add to resource tracker
+            try:
+                from unilabos.resources.resource_tracker import ResourceTreeInstance
+                tree = ResourceTreeInstance(res_dict)
+                for plr_resource in ResourceTreeSet([tree]).to_plr_resources():
+                    self.resource_tracker.add_resource(plr_resource)
+            except Exception as ex:
+                self.lab_logger().warning(f"[Workstation-DeviceMgr] PLR resource registration skipped: {ex}")
+
+            self.lab_logger().info(f"[Workstation-DeviceMgr] Sub-device {device_id} created")
+            return {"success": True, "device_id": device_id}
+
+        except Exception as e:
+            self.lab_logger().error(f"[Workstation-DeviceMgr] Failed to create {device_id}: {e}")
+            self.lab_logger().error(traceback.format_exc())
+            return {"success": False, "error": str(e)}
+
+    def destroy_device(self, device_id: str) -> dict:
+        """Dynamically remove a sub-device from this workstation."""
+        if not device_id:
+            return {"success": False, "error": "device_id required"}
+
+        if device_id not in self.sub_devices:
+            return {"success": False, "error": f"Sub-device {device_id} not found"}
+
+        try:
+            # Remove from children config list
+            self.children = [
+                c for c in self.children
+                if c.res_content.id != device_id
+            ]
+
+            # Remove from resource tracker
+            try:
+                tracked = self.resource_tracker.uuid_to_resources.copy()
+                for uid, res in tracked.items():
+                    res_id = res.get("id") if isinstance(res, dict) else getattr(res, "name", None)
+                    if res_id == device_id:
+                        self.resource_tracker.remove_resource(res)
+            except Exception as ex:
+                self.lab_logger().warning(f"[Workstation-DeviceMgr] Resource tracker cleanup: {ex}")
+
+            # Remove action clients for this sub-device
+            action_prefix = f"/devices/{device_id}/"
+            to_remove = [k for k in self._action_clients if k.startswith(action_prefix)]
+            for k in to_remove:
+                try:
+                    self._action_clients[k].destroy()
+                except Exception:
+                    pass
+                del self._action_clients[k]
+
+            # Destroy the ROS2 node
+            instance = self.sub_devices.pop(device_id, None)
+            if instance is not None:
+                ros_node = getattr(instance, "ros_node_instance", None)
+                if ros_node is not None:
+                    try:
+                        ros_node.destroy_node()
+                    except Exception as e:
+                        self.lab_logger().warning(
+                            f"[Workstation-DeviceMgr] Error destroying ROS node for {device_id}: {e}"
+                        )
+
+            # Remove from communication map if present
+            self.communication_node_id_to_instance.pop(device_id, None)
+
+            self.lab_logger().info(f"[Workstation-DeviceMgr] Sub-device {device_id} destroyed")
+            return {"success": True, "device_id": device_id}
+
+        except Exception as e:
+            self.lab_logger().error(f"[Workstation-DeviceMgr] Failed to destroy {device_id}: {e}")
+            self.lab_logger().error(traceback.format_exc())
+            return {"success": False, "error": str(e)}
+
     def create_ros_action_server(self, action_name, action_value_mapping):
         """创建ROS动作服务器"""
         if action_name not in self.protocol_names:

unilabos/test/experiments/comprehensive_protocol/comprehensive_station.json

@@ -339,13 +339,8 @@
                 "z": 0
             },
             "config": {
-                "max_volume": 500.0,
                 "type": "RegularContainer",
-                "category": "container",
-                "max_temp": 200.0,
-                "min_temp": -20.0,
-                "has_stirrer": true,
-                "has_heater": true
+                "category": "container"
             },
             "data": {
                 "liquids": [],
@@ -769,9 +764,7 @@
                 "size_y": 250,
                 "size_z": 0,
                 "type": "RegularContainer",
-                "category": "container",
-                "reagent": "sodium_chloride",
-                "physical_state": "solid"
+                "category": "container"
             },
             "data": {
                 "current_mass": 500.0,
@@ -792,14 +785,11 @@
                 "z": 0
             },
             "config": {
-                "volume": 500.0,
                 "size_x": 600,
                 "size_y": 250,
                 "size_z": 0,
                 "type": "RegularContainer",
-                "category": "container",
-                "reagent": "sodium_carbonate",
-                "physical_state": "solid"
+                "category": "container"
             },
             "data": {
                 "current_mass": 500.0,
@@ -820,14 +810,11 @@
                 "z": 0
             },
             "config": {
-                "volume": 500.0,
                 "size_x": 650,
                 "size_y": 250,
                 "size_z": 0,
                 "type": "RegularContainer",
-                "category": "container",
-                "reagent": "magnesium_chloride",
-                "physical_state": "solid"
+                "category": "container"
             },
             "data": {
                 "current_mass": 500.0,

unilabos/utils/decorator.py

@@ -19,74 +19,6 @@ def singleton(cls):
     return get_instance
 
 
-def topic_config(
-    period: Optional[float] = None,
-    print_publish: Optional[bool] = None,
-    qos: Optional[int] = None,
-) -> Callable[[F], F]:
-    """
-    Topic发布配置装饰器
-
-    用于装饰 get_{attr_name} 方法或 @property，控制对应属性的ROS topic发布行为。
-
-    Args:
-        period: 发布周期（秒）。None 表示使用默认值 5.0
-        print_publish: 是否打印发布日志。None 表示使用节点默认配置
-        qos: QoS深度配置。None 表示使用默认值 10
-
-    Example:
-        class MyDriver:
-            # 方式1: 装饰 get_{attr_name} 方法
-            @topic_config(period=1.0, print_publish=False, qos=5)
-            def get_temperature(self):
-                return self._temperature
-
-            # 方式2: 与 @property 连用（topic_config 放在下面）
-            @property
-            @topic_config(period=0.1)
-            def position(self):
-                return self._position
-
-    Note:
-        与 @property 连用时，@topic_config 必须放在 @property 下面，
-        这样装饰器执行顺序为：先 topic_config 添加配置，再 property 包装。
-    """
-
-    def decorator(func: F) -> F:
-        @wraps(func)
-        def wrapper(*args, **kwargs):
-            return func(*args, **kwargs)
-
-        # 在函数上附加配置属性 (type: ignore 用于动态属性)
-        wrapper._topic_period = period  # type: ignore[attr-defined]
-        wrapper._topic_print_publish = print_publish  # type: ignore[attr-defined]
-        wrapper._topic_qos = qos  # type: ignore[attr-defined]
-        wrapper._has_topic_config = True  # type: ignore[attr-defined]
-
-        return wrapper  # type: ignore[return-value]
-
-    return decorator
-
-
-def get_topic_config(func) -> dict:
-    """
-    获取函数上的topic配置
-
-    Args:
-        func: 被装饰的函数
-
-    Returns:
-        包含 period, print_publish, qos 的配置字典
-    """
-    if hasattr(func, "_has_topic_config") and getattr(func, "_has_topic_config", False):
-        return {
-            "period": getattr(func, "_topic_period", None),
-            "print_publish": getattr(func, "_topic_print_publish", None),
-            "qos": getattr(func, "_topic_qos", None),
-        }
-    return {}
-
-
 def subscribe(
     topic: str,
     msg_type: Optional[type] = None,
@@ -104,24 +36,6 @@ def subscribe(
             - {namespace}: 完整命名空间 (如 "/devices/pump_1")
         msg_type: ROS 消息类型。如果为 None，需要在回调函数的类型注解中指定
         qos: QoS 深度配置，默认为 10
-
-    Example:
-        from std_msgs.msg import String, Float64
-
-        class MyDriver:
-            @subscribe(topic="/devices/{device_id}/set_speed", msg_type=Float64)
-            def on_speed_update(self, msg: Float64):
-                self._speed = msg.data
-                print(f"Speed updated to: {self._speed}")
-
-            @subscribe(topic="{namespace}/command")
-            def on_command(self, msg: String):
-                # msg_type 可从类型注解推断
-                self.execute_command(msg.data)
-
-    Note:
-        - 回调方法的第一个参数是 self，第二个参数是收到的 ROS 消息
-        - topic 中的占位符会在创建订阅时被实际值替换
     """
 
     def decorator(func: F) -> F:
@@ -129,7 +43,6 @@ def subscribe(
         def wrapper(*args, **kwargs):
             return func(*args, **kwargs)
 
-        # 在函数上附加订阅配置
         wrapper._subscribe_topic = topic  # type: ignore[attr-defined]
         wrapper._subscribe_msg_type = msg_type  # type: ignore[attr-defined]
         wrapper._subscribe_qos = qos  # type: ignore[attr-defined]
@@ -141,15 +54,7 @@ def subscribe(
 
 
 def get_subscribe_config(func) -> dict:
-    """
-    获取函数上的订阅配置
-
-    Args:
-        func: 被装饰的函数
-
-    Returns:
-        包含 topic, msg_type, qos 的配置字典
-    """
+    """获取函数上的订阅配置 (topic, msg_type, qos)"""
     if hasattr(func, "_has_subscribe") and getattr(func, "_has_subscribe", False):
         return {
             "topic": getattr(func, "_subscribe_topic", None),
@@ -163,9 +68,6 @@ def get_all_subscriptions(instance) -> list:
     """
     扫描实例的所有方法，获取带有 @subscribe 装饰器的方法及其配置
 
-    Args:
-        instance: 要扫描的实例
-
     Returns:
         包含 (method_name, method, config) 元组的列表
     """
@@ -184,47 +86,14 @@ def get_all_subscriptions(instance) -> list:
     return subscriptions
 
 
-def not_action(func: F) -> F:
-    """
-    标记方法为非动作的装饰器
-
-    用于装饰 driver 类中的方法，使其在 complete_registry 时不被识别为动作。
-    适用于辅助方法、内部工具方法等不应暴露为设备动作的公共方法。
-
-    Example:
-        class MyDriver:
-            @not_action
-            def helper_method(self):
-                # 这个方法不会被注册为动作
-                pass
-
-            def actual_action(self, param: str):
-                # 这个方法会被注册为动作
-                self.helper_method()
-
-    Note:
-        - 可以与其他装饰器组合使用，@not_action 应放在最外层
-        - 仅影响 complete_registry 的动作识别，不影响方法的正常调用
-    """
-
-    @wraps(func)
-    def wrapper(*args, **kwargs):
-        return func(*args, **kwargs)
-
-    # 在函数上附加标记
-    wrapper._is_not_action = True  # type: ignore[attr-defined]
-
-    return wrapper  # type: ignore[return-value]
-
-
-def is_not_action(func) -> bool:
-    """
-    检查函数是否被标记为非动作
-
-    Args:
-        func: 被检查的函数
-
-    Returns:
-        如果函数被 @not_action 装饰则返回 True，否则返回 False
-    """
-    return getattr(func, "_is_not_action", False)
+# ---------------------------------------------------------------------------
+# 向后兼容重导出 -- 已迁移到 unilabos.registry.decorators
+# ---------------------------------------------------------------------------
+from unilabos.registry.decorators import (  # noqa: E402, F401
+    topic_config,
+    get_topic_config,
+    always_free,
+    is_always_free,
+    not_action,
+    is_not_action,
+)

unilabos/utils/environment_check.py

@@ -22,6 +22,7 @@ class EnvironmentChecker:
             # "pymodbus.framer.FramerType": "pymodbus==3.9.2",
             "websockets": "websockets",
             "msgcenterpy": "msgcenterpy",
+            "orjson": "orjson",
             "opentrons_shared_data": "opentrons_shared_data",
             "typing_extensions": "typing_extensions",
             "crcmod": "crcmod-plus",
@@ -32,7 +33,7 @@ class EnvironmentChecker:
 
         # 包版本要求（包名: 最低版本）
         self.version_requirements = {
-            "msgcenterpy": "0.1.5",  # msgcenterpy 最低版本要求
+            "msgcenterpy": "0.1.7",  # msgcenterpy 最低版本要求
         }
 
         self.missing_packages = []

unilabos/utils/import_manager.py

@@ -29,7 +29,7 @@ from ast import Constant
 
 from unilabos.resources.resource_tracker import PARAM_SAMPLE_UUIDS
 from unilabos.utils import logger
-from unilabos.utils.decorator import is_not_action
+from unilabos.registry.decorators import is_not_action, is_always_free
 
 
 class ImportManager:
@@ -282,6 +282,9 @@ class ImportManager:
                         continue
                     # 其他非_开头的方法归类为action
                     method_info = self._analyze_method_signature(method)
+                    # 检查是否被 @always_free 装饰器标记
+                    if is_always_free(method):
+                        method_info["always_free"] = True
                     result["action_methods"][name] = method_info
 
         return result
@@ -339,6 +342,9 @@ class ImportManager:
                     if self._is_not_action_method(node):
                         continue
                     # 其他非_开头的方法归类为action
+                    # 检查是否被 @always_free 装饰器标记
+                    if self._is_always_free_method(node):
+                        method_info["always_free"] = True
                     result["action_methods"][method_name] = method_info
         return result
 
@@ -474,6 +480,19 @@ class ImportManager:
                 return True
         return False
 
+    def _is_always_free_method(self, node: ast.FunctionDef) -> bool:
+        """检查是否是@always_free装饰的方法，或 @action(always_free=True) 装饰的方法"""
+        for decorator in node.decorator_list:
+            # 检查 @action(always_free=True)
+            if isinstance(decorator, ast.Call):
+                func = decorator.func
+                if isinstance(func, ast.Name) and func.id == "action":
+                    for keyword in decorator.keywords:
+                        if keyword.arg == "always_free":
+                            if isinstance(keyword.value, Constant) and keyword.value.value is True:
+                                return True
+        return False
+
     def _get_property_name_from_setter(self, node: ast.FunctionDef) -> str:
         """从setter装饰器中获取属性名"""
         for decorator in node.decorator_list:

unilabos/utils/log.py

@@ -193,6 +193,7 @@ def configure_logger(loglevel=None, working_dir=None):
     root_logger.addHandler(console_handler)
 
     # 如果指定了工作目录，添加文件处理器
+    log_filepath = None
     if working_dir is not None:
         logs_dir = os.path.join(working_dir, "logs")
         os.makedirs(logs_dir, exist_ok=True)
@@ -213,7 +214,7 @@ def configure_logger(loglevel=None, working_dir=None):
 
     logging.getLogger("asyncio").setLevel(logging.INFO)
     logging.getLogger("urllib3").setLevel(logging.INFO)
-
+    return log_filepath
 
 
 # 配置日志系统

unilabos/utils/requirements.txt

@@ -1,7 +1,8 @@
 networkx
 typing_extensions
 websockets
-msgcenterpy>=0.1.5
+msgcenterpy>=0.1.7
+orjson>=3.11
 opentrons_shared_data
 pint
 fastapi

unilabos/workflow/common.py

@@ -26,7 +26,7 @@
     res_id: plate_slot_{slot}
     device_id: /PRCXI
     class_name: PRCXI_BioER_96_wellplate
-    parent: /PRCXI/PRCXI_Deck/T{slot}
+    parent: /PRCXI/PRCXI_Deck
     slot_on_deck: "{slot}"
 - 输出端口: labware（用于连接 set_liquid_from_plate）
 - 控制流: create_resource 之间通过 ready 端口串联
@@ -122,7 +122,7 @@ NODE_TYPE_DEFAULT = "ILab"  # 所有节点的默认类型
 # create_resource 节点默认参数
 CREATE_RESOURCE_DEFAULTS = {
     "device_id": "/PRCXI",
-    "parent_template": "/PRCXI/PRCXI_Deck/T{slot}",  # {slot} 会被替换为实际的 slot 值
+    "parent_template": "/PRCXI/PRCXI_Deck",
     "class_name": "PRCXI_BioER_96_wellplate",
 }
 
@@ -362,14 +362,16 @@ def build_protocol_graph(
     protocol_steps: List[Dict[str, Any]],
     workstation_name: str,
     action_resource_mapping: Optional[Dict[str, str]] = None,
+    labware_defs: Optional[List[Dict[str, Any]]] = None,
 ) -> WorkflowGraph:
     """统一的协议图构建函数，根据设备类型自动选择构建逻辑
 
     Args:
-        labware_info: labware 信息字典，格式为 {name: {slot, well, labware, ...}, ...}
+        labware_info: reagent 信息字典，格式为 {name: {slot, well}, ...}，用于 set_liquid 和 well 查找
         protocol_steps: 协议步骤列表
         workstation_name: 工作站名称
         action_resource_mapping: action 到 resource_name 的映射字典，可选
+        labware_defs: labware 定义列表，格式为 [{"name": "...", "slot": "1", "type": "lab_xxx"}, ...]
     """
     G = WorkflowGraph()
     resource_last_writer = {}  # reagent_name -> "node_id:port"
@@ -377,18 +379,7 @@ def build_protocol_graph(
 
     protocol_steps = refactor_data(protocol_steps, action_resource_mapping)
 
-    # ==================== 第一步：按 slot 去重创建 create_resource 节点 ====================
-    # 收集所有唯一的 slot
-    slots_info = {}  # slot -> {labware, res_id}
-    for labware_id, item in labware_info.items():
-        slot = str(item.get("slot", ""))
-        if slot and slot not in slots_info:
-            res_id = f"plate_slot_{slot}"
-            slots_info[slot] = {
-                "labware": item.get("labware", ""),
-                "res_id": res_id,
-            }
-
+    # ==================== 第一步：按 slot 创建 create_resource 节点 ====================
     # 创建 Group 节点，包含所有 create_resource 节点
     group_node_id = str(uuid.uuid4())
     G.add_node(
@@ -404,38 +395,42 @@ def build_protocol_graph(
         param=None,
     )
 
-    # 为每个唯一的 slot 创建 create_resource 节点
+    # 直接使用 JSON 中的 labware 定义，每个 slot 一条记录，type 即 class_name
     res_index = 0
-    for slot, info in slots_info.items():
-        node_id = str(uuid.uuid4())
-        res_id = info["res_id"]
+    for lw in (labware_defs or []):
+        slot = str(lw.get("slot", ""))
+        if not slot or slot in slot_to_create_resource:
+            continue  # 跳过空 slot 或已处理的 slot
+
+        lw_name = lw.get("name", f"slot {slot}")
+        lw_type = lw.get("type", CREATE_RESOURCE_DEFAULTS["class_name"])
+        res_id = f"plate_slot_{slot}"
 
         res_index += 1
+        node_id = str(uuid.uuid4())
         G.add_node(
             node_id,
             template_name="create_resource",
             resource_name="host_node",
-            name=f"Plate {res_index}",
-            description=f"Create plate on slot {slot}",
+            name=lw_name,
+            description=f"Create {lw_name}",
             lab_node_type="Labware",
             footer="create_resource-host_node",
             device_name=DEVICE_NAME_HOST,
             type=NODE_TYPE_DEFAULT,
-            parent_uuid=group_node_id,  # 指向 Group 节点
-            minimized=True,  # 折叠显示
+            parent_uuid=group_node_id,
+            minimized=True,
             param={
                 "res_id": res_id,
                 "device_id": CREATE_RESOURCE_DEFAULTS["device_id"],
-                "class_name": CREATE_RESOURCE_DEFAULTS["class_name"],
-                "parent": CREATE_RESOURCE_DEFAULTS["parent_template"].format(slot=slot),
+                "class_name": lw_type,
+                "parent": CREATE_RESOURCE_DEFAULTS["parent_template"],
                 "bind_locations": {"x": 0.0, "y": 0.0, "z": 0.0},
                 "slot_on_deck": slot,
             },
         )
         slot_to_create_resource[slot] = node_id
 
-        # create_resource 之间不需要 ready 连接
-
     # ==================== 第二步：为每个 reagent 创建 set_liquid_from_plate 节点 ====================
     # 创建 Group 节点，包含所有 set_liquid_from_plate 节点
     set_liquid_group_id = str(uuid.uuid4())

unilabos/workflow/convert_from_json.py

@@ -1,16 +1,20 @@
 """
 JSON 工作流转换模块
 
-将 workflow/reagent 格式的 JSON 转换为统一工作流格式。
+将 workflow/reagent/labware 格式的 JSON 转换为统一工作流格式。
 
 输入格式:
 {
+    "labware": [
+        {"name": "...", "slot": "1", "type": "lab_xxx"},
+        ...
+    ],
     "workflow": [
         {"action": "...", "action_args": {...}},
         ...
     ],
     "reagent": {
-        "reagent_name": {"slot": int, "well": [...], "labware": "..."},
+        "reagent_name": {"slot": int, "well": [...]},
         ...
     }
 }
@@ -245,18 +249,18 @@ def convert_from_json(
     if "workflow" not in json_data or "reagent" not in json_data:
         raise ValueError(
             "不支持的 JSON 格式。请使用标准格式:\n"
-            '{"workflow": [{"action": "...", "action_args": {...}}, ...], '
-            '"reagent": {"name": {"slot": int, "well": [...], "labware": "..."}, ...}}'
+            '{"labware": [...], "workflow": [...], "reagent": {...}}'
         )
 
     # 提取数据
     workflow = json_data["workflow"]
     reagent = json_data["reagent"]
+    labware_defs = json_data.get("labware", [])  # 新的 labware 定义列表
 
     # 规范化步骤数据
     protocol_steps = normalize_workflow_steps(workflow)
 
-    # reagent 已经是字典格式，直接使用
+    # reagent 已经是字典格式，用于 set_liquid 和 well 数量查找
     labware_info = reagent
 
     # 构建工作流图
@@ -265,6 +269,7 @@ def convert_from_json(
         protocol_steps=protocol_steps,
         workstation_name=workstation_name,
         action_resource_mapping=ACTION_RESOURCE_MAPPING,
+        labware_defs=labware_defs,
     )
 
     # 校验句柄配置

unilabos/workflow/wf_utils.py

@@ -41,6 +41,7 @@ def upload_workflow(
     workflow_name: Optional[str] = None,
     tags: Optional[List[str]] = None,
     published: bool = False,
+    description: str = "",
 ) -> Dict[str, Any]:
     """
     上传工作流到服务器
@@ -56,6 +57,7 @@ def upload_workflow(
         workflow_name: 工作流名称，如果不提供则从文件中读取或使用文件名
         tags: 工作流标签列表，默认为空列表
         published: 是否发布工作流，默认为False
+        description: 工作流描述，发布时使用
 
     Returns:
         Dict: API响应数据
@@ -75,6 +77,14 @@ def upload_workflow(
         print_status(f"工作流文件JSON解析失败: {e}", "error")
         return {"code": -1, "message": f"JSON解析失败: {e}"}
 
+    # 从 JSON 文件中提取 description 和 tags（作为 fallback）
+    if not description and "description" in workflow_data:
+        description = workflow_data["description"]
+        print_status(f"从文件中读取 description", "info")
+    if not tags and "tags" in workflow_data:
+        tags = workflow_data["tags"]
+        print_status(f"从文件中读取 tags: {tags}", "info")
+
     # 自动检测并转换格式
     if not _is_node_link_format(workflow_data):
         try:
@@ -96,6 +106,7 @@ def upload_workflow(
     print_status(f"  - 节点数量: {len(nodes)}", "info")
     print_status(f"  - 边数量: {len(edges)}", "info")
     print_status(f"  - 标签: {tags or []}", "info")
+    print_status(f"  - 描述: {description[:50]}{'...' if len(description) > 50 else ''}", "info")
     print_status(f"  - 发布状态: {published}", "info")
 
     # 调用 http_client 上传
@@ -107,6 +118,7 @@ def upload_workflow(
         edges=edges,
         tags=tags,
         published=published,
+        description=description,
     )
 
     if result.get("code") == 0:
@@ -131,8 +143,9 @@ def handle_workflow_upload_command(args_dict: Dict[str, Any]) -> None:
     workflow_name = args_dict.get("workflow_name")
     tags = args_dict.get("tags", [])
     published = args_dict.get("published", False)
+    description = args_dict.get("description", "")
 
     if workflow_file:
-        upload_workflow(workflow_file, workflow_name, tags, published)
+        upload_workflow(workflow_file, workflow_name, tags, published, description)
     else:
         print_status("未指定工作流文件路径，请使用 -f/--workflow_file 参数", "error")

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.17</version>
+  <version>0.10.18</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>