Merge branch 'dev' into feat/3d_layout_and_visualize

unilabos/app/web/client.py

@@ -80,19 +80,20 @@ class HTTPClient:
             f.write(json.dumps(payload, indent=4))
         # 从序列化数据中提取所有节点的UUID（保存旧UUID）
         old_uuids = {n.res_content.uuid: n for n in resources.all_nodes}
+        nodes_info = [x for xs in resources.dump() for x in xs]
         if not self.initialized or first_add:
             self.initialized = True
             info(f"首次添加资源，当前远程地址: {self.remote_addr}")
             response = requests.post(
                 f"{self.remote_addr}/edge/material",
-                json={"nodes": [x for xs in resources.dump() for x in xs], "mount_uuid": mount_uuid},
+                json={"nodes": nodes_info, "mount_uuid": mount_uuid},
                 headers={"Authorization": f"Lab {self.auth}"},
                 timeout=60,
             )
         else:
             response = requests.put(
                 f"{self.remote_addr}/edge/material",
-                json={"nodes": [x for xs in resources.dump() for x in xs], "mount_uuid": mount_uuid},
+                json={"nodes": nodes_info, "mount_uuid": mount_uuid},
                 headers={"Authorization": f"Lab {self.auth}"},
                 timeout=10,
             )
@@ -111,6 +112,7 @@ class HTTPClient:
                     uuid_mapping[i["uuid"]] = i["cloud_uuid"]
         else:
             logger.error(f"添加物料失败: {response.text}")
+            logger.trace(f"添加物料失败: {nodes_info}")
         for u, n in old_uuids.items():
             if u in uuid_mapping:
                 n.res_content.uuid = uuid_mapping[u]

unilabos/app/ws_client.py

@@ -1113,7 +1113,7 @@ class MessageProcessor:
                 "task_id": task_id,
                 "job_id": job_id,
                 "free": free,
-                "need_more": need_more,
+                "need_more": need_more + 1,
             },
         }
 
@@ -1253,7 +1253,7 @@ class QueueProcessor:
                     "task_id": job_info.task_id,
                     "job_id": job_info.job_id,
                     "free": False,
-                    "need_more": 10,
+                    "need_more": 10 + 1,
                 },
             }
             self.message_processor.send_message(message)
@@ -1286,7 +1286,7 @@ class QueueProcessor:
                     "task_id": job_info.task_id,
                     "job_id": job_info.job_id,
                     "free": False,
-                    "need_more": 10,
+                    "need_more": 10 + 1,
                 },
             }
             success = self.message_processor.send_message(message)
@@ -1369,6 +1369,10 @@ class WebSocketClient(BaseCommunicationClient):
         self.message_processor = MessageProcessor(self.websocket_url, self.send_queue, self.device_manager)
         self.queue_processor = QueueProcessor(self.device_manager, self.message_processor)
 
+        # running状态debounce缓存: {job_id: (last_send_timestamp, last_feedback_data)}
+        self._job_running_last_sent: Dict[str, tuple] = {}
+        self._job_running_debounce_interval: float = 10.0  # 秒
+
         # 设置相互引用
         self.message_processor.set_queue_processor(self.queue_processor)
         self.message_processor.set_websocket_client(self)
@@ -1468,22 +1472,32 @@ class WebSocketClient(BaseCommunicationClient):
             logger.debug(f"[WebSocketClient] Not connected, cannot publish job status for job_id: {item.job_id}")
             return
 
+        job_log = format_job_log(item.job_id, item.task_id, item.device_id, item.action_name)
+
         # 拦截最终结果状态，与原版本逻辑一致
         if status in ["success", "failed"]:
+            self._job_running_last_sent.pop(item.job_id, None)
+
             host_node = HostNode.get_instance(0)
             if host_node:
-                # 从HostNode的device_action_status中移除job_id
                 try:
                     host_node._device_action_status[item.device_action_key].job_ids.pop(item.job_id, None)
                 except (KeyError, AttributeError):
                     logger.warning(f"[WebSocketClient] Failed to remove job {item.job_id} from HostNode status")
 
-            # logger.debug(f"[WebSocketClient] Intercepting final status for job_id: {item.job_id} - {status}")
-
-            # 通知队列处理器job完成（包括timeout的job）
             self.queue_processor.handle_job_completed(item.job_id, status)
 
-        # 发送job状态消息
+        # running状态按job_id做debounce，内容变化时仍然上报
+        if status == "running":
+            now = time.time()
+            cached = self._job_running_last_sent.get(item.job_id)
+            if cached is not None:
+                last_ts, last_data = cached
+                if now - last_ts < self._job_running_debounce_interval and last_data == feedback_data:
+                    logger.trace(f"[WebSocketClient] Job status debounced (skip): {job_log} - {status}")
+                    return
+            self._job_running_last_sent[item.job_id] = (now, feedback_data)
+
         message = {
             "action": "job_status",
             "data": {
@@ -1499,7 +1513,6 @@ class WebSocketClient(BaseCommunicationClient):
         }
         self.message_processor.send_message(message)
 
-        job_log = format_job_log(item.job_id, item.task_id, item.device_id, item.action_name)
         logger.trace(f"[WebSocketClient] Job status published: {job_log} - {status}")
 
     def send_ping(self, ping_id: str, timestamp: float) -> None:

unilabos/ros/main_slave_run.py

@@ -1,4 +1,5 @@
 import json
+import os
 
 # from nt import device_encoding
 import threading
@@ -61,7 +62,7 @@ def main(
         rclpy.init(args=rclpy_init_args)
     else:
         logger.info("[ROS] rclpy already initialized, reusing context")
-    executor = rclpy.__executor = MultiThreadedExecutor()
+    executor = rclpy.__executor = MultiThreadedExecutor(num_threads=max(os.cpu_count() * 4, 48))
     # 创建主机节点
     host_node = HostNode(
         "host_node",
@@ -122,7 +123,7 @@ def slave(
         rclpy.init(args=rclpy_init_args)
     executor = rclpy.__executor
     if not executor:
-        executor = rclpy.__executor = MultiThreadedExecutor()
+        executor = rclpy.__executor = MultiThreadedExecutor(num_threads=max(os.cpu_count() * 4, 48))
 
     # 1.5 启动 executor 线程
     thread = threading.Thread(target=executor.spin, daemon=True, name="slave_executor_thread")

unilabos/ros/nodes/base_device_node.py

@@ -486,18 +486,12 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                 if len(rts.root_nodes) == 1 and parent_resource is not None:
                     plr_instance = plr_instances[0]
                     if isinstance(plr_instance, Plate):
-                        empty_liquid_info_in: List[Tuple[Optional[str], float]] = [(None, 0)] * plr_instance.num_items
                         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)} 个"
                             )
-                        for liquid_type, liquid_volume, liquid_input_slot in zip(
-                            ADD_LIQUID_TYPE, LIQUID_VOLUME, LIQUID_INPUT_SLOT
-                        ):
-                            empty_liquid_info_in[liquid_input_slot] = (liquid_type, liquid_volume)
-                        plr_instance.set_well_liquids(empty_liquid_info_in)
                         try:
                             # noinspection PyProtectedMember
                             keys = list(plr_instance._ordering.keys())
@@ -511,6 +505,10 @@ class BaseROS2DeviceNode(Node, Generic[T]):
                             input_wells = []
                             for r in LIQUID_INPUT_SLOT:
                                 input_wells.append(plr_instance.children[r])
+                        for input_well, liquid_type, liquid_volume, liquid_input_slot in zip(
+                            input_wells, ADD_LIQUID_TYPE, LIQUID_VOLUME, LIQUID_INPUT_SLOT
+                        ):
+                            input_well.set_liquids([(liquid_type, liquid_volume, "uL")])
                         final_response["liquid_input_resource_tree"] = ResourceTreeSet.from_plr_resources(
                             input_wells
                         ).dump()
@@ -1256,9 +1254,8 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         return self._lab_logger
 
     def create_ros_publisher(self, attr_name, msg_type, initial_period=5.0):
-        """创建ROS发布者，仅当方法/属性有 @topic_config 装饰器时才创建。"""
-        # 检测 @topic_config 装饰器配置
-        topic_config = {}
+        """创建ROS发布者。已在 status_types 中声明的属性直接创建；@topic_config 用于覆盖默认参数。"""
+        topic_cfg = {}
         driver_class = type(self.driver_instance)
 
         # 区分 @property 和普通方法两种情况
@@ -1267,23 +1264,17 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         )
 
         if is_prop:
-            # @property: 检测 fget 上的 @topic_config
             class_attr = getattr(driver_class, attr_name)
             if class_attr.fget is not None:
-                topic_config = get_topic_config(class_attr.fget)
+                topic_cfg = get_topic_config(class_attr.fget)
         else:
-            # 普通方法: 直接检测 attr_name 方法上的 @topic_config
             if hasattr(self.driver_instance, attr_name):
                 method = getattr(self.driver_instance, attr_name)
                 if callable(method):
-                    topic_config = get_topic_config(method)
-
-        # 没有 @topic_config 装饰器则跳过发布
-        if not topic_config:
-            return
+                    topic_cfg = get_topic_config(method)
 
         # 发布名称优先级: @topic_config(name=...) > get_ 前缀去除 > attr_name
-        cfg_name = topic_config.get("name")
+        cfg_name = topic_cfg.get("name")
         if cfg_name:
             publish_name = cfg_name
         elif attr_name.startswith("get_"):
@@ -1291,10 +1282,10 @@ class BaseROS2DeviceNode(Node, Generic[T]):
         else:
             publish_name = attr_name
 
-        # 使用装饰器配置或默认值
-        cfg_period = topic_config.get("period")
-        cfg_print = topic_config.get("print_publish")
-        cfg_qos = topic_config.get("qos")
+        # @topic_config 参数覆盖默认值
+        cfg_period = topic_cfg.get("period")
+        cfg_print = topic_cfg.get("print_publish")
+        cfg_qos = topic_cfg.get("qos")
         period: float = cfg_period if cfg_period is not None else initial_period
         print_publish: bool = cfg_print if cfg_print is not None else self._print_publish
         qos: int = cfg_qos if cfg_qos is not None else 10

unilabos/ros/nodes/presets/host_node.py

@@ -1632,6 +1632,7 @@ class HostNode(BaseROS2DeviceNode):
     def manual_confirm(self, timeout_seconds: int, assignee_user_ids: list[str], **kwargs) -> dict:
         """
         timeout_seconds: 超时时间（秒），默认3600秒
+        修改的结果无效，是只读的
         """
         return kwargs
 

unilabos/workflow/common.py

@@ -346,7 +346,7 @@ def refactor_data(
             "template_name": template_name,
             "resource_name": resource_name,
             "description": step.get("description", step.get("purpose", f"{operation} operation")),
-            "lab_node_type": "Device",
+            "lab_node_type": "ILab",
             "param": step.get("parameters", step.get("action_args", {})),
             "footer": f"{template_name}-{resource_name}",
         }