add timeout option for ros2param to find node. (#802)

Signed-off-by: Tomoya Fujita <[email protected]>

Author: fujitatomoya

ros2cli/ros2cli/helpers.py

@@ -39,10 +39,9 @@ def wait_for(predicate, timeout, period=0.1):
     deadline = time.time() + timeout
     while not predicate():
         if time.time() > deadline:
-            break
+            return predicate()
         time.sleep(period)
-    return predicate()
-
+    return True
 
 def bind(func, *args, **kwargs):
     """

ros2node/ros2node/api/__init__.py

@@ -12,11 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
+
 from collections import namedtuple
 from typing import Any
 from typing import List
 
 from rclpy.node import HIDDEN_NODE_PREFIX
+from ros2cli.helpers import wait_for
 from ros2cli.node.strategy import NodeStrategy
 
 INFO_NONUNIQUE_WARNING_TEMPLATE = (
@@ -71,6 +74,15 @@ def get_node_names(*, node, include_hidden_nodes=False):
     ]
 
 
+def wait_for_node(node, node_name, include_hidden_nodes=False, timeout=1.0) -> bool:
+    def node_available():
+        node_names = get_node_names(
+            node=node, include_hidden_nodes=include_hidden_nodes)
+        return True if node_name in {n.full_name for n in node_names} else False
+
+    return wait_for(node_available, timeout)
+
+
 def get_topics(remote_node_name, func, *, include_hidden_topics=False):
     node = parse_node_name(remote_node_name)
     names_and_types = func(node.name, node.namespace)

ros2param/ros2param/verb/delete.py

@@ -21,8 +21,8 @@
 from ros2cli.node.strategy import add_arguments
 from ros2cli.node.strategy import NodeStrategy
 from ros2node.api import get_absolute_node_name
-from ros2node.api import get_node_names
 from ros2node.api import NodeNameCompleter
+from ros2node.api import wait_for_node
 from ros2param.api import call_set_parameters
 from ros2param.api import ParameterNameCompleter
 from ros2param.verb import VerbExtension
@@ -43,15 +43,15 @@ def add_arguments(self, parser, cli_name):  # noqa: D102
         arg = parser.add_argument(
             'parameter_name', help='Name of the parameter')
         arg.completer = ParameterNameCompleter()
+        parser.add_argument(
+            '--timeout', metavar='N', type=int, default=1,
+            help='Wait for N seconds until node becomes available (default %(default)s sec)')
 
     def main(self, *, args):  # noqa: D102
-        with NodeStrategy(args) as node:
-            node_names = get_node_names(
-                node=node, include_hidden_nodes=args.include_hidden_nodes)
-
         node_name = get_absolute_node_name(args.node_name)
-        if node_name not in [n.full_name for n in node_names]:
-            return 'Node not found'
+        with NodeStrategy(args) as node:
+            if not wait_for_node(node, node_name, args.include_hidden_nodes, args.timeout):
+                return 'Node not found'
 
         with DirectNode(args) as node:
             parameter = Parameter()

ros2param/ros2param/verb/describe.py

@@ -16,8 +16,8 @@
 from ros2cli.node.strategy import add_arguments
 from ros2cli.node.strategy import NodeStrategy
 from ros2node.api import get_absolute_node_name
-from ros2node.api import get_node_names
 from ros2node.api import NodeNameCompleter
+from ros2node.api import wait_for_node
 from ros2param.api import call_describe_parameters
 from ros2param.api import get_parameter_type_string
 from ros2param.api import ParameterNameCompleter
@@ -38,16 +38,16 @@ def add_arguments(self, parser, cli_name):  # noqa: D102
             help='Consider hidden nodes as well')
         arg = parser.add_argument(
             'parameter_names', nargs='+', help='Names of the parameters')
+        parser.add_argument(
+            '--timeout', metavar='N', type=int, default=1,
+            help='Wait for N seconds until node becomes available (default %(default)s sec)')
         arg.completer = ParameterNameCompleter()
 
     def main(self, *, args):  # noqa: D102
-        with NodeStrategy(args) as node:
-            node_names = get_node_names(
-                node=node, include_hidden_nodes=args.include_hidden_nodes)
-
         node_name = get_absolute_node_name(args.node_name)
-        if node_name not in {n.full_name for n in node_names}:
-            return 'Node not found'
+        with NodeStrategy(args) as node:
+            if not wait_for_node(node, node_name, args.include_hidden_nodes, args.timeout):
+                return 'Node not found'
 
         with DirectNode(args) as node:
             response = call_describe_parameters(

ros2param/ros2param/verb/dump.py

@@ -20,9 +20,9 @@
 from ros2cli.node.strategy import add_arguments
 from ros2cli.node.strategy import NodeStrategy
 from ros2node.api import get_absolute_node_name
-from ros2node.api import get_node_names
 from ros2node.api import NodeNameCompleter
 from ros2node.api import parse_node_name
+from ros2node.api import wait_for_node
 
 from ros2param.api import call_get_parameters
 from ros2param.api import call_list_parameters
@@ -50,6 +50,9 @@ def add_arguments(self, parser, cli_name):  # noqa: D102
         parser.add_argument(
             '--print', action='store_true',
             help='DEPRECATED: Does nothing.')
+        parser.add_argument(
+            '--timeout', metavar='N', type=int, default=1,
+            help='Wait for N seconds until node becomes available (default %(default)s sec)')
 
     @staticmethod
     def get_parameter_values(node, node_name, params):
@@ -74,16 +77,13 @@ def insert_dict(self, dictionary, key, value):
             dictionary[key] = value
 
     def main(self, *, args):  # noqa: D102
-
-        with NodeStrategy(args) as node:
-            node_names = get_node_names(node=node, include_hidden_nodes=args.include_hidden_nodes)
-
         absolute_node_name = get_absolute_node_name(args.node_name)
-        node_name = parse_node_name(absolute_node_name)
-        if absolute_node_name:
-            if absolute_node_name not in [n.full_name for n in node_names]:
+        with NodeStrategy(args) as node:
+            if not wait_for_node(node, absolute_node_name,
+                                 args.include_hidden_nodes, args.timeout):
                 return 'Node not found'
 
+        node_name = parse_node_name(absolute_node_name)
         if not os.path.isdir(args.output_dir):
             raise RuntimeError(
                 f"'{args.output_dir}' is not a valid directory.")

ros2param/ros2param/verb/get.py

@@ -17,8 +17,8 @@
 from ros2cli.node.strategy import add_arguments
 from ros2cli.node.strategy import NodeStrategy
 from ros2node.api import get_absolute_node_name
-from ros2node.api import get_node_names
 from ros2node.api import NodeNameCompleter
+from ros2node.api import wait_for_node
 from ros2param.api import call_get_parameters
 from ros2param.api import ParameterNameCompleter
 from ros2param.verb import VerbExtension
@@ -42,15 +42,15 @@ def add_arguments(self, parser, cli_name):  # noqa: D102
         parser.add_argument(
             '--hide-type', action='store_true',
             help='Hide the type information')
+        parser.add_argument(
+            '--timeout', metavar='N', type=int, default=1,
+            help='Wait for N seconds until node becomes available (default %(default)s sec)')
 
     def main(self, *, args):  # noqa: D102
-        with NodeStrategy(args) as node:
-            node_names = get_node_names(
-                node=node, include_hidden_nodes=args.include_hidden_nodes)
-
         node_name = get_absolute_node_name(args.node_name)
-        if node_name not in {n.full_name for n in node_names}:
-            return 'Node not found'
+        with NodeStrategy(args) as node:
+            if not wait_for_node(node, node_name, args.include_hidden_nodes, args.timeout):
+                return 'Node not found'
 
         with DirectNode(args) as node:
             response = call_get_parameters(

ros2param/ros2param/verb/load.py

@@ -16,8 +16,8 @@
 from ros2cli.node.strategy import add_arguments
 from ros2cli.node.strategy import NodeStrategy
 from ros2node.api import get_absolute_node_name
-from ros2node.api import get_node_names
 from ros2node.api import NodeNameCompleter
+from ros2node.api import wait_for_node
 from ros2param.api import load_parameter_file
 from ros2param.verb import VerbExtension
 
@@ -39,15 +39,15 @@ def add_arguments(self, parser, cli_name):  # noqa: D102
         parser.add_argument(
             '--no-use-wildcard', action='store_true',
             help="Do not load parameters in the '/**' namespace into the node")
+        parser.add_argument(
+            '--timeout', metavar='N', type=int, default=1,
+            help='Wait for N seconds until node becomes available (default %(default)s sec)')
 
     def main(self, *, args):  # noqa: D102
-        with NodeStrategy(args) as node:
-            node_names = get_node_names(
-                node=node, include_hidden_nodes=args.include_hidden_nodes)
-
         node_name = get_absolute_node_name(args.node_name)
-        if node_name not in {n.full_name for n in node_names}:
-            return 'Node not found'
+        with NodeStrategy(args) as node:
+            if not wait_for_node(node, node_name, args.include_hidden_nodes, args.timeout):
+                return 'Node not found'
 
         with DirectNode(args) as node:
             load_parameter_file(node=node, node_name=node_name, parameter_file=args.parameter_file,

ros2param/ros2param/verb/set.py

@@ -20,8 +20,8 @@
 from ros2cli.node.strategy import add_arguments
 from ros2cli.node.strategy import NodeStrategy
 from ros2node.api import get_absolute_node_name
-from ros2node.api import get_node_names
 from ros2node.api import NodeNameCompleter
+from ros2node.api import wait_for_node
 
 from ros2param.api import call_set_parameters
 from ros2param.api import ParameterNameCompleter
@@ -45,15 +45,15 @@ def add_arguments(self, parser, cli_name):  # noqa: D102
         arg.completer = ParameterNameCompleter()
         parser.add_argument(
             'value', help='Value of the parameter')
+        parser.add_argument(
+            '--timeout', metavar='N', type=int, default=1,
+            help='Wait for N seconds until node becomes available (default %(default)s sec)')
 
     def main(self, *, args):  # noqa: D102
-        with NodeStrategy(args) as node:
-            node_names = get_node_names(
-                node=node, include_hidden_nodes=args.include_hidden_nodes)
-
         node_name = get_absolute_node_name(args.node_name)
-        if node_name not in {n.full_name for n in node_names}:
-            return 'Node not found'
+        with NodeStrategy(args) as node:
+            if not wait_for_node(node, node_name, args.include_hidden_nodes, args.timeout):
+                return 'Node not found'
 
         with DirectNode(args) as node:
             parameter = Parameter()

ros2param/test/test_verb_load.py

@@ -269,11 +269,23 @@ def test_verb_load_invalid_path(self):
             strict=False
         )
 
+    def test_verb_load_timeout(self):
+        with self.launch_param_load_command(
+            arguments=['invalid_node', 'invalid_path', '--timeout', '2']
+        ) as param_load_command:
+            assert param_load_command.wait_for_shutdown(timeout=TEST_TIMEOUT)
+        assert param_load_command.exit_code != launch_testing.asserts.EXIT_OK
+        assert launch_testing.tools.expect_output(
+            expected_lines=['Node not found'],
+            text=param_load_command.output,
+            strict=False
+        )
+
     def test_verb_load(self):
         with tempfile.TemporaryDirectory() as tmpdir:
             filepath = self._write_param_file(tmpdir, 'params.yaml')
             with self.launch_param_load_command(
-                arguments=[f'{TEST_NAMESPACE}/{TEST_NODE}', filepath]
+                arguments=[f'{TEST_NAMESPACE}/{TEST_NODE}', filepath, '--timeout', '3']
             ) as param_load_command:
                 assert param_load_command.wait_for_shutdown(timeout=TEST_TIMEOUT)
             assert param_load_command.exit_code == launch_testing.asserts.EXIT_OK
@@ -300,7 +312,7 @@ def test_verb_load_wildcard(self):
             filepath = self._write_param_file(tmpdir, 'params.yaml', INPUT_WILDCARD_PARAMETER_FILE)
             with self.launch_param_load_command(
                 arguments=[f'{TEST_NAMESPACE}/{TEST_NODE}', filepath,
-                           '--no-use-wildcard']
+                           '--no-use-wildcard', '--timeout', '3']
             ) as param_load_command:
                 assert param_load_command.wait_for_shutdown(timeout=TEST_TIMEOUT)
             assert param_load_command.exit_code != launch_testing.asserts.EXIT_OK