Merge branch 'dev' into prcix9320

docs/user_guide/best_practice.md

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

unilabos/app/model.py

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

unilabos/app/web/controller.py

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

unilabos/app/ws_client.py

@@ -540,7 +540,7 @@ class MessageProcessor:
                         try:
                             message_str = json.dumps(msg, ensure_ascii=False)
                             await self.websocket.send(message_str)
-                            logger.trace(f"[MessageProcessor] Message sent: {msg.get('action', 'unknown')}")  # type: ignore  # noqa: E501
+                            # logger.trace(f"[MessageProcessor] Message sent: {msg.get('action', 'unknown')}")  # type: ignore  # noqa: E501
                         except Exception as e:
                             logger.error(f"[MessageProcessor] Failed to send message: {str(e)}")
                             logger.error(traceback.format_exc())
@@ -652,6 +652,8 @@ class MessageProcessor:
     async def _handle_job_start(self, data: Dict[str, Any]):
         """处理job_start消息"""
         try:
+            if not data.get("sample_material"):
+                data["sample_material"] = {}
             req = JobAddReq(**data)
 
             job_log = format_job_log(req.job_id, req.task_id, req.device_id, req.action)
@@ -683,6 +685,7 @@ class MessageProcessor:
                 queue_item,
                 action_type=req.action_type,
                 action_kwargs=req.action_args,
+                sample_material=req.sample_material,
                 server_info=req.server_info,
             )
 
@@ -1294,7 +1297,7 @@ class WebSocketClient(BaseCommunicationClient):
             },
         }
         self.message_processor.send_message(message)
-        logger.trace(f"[WebSocketClient] Device status published: {device_id}.{property_name}")
+        # logger.trace(f"[WebSocketClient] Device status published: {device_id}.{property_name}")
 
     def publish_job_status(
         self, feedback_data: dict, item: QueueItem, status: str, return_info: Optional[dict] = None

unilabos/compile/pump_protocol.py

@@ -95,8 +95,29 @@ def get_vessel_liquid_volume(G: nx.DiGraph, vessel: str) -> float:
     return total_volume
 
 
-def is_integrated_pump(node_name):
+def is_integrated_pump(node_class: str, node_name: str = "") -> bool:
-    return "pump" in node_name and "valve" in node_name
+    """
+    判断是否为泵阀一体设备
+    """
+    class_lower = (node_class or "").lower()
+    name_lower = (node_name or "").lower()
+
+    if "pump" not in class_lower and "pump" not in name_lower:
+        return False
+
+    integrated_markers = [
+        "valve",
+        "pump_valve",
+        "pumpvalve",
+        "integrated",
+        "transfer_pump",
+    ]
+
+    for marker in integrated_markers:
+        if marker in class_lower or marker in name_lower:
+            return True
+
+    return False
 
 
 def find_connected_pump(G, valve_node):
@@ -186,7 +207,9 @@ def build_pump_valve_maps(G, pump_backbone):
     debug_print(f"🔧 过滤后的骨架: {filtered_backbone}")
 
     for node in filtered_backbone:
-        if is_integrated_pump(G.nodes[node]["class"]):
+        node_data = G.nodes.get(node, {})
+        node_class = node_data.get("class", "") or ""
+        if is_integrated_pump(node_class, node):
             pumps_from_node[node] = node
             valve_from_node[node] = node
             debug_print(f"  - 集成泵-阀: {node}")

unilabos/devices/liquid_handling/liquid_handler_abstract.py

@@ -27,7 +27,12 @@ from typing_extensions import TypedDict
 
 from unilabos.devices.liquid_handling.rviz_backend import UniLiquidHandlerRvizBackend
 from unilabos.registry.placeholder_type import ResourceSlot
-from unilabos.resources.resource_tracker import ResourceTreeSet, ResourceDict
+from unilabos.resources.resource_tracker import (
+    ResourceTreeSet,
+    ResourceDict,
+    EXTRA_SAMPLE_UUID,
+    EXTRA_UNILABOS_SAMPLE_UUID,
+)
 from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode
 
 
@@ -241,7 +246,7 @@ class LiquidHandlerMiddleware(LiquidHandler):
             res_samples.append({"name": resource.name, "sample_uuid": resource.unilabos_extra.get("sample_uuid", None)})
             res_volumes.append(volume)
             self.pending_liquids_dict[channel] = {
-                "sample_uuid": resource.unilabos_extra.get("sample_uuid", None),
+                EXTRA_SAMPLE_UUID: sample_uuid_value,
                 "volume": volume,
             }
         return SimpleReturn(samples=res_samples, volumes=res_volumes)
@@ -283,10 +288,10 @@ class LiquidHandlerMiddleware(LiquidHandler):
         res_samples = []
         res_volumes = []
         for resource, volume, channel in zip(resources, vols, use_channels):
-            res_uuid = self.pending_liquids_dict[channel]["sample_uuid"]
+            res_uuid = self.pending_liquids_dict[channel][EXTRA_SAMPLE_UUID]
             self.pending_liquids_dict[channel]["volume"] -= volume
-            resource.unilabos_extra["sample_uuid"] = res_uuid
+            resource.unilabos_extra[EXTRA_SAMPLE_UUID] = res_uuid
-            res_samples.append({"name": resource.name, "sample_uuid": res_uuid})
+            res_samples.append({"name": resource.name, EXTRA_SAMPLE_UUID: res_uuid})
             res_volumes.append(volume)
 
         return SimpleReturn(samples=res_samples, volumes=res_volumes)
@@ -691,16 +696,14 @@ class LiquidHandlerAbstract(LiquidHandlerMiddleware):
         )
 
     def set_liquid_from_plate(
-        self, plate: List[ResourceSlot], well_names: list[str], liquid_names: list[str], volumes: list[float]
+        self, plate: ResourceSlot, well_names: list[str], liquid_names: list[str], volumes: list[float]
     ) -> SetLiquidFromPlateReturn:
         """Set the liquid in wells of a plate by well names (e.g., A1, A2, B3).
 
         如果 liquid_names 和 volumes 为空，但 plate 和 well_names 不为空，直接返回 plate 和 wells。
         """
-        if isinstance(plate, list):  # 未来移除
-            plate = plate[0]
         assert issubclass(plate.__class__, Plate), "plate must be a Plate"
-        plate: Plate = cast(Plate, plate)
+        plate: Plate = cast(Plate, cast(Resource, plate))
         # 根据 well_names 获取对应的 Well 对象
         wells = [plate.get_well(name) for name in well_names]
         res_volumes = []

unilabos/devices/liquid_handling/prcxi/prcxi.py

@@ -803,7 +803,7 @@ class PRCXI9300Handler(LiquidHandlerAbstract):
         return super().set_liquid(wells, liquid_names, volumes)
 
     def set_liquid_from_plate(
-        self, plate: List[ResourceSlot], well_names: list[str], liquid_names: list[str], volumes: list[float]
+        self, plate: ResourceSlot, well_names: list[str], liquid_names: list[str], volumes: list[float]
     ) -> SetLiquidFromPlateReturn:
         return super().set_liquid_from_plate(plate, well_names, liquid_names, volumes)
 

unilabos/devices/virtual/virtual_transferpump.py

@@ -31,14 +31,14 @@ class VirtualTransferPump:
 
         # 从config或kwargs中获取参数，确保类型正确
         if config:
-            self.max_volume = float(config.get('max_volume', 25.0))
+            self.max_volume = float(config.get("max_volume", 25.0))
-            self.port = config.get('port', 'VIRTUAL')
+            self.port = config.get("port", "VIRTUAL")
         else:
-            self.max_volume = float(kwargs.get('max_volume', 25.0))
+            self.max_volume = float(kwargs.get("max_volume", 25.0))
-            self.port = kwargs.get('port', 'VIRTUAL')
+            self.port = kwargs.get("port", "VIRTUAL")
 
-        self._transfer_rate = float(kwargs.get('transfer_rate', 0))
+        self._transfer_rate = float(kwargs.get("transfer_rate", 0))
-        self.mode = kwargs.get('mode', VirtualPumpMode.Normal)
+        self.mode = kwargs.get("mode", VirtualPumpMode.Normal)
 
         # 状态变量 - 确保都是正确类型
         self._status = "Idle"
@@ -54,7 +54,9 @@ class VirtualTransferPump:
         self.logger = logging.getLogger(f"VirtualTransferPump.{self.device_id}")
 
         print(f"🚰 === 虚拟转移泵 {self.device_id} 已创建 === ✨")
-        print(f"💨 快速模式: {'启用' if self._fast_mode else '禁用'} | 移动时间: {self._fast_move_time}s | 喷射时间: {self._fast_dispense_time}s")
+        print(
+            f"💨 快速模式: {'启用' if self._fast_mode else '禁用'} | 移动时间: {self._fast_move_time}s | 喷射时间: {self._fast_dispense_time}s"
+        )
         print(f"📊 最大容量: {self.max_volume}mL | 端口: {self.port}")
 
     def post_init(self, ros_node: BaseROS2DeviceNode):
@@ -189,7 +191,9 @@ class VirtualTransferPump:
                 operation_emoji = "📍"
 
             self.logger.info(f"🎯 SET_POSITION: {operation_type} {operation_emoji}")
-            self.logger.info(f"  📍 位置: {self._position:.2f}mL → {target_position:.2f}mL (移动 {volume_to_move:.2f}mL)")
+            self.logger.info(
+                f"  📍 位置: {self._position:.2f}mL → {target_position:.2f}mL (移动 {volume_to_move:.2f}mL)"
+            )
             self.logger.info(f"  🌊 速度: {velocity:.2f} mL/s")
             self.logger.info(f"  ⏰ 预计时间: {display_duration:.2f}s")
 
@@ -207,7 +211,11 @@ class VirtualTransferPump:
                 for i in range(steps + 1):
                     # 计算当前位置和进度
                     progress = (i / steps) * 100 if steps > 0 else 100
-                    current_pos = start_position + (target_position - start_position) * (i / steps) if steps > 0 else target_position
+                    current_pos = (
+                        start_position + (target_position - start_position) * (i / steps)
+                        if steps > 0
+                        else target_position
+                    )
 
                     # 更新状态
                     if i < steps:
@@ -244,7 +252,9 @@ class VirtualTransferPump:
 
             # 📊 最终状态日志
             if volume_to_move > 0.01:
-                self.logger.info(f"🎉 SET_POSITION 完成! 📍 最终位置: {self._position:.2f}mL | 💧 当前体积: {self._current_volume:.2f}mL")
+                self.logger.info(
+                    f"🎉 SET_POSITION 完成! 📍 最终位置: {self._position:.2f}mL | 💧 当前体积: {self._current_volume:.2f}mL"
+                )
 
             # 返回符合action定义的结果
             return {
@@ -252,7 +262,7 @@ class VirtualTransferPump:
                 "message": f"✅ 成功移动到位置 {self._position:.2f}mL ({operation_type})",
                 "final_position": self._position,
                 "final_volume": self._current_volume,
-                "operation_type": operation_type
+                "operation_type": operation_type,
             }
 
         except Exception as e:
@@ -262,7 +272,7 @@ class VirtualTransferPump:
                 "success": False,
                 "message": error_msg,
                 "final_position": self._position,
-                "final_volume": self._current_volume
+                "final_volume": self._current_volume,
             }
 
     # 其他泵操作方法
@@ -388,7 +398,9 @@ class VirtualTransferPump:
         return self._current_volume >= (self.max_volume - 0.01)  # 允许小量误差
 
     def __str__(self):
-        return f"VirtualTransferPump({self.device_id}: {self._current_volume:.2f}/{self.max_volume} ml, {self._status})"
+        return (
+            f"VirtualTransferPump({self.device_id}: {self._current_volume:.2f}/{self.max_volume} ml, {self._status})"
+        )
 
     def __repr__(self):
         return self.__str__()

unilabos/devices/virtual/workbench.py

@@ -0,0 +1,742 @@
+"""
+Virtual Workbench Device - 模拟工作台设备
+包含：
+- 1个机械臂 (每次操作3s, 独占锁)
+- 3个加热台 (每次加热10s, 可并行)
+
+工作流程：
+1. A1-A5 物料同时启动，竞争机械臂
+2. 机械臂将物料移动到空闲加热台
+3. 加热完成后，机械臂将物料移动到C1-C5
+
+注意：调用来自线程池，使用 threading.Lock 进行同步
+"""
+
+import logging
+import time
+from typing import Dict, Any, Optional, List
+from dataclasses import dataclass
+from enum import Enum
+from threading import Lock, RLock
+
+from typing_extensions import TypedDict
+
+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
+
+
+# ============ TypedDict 返回类型定义 ============
+
+
+class MoveToHeatingStationResult(TypedDict):
+    """move_to_heating_station 返回类型"""
+
+    success: bool
+    station_id: int
+    material_id: str
+    material_number: int
+    message: str
+    unilabos_samples: List[LabSample]
+
+
+class StartHeatingResult(TypedDict):
+    """start_heating 返回类型"""
+
+    success: bool
+    station_id: int
+    material_id: str
+    material_number: int
+    message: str
+    unilabos_samples: List[LabSample]
+
+
+class MoveToOutputResult(TypedDict):
+    """move_to_output 返回类型"""
+
+    success: bool
+    station_id: int
+    material_id: str
+    unilabos_samples: List[LabSample]
+
+
+class PrepareMaterialsResult(TypedDict):
+    """prepare_materials 返回类型 - 批量准备物料"""
+
+    success: bool
+    count: int
+    material_1: int  # 物料编号1
+    material_2: int  # 物料编号2
+    material_3: int  # 物料编号3
+    material_4: int  # 物料编号4
+    material_5: int  # 物料编号5
+    message: str
+    unilabos_samples: List[LabSample]
+
+
+# ============ 状态枚举 ============
+
+
+class HeatingStationState(Enum):
+    """加热台状态枚举"""
+
+    IDLE = "idle"  # 空闲
+    OCCUPIED = "occupied"  # 已放置物料，等待加热
+    HEATING = "heating"  # 加热中
+    COMPLETED = "completed"  # 加热完成，等待取走
+
+
+class ArmState(Enum):
+    """机械臂状态枚举"""
+
+    IDLE = "idle"  # 空闲
+    BUSY = "busy"  # 工作中
+
+
+@dataclass
+class HeatingStation:
+    """加热台数据结构"""
+
+    station_id: int
+    state: HeatingStationState = HeatingStationState.IDLE
+    current_material: Optional[str] = None  # 当前物料 (如 "A1", "A2")
+    material_number: Optional[int] = None  # 物料编号 (1-5)
+    heating_start_time: Optional[float] = None
+    heating_progress: float = 0.0
+
+
+class VirtualWorkbench:
+    """
+    Virtual Workbench Device - 虚拟工作台设备
+
+    模拟一个包含1个机械臂和3个加热台的工作站
+    - 机械臂操作耗时3秒，同一时间只能执行一个操作
+    - 加热台加热耗时10秒，3个加热台可并行工作
+
+    工作流:
+    1. 物料A1-A5并发启动（线程池），竞争机械臂使用权
+    2. 获取机械臂后，查找空闲加热台
+    3. 机械臂将物料放入加热台，开始加热
+    4. 加热完成后，机械臂将物料移动到目标位置Cn
+    """
+
+    _ros_node: BaseROS2DeviceNode
+
+    # 配置常量
+    ARM_OPERATION_TIME: float = 3.0  # 机械臂操作时间(秒)
+    HEATING_TIME: float = 10.0  # 加热时间(秒)
+    NUM_HEATING_STATIONS: int = 3  # 加热台数量
+
+    def __init__(self, device_id: Optional[str] = None, config: Optional[Dict[str, Any]] = None, **kwargs):
+        # 处理可能的不同调用方式
+        if device_id is None and "id" in kwargs:
+            device_id = kwargs.pop("id")
+        if config is None and "config" in kwargs:
+            config = kwargs.pop("config")
+
+        self.device_id = device_id or "virtual_workbench"
+        self.config = config or {}
+
+        self.logger = logging.getLogger(f"VirtualWorkbench.{self.device_id}")
+        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))
+
+        # 机械臂状态和锁 (使用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._active_tasks: Dict[str, Dict[str, Any]] = {}  # material_id -> task_info
+        self._tasks_lock = Lock()
+
+        # 处理其他kwargs参数
+        skip_keys = {"arm_operation_time", "heating_time", "num_heating_stations"}
+        for key, value in kwargs.items():
+            if key not in skip_keys and not hasattr(self, key):
+                setattr(self, key, value)
+
+        self.logger.info(f"=== 虚拟工作台 {self.device_id} 已创建 ===")
+        self.logger.info(
+            f"机械臂操作时间: {self.ARM_OPERATION_TIME}s | "
+            f"加热时间: {self.HEATING_TIME}s | "
+            f"加热台数量: {self.NUM_HEATING_STATIONS}"
+        )
+
+    @not_action
+    def post_init(self, ros_node: BaseROS2DeviceNode):
+        """ROS节点初始化后回调"""
+        self._ros_node = ros_node
+
+    @not_action
+    def initialize(self) -> bool:
+        """初始化虚拟工作台"""
+        self.logger.info(f"初始化虚拟工作台 {self.device_id}")
+
+        # 重置加热台状态 (已在__init__中创建，这里重置为初始状态)
+        with self._stations_lock:
+            for station in self._heating_stations.values():
+                station.state = HeatingStationState.IDLE
+                station.current_material = None
+                station.material_number = None
+                station.heating_progress = 0.0
+
+        # 初始化状态
+        self.data.update(
+            {
+                "status": "Ready",
+                "arm_state": ArmState.IDLE.value,
+                "arm_current_task": None,
+                "heating_stations": self._get_stations_status(),
+                "active_tasks_count": 0,
+                "message": "工作台就绪",
+            }
+        )
+
+        self.logger.info(f"工作台初始化完成: {self.NUM_HEATING_STATIONS}个加热台就绪")
+        return True
+
+    @not_action
+    def cleanup(self) -> bool:
+        """清理虚拟工作台"""
+        self.logger.info(f"清理虚拟工作台 {self.device_id}")
+
+        self._arm_state = ArmState.IDLE
+        self._arm_current_task = None
+
+        with self._stations_lock:
+            self._heating_stations.clear()
+
+        with self._tasks_lock:
+            self._active_tasks.clear()
+
+        self.data.update(
+            {
+                "status": "Offline",
+                "arm_state": ArmState.IDLE.value,
+                "heating_stations": {},
+                "message": "工作台已关闭",
+            }
+        )
+        return True
+
+    def _get_stations_status(self) -> Dict[int, Dict[str, Any]]:
+        """获取所有加热台状态"""
+        with self._stations_lock:
+            return {
+                station_id: {
+                    "state": station.state.value,
+                    "current_material": station.current_material,
+                    "material_number": station.material_number,
+                    "heating_progress": station.heating_progress,
+                }
+                for station_id, station in self._heating_stations.items()
+            }
+
+    def _update_data_status(self, message: Optional[str] = None):
+        """更新状态数据"""
+        self.data.update(
+            {
+                "arm_state": self._arm_state.value,
+                "arm_current_task": self._arm_current_task,
+                "heating_stations": self._get_stations_status(),
+                "active_tasks_count": len(self._active_tasks),
+            }
+        )
+        if message:
+            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:
+                    return station_id
+        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
+
+    def _release_arm(self):
+        """释放机械臂"""
+        task = self._arm_current_task
+        self._arm_state = ArmState.IDLE
+        self._arm_current_task = None
+        self._arm_lock.release()
+        self._update_data_status(f"机械臂已释放 (完成: {task})")
+        self.logger.info(f"机械臂已释放 (完成: {task})")
+
+    def prepare_materials(
+        self,
+        sample_uuids: SampleUUIDsType,
+        count: int = 5,
+    ) -> PrepareMaterialsResult:
+        """
+        批量准备物料 - 虚拟起始节点
+
+        作为工作流的起始节点，生成指定数量的物料编号供后续节点使用。
+        输出5个handle (material_1 ~ material_5)，分别对应实验1~5。
+
+        Args:
+            count: 待生成的物料数量，默认5 (生成 A1-A5)
+
+        Returns:
+            PrepareMaterialsResult: 包含 material_1 ~ material_5 用于传递给 move_to_heating_station
+        """
+        # 生成物料列表 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}")
+
+        return {
+            "success": True,
+            "count": count,
+            "material_1": materials[0] if len(materials) > 0 else 0,
+            "material_2": materials[1] if len(materials) > 1 else 0,
+            "material_3": materials[2] if len(materials) > 2 else 0,
+            "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()]
+        }
+
+    def move_to_heating_station(
+        self,
+        sample_uuids: SampleUUIDsType,
+        material_number: int,
+    ) -> MoveToHeatingStationResult:
+        """
+        将物料从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",
+                "start_time": time.time(),
+            }
+
+        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._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
+            self.logger.info(f"[{material_id}] 机械臂正在移动到加热台{station_id}...")
+
+            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._release_arm()
+
+            with self._tasks_lock:
+                self._active_tasks[material_id]["status"] = "placed_on_station"
+
+            return {
+                "success": True,
+                "station_id": station_id,
+                "material_id": material_id,
+                "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()]
+            }
+
+        except Exception as e:
+            self.logger.error(f"[{material_id}] 移动失败: {str(e)}")
+            if self._arm_lock.locked():
+                self._release_arm()
+            return {
+                "success": False,
+                "station_id": -1,
+                "material_id": material_id,
+                "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()]
+            }
+
+    def start_heating(
+        self,
+        sample_uuids: SampleUUIDsType,
+        station_id: int,
+        material_number: int,
+    ) -> 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}] 开始加热")
+
+        if station_id not in self._heating_stations:
+            return {
+                "success": False,
+                "station_id": station_id,
+                "material_id": "",
+                "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()]
+            }
+
+        with self._stations_lock:
+            station = self._heating_stations[station_id]
+
+            if station.current_material is None:
+                return {
+                    "success": False,
+                    "station_id": station_id,
+                    "material_id": "",
+                    "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()]
+                }
+
+            if station.state == HeatingStationState.HEATING:
+                return {
+                    "success": False,
+                    "station_id": station_id,
+                    "material_id": station.current_material,
+                    "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()]
+                }
+
+            material_id = station.current_material
+
+            # 开始加热
+            station.state = HeatingStationState.HEATING
+            station.heating_start_time = time.time()
+            station.heating_progress = 0.0
+
+        with self._tasks_lock:
+            if material_id in self._active_tasks:
+                self._active_tasks[material_id]["status"] = "heating"
+
+        self._update_data_status(f"加热台{station_id}开始加热{material_id}")
+
+        # 模拟加热过程 (10秒)
+        start_time = time.time()
+        while True:
+            elapsed = time.time() - start_time
+            progress = min(100.0, (elapsed / self.HEATING_TIME) * 100)
+
+            with self._stations_lock:
+                self._heating_stations[station_id].heating_progress = progress
+
+            self._update_data_status(f"加热台{station_id}加热中: {progress:.1f}%")
+
+            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
+
+        with self._tasks_lock:
+            if material_id in self._active_tasks:
+                self._active_tasks[material_id]["status"] = "heating_completed"
+
+        self._update_data_status(f"加热台{station_id}加热完成")
+        self.logger.info(f"[加热台{station_id}] {material_id}加热完成 (用时{self.HEATING_TIME}s)")
+
+        return {
+            "success": True,
+            "station_id": station_id,
+            "material_id": material_id,
+            "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()]
+        }
+
+    def move_to_output(
+        self,
+        sample_uuids: SampleUUIDsType,
+        station_id: int,
+        material_number: int,
+    ) -> MoveToOutputResult:
+        """
+        将物料从加热台移动到输出位置Cn
+
+        Args:
+            station_id: 加热台ID (1-3)，从 start_heating 的 handle 传入
+            material_number: 物料编号，从 start_heating 的 handle 传入，用于确定输出位置 Cn
+
+        Returns:
+            MoveToOutputResult: 包含执行结果
+        """
+        output_number = material_number  # 物料编号决定输出位置
+
+        if station_id not in self._heating_stations:
+            return {
+                "success": False,
+                "station_id": station_id,
+                "material_id": "",
+                "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()]
+            }
+
+        with self._stations_lock:
+            station = self._heating_stations[station_id]
+            material_id = station.current_material
+
+            if material_id is None:
+                return {
+                    "success": False,
+                    "station_id": station_id,
+                    "material_id": "",
+                    "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()]
+                }
+
+            if station.state != HeatingStationState.COMPLETED:
+                return {
+                    "success": False,
+                    "station_id": station_id,
+                    "material_id": material_id,
+                    "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()]
+                }
+
+        output_position = f"C{output_number}"
+        task_desc = f"从加热台{station_id}移动{material_id}到{output_position}"
+        self.logger.info(f"[任务] {task_desc}")
+
+        try:
+            with self._tasks_lock:
+                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}...")
+            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
+                self._heating_stations[station_id].material_number = None
+                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)")
+
+            return {
+                "success": True,
+                "station_id": station_id,
+                "material_id": material_id,
+                "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()]
+            }
+
+        except Exception as e:
+            self.logger.error(f"移动到输出位置失败: {str(e)}")
+            if self._arm_lock.locked():
+                self._release_arm()
+            return {
+                "success": False,
+                "station_id": station_id,
+                "material_id": "",
+                "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()]
+            }
+
+    # ============ 状态属性 ============
+
+    @property
+    def status(self) -> str:
+        return self.data.get("status", "Unknown")
+
+    @property
+    def arm_state(self) -> str:
+        return self._arm_state.value
+
+    @property
+    def arm_current_task(self) -> str:
+        return self._arm_current_task or ""
+
+    @property
+    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
+    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
+    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
+    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
+    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
+    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
+    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
+    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
+    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
+    def active_tasks_count(self) -> int:
+        with self._tasks_lock:
+            return len(self._active_tasks)
+
+    @property
+    def message(self) -> str:
+        return self.data.get("message", "")

unilabos/registry/devices/liquid_handler.yaml

@@ -9592,7 +9592,7 @@ liquid_handler.prcxi:
           well_names: null
         handles:
           input:
-          - data_key: plate
+          - data_key: '@this.0@@@plate'
             data_source: handle
             data_type: resource
             handler_key: input_plate
@@ -9627,7 +9627,6 @@ liquid_handler.prcxi:
                     type: string
                   type: array
                 plate:
-                  items:
                   properties:
                     category:
                       type: string
@@ -9700,8 +9699,6 @@ liquid_handler.prcxi:
                   - data
                   title: plate
                   type: object
-                  title: plate
-                  type: array
                 volumes:
                   items:
                     type: number
@@ -9717,17 +9714,207 @@ liquid_handler.prcxi:
               - volumes
               type: object
             result:
+              $defs:
+                ResourceDict:
+                  properties:
+                    class:
+                      description: Resource class name
+                      title: Class
+                      type: string
+                    config:
+                      additionalProperties: true
+                      description: Resource configuration
+                      title: Config
+                      type: object
+                    data:
+                      additionalProperties: true
+                      description: 'Resource data, eg: container liquid data'
+                      title: Data
+                      type: object
+                    description:
+                      default: ''
+                      description: Resource description
+                      title: Description
+                      type: string
+                    extra:
+                      additionalProperties: true
+                      description: 'Extra data, eg: slot index'
+                      title: Extra
+                      type: object
+                    icon:
+                      default: ''
+                      description: Resource icon
+                      title: Icon
+                      type: string
+                    id:
+                      description: Resource ID
+                      title: Id
+                      type: string
+                    model:
+                      additionalProperties: true
+                      description: Resource model
+                      title: Model
+                      type: object
+                    name:
+                      description: Resource name
+                      title: Name
+                      type: string
+                    parent:
+                      anyOf:
+                      - $ref: '#/$defs/ResourceDict'
+                      - type: 'null'
+                      default: null
+                      description: Parent resource object
+                    parent_uuid:
+                      anyOf:
+                      - type: string
+                      - type: 'null'
+                      default: null
+                      description: Parent resource uuid
+                      title: Parent Uuid
+                    pose:
+                      $ref: '#/$defs/ResourceDictPosition'
+                      description: Resource position
+                    schema:
+                      additionalProperties: true
+                      description: Resource schema
+                      title: Schema
+                      type: object
+                    type:
+                      anyOf:
+                      - const: device
+                        type: string
+                      - type: string
+                      description: Resource type
+                      title: Type
+                    uuid:
+                      description: Resource UUID
+                      title: Uuid
+                      type: string
+                  required:
+                  - id
+                  - uuid
+                  - name
+                  - type
+                  - class
+                  - config
+                  - data
+                  - extra
+                  title: ResourceDict
+                  type: object
+                ResourceDictPosition:
+                  properties:
+                    cross_section_type:
+                      default: rectangle
+                      description: Cross section type
+                      enum:
+                      - rectangle
+                      - circle
+                      - rounded_rectangle
+                      title: Cross Section Type
+                      type: string
+                    layout:
+                      default: x-y
+                      description: Resource layout
+                      enum:
+                      - 2d
+                      - x-y
+                      - z-y
+                      - x-z
+                      title: Layout
+                      type: string
+                    position:
+                      $ref: '#/$defs/ResourceDictPositionObject'
+                      description: Resource position
+                    position3d:
+                      $ref: '#/$defs/ResourceDictPositionObject'
+                      description: Resource position in 3D space
+                    rotation:
+                      $ref: '#/$defs/ResourceDictPositionObject'
+                      description: Resource rotation
+                    scale:
+                      $ref: '#/$defs/ResourceDictPositionScale'
+                      description: Resource scale
+                    size:
+                      $ref: '#/$defs/ResourceDictPositionSize'
+                      description: Resource size
+                  title: ResourceDictPosition
+                  type: object
+                ResourceDictPositionObject:
+                  properties:
+                    x:
+                      default: 0.0
+                      description: X coordinate
+                      title: X
+                      type: number
+                    y:
+                      default: 0.0
+                      description: Y coordinate
+                      title: Y
+                      type: number
+                    z:
+                      default: 0.0
+                      description: Z coordinate
+                      title: Z
+                      type: number
+                  title: ResourceDictPositionObject
+                  type: object
+                ResourceDictPositionScale:
+                  properties:
+                    x:
+                      default: 0.0
+                      description: x scale
+                      title: X
+                      type: number
+                    y:
+                      default: 0.0
+                      description: y scale
+                      title: Y
+                      type: number
+                    z:
+                      default: 0.0
+                      description: z scale
+                      title: Z
+                      type: number
+                  title: ResourceDictPositionScale
+                  type: object
+                ResourceDictPositionSize:
+                  properties:
+                    depth:
+                      default: 0.0
+                      description: Depth
+                      title: Depth
+                      type: number
+                    height:
+                      default: 0.0
+                      description: Height
+                      title: Height
+                      type: number
+                    width:
+                      default: 0.0
+                      description: Width
+                      title: Width
+                      type: number
+                  title: ResourceDictPositionSize
+                  type: object
               properties:
                 plate:
-                  items: {}
+                  items:
+                    items:
+                      $ref: '#/$defs/ResourceDict'
+                    type: array
                   title: Plate
                   type: array
                 volumes:
-                  items: {}
+                  items:
+                    type: number
                   title: Volumes
                   type: array
                 wells:
-                  items: {}
+                  items:
+                    items:
+                      $ref: '#/$defs/ResourceDict'
+                    type: array
                   title: Wells
                   type: array
               required:

unilabos/registry/devices/virtual_device.yaml

@@ -5835,6 +5835,25 @@ virtual_workbench:
               - material_number
               type: object
             result:
+              $defs:
+                LabSample:
+                  properties:
+                    extra:
+                      additionalProperties: true
+                      title: Extra
+                      type: object
+                    oss_path:
+                      title: Oss Path
+                      type: string
+                    sample_uuid:
+                      title: Sample Uuid
+                      type: string
+                  required:
+                  - sample_uuid
+                  - oss_path
+                  - extra
+                  title: LabSample
+                  type: object
               description: move_to_heating_station 返回类型
               properties:
                 material_id:
@@ -5853,12 +5872,18 @@ virtual_workbench:
                 success:
                   title: Success
                   type: boolean
+                unilabos_samples:
+                  items:
+                    $ref: '#/$defs/LabSample'
+                  title: Unilabos Samples
+                  type: array
               required:
               - success
               - station_id
               - material_id
               - material_number
               - message
+              - unilabos_samples
               title: MoveToHeatingStationResult
               type: object
           required:
@@ -5903,6 +5928,25 @@ virtual_workbench:
               - material_number
               type: object
             result:
+              $defs:
+                LabSample:
+                  properties:
+                    extra:
+                      additionalProperties: true
+                      title: Extra
+                      type: object
+                    oss_path:
+                      title: Oss Path
+                      type: string
+                    sample_uuid:
+                      title: Sample Uuid
+                      type: string
+                  required:
+                  - sample_uuid
+                  - oss_path
+                  - extra
+                  title: LabSample
+                  type: object
               description: move_to_output 返回类型
               properties:
                 material_id:
@@ -5914,10 +5958,16 @@ virtual_workbench:
                 success:
                   title: Success
                   type: boolean
+                unilabos_samples:
+                  items:
+                    $ref: '#/$defs/LabSample'
+                  title: Unilabos Samples
+                  type: array
               required:
               - success
               - station_id
               - material_id
+              - unilabos_samples
               title: MoveToOutputResult
               type: object
           required:
@@ -5972,6 +6022,25 @@ virtual_workbench:
               required: []
               type: object
             result:
+              $defs:
+                LabSample:
+                  properties:
+                    extra:
+                      additionalProperties: true
+                      title: Extra
+                      type: object
+                    oss_path:
+                      title: Oss Path
+                      type: string
+                    sample_uuid:
+                      title: Sample Uuid
+                      type: string
+                  required:
+                  - sample_uuid
+                  - oss_path
+                  - extra
+                  title: LabSample
+                  type: object
               description: prepare_materials 返回类型 - 批量准备物料
               properties:
                 count:
@@ -5998,6 +6067,11 @@ virtual_workbench:
                 success:
                   title: Success
                   type: boolean
+                unilabos_samples:
+                  items:
+                    $ref: '#/$defs/LabSample'
+                  title: Unilabos Samples
+                  type: array
               required:
               - success
               - count
@@ -6007,6 +6081,7 @@ virtual_workbench:
               - material_4
               - material_5
               - message
+              - unilabos_samples
               title: PrepareMaterialsResult
               type: object
           required:
@@ -6062,6 +6137,25 @@ virtual_workbench:
               - material_number
               type: object
             result:
+              $defs:
+                LabSample:
+                  properties:
+                    extra:
+                      additionalProperties: true
+                      title: Extra
+                      type: object
+                    oss_path:
+                      title: Oss Path
+                      type: string
+                    sample_uuid:
+                      title: Sample Uuid
+                      type: string
+                  required:
+                  - sample_uuid
+                  - oss_path
+                  - extra
+                  title: LabSample
+                  type: object
               description: start_heating 返回类型
               properties:
                 material_id:
@@ -6079,12 +6173,18 @@ virtual_workbench:
                 success:
                   title: Success
                   type: boolean
+                unilabos_samples:
+                  items:
+                    $ref: '#/$defs/LabSample'
+                  title: Unilabos Samples
+                  type: array
               required:
               - success
               - station_id
               - material_id
               - material_number
               - message
+              - unilabos_samples
               title: StartHeatingResult
               type: object
           required:

unilabos/resources/resource_tracker.py

@@ -5,6 +5,8 @@ from pydantic import BaseModel, field_serializer, field_validator, ValidationErr
 from pydantic import Field
 from typing import List, Tuple, Any, Dict, Literal, Optional, cast, TYPE_CHECKING, Union
 
+from typing_extensions import TypedDict
+
 from unilabos.resources.plr_additional_res_reg import register
 from unilabos.utils.log import logger
 
@@ -14,6 +16,26 @@ if TYPE_CHECKING:
 
 
 EXTRA_CLASS = "unilabos_resource_class"
+EXTRA_SAMPLE_UUID = "sample_uuid"
+EXTRA_UNILABOS_SAMPLE_UUID = "unilabos_sample_uuid"
+
+# 函数参数名常量 - 用于自动注入 sample_uuids 列表
+PARAM_SAMPLE_UUIDS = "sample_uuids"
+
+# JSON Command 中的系统参数字段名
+JSON_UNILABOS_PARAM = "unilabos_param"
+
+# 返回值中的 samples 字段名
+RETURN_UNILABOS_SAMPLES = "unilabos_samples"
+
+# sample_uuids 参数类型 (用于 virtual bench 等设备添加 sample_uuids 参数)
+SampleUUIDsType = Dict[str, Optional["PLRResource"]]
+
+
+class LabSample(TypedDict):
+    sample_uuid: str
+    oss_path: str
+    extra: Dict[str, Any]
 
 
 class ResourceDictPositionSize(BaseModel):
@@ -529,6 +551,7 @@ class ResourceTreeSet(object):
                 plr_resource = sub_cls.deserialize(plr_dict, allow_marshal=True)
                 from pylabrobot.resources import Coordinate
                 from pylabrobot.serializer import deserialize
+
                 location = cast(Coordinate, deserialize(plr_dict["location"]))
                 plr_resource.location = location
                 plr_resource.load_all_state(all_states)

unilabos/ros/nodes/base_device_node.py

@@ -1,12 +1,23 @@
-from ast import Try
 import inspect
 import io
 import json
 import threading
 import time
 import traceback
-from typing import get_type_hints, TypeVar, Generic, Dict, Any, Type, TypedDict, Optional, List, TYPE_CHECKING, Union, \
+from typing import (
-    Tuple
+    get_type_hints,
+    TypeVar,
+    Generic,
+    Dict,
+    Any,
+    Type,
+    TypedDict,
+    Optional,
+    List,
+    TYPE_CHECKING,
+    Union,
+    Tuple,
+)
 
 from concurrent.futures import ThreadPoolExecutor
 import asyncio
@@ -49,8 +60,10 @@ from unilabos.resources.resource_tracker import (
     ResourceTreeSet,
     ResourceTreeInstance,
     ResourceDictInstance,
+    EXTRA_SAMPLE_UUID,
+    PARAM_SAMPLE_UUIDS,
+    JSON_UNILABOS_PARAM,
 )
-from unilabos.ros.x.rclpyx import get_event_loop
 from unilabos.ros.utils.driver_creator import WorkstationNodeCreator, PyLabRobotCreator, DeviceClassCreator
 from rclpy.task import Task, Future
 from unilabos.utils.import_manager import default_manager
@@ -186,7 +199,7 @@ class PropertyPublisher:
                 f"创建发布者 {name} 失败，可能由于注册表有误，类型: {msg_type}，错误: {ex}\n{traceback.format_exc()}"
             )
         self.timer = node.create_timer(self.timer_period, self.publish_property)
-        self.__loop = get_event_loop()
+        self.__loop = ROS2DeviceNode.get_asyncio_loop()
         str_msg_type = str(msg_type)[8:-2]
         self.node.lab_logger().trace(f"发布属性: {name}, 类型: {str_msg_type}, 周期: {initial_period}秒, QoS: {qos}")
 
@@ -218,14 +231,15 @@ class PropertyPublisher:
 
     def publish_property(self):
         try:
-            self.node.lab_logger().trace(f"【.publish_property】开始发布属性: {self.name}")
+            # self.node.lab_logger().trace(f"【.publish_property】开始发布属性: {self.name}")
             value = self.get_property()
             if self.print_publish:
-                self.node.lab_logger().trace(f"【.publish_property】发布 {self.msg_type}: {value}")
+                pass
+                # self.node.lab_logger().trace(f"【.publish_property】发布 {self.msg_type}: {value}")
             if value is not None:
                 msg = convert_to_ros_msg(self.msg_type, value)
                 self.publisher_.publish(msg)
-                self.node.lab_logger().trace(f"【.publish_property】属性 {self.name} 发布成功")
+                # self.node.lab_logger().trace(f"【.publish_property】属性 {self.name} 发布成功")
         except Exception as e:
             self.node.lab_logger().error(
                 f"【.publish_property】发布属性 {self.publisher_.topic} 出错: {str(e)}\n{traceback.format_exc()}"
@@ -363,6 +377,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
             from pylabrobot.resources.deck import Deck
             from pylabrobot.resources import Coordinate
             from pylabrobot.resources import Plate
+
             # 物料传输到对应的node节点
             client = self._resource_clients["c2s_update_resource_tree"]
             request = SerialCommand.Request()
@@ -390,33 +405,27 @@ class BaseROS2DeviceNode(Node, Generic[T]):
             rts: ResourceTreeSet = ResourceTreeSet.from_raw_dict_list(input_resources)
             parent_resource = None
             if bind_parent_id != self.node_name:
-                parent_resource = self.resource_tracker.figure_resource(
+                parent_resource = self.resource_tracker.figure_resource({"name": bind_parent_id})
-                    {"name": bind_parent_id}
-                )
                 for r in rts.root_nodes:
                     # noinspection PyUnresolvedReferences
                     r.res_content.parent_uuid = parent_resource.unilabos_uuid
             else:
                 for r in rts.root_nodes:
                     r.res_content.parent_uuid = self.uuid
-
+            rts_plr_instances = rts.to_plr_resources()
-            if len(LIQUID_INPUT_SLOT) and LIQUID_INPUT_SLOT[0] == -1 and len(rts.root_nodes) == 1 and isinstance(rts.root_nodes[0], RegularContainer):
+            if len(rts.root_nodes) == 1 and isinstance(rts_plr_instances[0], RegularContainer):
                 # noinspection PyTypeChecker
-                container_instance: RegularContainer = rts.root_nodes[0]
+                container_instance: RegularContainer = rts_plr_instances[0]
-                found_resources = self.resource_tracker.figure_resource(
+                found_resources = self.resource_tracker.figure_resource({"name": container_instance.name}, try_mode=True)
-                    {"id": container_instance.name}, try_mode=True
-                )
                 if not len(found_resources):
                     self.resource_tracker.add_resource(container_instance)
                     logger.info(f"添加物料{container_instance.name}到资源跟踪器")
                 else:
-                    assert (
+                    assert len(found_resources) == 1, f"找到多个同名物料: {container_instance.name}, 请检查物料系统"
-                        len(found_resources) == 1
-                    ), f"找到多个同名物料: {container_instance.name}, 请检查物料系统"
                     found_resource = found_resources[0]
                     if isinstance(found_resource, RegularContainer):
                         logger.info(f"更新物料{container_instance.name}的数据{found_resource.state}")
-                        found_resource.state.update(json.loads(container_instance.state))
+                        found_resource.state.update(container_instance.state)
                     elif isinstance(found_resource, dict):
                         raise ValueError("已不支持 字典 版本的RegularContainer")
                     else:
@@ -424,14 +433,16 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                             f"更新物料{container_instance.name}出现不支持的数据类型{type(found_resource)} {found_resource}"
                         )
             # noinspection PyUnresolvedReferences
-            request.command = json.dumps({
+            request.command = json.dumps(
+                {
                     "action": "add",
                     "data": {
                         "data": rts.dump(),
-                    "mount_uuid": parent_resource.unilabos_uuid if parent_resource is not None else "",
+                        "mount_uuid": parent_resource.unilabos_uuid if parent_resource is not None else self.uuid,
                         "first_add": False,
                     },
-            })
+                }
+            )
             tree_response: SerialCommand.Response = await client.call_async(request)
             uuid_maps = json.loads(tree_response.response)
             plr_instances = rts.to_plr_resources()
@@ -473,7 +484,9 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                         if len(ADD_LIQUID_TYPE) == 1 and len(LIQUID_VOLUME) == 1 and len(LIQUID_INPUT_SLOT) > 1:
                             ADD_LIQUID_TYPE = ADD_LIQUID_TYPE * len(LIQUID_INPUT_SLOT)
                             LIQUID_VOLUME = LIQUID_VOLUME * len(LIQUID_INPUT_SLOT)
-                            self.lab_logger().warning(f"增加液体资源时，数量为1，自动补全为 {len(LIQUID_INPUT_SLOT)} 个")
+                            self.lab_logger().warning(
+                                f"增加液体资源时，数量为1，自动补全为 {len(LIQUID_INPUT_SLOT)} 个"
+                            )
                         for liquid_type, liquid_volume, liquid_input_slot in zip(
                             ADD_LIQUID_TYPE, LIQUID_VOLUME, LIQUID_INPUT_SLOT
                         ):
@@ -492,9 +505,15 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                             input_wells = []
                             for r in LIQUID_INPUT_SLOT:
                                 input_wells.append(plr_instance.children[r])
-                        final_response["liquid_input_resource_tree"] = ResourceTreeSet.from_plr_resources(input_wells).dump()
+                        final_response["liquid_input_resource_tree"] = ResourceTreeSet.from_plr_resources(
+                            input_wells
+                        ).dump()
                         res.response = json.dumps(final_response)
-                    if issubclass(parent_resource.__class__, Deck) and hasattr(parent_resource, "assign_child_at_slot") and "slot" in other_calling_param:
+                    if (
+                        issubclass(parent_resource.__class__, Deck)
+                        and hasattr(parent_resource, "assign_child_at_slot")
+                        and "slot" in other_calling_param
+                    ):
                         other_calling_param["slot"] = int(other_calling_param["slot"])
                         parent_resource.assign_child_at_slot(plr_instance, **other_calling_param)
                     else:
@@ -509,14 +528,16 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                     rts_with_parent = ResourceTreeSet.from_plr_resources([parent_resource])
                     if rts_with_parent.root_nodes[0].res_content.uuid_parent is None:
                         rts_with_parent.root_nodes[0].res_content.parent_uuid = self.uuid
-                    request.command = json.dumps({
+                    request.command = json.dumps(
+                        {
                             "action": "add",
                             "data": {
                                 "data": rts_with_parent.dump(),
                                 "mount_uuid": rts_with_parent.root_nodes[0].res_content.uuid_parent,
                                 "first_add": False,
                             },
-                    })
+                        }
+                    )
                     tree_response: SerialCommand.Response = await client.call_async(request)
                     uuid_maps = json.loads(tree_response.response)
                     self.resource_tracker.loop_update_uuid(input_resources, uuid_maps)
@@ -626,7 +647,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         )  # type: ignore
         raw_nodes = json.loads(response.response)
         tree_set = ResourceTreeSet.from_raw_dict_list(raw_nodes)
-        self.lab_logger().debug(f"获取资源结果: {len(tree_set.trees)} 个资源树")
+        self.lab_logger().trace(f"获取资源结果: {len(tree_set.trees)} 个资源树 {tree_set.root_nodes}")
         return tree_set
 
     async def get_resource_with_dir(self, resource_id: str, with_children: bool = True) -> "ResourcePLR":
@@ -813,7 +834,9 @@ class BaseROS2DeviceNode(Node, Generic[T]):
             }
 
         def _handle_update(
-            plr_resources: List[Union[ResourcePLR, ResourceDictInstance]], tree_set: ResourceTreeSet, additional_add_params: Dict[str, Any]
+            plr_resources: List[Union[ResourcePLR, ResourceDictInstance]],
+            tree_set: ResourceTreeSet,
+            additional_add_params: Dict[str, Any],
         ) -> Tuple[Dict[str, Any], List[ResourcePLR]]:
             """
             处理资源更新操作的内部函数
@@ -838,7 +861,10 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                 original_parent_resource = original_instance.parent
                 original_parent_resource_uuid = getattr(original_parent_resource, "unilabos_uuid", None)
                 target_parent_resource_uuid = tree.root_node.res_content.uuid_parent
-                not_same_parent = original_parent_resource_uuid != target_parent_resource_uuid and original_parent_resource is not None
+                not_same_parent = (
+                    original_parent_resource_uuid != target_parent_resource_uuid
+                    and original_parent_resource is not None
+                )
                 old_name = original_instance.name
                 new_name = plr_resource.name
                 parent_appended = False
@@ -874,8 +900,16 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                 else:
                     # 判断是否变更了resource_site，重新登记
                     target_site = original_instance.unilabos_extra.get("update_resource_site")
-                    sites = original_instance.parent.sites if original_instance.parent is not None and hasattr(original_instance.parent, "sites") else None
+                    sites = (
-                    site_names = list(original_instance.parent._ordering.keys()) if original_instance.parent is not None and hasattr(original_instance.parent, "sites") else []
+                        original_instance.parent.sites
+                        if original_instance.parent is not None and hasattr(original_instance.parent, "sites")
+                        else None
+                    )
+                    site_names = (
+                        list(original_instance.parent._ordering.keys())
+                        if original_instance.parent is not None and hasattr(original_instance.parent, "sites")
+                        else []
+                    )
                     if target_site is not None and sites is not None and site_names is not None:
                         site_index = sites.index(original_instance)
                         site_name = site_names[site_index]
@@ -912,9 +946,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                 action = i.get("action")  # remove, add, update
                 resources_uuid: List[str] = i.get("data")  # 资源数据
                 additional_add_params = i.get("additional_add_params", {})  # 额外参数
-                self.lab_logger().trace(
+                self.lab_logger().trace(f"[资源同步] 处理 {action}, " f"resources count: {len(resources_uuid)}")
-                    f"[资源同步] 处理 {action}, " f"resources count: {len(resources_uuid)}"
-                )
                 tree_set = None
                 if action in ["add", "update"]:
                     tree_set = await self.get_resource(
@@ -941,9 +973,13 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                                 tree.root_node.res_content.parent_uuid = self.uuid
                         r = SerialCommand.Request()
                         r.command = json.dumps(
-                            {"data": {"data": new_tree_set.dump()}, "action": "update"})  # 和Update Resource一致
+                            {"data": {"data": new_tree_set.dump()}, "action": "update"}
+                        )  # 和Update Resource一致
                         response: SerialCommand_Response = await self._resource_clients[
-                            "c2s_update_resource_tree"].call_async(r)  # type: ignore
+                            "c2s_update_resource_tree"
+                        ].call_async(
+                            r
+                        )  # type: ignore
                         self.lab_logger().info(f"确认资源云端 Add 结果: {response.response}")
                         results.append(result)
                     elif action == "update":
@@ -963,9 +999,13 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                                     tree.root_node.res_content.parent_uuid = self.uuid
                             r = SerialCommand.Request()
                             r.command = json.dumps(
-                                {"data": {"data": new_tree_set.dump()}, "action": "update"})  # 和Update Resource一致
+                                {"data": {"data": new_tree_set.dump()}, "action": "update"}
+                            )  # 和Update Resource一致
                             response: SerialCommand_Response = await self._resource_clients[
-                                "c2s_update_resource_tree"].call_async(r)  # type: ignore
+                                "c2s_update_resource_tree"
+                            ].call_async(
+                                r
+                            )  # type: ignore
                             self.lab_logger().info(f"确认资源云端 Update 结果: {response.response}")
                         results.append(result)
                     elif action == "remove":
@@ -1335,7 +1375,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                                         resource_id=resource_data["id"], with_children=True
                                     )
                                     if "sample_id" in resource_data:
-                                        plr_resource.unilabos_extra["sample_uuid"] = resource_data["sample_id"]
+                                        plr_resource.unilabos_extra[EXTRA_SAMPLE_UUID] = resource_data["sample_id"]
                                     queried_resources[idx] = plr_resource
                                 else:
                                     uuid_indices.append((idx, unilabos_uuid, resource_data))
@@ -1348,31 +1388,17 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                                 for i, (idx, _, resource_data) in enumerate(uuid_indices):
                                     plr_resource = plr_resources[i]
                                     if "sample_id" in resource_data:
-                                        plr_resource.unilabos_extra["sample_uuid"] = resource_data["sample_id"]
+                                        plr_resource.unilabos_extra[EXTRA_SAMPLE_UUID] = resource_data["sample_id"]
                                     queried_resources[idx] = plr_resource
 
-                            # 第二遍：批量查询有uuid的资源
-                            if uuid_indices:
-                                uuids = [item[1] for item in uuid_indices]
-                                resource_tree = await self.get_resource(uuids)
-                                plr_resources = resource_tree.to_plr_resources()
-                                # 通过uuid查找对应的plr_resource
-                                tracker = self.resource_tracker
-                                for idx, uuid, resource_data in uuid_indices:
-                                    try:
-                                        plr_resource = tracker.loop_find_with_uuid(plr_resources, uuid)
-                                        if "sample_id" in resource_data:
-                                            plr_resource.unilabos_extra["sample_uuid"] = resource_data["sample_id"]
-                                        queried_resources[idx] = plr_resource
-                                    except Exception as e:
-                                        self.lab_logger().error(f"资源查询失败: {e}\n{traceback.format_exc()}")
-                                        continue
                             self.lab_logger().debug(f"资源查询结果: 共 {len(queried_resources)} 个资源")
 
                             # 通过资源跟踪器获取本地实例
                             final_resources = queried_resources if is_sequence else queried_resources[0]
                             if not is_sequence:
-                                plr = self.resource_tracker.figure_resource({"name": final_resources.name}, try_mode=False)
+                                plr = self.resource_tracker.figure_resource(
+                                    {"name": final_resources.name}, try_mode=False
+                                )
                                 # 保留unilabos_extra
                                 if hasattr(final_resources, "unilabos_extra") and hasattr(plr, "unilabos_extra"):
                                     plr.unilabos_extra = getattr(final_resources, "unilabos_extra", {}).copy()
@@ -1411,8 +1437,12 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                                 execution_success = True
                             except Exception as _:
                                 execution_error = traceback.format_exc()
-                                error(f"异步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{str(action_kwargs)[:1000]}")
+                                error(
-                                trace(f"异步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{action_kwargs}")
+                                    f"异步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{str(action_kwargs)[:1000]}"
+                                )
+                                trace(
+                                    f"异步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{action_kwargs}"
+                                )
 
                         future = ROS2DeviceNode.run_async_func(ACTION, trace_error=False, **action_kwargs)
                         future.add_done_callback(_handle_future_exception)
@@ -1432,9 +1462,11 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                         except Exception as _:
                             execution_error = traceback.format_exc()
                             error(
-                                f"同步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{str(action_kwargs)[:1000]}")
+                                f"同步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{str(action_kwargs)[:1000]}"
+                            )
                             trace(
-                                f"同步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{action_kwargs}")
+                                f"同步任务 {ACTION.__name__} 报错了\n{traceback.format_exc()}\n原始输入：{action_kwargs}"
+                            )
 
                     future.add_done_callback(_handle_future_exception)
 
@@ -1501,10 +1533,15 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                             if isinstance(rs, list):
                                 for r in rs:
                                     res = self.resource_tracker.parent_resource(r)  # 获取 resource 对象
-                            elif type(rs).__name__ == "ResourceHolder":
+                                    if res is None:
-                                pass
+                                        res = rs
+                                    if id(res) not in seen:
+                                        seen.add(id(res))
+                                        unique_resources.append(res)
                             else:
                                 res = self.resource_tracker.parent_resource(rs)
+                                if res is None:
+                                    res = rs
                                 if id(res) not in seen:
                                     seen.add(id(res))
                                     unique_resources.append(res)
@@ -1559,20 +1596,39 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         try:
             function_name = target["function_name"]
             function_args = target["function_args"]
+            # 获取 unilabos 系统参数
+            unilabos_param: Dict[str, Any] = target[JSON_UNILABOS_PARAM]
+
             assert isinstance(function_args, dict), "执行动作时JSON必须为dict类型\n原JSON: {string}"
             function = getattr(self.driver_instance, function_name)
             assert callable(
                 function
             ), f"执行动作时JSON中的function_name对应的函数不可调用: {function_name}\n原JSON: {string}"
 
-            # 处理 ResourceSlot 类型参数
+            # 处理参数（包含 unilabos 系统参数如 sample_uuids）
-            args_list = default_manager._analyze_method_signature(function)["args"]
+            args_list = default_manager._analyze_method_signature(function, skip_unilabos_params=False)["args"]
             for arg in args_list:
                 arg_name = arg["name"]
                 arg_type = arg["type"]
 
                 # 跳过不在 function_args 中的参数
                 if arg_name not in function_args:
+                    # 处理 sample_uuids 参数注入
+                    if arg_name == PARAM_SAMPLE_UUIDS:
+                        raw_sample_uuids = unilabos_param.get(PARAM_SAMPLE_UUIDS, {})
+                        # 将 material uuid 转换为 resource 实例
+                        # key: sample_uuid, value: material_uuid -> resource 实例
+                        resolved_sample_uuids: Dict[str, Any] = {}
+                        for sample_uuid, material_uuid in raw_sample_uuids.items():
+                            if material_uuid and self.resource_tracker:
+                                resource = self.resource_tracker.uuid_to_resources.get(material_uuid)
+                                resolved_sample_uuids[sample_uuid] = resource if resource else material_uuid
+                            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}"
+                        )
                     continue
 
                 # 处理单个 ResourceSlot
@@ -1601,6 +1657,7 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                                     f"转换ResourceSlot列表参数 {arg_name} 失败: {e}\n{traceback.format_exc()}"
                                 )
                                 raise JsonCommandInitError(f"ResourceSlot列表参数转换失败: {arg_name}")
+
             # todo: 默认反报送
             return function(**function_args)
         except KeyError as ex:
@@ -1621,14 +1678,16 @@ class BaseROS2DeviceNode(Node, Generic[T]):
             raise ValueError("至少需要提供一个 UUID")
 
         uuids_list = list(uuids)
-        future = self._resource_clients["c2s_update_resource_tree"].call_async(SerialCommand.Request(
+        future = self._resource_clients["c2s_update_resource_tree"].call_async(
+            SerialCommand.Request(
                 command=json.dumps(
                     {
                         "data": {"data": uuids_list, "with_children": True},
                         "action": "get",
                     }
                 )
-        ))
+            )
+        )
 
         # 等待结果（使用while循环，每次sleep 0.05秒，最多等待30秒）
         timeout = 30.0
@@ -1686,6 +1745,9 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         try:
             function_name = target["function_name"]
             function_args = target["function_args"]
+            # 获取 unilabos 系统参数
+            unilabos_param: Dict[str, Any] = target.get(JSON_UNILABOS_PARAM, {})
+
             assert isinstance(function_args, dict), "执行动作时JSON必须为dict类型\n原JSON: {string}"
             function = getattr(self.driver_instance, function_name)
             assert callable(
@@ -1695,14 +1757,30 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                 function
             ), f"执行动作时JSON中的function并非异步: {function_name}\n原JSON: {string}"
 
-            # 处理 ResourceSlot 类型参数
+            # 处理参数（包含 unilabos 系统参数如 sample_uuids）
-            args_list = default_manager._analyze_method_signature(function)["args"]
+            args_list = default_manager._analyze_method_signature(function, skip_unilabos_params=False)["args"]
             for arg in args_list:
                 arg_name = arg["name"]
                 arg_type = arg["type"]
 
                 # 跳过不在 function_args 中的参数
                 if arg_name not in function_args:
+                    # 处理 sample_uuids 参数注入
+                    if arg_name == PARAM_SAMPLE_UUIDS:
+                        raw_sample_uuids = unilabos_param.get(PARAM_SAMPLE_UUIDS, {})
+                        # 将 material uuid 转换为 resource 实例
+                        # key: sample_uuid, value: material_uuid -> resource 实例
+                        resolved_sample_uuids: Dict[str, Any] = {}
+                        for sample_uuid, material_uuid in raw_sample_uuids.items():
+                            if material_uuid and self.resource_tracker:
+                                resource = self.resource_tracker.uuid_to_resources.get(material_uuid)
+                                resolved_sample_uuids[sample_uuid] = resource if resource else material_uuid
+                            else:
+                                resolved_sample_uuids[sample_uuid] = material_uuid
+                        function_args[PARAM_SAMPLE_UUIDS] = resolved_sample_uuids
+                        self.lab_logger().debug(
+                            f"[JsonCommandAsync] 注入 {PARAM_SAMPLE_UUIDS}: {resolved_sample_uuids}"
+                        )
                     continue
 
                 # 处理单个 ResourceSlot
@@ -1792,6 +1870,15 @@ class ROS2DeviceNode:
     它不继承设备类，而是通过代理模式访问设备类的属性和方法。
     """
 
+    # 类变量，用于循环管理
+    _asyncio_loop = None
+    _asyncio_loop_running = False
+    _asyncio_loop_thread = None
+
+    @classmethod
+    def get_asyncio_loop(cls):
+        return cls._asyncio_loop
+
     @staticmethod
     async def safe_task_wrapper(trace_callback, func, **kwargs):
         try:
@@ -1868,6 +1955,11 @@ class ROS2DeviceNode:
             print_publish: 是否打印发布信息
             driver_is_ros:
         """
+        # 在初始化时检查循环状态
+        if ROS2DeviceNode._asyncio_loop_running and ROS2DeviceNode._asyncio_loop_thread is not None:
+            pass
+        elif ROS2DeviceNode._asyncio_loop_thread is None:
+            self._start_loop()
 
         # 保存设备类是否支持异步上下文
         self._has_async_context = hasattr(driver_class, "__aenter__") and hasattr(driver_class, "__aexit__")
@@ -1959,6 +2051,19 @@ class ROS2DeviceNode:
             except Exception as e:
                 self._ros_node.lab_logger().error(f"设备后初始化失败: {e}")
 
+    def _start_loop(self):
+        def run_event_loop():
+            loop = asyncio.new_event_loop()
+            ROS2DeviceNode._asyncio_loop = loop
+            asyncio.set_event_loop(loop)
+            loop.run_forever()
+
+        ROS2DeviceNode._asyncio_loop_thread = threading.Thread(
+            target=run_event_loop, daemon=True, name="ROS2DeviceNode"
+        )
+        ROS2DeviceNode._asyncio_loop_thread.start()
+        logger.info(f"循环线程已启动")
+
 
 class DeviceInfoType(TypedDict):
     id: str

unilabos/ros/nodes/presets/host_node.py

@@ -1,16 +1,17 @@
 import collections
-from dataclasses import dataclass, field
 import json
 import threading
 import time
 import traceback
 import uuid
-from typing import TYPE_CHECKING, Optional, Dict, Any, List, ClassVar, Set, TypedDict, Union
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Optional, Dict, Any, List, ClassVar, Set, Union
 
 from action_msgs.msg import GoalStatus
 from geometry_msgs.msg import Point
 from rclpy.action import ActionClient, get_action_server_names_and_types_by_node
 from rclpy.service import Service
+from typing_extensions import TypedDict
 from unilabos_msgs.msg import Resource  # type: ignore
 from unilabos_msgs.srv import (
     ResourceAdd,
@@ -22,10 +23,20 @@ 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.placeholder_type import ResourceSlot, DeviceSlot
 from unilabos.registry.registry import lab_registry
 from unilabos.resources.container import RegularContainer
 from unilabos.resources.graphio import initialize_resource
 from unilabos.resources.registry import add_schema
+from unilabos.resources.resource_tracker import (
+    ResourceDict,
+    ResourceDictInstance,
+    ResourceTreeSet,
+    ResourceTreeInstance,
+    RETURN_UNILABOS_SAMPLES,
+    JSON_UNILABOS_PARAM,
+    PARAM_SAMPLE_UUIDS,
+)
 from unilabos.ros.initialize_device import initialize_device_from_dict
 from unilabos.ros.msgs.message_converter import (
     get_msg_type,
@@ -36,17 +47,10 @@ from unilabos.ros.msgs.message_converter import (
 )
 from unilabos.ros.nodes.base_device_node import BaseROS2DeviceNode, ROS2DeviceNode, DeviceNodeResourceTracker
 from unilabos.ros.nodes.presets.controller_node import ControllerNode
-from unilabos.resources.resource_tracker import (
-    ResourceDict,
-    ResourceDictInstance,
-    ResourceTreeSet,
-    ResourceTreeInstance,
-)
 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.registry.placeholder_type import ResourceSlot, DeviceSlot
 
 if TYPE_CHECKING:
     from unilabos.app.ws_client import QueueItem
@@ -62,6 +66,18 @@ class TestResourceReturn(TypedDict):
     devices: List[DeviceSlot]
 
 
+class TestLatencyReturn(TypedDict):
+    """test_latency方法的返回值类型"""
+
+    avg_rtt_ms: float
+    avg_time_diff_ms: float
+    max_time_error_ms: float
+    task_delay_ms: float
+    raw_delay_ms: float
+    test_count: int
+    status: str
+
+
 class HostNode(BaseROS2DeviceNode):
     """
     主机节点类，负责管理设备、资源和控制器
@@ -735,13 +751,14 @@ class HostNode(BaseROS2DeviceNode):
                         if bCreate:
                             self.lab_logger().trace(f"Status created: {device_id}.{property_name} = {msg.data}")
                         else:
-                            self.lab_logger().debug(f"Status updated: {device_id}.{property_name} = {msg.data}")
+                            self.lab_logger().trace(f"Status updated: {device_id}.{property_name} = {msg.data}")
 
     def send_goal(
         self,
         item: "QueueItem",
         action_type: str,
         action_kwargs: Dict[str, Any],
+        sample_material: Dict[str, str],
         server_info: Optional[Dict[str, Any]] = None,
     ) -> None:
         """
@@ -759,14 +776,14 @@ class HostNode(BaseROS2DeviceNode):
             if action_name.startswith("auto-"):
                 action_name = action_name[5:]
             action_id = f"/devices/{device_id}/_execute_driver_command"
-            action_kwargs = {
+            json_command: Dict[str, Any] = {
-                "string": json.dumps(
-                    {
                 "function_name": action_name,
                 "function_args": action_kwargs,
+                JSON_UNILABOS_PARAM: {
+                    PARAM_SAMPLE_UUIDS: sample_material,
+                },
             }
-                )
+            action_kwargs = {"string": json.dumps(json_command)}
-            }
             if action_type.startswith("UniLabJsonCommandAsync"):
                 action_id = f"/devices/{device_id}/_execute_driver_command_async"
         else:
@@ -777,21 +794,6 @@ class HostNode(BaseROS2DeviceNode):
             raise ValueError(f"ActionClient {action_id} not found.")
 
         action_client: ActionClient = self._action_clients[action_id]
-
-        # 遍历action_kwargs下的所有子dict，将"sample_uuid"的值赋给"sample_id"
-        def assign_sample_id(obj):
-            if isinstance(obj, dict):
-                if "sample_uuid" in obj:
-                    obj["sample_id"] = obj["sample_uuid"]
-                    obj.pop("sample_uuid")
-                for k, v in obj.items():
-                    if k != "unilabos_extra":
-                        assign_sample_id(v)
-            elif isinstance(obj, list):
-                for item in obj:
-                    assign_sample_id(item)
-
-        assign_sample_id(action_kwargs)
         goal_msg = convert_to_ros_msg(action_client._action_type.Goal(), action_kwargs)
 
         # self.lab_logger().trace(f"[Host Node] Sending goal for {action_id}: {str(goal_msg)[:1000]}")
@@ -854,9 +856,14 @@ class HostNode(BaseROS2DeviceNode):
                         # 适配后端的一些额外处理
                         return_value = return_info.get("return_value")
                         if isinstance(return_value, dict):
-                            unilabos_samples = return_info.get("unilabos_samples")
+                            unilabos_samples = return_value.pop(RETURN_UNILABOS_SAMPLES, None)
-                            if isinstance(unilabos_samples, list):
+                            if isinstance(unilabos_samples, list) and unilabos_samples:
-                                return_info["unilabos_samples"] = unilabos_samples
+                                self.lab_logger().info(
+                                    f"[Host Node] Job {job_id[:8]} returned {len(unilabos_samples)} sample(s): "
+                                    f"{[s.get('name', s.get('id', 'unknown')) if isinstance(s, dict) else str(s)[:20] for s in unilabos_samples[:5]]}"
+                                    f"{'...' if len(unilabos_samples) > 5 else ''}"
+                                )
+                                return_info["samples"] = unilabos_samples
                         suc = return_info.get("suc", False)
                         if not suc:
                             status = "failed"
@@ -882,7 +889,7 @@ class HostNode(BaseROS2DeviceNode):
             # 清理 _goals 中的记录
             if job_id in self._goals:
                 del self._goals[job_id]
-                self.lab_logger().debug(f"[Host Node] Removed goal {job_id[:8]} from _goals")
+                self.lab_logger().trace(f"[Host Node] Removed goal {job_id[:8]} from _goals")
 
             # 存储结果供 HTTP API 查询
             try:
@@ -1327,10 +1334,20 @@ class HostNode(BaseROS2DeviceNode):
         self.lab_logger().debug(f"[Host Node-Resource] List parameters: {request}")
         return response
 
-    def test_latency(self):
+    def test_latency(self) -> TestLatencyReturn:
         """
         测试网络延迟的action实现
         通过5次ping-pong机制校对时间误差并计算实际延迟
+
+        Returns:
+            TestLatencyReturn: 包含延迟测试结果的字典，包括：
+                - avg_rtt_ms: 平均往返时间（毫秒）
+                - avg_time_diff_ms: 平均时间差（毫秒）
+                - max_time_error_ms: 最大时间误差（毫秒）
+                - task_delay_ms: 实际任务延迟（毫秒），-1表示无法计算
+                - raw_delay_ms: 原始时间差（毫秒），-1表示无法计算
+                - test_count: 有效测试次数
+                - status: 测试状态，"success"表示成功，"all_timeout"表示全部超时
         """
         import uuid as uuid_module
 
@@ -1393,7 +1410,15 @@ class HostNode(BaseROS2DeviceNode):
 
         if not ping_results:
             self.lab_logger().error("❌ 所有ping-pong测试都失败了")
-            return {"status": "all_timeout"}
+            return {
+                "avg_rtt_ms": -1.0,
+                "avg_time_diff_ms": -1.0,
+                "max_time_error_ms": -1.0,
+                "task_delay_ms": -1.0,
+                "raw_delay_ms": -1.0,
+                "test_count": 0,
+                "status": "all_timeout",
+            }
 
         # 统计分析
         rtts = [r["rtt_ms"] for r in ping_results]
@@ -1401,7 +1426,7 @@ class HostNode(BaseROS2DeviceNode):
 
         avg_rtt_ms = sum(rtts) / len(rtts)
         avg_time_diff_ms = sum(time_diffs) / len(time_diffs)
-        max_time_diff_error_ms = max(abs(min(time_diffs)), abs(max(time_diffs)))
+        max_time_diff_error_ms: float = max(abs(min(time_diffs)), abs(max(time_diffs)))
 
         self.lab_logger().info("-" * 50)
         self.lab_logger().info("[测试统计]")
@@ -1441,7 +1466,7 @@ class HostNode(BaseROS2DeviceNode):
 
         self.lab_logger().info("=" * 60)
 
-        return {
+        res: TestLatencyReturn = {
             "avg_rtt_ms": avg_rtt_ms,
             "avg_time_diff_ms": avg_time_diff_ms,
             "max_time_error_ms": max_time_diff_error_ms,
@@ -1452,9 +1477,14 @@ class HostNode(BaseROS2DeviceNode):
             "test_count": len(ping_results),
             "status": "success",
         }
+        return res
 
     def test_resource(
-        self, resource: ResourceSlot = None, resources: List[ResourceSlot] = None, device: DeviceSlot = None, devices: List[DeviceSlot] = None
+        self,
+        resource: ResourceSlot = None,
+        resources: List[ResourceSlot] = None,
+        device: DeviceSlot = None,
+        devices: List[DeviceSlot] = None,
     ) -> TestResourceReturn:
         if resources is None:
             resources = []
@@ -1515,7 +1545,9 @@ class HostNode(BaseROS2DeviceNode):
 
             # 构建服务地址
             srv_address = f"/srv{namespace}/s2c_resource_tree"
-            self.lab_logger().trace(f"[Host Node-Resource] Host -> {device_id} ResourceTree {action} operation started -------")
+            self.lab_logger().trace(
+                f"[Host Node-Resource] Host -> {device_id} ResourceTree {action} operation started -------"
+            )
 
             # 创建服务客户端
             sclient = self.create_client(SerialCommand, srv_address)
@@ -1550,7 +1582,9 @@ class HostNode(BaseROS2DeviceNode):
                 time.sleep(0.05)
 
             response = future.result()
-            self.lab_logger().trace(f"[Host Node-Resource] Host -> {device_id} ResourceTree {action} operation completed -------")
+            self.lab_logger().trace(
+                f"[Host Node-Resource] Host -> {device_id} ResourceTree {action} operation completed -------"
+            )
             return True
 
         except Exception as e:

unilabos/ros/nodes/presets/workstation.py

@@ -340,6 +340,8 @@ class ROS2WorkstationNode(BaseROS2DeviceNode):
                             plr = self.resource_tracker.figure_resource({"name": res_name}, try_mode=False)
                             # 获取父资源
                             res = self.resource_tracker.parent_resource(plr)
+                            if res is None:
+                                res = plr
                             if id(res) not in seen:
                                 seen.add(id(res))
                                 unique_resources.append(res)

unilabos/ros/utils/driver_creator.py

@@ -52,7 +52,8 @@ class DeviceClassCreator(Generic[T]):
         if self.device_instance is not None:
             for c in self.children:
                 if c.res_content.type != "device":
-                    self.resource_tracker.add_resource(c.get_plr_nested_dict())
+                    res = ResourceTreeSet([ResourceTreeInstance(c)]).to_plr_resources()[0]
+                    self.resource_tracker.add_resource(res)
 
     def create_instance(self, data: Dict[str, Any]) -> T:
         """
@@ -119,7 +120,7 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
     #     return resource, source_type
 
     def _process_resource_references(
-        self, data: Any, to_dict=False, states=None, prefix_path="", name_to_uuid=None
+        self, data: Any, processed_child_names: Optional[Dict[str, Any]], to_dict=False, states=None, prefix_path="", name_to_uuid=None
     ) -> Any:
         """
         递归处理资源引用，替换_resource_child_name对应的资源
@@ -164,6 +165,7 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
                                 states[prefix_path] = resource_instance.serialize_all_state()
                                 return serialized
                             else:
+                                processed_child_names[child_name] = resource_instance
                                 self.resource_tracker.add_resource(resource_instance)
                                 # 立即设置UUID，state已经在resource_ulab_to_plr中处理过了
                                 if name_to_uuid:
@@ -182,12 +184,12 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
             result = {}
             for key, value in data.items():
                 new_prefix = f"{prefix_path}.{key}" if prefix_path else key
-                result[key] = self._process_resource_references(value, to_dict, states, new_prefix, name_to_uuid)
+                result[key] = self._process_resource_references(value, processed_child_names, to_dict, states, new_prefix, name_to_uuid)
             return result
 
         elif isinstance(data, list):
             return [
-                self._process_resource_references(item, to_dict, states, f"{prefix_path}[{i}]", name_to_uuid)
+                self._process_resource_references(item, processed_child_names, to_dict, states, f"{prefix_path}[{i}]", name_to_uuid)
                 for i, item in enumerate(data)
             ]
 
@@ -234,7 +236,7 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
             # 首先处理资源引用
             states = {}
             processed_data = self._process_resource_references(
-                data, to_dict=True, states=states, name_to_uuid=name_to_uuid
+                data, {}, to_dict=True, states=states, name_to_uuid=name_to_uuid
             )
 
             try:
@@ -270,7 +272,12 @@ class PyLabRobotCreator(DeviceClassCreator[T]):
                         arg_value = spec_args[param_name].annotation
                         data[param_name]["_resource_type"] = self.device_cls.__module__ + ":" + arg_value
                         logger.debug(f"自动补充 _resource_type: {data[param_name]['_resource_type']}")
-                processed_data = self._process_resource_references(data, to_dict=False, name_to_uuid=name_to_uuid)
+                processed_child_names = {}
+                processed_data = self._process_resource_references(data, processed_child_names, to_dict=False, name_to_uuid=name_to_uuid)
+                for child_name, resource_instance in processed_data.items():
+                    for ind, name in enumerate([child.res_content.name for child in self.children]):
+                        if name == child_name:
+                            self.children.pop(ind)
                 self.device_instance = super(PyLabRobotCreator, self).create_instance(processed_data)  # 补全变量后直接调用，调用的自身的attach_resource
             except Exception as e:
                 logger.error(f"PyLabRobot创建实例失败: {e}")
@@ -342,9 +349,10 @@ class WorkstationNodeCreator(DeviceClassCreator[T]):
         try:
             # 创建实例，额外补充一个给protocol node的字段，后面考虑取消
             data["children"] = self.children
-            for child in self.children:
+            # super(WorkstationNodeCreator, self).create_instance(data)的时候会attach
-                if child.res_content.type != "device":
+            # for child in self.children:
-                    self.resource_tracker.add_resource(child.get_plr_nested_dict())
+            #     if child.res_content.type != "device":
+            #         self.resource_tracker.add_resource(child.get_plr_nested_dict())
             deck_dict = data.get("deck")
             if deck_dict:
                 from pylabrobot.resources import Deck, Resource

unilabos/test/experiments/comprehensive_protocol/comprehensive_station.json

@@ -339,13 +339,8 @@
                 "z": 0
             },
             "config": {
-                "max_volume": 500.0,
                 "type": "RegularContainer",
-                "category": "container",
+                "category": "container"
-                "max_temp": 200.0,
-                "min_temp": -20.0,
-                "has_stirrer": true,
-                "has_heater": true
             },
             "data": {
                 "liquids": [],
@@ -769,9 +764,7 @@
                 "size_y": 250,
                 "size_z": 0,
                 "type": "RegularContainer",
-                "category": "container",
+                "category": "container"
-                "reagent": "sodium_chloride",
-                "physical_state": "solid"
             },
             "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",
+                "category": "container"
-                "reagent": "sodium_carbonate",
-                "physical_state": "solid"
             },
             "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",
+                "category": "container"
-                "reagent": "magnesium_chloride",
-                "physical_state": "solid"
             },
             "data": {
                 "current_mass": 500.0,

unilabos/utils/import_manager.py

@@ -27,6 +27,7 @@ __all__ = [
 
 from ast import Constant
 
+from unilabos.resources.resource_tracker import PARAM_SAMPLE_UUIDS
 from unilabos.utils import logger
 
 
@@ -334,13 +335,18 @@ class ImportManager:
                     result["action_methods"][method_name] = method_info
         return result
 
-    def _analyze_method_signature(self, method) -> Dict[str, Any]:
+    def _analyze_method_signature(self, method, skip_unilabos_params: bool = True) -> Dict[str, Any]:
         """
         分析方法签名，提取具体的命名参数信息
 
         注意：此方法会跳过*args和**kwargs，只提取具体的命名参数
         这样可以确保通过**dict方式传参时的准确性
 
+        Args:
+            method: 要分析的方法
+            skip_unilabos_params: 是否跳过 unilabos 系统参数（如 sample_uuids），
+                                  registry 补全时为 True，JsonCommand 执行时为 False
+
         示例用法：
             method_info = self._analyze_method_signature(some_method)
             params = {"param1": "value1", "param2": "value2"}
@@ -361,6 +367,10 @@ class ImportManager:
             if param.kind == param.VAR_KEYWORD:  # **kwargs
                 continue
 
+            # 跳过 sample_uuids 参数（由系统自动注入，registry 补全时跳过）
+            if skip_unilabos_params and param_name == PARAM_SAMPLE_UUIDS:
+                continue
+
             is_required = param.default == inspect.Parameter.empty
             if is_required:
                 num_required += 1
@@ -549,6 +559,9 @@ class ImportManager:
         for i, arg in enumerate(node.args.args):
             if arg.arg == "self":
                 continue
+            # 跳过 sample_uuids 参数（由系统自动注入）
+            if arg.arg == PARAM_SAMPLE_UUIDS:
+                continue
             arg_info = {
                 "name": arg.arg,
                 "type": None,

unilabos/workflow/common.py

@@ -60,7 +60,11 @@
 ==================== 连接关系图 ====================
 
 控制流 (ready 端口串联):
-    create_resource_1 -> create_resource_2 -> ... -> set_liquid_1 -> set_liquid_2 -> ... -> transfer_liquid_1 -> transfer_liquid_2 -> ...
+    - create_resource 之间: 无 ready 连接
+    - set_liquid_from_plate 之间: 无 ready 连接
+    - create_resource 与 set_liquid_from_plate 之间: 无 ready 连接
+    - transfer_liquid 之间: 通过 ready 端口串联
+        transfer_liquid_1 -> transfer_liquid_2 -> transfer_liquid_3 -> ...
 
 物料流:
     [create_resource] --labware--> [set_liquid_from_plate] --output_wells--> [transfer_liquid] --sources_out/targets_out--> [下一个 transfer_liquid]
@@ -402,7 +406,6 @@ def build_protocol_graph(
 
     # 为每个唯一的 slot 创建 create_resource 节点
     res_index = 0
-    last_create_resource_id = None
     for slot, info in slots_info.items():
         node_id = str(uuid.uuid4())
         res_id = info["res_id"]
@@ -431,10 +434,7 @@ def build_protocol_graph(
         )
         slot_to_create_resource[slot] = node_id
 
-        # create_resource 之间通过 ready 串联
+        # create_resource 之间不需要 ready 连接
-        if last_create_resource_id is not None:
-            G.add_edge(last_create_resource_id, node_id, source_port="ready", target_port="ready")
-        last_create_resource_id = node_id
 
     # ==================== 第二步：为每个 reagent 创建 set_liquid_from_plate 节点 ====================
     # 创建 Group 节点，包含所有 set_liquid_from_plate 节点
@@ -453,7 +453,6 @@ def build_protocol_graph(
     )
 
     set_liquid_index = 0
-    last_set_liquid_id = last_create_resource_id  # set_liquid_from_plate 连接在 create_resource 之后
 
     for labware_id, item in labware_info.items():
         # 跳过 Tip/Rack 类型
@@ -494,10 +493,7 @@ def build_protocol_graph(
             },
         )
 
-        # ready 连接：上一个节点 -> set_liquid_from_plate
+        # set_liquid_from_plate 之间不需要 ready 连接
-        if last_set_liquid_id is not None:
-            G.add_edge(last_set_liquid_id, node_id, source_port="ready", target_port="ready")
-        last_set_liquid_id = node_id
 
         # 物料流：create_resource 的 labware -> set_liquid_from_plate 的 input_plate
         create_res_node_id = slot_to_create_resource.get(slot)
@@ -507,7 +503,8 @@ def build_protocol_graph(
         # set_liquid_from_plate 的输出 output_wells 用于连接 transfer_liquid
         resource_last_writer[labware_id] = f"{node_id}:output_wells"
 
-    last_control_node_id = last_set_liquid_id
+    # transfer_liquid 之间通过 ready 串联，从 None 开始
+    last_control_node_id = None
 
     # 端口名称映射：JSON 字段名 -> 实际 handle key
     INPUT_PORT_MAPPING = {