Add verbose info for topic list

ros2topic/ros2topic/verb/list.py

@@ -34,19 +34,46 @@ def add_arguments(self, parser, cli_name):
         parser.add_argument(
             '--include-hidden-topics', action='store_true',
             help='Consider hidden topics as well')
+        parser.add_argument(
+            '-v', '--verbose', action='store_true',
+            help='List full details about each topic')
+
+    @staticmethod
+    def show_topic_info(topic_info, is_publisher):
+        message = ('Published' if is_publisher else 'Subscribed') + ' topics:\n'
+        for (topic_name, topic_types, pub_count, sub_count) in topic_info:
+            count = pub_count if is_publisher else sub_count
+            if count:
+                topic_types_formatted = ', '.join(topic_types)
+                count_str = str(count) + ' ' + ('publisher' if is_publisher else 'subscriber') \
+                    + ('s' if count > 1 else '')
+                message += f' * {topic_name} [{topic_types_formatted}] {count_str}\n'
+        return message
 
     def main(self, *, args):
+        topic_info = []
         with NodeStrategy(args) as node:
             topic_names_and_types = get_topic_names_and_types(
                 node=node,
                 include_hidden_topics=args.include_hidden_topics)
+            for (topic_name, topic_types) in topic_names_and_types:
+                if args.verbose:
+                    pub_count = node.count_publishers(topic_name)
+                    sub_count = node.count_subscribers(topic_name)
+                    topic_info.append((topic_name, topic_types, pub_count, sub_count))
+                else:
+                    topic_info.append((topic_name, topic_types, 0, 0))
 
         if args.count_topics:
             print(len(topic_names_and_types))
         elif topic_names_and_types:
-            for (topic_name, topic_types) in topic_names_and_types:
-                msg = f'{topic_name}'
-                topic_types_formatted = ', '.join(topic_types)
-                if args.show_types:
-                    msg += f' [{topic_types_formatted}]'
-                print(msg)
+            if args.verbose:
+                print(self.show_topic_info(topic_info, is_publisher=True))
+                print(self.show_topic_info(topic_info, is_publisher=False))
+            else:
+                for (topic_name, topic_types, _, _) in topic_info:
+                    msg = '{topic_name}'
+                    topic_types_formatted = ', '.join(topic_types)
+                    if args.show_types:
+                        msg += ' [{topic_types_formatted}]'
+                    print(msg.format_map(locals()))

ros2topic/test/test_cli.py

@@ -254,6 +254,31 @@ def test_list_with_types(self):
             strict=True
         )
 
+    @launch_testing.markers.retry_on_failure(times=5, delay=1)
+    def test_list_with_verbose(self):
+        with self.launch_topic_command(arguments=['list', '-v']) as topic_command:
+            assert topic_command.wait_for_shutdown(timeout=10)
+        assert topic_command.exit_code == launch_testing.asserts.EXIT_OK
+        assert launch_testing.tools.expect_output(
+            expected_lines=[
+                'Published topics:',
+                ' * /arrays [test_msgs/msg/Arrays] 1 publisher',
+                ' * /bounded_sequences [test_msgs/msg/BoundedSequences] 1 publisher',
+                ' * /chatter [std_msgs/msg/String] 1 publisher',
+                ' * /cmd_vel [geometry_msgs/msg/TwistStamped] 1 publisher',
+                ' * /defaults [test_msgs/msg/Defaults] 1 publisher',
+                ' * /parameter_events [rcl_interfaces/msg/ParameterEvent] 9 publishers',
+                ' * /rosout [rcl_interfaces/msg/Log] 9 publishers',
+                ' * /unbounded_sequences [test_msgs/msg/UnboundedSequences] 1 publisher',
+                '',
+                'Subscribed topics:',
+                ' * /chit_chatter [std_msgs/msg/String] 1 subscriber',
+                '',
+            ],
+            text=topic_command.output,
+            strict=True
+        )
+
     @launch_testing.markers.retry_on_failure(times=5, delay=1)
     def test_list_count(self):
         with self.launch_topic_command(arguments=['list', '-c']) as topic_command: