移动内部action以兼容host node

docs/developer_guide/add_yaml.md

@@ -111,8 +111,8 @@ new_device: # 设备名，要唯一
 
 1. 以 `auto-` 开头的动作：从你 Python 类的方法自动生成
 2. 通用的驱动动作：
-   - `_execute_driver_command`：同步执行驱动命令
-   - `_execute_driver_command_async`：异步执行驱动命令
+   - `_execute_driver_command`：同步执行驱动命令（仅本地可用）
+   - `_execute_driver_command_async`：异步执行驱动命令（仅本地可用）
 
 ### 如果要手动定义动作
 

unilabos/registry/registry.py

@@ -55,6 +55,7 @@ class Registry:
             "ResourceCreateFromOuterEasy", "host_node", f"动作 create_resource"
         )
         self.EmptyIn = self._replace_type_with_class("EmptyIn", "host_node", f"")
+        self.StrSingleInput = self._replace_type_with_class("StrSingleInput", "host_node", f"")
         self.device_type_registry = {}
         self.device_module_to_registry = {}
         self.resource_type_registry = {}

unilabos/ros/nodes/base_device_node.py

@@ -1078,6 +1078,57 @@ class BaseROS2DeviceNode(Node, Generic[T]):
 
         return execute_callback
 
+    def _execute_driver_command(self, string: str):
+        try:
+            target = json.loads(string)
+        except Exception as ex:
+            try:
+                target = yaml.safe_load(io.StringIO(string))
+            except Exception as ex2:
+                raise JsonCommandInitError(
+                    f"执行动作时JSON/YAML解析失败: \n{ex}\n{ex2}\n原内容: {string}\n{traceback.format_exc()}"
+                )
+        try:
+            function_name = target["function_name"]
+            function_args = target["function_args"]
+            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}"
+            return function(**function_args)
+        except KeyError as ex:
+            raise JsonCommandInitError(
+                f"执行动作时JSON缺少function_name或function_args: {ex}\n原JSON: {string}\n{traceback.format_exc()}"
+            )
+
+    async def _execute_driver_command_async(self, string: str):
+        try:
+            target = json.loads(string)
+        except Exception as ex:
+            try:
+                target = yaml.safe_load(io.StringIO(string))
+            except Exception as ex2:
+                raise JsonCommandInitError(
+                    f"执行动作时JSON/YAML解析失败: \n{ex}\n{ex2}\n原内容: {string}\n{traceback.format_exc()}"
+                )
+        try:
+            function_name = target["function_name"]
+            function_args = target["function_args"]
+            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}"
+            assert asyncio.iscoroutinefunction(
+                function
+            ), f"执行动作时JSON中的function并非异步: {function_name}\n原JSON: {string}"
+            return await function(**function_args)
+        except KeyError as ex:
+            raise JsonCommandInitError(
+                f"执行动作时JSON缺少function_name或function_args: {ex}\n原JSON: {string}\n{traceback.format_exc()}"
+            )
+
     # 异步上下文管理方法
     async def __aenter__(self):
         """进入异步上下文"""
@@ -1244,65 +1295,14 @@ class ROS2DeviceNode:
         self._ros_node: BaseROS2DeviceNode
         self._ros_node.lab_logger().info(f"初始化完成 {self._ros_node.uuid} {self.driver_is_ros}")
         self.driver_instance._ros_node = self._ros_node  # type: ignore
-        self.driver_instance._execute_driver_command = self._execute_driver_command  # type: ignore
-        self.driver_instance._execute_driver_command_async = self._execute_driver_command_async  # type: ignore
+        self.driver_instance._execute_driver_command = self._ros_node._execute_driver_command  # type: ignore
+        self.driver_instance._execute_driver_command_async = self._ros_node._execute_driver_command_async  # type: ignore
         if hasattr(self.driver_instance, "post_init"):
             try:
                 self.driver_instance.post_init(self._ros_node)  # type: ignore
             except Exception as e:
                 self._ros_node.lab_logger().error(f"设备后初始化失败: {e}")
 
-    def _execute_driver_command(self, string: str):
-        try:
-            target = json.loads(string)
-        except Exception as ex:
-            try:
-                target = yaml.safe_load(io.StringIO(string))
-            except Exception as ex2:
-                raise JsonCommandInitError(
-                    f"执行动作时JSON/YAML解析失败: \n{ex}\n{ex2}\n原内容: {string}\n{traceback.format_exc()}"
-                )
-        try:
-            function_name = target["function_name"]
-            function_args = target["function_args"]
-            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}"
-            return function(**function_args)
-        except KeyError as ex:
-            raise JsonCommandInitError(
-                f"执行动作时JSON缺少function_name或function_args: {ex}\n原JSON: {string}\n{traceback.format_exc()}"
-            )
-
-    async def _execute_driver_command_async(self, string: str):
-        try:
-            target = json.loads(string)
-        except Exception as ex:
-            try:
-                target = yaml.safe_load(io.StringIO(string))
-            except Exception as ex2:
-                raise JsonCommandInitError(
-                    f"执行动作时JSON/YAML解析失败: \n{ex}\n{ex2}\n原内容: {string}\n{traceback.format_exc()}"
-                )
-        try:
-            function_name = target["function_name"]
-            function_args = target["function_args"]
-            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}"
-            assert asyncio.iscoroutinefunction(
-                function
-            ), f"执行动作时JSON中的function并非异步: {function_name}\n原JSON: {string}"
-            return await function(**function_args)
-        except KeyError as ex:
-            raise JsonCommandInitError(
-                f"执行动作时JSON缺少function_name或function_args: {ex}\n原JSON: {string}\n{traceback.format_exc()}"
-            )
-
     def _start_loop(self):
         def run_event_loop():
             loop = asyncio.new_event_loop()

unilabos/ros/nodes/presets/host_node.py

@@ -160,13 +160,13 @@ class HostNode(BaseROS2DeviceNode):
             ),
             "/devices/host_node/_execute_driver_command": ActionClient(
                 self,
-                lab_registry.EmptyIn,
+                lab_registry.StrSingleInput,
                 "/devices/host_node/_execute_driver_command",
                 callback_group=self.callback_group,
             ),
             "/devices/host_node/_execute_driver_command_async": ActionClient(
                 self,
-                lab_registry.EmptyIn,
+                lab_registry.StrSingleInput,
                 "/devices/host_node/_execute_driver_command_async",
                 callback_group=self.callback_group,
             ),