Add ros2 service info command

ros2service/ros2service/api/__init__.py

@@ -12,13 +12,69 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from rclpy.expand_topic_name import expand_topic_name
 from rclpy.topic_or_service_is_hidden import topic_or_service_is_hidden
+from rclpy.validate_full_topic_name import validate_full_topic_name
 from ros2cli.node.strategy import NodeStrategy
 from rosidl_runtime_py import get_service_interfaces
 from rosidl_runtime_py import message_to_yaml
 from rosidl_runtime_py.utilities import get_service
 
 
+def get_service_clients_and_servers(*, node, service_name):
+    service_clients = []
+    service_servers = []
+
+    expanded_name = expand_topic_name(service_name, node.get_name(), node.get_namespace())
+    validate_full_topic_name(expanded_name)
+
+    node_names_and_ns = node.get_node_names_and_namespaces()
+    for node_name, node_ns in node_names_and_ns:
+        node_fqn = '/'.join(node_ns) + node_name
+
+        client_names_and_types = get_client_names_and_types_for_node(
+            node=node,
+            name=node_name,
+            namespace=node_ns,
+            include_hidden_services=True
+        )
+
+        for name, types in client_names_and_types:
+            if name == expanded_name:
+                service_clients.append((node_fqn, types))
+
+        service_names_and_types = get_service_names_and_types_for_node(
+            node=node,
+            name=node_name,
+            namespace=node_ns,
+            include_hidden_services=True
+        )
+
+        for name, types in service_names_and_types:
+            if name == expanded_name:
+                service_servers.append((node_fqn, types))
+
+    return service_clients, service_servers
+
+
+def get_client_names_and_types_for_node(*, node, name, namespace, include_hidden_services=False):
+    client_names_and_types = node.get_client_names_and_types_by_node(name, namespace)
+    if not include_hidden_services:
+        client_names_and_types = [
+            (n, t) for (n, t) in client_names_and_types
+            if not topic_or_service_is_hidden(n)]
+    return client_names_and_types
+
+
+def get_service_names_and_types_for_node(*, node, name, namespace, include_hidden_services=False):
+    service_names_and_types = node.get_service_names_and_types_by_node(name, namespace)
+    if not include_hidden_services:
+        service_names_and_types = [
+            (n, t) for (n, t) in service_names_and_types
+            if not topic_or_service_is_hidden(n)]
+    return service_names_and_types
+
+
 def get_service_names_and_types(*, node, include_hidden_services=False):
     service_names_and_types = node.get_service_names_and_types()
     if not include_hidden_services:

ros2service/ros2service/verb/info.py

@@ -0,0 +1,61 @@
+# Copyright 2022 PickNik Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ros2cli.node.strategy import NodeStrategy
+from ros2service.api import get_service_clients_and_servers
+from ros2service.api import ServiceNameCompleter
+from ros2service.verb import VerbExtension
+
+
+class InfoVerb(VerbExtension):
+    """Print information about a service."""
+
+    def add_arguments(self, parser, cli_name):
+        arg = parser.add_argument(
+            'service_name',
+            help="Name of the ROS service to print info about (e.g. '/talker/list_parameters')")
+        arg.completer = ServiceNameCompleter(
+            include_hidden_services_key='include_hidden_services')
+        parser.add_argument(
+            '-t', '--show-types', action='store_true',
+            help='Additionally show the service type')
+        parser.add_argument(
+            '-c', '--count', action='store_true',
+            help='Only display the number of service clients and service servers')
+
+    def main(self, *, args):
+        with NodeStrategy(args) as node:
+
+            service_clients, service_servers = get_service_clients_and_servers(
+                node=node,
+                service_name=args.service_name
+            )
+
+            print('Service: {}'.format(args.service_name))
+            print('Service clients: {}'.format(len(service_clients)))
+            if not args.count:
+                for client_name, client_types in service_clients:
+                    if args.show_types:
+                        types_formatted = ', '.join(client_types)
+                        print(f'    {client_name} [{types_formatted}]')
+                    else:
+                        print(f'    {client_name}')
+            print('Service servers: {}'.format(len(service_servers)))
+            if not args.count:
+                for server_name, server_types in service_servers:
+                    if args.show_types:
+                        types_formatted = ', '.join(server_types)
+                        print(f'    {server_name} [{types_formatted}]')
+                    else:
+                        print(f'    {server_name}')

ros2service/setup.py

@@ -42,6 +42,7 @@
         'ros2service.verb': [
             'call = ros2service.verb.call:CallVerb',
             'find = ros2service.verb.find:FindVerb',
+            'info = ros2service.verb.info:InfoVerb',
             'list = ros2service.verb.list:ListVerb',
             'type = ros2service.verb.type:TypeVerb',
         ],

ros2service/test/fixtures/echo_client.py

@@ -0,0 +1,49 @@
+# Copyright 2022 PickNik, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+import rclpy
+from rclpy.node import Node
+
+from test_msgs.srv import BasicTypes
+
+
+class EchoClient(Node):
+
+    def __init__(self):
+        super().__init__('echo_server')
+        self.client = self.create_client(BasicTypes, 'echo')
+
+
+def main(args=None):
+    rclpy.init(args=args)
+
+    node = EchoClient()
+    try:
+        rclpy.spin(node)
+    except KeyboardInterrupt:
+        print('server stopped cleanly')
+    except BaseException:
+        print('exception in server:', file=sys.stderr)
+        raise
+    finally:
+        # Destroy the node explicitly
+        # (optional - Done automatically when node is garbage collected)
+        node.destroy_node()
+        rclpy.shutdown()
+
+
+if __name__ == '__main__':
+    main()

ros2service/test/test_cli.py

@@ -68,6 +68,9 @@ def generate_test_description(rmw_implementation):
     path_to_echo_server_script = os.path.join(
         os.path.dirname(__file__), 'fixtures', 'echo_server.py'
     )
+    path_to_echo_client_script = os.path.join(
+        os.path.dirname(__file__), 'fixtures', 'echo_client.py'
+    )
     additional_env = {'RMW_IMPLEMENTATION': rmw_implementation}
     return LaunchDescription([
         # Always restart daemon to isolate tests.
@@ -95,6 +98,13 @@ def generate_test_description(rmw_implementation):
                             remappings=[('echo', '_echo')],
                             additional_env=additional_env,
                         ),
+                        Node(
+                            executable=sys.executable,
+                            arguments=[path_to_echo_client_script],
+                            name='echo_client',
+                            namespace='my_ns',
+                            additional_env=additional_env,
+                        ),
                         launch_testing.actions.ReadyToTest()
                     ],
                     additional_env=additional_env
@@ -146,9 +156,12 @@ def test_list_services(self):
         assert launch_testing.tools.expect_output(
             expected_lines=itertools.chain(
                 ['/my_ns/echo'],
+                itertools.repeat(re.compile(
+                    r'/my_ns/echo_client/.*parameter.*'
+                ), 6),
                 itertools.repeat(re.compile(
                     r'/my_ns/echo_server/.*parameter.*'
-                ), 6)
+                ), 6),
             ),
             text=service_command.output,
             strict=True
@@ -168,6 +181,9 @@ def test_list_hidden(self):
                     r'/my_ns/_hidden_echo_server/.*parameter.*'
                 ), 6),
                 ['/my_ns/echo'],
+                itertools.repeat(re.compile(
+                    r'/my_ns/echo_client/.*parameter.*'
+                ), 6),
                 itertools.repeat(re.compile(
                     r'/my_ns/echo_server/.*parameter.*'
                 ), 6)
@@ -184,10 +200,14 @@ def test_list_with_types(self):
         assert launch_testing.tools.expect_output(
             expected_lines=itertools.chain(
                 ['/my_ns/echo [test_msgs/srv/BasicTypes]'],
+                itertools.repeat(re.compile(
+                    r'/my_ns/echo_client/.*parameter.*'
+                    r' \[rcl_interfaces/srv/.*Parameter.*\]'
+                ), 6),
                 itertools.repeat(re.compile(
                     r'/my_ns/echo_server/.*parameter.*'
                     r' \[rcl_interfaces/srv/.*Parameter.*\]'
-                ), 6)
+                ), 6),
             ),
             text=service_command.output,
             strict=True
@@ -200,7 +220,58 @@ def test_list_count(self):
         assert service_command.exit_code == launch_testing.asserts.EXIT_OK
         output_lines = service_command.output.splitlines()
         assert len(output_lines) == 1
-        assert int(output_lines[0]) == 7
+        assert int(output_lines[0]) == 13
+
+    @launch_testing.markers.retry_on_failure(times=5, delay=1)
+    def test_info(self):
+        with self.launch_service_command(arguments=['info', '/my_ns/echo']) as service_command:
+            assert service_command.wait_for_shutdown(timeout=10)
+        assert service_command.exit_code == launch_testing.asserts.EXIT_OK
+        assert launch_testing.tools.expect_output(
+            expected_lines=[
+                'Service: /my_ns/echo',
+                'Service clients: 1',
+                '    /my_ns/echo_client',
+                'Service servers: 1',
+                '    /my_ns/echo_server',
+            ],
+            text=service_command.output,
+            strict=True
+        )
+
+    def test_info_types(self):
+        with self.launch_service_command(
+                arguments=['info', '-t', '/my_ns/echo']
+        ) as service_command:
+            assert service_command.wait_for_shutdown(timeout=10)
+        assert service_command.exit_code == launch_testing.asserts.EXIT_OK
+        assert launch_testing.tools.expect_output(
+            expected_lines=[
+                'Service: /my_ns/echo',
+                'Service clients: 1',
+                '    /my_ns/echo_client [test_msgs/srv/BasicTypes]',
+                'Service servers: 1',
+                '    /my_ns/echo_server [test_msgs/srv/BasicTypes]',
+            ],
+            text=service_command.output,
+            strict=True
+        )
+
+    def test_info_count(self):
+        with self.launch_service_command(
+                arguments=['info', '-c', '/my_ns/echo']
+        ) as service_command:
+            assert service_command.wait_for_shutdown(timeout=10)
+        assert service_command.exit_code == launch_testing.asserts.EXIT_OK
+        assert launch_testing.tools.expect_output(
+            expected_lines=[
+                'Service: /my_ns/echo',
+                'Service clients: 1',
+                'Service servers: 1',
+            ],
+            text=service_command.output,
+            strict=True
+        )
 
     @launch_testing.markers.retry_on_failure(times=5, delay=1)
     def test_find(self):