add --atomic option for ros2 param set

Signed-off-by: Brian Chen <[email protected]>

Author: ihasdapie

ros2param/ros2param/api/__init__.py

@@ -99,6 +99,18 @@ def call_set_parameters(*, node, node_name, parameters):
     return response
 
 
+def call_set_parameters_atomically(*, node, node_name, parameters):
+    client = AsyncParameterClient(node, node_name)
+    client.wait_for_services(timeout_sec=5.0)
+    future = client.set_parameters_atomically(parameters)
+    rclpy.spin_until_future_complete(node, future)
+    response = future.result()
+    if response is None:
+        raise RuntimeError('Exception while calling service of node '
+                           f'{node_name}: {future.exception()}')
+    return response
+
+
 def call_list_parameters(*, node, node_name, prefixes=None):
     client = AsyncParameterClient(node, node_name)
     client.wait_for_services(timeout_sec=5.0)

ros2param/ros2param/verb/set.py

@@ -22,8 +22,8 @@
 from ros2node.api import get_absolute_node_name
 from ros2node.api import get_node_names
 from ros2node.api import NodeNameCompleter
-
 from ros2param.api import call_set_parameters
+from ros2param.api import call_set_parameters_atomically
 from ros2param.api import ParameterNameCompleter
 from ros2param.verb import VerbExtension
 
@@ -40,6 +40,10 @@ def add_arguments(self, parser, cli_name):  # noqa: D102
         parser.add_argument(
             '--include-hidden-nodes', action='store_true',
             help='Consider hidden nodes as well')
+        parser.add_argument(
+            '--atomic', action='store_true',
+            help='Set parameters atomically'
+        )
         arg = parser.add_argument(
             'parameter_name', help='Name of the parameter')
         arg.completer = ParameterNameCompleter()
@@ -60,12 +64,16 @@ def main(self, *, args):  # noqa: D102
             Parameter.name = args.parameter_name
             parameter.value = get_parameter_value(string_value=args.value)
 
-            response = call_set_parameters(
-                node=node, node_name=args.node_name, parameters=[parameter])
+            if args.atomic:
+                response = call_set_parameters_atomically(
+                    node=node, node_name=args.node_name, parameters=[parameter])
+                result = response.result
+            else:
+                response = call_set_parameters(
+                    node=node, node_name=args.node_name, parameters=[parameter])
+                assert len(response.results) == 1
+                result = response.results[0]
 
-            # output response
-            assert len(response.results) == 1
-            result = response.results[0]
             if result.successful:
                 msg = 'Set parameter successful'
                 if result.reason: