Move rosidl implementation details to rosidl_runtime_py

ros2interface/ros2interface/api/__init__.py

@@ -11,159 +11,37 @@
 # 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 os
-
-from ament_index_python import get_resource
-from ament_index_python import get_resources
-from ament_index_python import has_resource
 
+from rosidl_runtime_py import get_action_interfaces
+from rosidl_runtime_py import get_interface_packages
+from rosidl_runtime_py import get_message_interfaces
+from rosidl_runtime_py import get_service_interfaces
+from rosidl_runtime_py import message_to_yaml
 from rosidl_runtime_py import utilities
-from rosidl_runtime_py.convert import message_to_yaml
-
-
-def get_all_interface_packages():
-    return get_resources('rosidl_interfaces')
-
-
-def get_interfaces(package_name):
-    if not has_resource('packages', package_name):
-        raise LookupError('Unknown package {}'.format(package_name))
-    try:
-        content, _ = get_resource('rosidl_interfaces', package_name)
-    except LookupError:
-        return []
-    interface_names = content.splitlines()
-    return list(sorted({
-        n.rsplit('.', 1)[0]
-        for n in interface_names
-        if '_' not in n}))
-
-
-def get_interface_path(parts):
-    prefix_path = has_resource('packages', parts[0])
-    if not prefix_path:
-        raise LookupError('Unknown package {}'.format(parts[0]))
-    joined = '/'.join(parts)
-    if len(parts[-1].rsplit('.', 1)) == 1:
-        joined += '.idl'
-    interface_path = os.path.join(
-        prefix_path, 'share', joined)
-    if not os.path.exists(interface_path):
-        raise LookupError('Could not find the interface {!r}'.format(interface_path))
-    return interface_path
 
 
 def package_name_completer(**kwargs):
-    """Callable returning a list of types containing messages, services, and action."""
-    return get_all_interface_packages()
+    """Callable returning a list of packages containing messages, services, and action."""
+    return get_interface_packages()
 
 
 def type_completer(**kwargs):
     """Callable returning a list of message, service, and action types."""
     types = []
-    for package_name, service_names in get_all_service_types().items():
+    for package_name, service_names in get_service_interfaces().items():
         for service_name in service_names:
-            types.append(
-                '{package_name}/srv/{service_name}'.format_map(locals()))
+            types.append(f'{package_name}/{service_name}')
 
-    for package_name, message_names in get_all_message_types().items():
+    for package_name, message_names in get_message_interfaces().items():
         for message_name in message_names:
-            types.append(
-                '{package_name}/msg/{message_name}'.format_map(locals()))
+            types.append(f'{package_name}/{message_name}')
 
-    for package_name, action_names in get_all_action_types().items():
+    for package_name, action_names in get_action_interfaces().items():
         for action_name in action_names:
-            types.append(
-                '{package_name}/action/{action_name}'.format_map(locals()))
-
+            types.append(f'{package_name}/{action_name}')
     return sorted(types)
 
 
-def get_all_action_types():
-    all_action_types = {}
-    for package_name in get_resources('rosidl_interfaces'):
-        action_types = get_action_types(package_name)
-        if action_types:
-            all_action_types[package_name] = action_types
-    return all_action_types
-
-
-def get_action_types(package_name):
-    if not has_resource('packages', package_name):
-        raise LookupError('Unknown package name')
-    try:
-        content, _ = get_resource('rosidl_interfaces', package_name)
-    except LookupError:
-        return []
-    interface_names = content.splitlines()
-    # TODO(jacobperron) this logic should come from a rosidl related package
-    # Only return actions in action folder
-    return list(sorted({
-        n[7:].rsplit('.', 1)[0]
-        for n in interface_names
-        if n.startswith('action/') and (n[-4:] == '.idl' or n[-7:] == '.action')}))
-
-
-def get_all_message_types():
-    all_message_types = {}
-    for package_name in get_resources('rosidl_interfaces'):
-        message_types = get_message_types(package_name)
-        if message_types:
-            all_message_types[package_name] = message_types
-    return all_message_types
-
-
-def get_message_types(package_name):
-    if not has_resource('packages', package_name):
-        raise LookupError('Unknown package name')
-    try:
-        content, _ = get_resource('rosidl_interfaces', package_name)
-    except LookupError:
-        return []
-    interface_names = content.splitlines()
-    # TODO(dirk-thomas) this logic should come from a rosidl related package
-    # Only return messages in msg folder
-    return list(sorted({
-        n[4:-4]
-        for n in interface_names
-        if n.startswith('msg/') and n[-4:] in ('.idl', '.msg')}))
-
-
-def get_all_service_types():
-    all_service_types = {}
-    for package_name in get_resources('rosidl_interfaces'):
-        service_types = get_service_types(package_name)
-        if service_types:
-            all_service_types[package_name] = service_types
-    return all_service_types
-
-
-def get_service_types(package_name):
-    if not has_resource('packages', package_name):
-        raise LookupError('Unknown package name')
-    try:
-        content, _ = get_resource('rosidl_interfaces', package_name)
-    except LookupError:
-        return []
-    interface_names = content.splitlines()
-    # TODO(dirk-thomas) this logic should come from a rosidl related package
-    # Only return services in srv folder
-    return list(sorted({
-        n[4:-4]
-        for n in interface_names
-        if n.startswith('srv/') and n[-4:] in ('.idl', '.srv')}))
-
-
-def get_message_path(package_name, message_name):
-    message_types = get_message_types(package_name)
-    if message_name not in message_types:
-        raise LookupError('Unknown message name')
-    prefix_path = has_resource('packages', package_name)
-    # TODO(dirk-thomas) this logic should come from a rosidl related package
-    return os.path.join(
-        prefix_path, 'share', package_name, 'msg', message_name + '.msg')
-
-
 def interface_to_yaml(identifier):
     interface = utilities.get_interface(identifier)
     if utilities.is_action(interface):

ros2interface/ros2interface/verb/list.py

@@ -12,34 +12,34 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ros2interface.api import get_all_action_types
-from ros2interface.api import get_all_message_types
-from ros2interface.api import get_all_service_types
 from ros2interface.verb import VerbExtension
+from rosidl_runtime_py import get_action_interfaces
+from rosidl_runtime_py import get_message_interfaces
+from rosidl_runtime_py import get_service_interfaces
 
 
 def print_messages():
     print('Messages:')
-    message_types = get_all_message_types()
-    for package_name in sorted(message_types):
-        for message_name in sorted(message_types[package_name]):
-            print('    {package_name}/msg/{message_name}'.format_map(locals()))
+    message_interfaces = get_message_interfaces()
+    for package_name in sorted(message_interfaces):
+        for message_name in sorted(message_interfaces[package_name]):
+            print(f'    {package_name}/{message_name}')
 
 
 def print_services():
     print('Services:')
-    service_types = get_all_service_types()
-    for package_name in sorted(service_types):
-        for service_name in sorted(service_types[package_name]):
-            print('    {package_name}/srv/{service_name}'.format_map(locals()))
+    service_interfaces = get_service_interfaces()
+    for package_name in sorted(service_interfaces):
+        for service_name in sorted(service_interfaces[package_name]):
+            print(f'    {package_name}/{service_name}')
 
 
 def print_actions():
     print('Actions:')
-    action_types = get_all_action_types()
-    for package_name in sorted(action_types):
-        for action_name in sorted(action_types[package_name]):
-            print('    {package_name}/action/{action_name}'.format_map(locals()))
+    action_interfaces = get_action_interfaces()
+    for package_name in sorted(action_interfaces):
+        for action_name in sorted(action_interfaces[package_name]):
+            print(f'    {package_name}/{action_name}')
 
 
 class ListVerb(VerbExtension):

ros2interface/ros2interface/verb/package.py

@@ -12,9 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ros2interface.api import get_interfaces
 from ros2interface.api import package_name_completer
 from ros2interface.verb import VerbExtension
+from rosidl_runtime_py import get_interfaces
 
 
 class PackageVerb(VerbExtension):
@@ -28,8 +28,9 @@ def add_arguments(self, parser, cli_name):
 
     def main(self, *, args):
         try:
-            names = get_interfaces(args.package_name)
+            interfaces = get_interfaces([args.package_name])
         except LookupError as e:
             return str(e)
-        for name in names:
-            print('{args.package_name}/{name}'.format_map(locals()))
+        for package_name in sorted(interfaces):
+            for interface_name in interfaces[package_name]:
+                print(f'{package_name}/{interface_name}')

ros2interface/ros2interface/verb/packages.py

@@ -12,11 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ros2interface.api import get_all_action_types
-from ros2interface.api import get_all_interface_packages
-from ros2interface.api import get_all_message_types
-from ros2interface.api import get_all_service_types
 from ros2interface.verb import VerbExtension
+from rosidl_runtime_py import get_action_interfaces
+from rosidl_runtime_py import get_interface_packages
+from rosidl_runtime_py import get_message_interfaces
+from rosidl_runtime_py import get_service_interfaces
 
 
 def print_packages(packs):
@@ -39,11 +39,10 @@ def add_arguments(self, parser, cli_name):
 
     def main(self, *, args):
         if args.only_msgs:
-            print_packages(get_all_message_types().keys())
+            print_packages(sorted(get_message_interfaces()))
         elif args.only_srvs:
-            print_packages(get_all_service_types().keys())
+            print_packages(sorted(get_service_interfaces()))
         elif args.only_actions:
-            print_packages(get_all_action_types().keys())
+            print_packages(sorted(get_action_interfaces()))
         else:
-            all_interface_packages = get_all_interface_packages()
-            print_packages(all_interface_packages)
+            print_packages(sorted(get_interface_packages()))

ros2interface/ros2interface/verb/show.py

@@ -12,9 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ros2interface.api import get_interface_path
 from ros2interface.api import type_completer
 from ros2interface.verb import VerbExtension
+from rosidl_runtime_py import get_interface_path
 
 
 class ShowVerb(VerbExtension):
@@ -27,17 +27,6 @@ def add_arguments(self, parser, cli_name):
         arg.completer = type_completer
 
     def main(self, *, args):
-        # TODO(kucheria) this logic should come from a rosidl related package
-        try:
-            parts = args.type.split('/')
-            if len(parts) < 2:
-                raise ValueError()
-            if not all(parts):
-                raise ValueError()
-            file_path = get_interface_path(parts)
-        except ValueError:
-            raise RuntimeError('The passed interface type is invalid')
-        except LookupError as e:
-            return str(e)
+        file_path = get_interface_path(args.type)
         with open(file_path, 'r', encoding='utf-8') as h:
             print(h.read().rstrip())

ros2service/package.xml

@@ -13,7 +13,6 @@
   <depend>ros2cli</depend>
 
   <exec_depend>python3-yaml</exec_depend>
-  <exec_depend>ros2srv</exec_depend>
   <exec_depend>rosidl_runtime_py</exec_depend>
 
   <test_depend>ament_copyright</test_depend>

ros2service/ros2service/api/__init__.py

@@ -14,8 +14,8 @@
 
 from rclpy.topic_or_service_is_hidden import topic_or_service_is_hidden
 from ros2cli.node.strategy import NodeStrategy
-from ros2srv.api import service_type_completer
-from rosidl_runtime_py.convert import message_to_yaml
+from rosidl_runtime_py import get_service_interfaces
+from rosidl_runtime_py import message_to_yaml
 from rosidl_runtime_py.utilities import get_service
 
 
@@ -34,6 +34,15 @@ def get_service_names(*, node, include_hidden_services=False):
     return [n for (n, t) in service_names_and_types]
 
 
+def service_type_completer(**kwargs):
+    """Callable returning a list of service types."""
+    service_types = []
+    for package_name, service_names in get_service_interfaces().items():
+        for service_name in service_names:
+            service_types.append(f'{package_name}/{service_name}')
+    return service_types
+
+
 class ServiceNameCompleter:
     """Callable returning a list of service names."""