Add interface type filters to ros2 interface package

ros2interface/ros2interface/verb/package.py

@@ -21,16 +21,41 @@ class PackageVerb(VerbExtension):
     """Output a list of available interface types within one package."""
 
     def add_arguments(self, parser, cli_name):
+        parser.add_argument(
+            '-m', '--only-msgs', action='store_true',
+            help='Print out only the message types')
+
+        parser.add_argument(
+            '-s', '--only-srvs', action='store_true',
+            help='Print out only the service types')
+
+        parser.add_argument(
+            '-a', '--only-actions', action='store_true',
+            help='Print out only the action types')
+
         arg = parser.add_argument(
             'package_name',
             help="Name of the ROS package (e.g. 'example_interfaces')")
         arg.completer = package_name_completer
 
     def main(self, *, args):
+        types_to_filter = []
+        if args.only_msgs or args.only_srvs or args.only_actions:
+            if not args.only_msgs:
+                types_to_filter.append('msg')
+            if not args.only_srvs:
+                types_to_filter.append('srv')
+            if not args.only_actions:
+                types_to_filter.append('action')
+
         try:
             interfaces = get_interfaces([args.package_name])
         except LookupError as e:
             return str(e)
         for package_name in sorted(interfaces):
             for interface_name in interfaces[package_name]:
+                if types_to_filter:
+                    interface_type = interface_name[:interface_name.index('/')]
+                    if interface_type in types_to_filter:
+                        continue
                 print(f'{package_name}/{interface_name}')

ros2interface/test/test_cli.py

@@ -200,6 +200,22 @@ def test_package_on_test_msgs(self):
         )
         assert all(interface in output_lines for interface in some_interfaces)
 
+    def test_package_on_test_msgs_only_srvs(self):
+        with self.launch_interface_command(
+            arguments=['package', 'test_msgs', '-s']
+        ) as interface_command:
+            assert interface_command.wait_for_shutdown(timeout=2)
+        assert interface_command.exit_code == launch_testing.asserts.EXIT_OK
+        output_lines = interface_command.output.splitlines()
+        assert launch_testing.tools.expect_output(
+            expected_lines=itertools.repeat(
+                re.compile(r'test_msgs/srv/[A-z0-9_]+'), len(output_lines)
+            ),
+            lines=output_lines,
+            strict=True
+        )
+        assert all(interface in output_lines for interface in some_services_from_test_msgs)
+
     def test_packages(self):
         with self.launch_interface_command(arguments=['packages']) as interface_command:
             assert interface_command.wait_for_shutdown(timeout=2)