New async rosapi params module implementation

rosapi/scripts/rosapi_node

@@ -35,8 +35,10 @@ import os
 import sys
 
 import rclpy
+from rclpy.callback_groups import ReentrantCallbackGroup
 from rclpy.clock import Clock, ClockType
 from rclpy.node import Node
+from rclpy.qos import qos_profile_parameters
 
 from rosapi import glob_helper, objectutils, params, proxy
 from rosapi_msgs.msg import TypeDef
@@ -85,13 +87,9 @@ class Rosapi(Node):
     # Initialises the ROS node
     def register_services(self):
         proxy.init(self)
-        if self.get_namespace() == "/":
-            full_name = self.get_namespace() + self.get_name()
-        else:
-            full_name = self.get_namespace() + "/" + self.get_name()
 
         timeout_sec = self.get_parameter("params_timeout").value
-        params.init(full_name, timeout_sec)
+        params.init(self, timeout_sec)
 
         self.create_service(Topics, "~/topics", self.get_topics)
         self.create_service(Interfaces, "~/interfaces", self.get_interfaces)
@@ -131,11 +129,41 @@ class Rosapi(Node):
             "~/service_response_details",
             self.get_service_response_details,
         )
-        self.create_service(SetParam, "~/set_param", self.set_param)
-        self.create_service(GetParam, "~/get_param", self.get_param)
-        self.create_service(HasParam, "~/has_param", self.has_param)
-        self.create_service(DeleteParam, "~/delete_param", self.delete_param)
-        self.create_service(GetParamNames, "~/get_param_names", self.get_param_names)
+        self.create_service(
+            SetParam,
+            "~/set_param",
+            self.set_param,
+            callback_group=ReentrantCallbackGroup(),
+            qos_profile=qos_profile_parameters,
+        )
+        self.create_service(
+            GetParam,
+            "~/get_param",
+            self.get_param,
+            callback_group=ReentrantCallbackGroup(),
+            qos_profile=qos_profile_parameters,
+        )
+        self.create_service(
+            HasParam,
+            "~/has_param",
+            self.has_param,
+            callback_group=ReentrantCallbackGroup(),
+            qos_profile=qos_profile_parameters,
+        )
+        self.create_service(
+            DeleteParam,
+            "~/delete_param",
+            self.delete_param,
+            callback_group=ReentrantCallbackGroup(),
+            qos_profile=qos_profile_parameters,
+        )
+        self.create_service(
+            GetParamNames,
+            "~/get_param_names",
+            self.get_param_names,
+            callback_group=ReentrantCallbackGroup(),
+            qos_profile=qos_profile_parameters,
+        )
         self.create_service(GetTime, "~/get_time", self.get_time)
         self.create_service(GetROSVersion, "~/get_ros_version", self.get_ros_version)
 
@@ -262,38 +290,68 @@ class Rosapi(Node):
         ]
         return response
 
-    def set_param(self, request, response):
+    async def set_param(self, request, response):
         try:
             node_name, param_name = self._get_node_and_param_name(request.name)
-            params.set_param(node_name, param_name, request.value, self.globs.params)
         except ValueError:
             self._print_malformed_param_name_warning(request.name)
+
+        try:
+            await params.set_param(node_name, param_name, request.value, self.globs.params)
+        except Exception as e:
+            response.successful = False
+            response.reason = str(e)
+        else:
+            response.successful = True
+
         return response
 
-    def get_param(self, request, response):
+    async def get_param(self, request, response):
         try:
             node_name, param_name = self._get_node_and_param_name(request.name)
-            response.value = params.get_param(
-                node_name, param_name, request.default_value, self.globs.params
-            )
         except ValueError:
             self._print_malformed_param_name_warning(request.name)
+
+        try:
+            response.value = await params.get_param(node_name, param_name, self.globs.params)
+        except Exception as e:
+            response.successful = False
+            response.reason = str(e)
+            # return empty JSON string so the client doesn't fail if it tries to parse it
+            response.value = '""'
+        else:
+            response.successful = True
+
         return response
 
-    def has_param(self, request, response):
+    async def has_param(self, request, response):
         try:
             node_name, param_name = self._get_node_and_param_name(request.name)
-            response.exists = params.has_param(node_name, param_name, self.globs.params)
         except ValueError:
             self._print_malformed_param_name_warning(request.name)
+
+        response.exists = await params.has_param(node_name, param_name, self.globs.params)
+
         return response
 
-    def delete_param(self, request, response):
-        params.delete_param(request.node_name, request.name, self.globs.params)
+    async def delete_param(self, request, response):
+        try:
+            node_name, param_name = self._get_node_and_param_name(request.name)
+        except ValueError:
+            self._print_malformed_param_name_warning(request.name)
+
+        try:
+            await params.delete_param(node_name, param_name, self.globs.params)
+        except Exception as e:
+            response.successful = False
+            response.reason = str(e)
+        else:
+            response.successful = True
+
         return response
 
-    def get_param_names(self, request, response):
-        response.names = params.get_param_names(self.globs.params)
+    async def get_param_names(self, request, response):
+        response.names = await params.get_param_names(self.globs.params)
         return response
 
     def get_time(self, request, response):

rosapi/src/rosapi/async_helper.py

@@ -0,0 +1,71 @@
+# Software License Agreement (BSD License)
+#
+# Copyright (c) 2025, Fictionlab sp. z o.o.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above
+#    copyright notice, this list of conditions and the following
+#    disclaimer in the documentation and/or other materials provided
+#    with the distribution.
+#  * Neither the name of Willow Garage, Inc. nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+from rclpy.node import Node
+from rclpy.task import Future
+
+
+async def futures_wait_for(node: Node, futures: list[Future], timeout_sec: float):
+    """Await a list of futures with a timeout."""
+    first_done_future: Future = Future()
+
+    def timeout_callback():
+        first_done_future.set_result(None)
+
+    timer = node.create_timer(timeout_sec, timeout_callback)
+
+    def future_done_callback(arg):
+        if all(future.done() for future in futures):
+            first_done_future.set_result(None)
+
+    for future in futures:
+        future.add_done_callback(future_done_callback)
+
+    await first_done_future
+
+    timer.cancel()
+    timer.destroy()
+
+
+async def async_sleep(node: Node, delay_sec: float):
+    """Block the coroutine for a given time."""
+    sleep_future: Future = Future()
+
+    def timeout_callback():
+        sleep_future.set_result(None)
+
+    timer = node.create_timer(delay_sec, timeout_callback)
+
+    await sleep_future
+
+    timer.cancel()
+    timer.destroy()