[ros2topic] Use import message logic provided by rosidl_runtime_py

ros2topic/ros2topic/api/__init__.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 
 import argparse
-import importlib
 
 from time import sleep
 
@@ -57,24 +56,6 @@ def __call__(self, prefix, parsed_args, **kwargs):
                     parsed_args, self.include_hidden_topics_key))
 
 
-def import_message_type(topic_name, message_type):
-    # TODO(dirk-thomas) this logic should come from a rosidl related package
-    try:
-        package_name, *message_name = message_type.split('/')
-        if not package_name or not message_name or not all(message_name):
-            raise ValueError()
-    except ValueError:
-        raise RuntimeError('The passed message type is invalid')
-
-    # TODO(sloretz) node API to get topic types should indicate if action or msg
-    middle_module = 'msg'
-    if topic_name.endswith('/_action/feedback'):
-        middle_module = 'action'
-
-    module = importlib.import_module(package_name + '.' + middle_module)
-    return getattr(module, message_name[-1])
-
-
 def message_type_completer(**kwargs):
     """Callable returning a list of message types."""
     message_types = []
@@ -149,7 +130,7 @@ def _get_msg_class(node, topic, include_hidden_topics):
         # Could not determine the type for the passed topic
         return None
 
-    return import_message_type(topic, message_type)
+    return get_message(message_type)
 
 
 class TopicMessagePrototypeCompleter:

ros2topic/ros2topic/verb/echo.py

@@ -26,12 +26,12 @@
 from ros2cli.node.direct import DirectNode
 from ros2topic.api import add_qos_arguments_to_argument_parser
 from ros2topic.api import get_topic_names_and_types
-from ros2topic.api import import_message_type
 from ros2topic.api import qos_profile_from_short_keys
 from ros2topic.api import TopicNameCompleter
 from ros2topic.verb import VerbExtension
 from rosidl_runtime_py import message_to_csv
 from rosidl_runtime_py import message_to_yaml
+from rosidl_runtime_py.utilities import get_message
 
 DEFAULT_TRUNCATE_LENGTH = 128
 MsgType = TypeVar('MsgType')
@@ -58,7 +58,7 @@ def add_arguments(self, parser, cli_name):
             include_hidden_topics_key='include_hidden_topics')
         parser.add_argument(
             'message_type', nargs='?',
-            help="Type of the ROS message (e.g. 'std_msgs/String')")
+            help="Type of the ROS message (e.g. 'std_msgs/msg/String')")
         add_qos_arguments_to_argument_parser(
             parser, is_publisher=False, default_preset='sensor_data')
         parser.add_argument(
@@ -128,7 +128,7 @@ def subscriber(
             raise RuntimeError(
                 'Could not determine the type for the passed topic')
 
-    msg_module = import_message_type(topic_name, message_type)
+    msg_module = get_message(message_type)
 
     node.create_subscription(
         msg_module, topic_name, callback, qos_profile)

ros2topic/ros2topic/verb/pub.py

@@ -21,13 +21,13 @@
 from rclpy.qos import QoSProfile
 from ros2cli.node.direct import DirectNode
 from ros2topic.api import add_qos_arguments_to_argument_parser
-from ros2topic.api import import_message_type
 from ros2topic.api import qos_profile_from_short_keys
 from ros2topic.api import TopicMessagePrototypeCompleter
 from ros2topic.api import TopicNameCompleter
 from ros2topic.api import TopicTypeCompleter
 from ros2topic.verb import VerbExtension
 from rosidl_runtime_py import set_message_fields
+from rosidl_runtime_py.utilities import get_message
 import yaml
 
 MsgType = TypeVar('MsgType')
@@ -102,7 +102,7 @@ def publisher(
     qos_profile: QoSProfile,
 ) -> Optional[str]:
     """Initialize a node with a single publisher and run its publish loop (maybe only once)."""
-    msg_module = import_message_type(topic_name, message_type)
+    msg_module = get_message(message_type)
     values_dictionary = yaml.safe_load(values)
     if not isinstance(values_dictionary, dict):
         return 'The passed value needs to be a dictionary in YAML format'

ros2topic/test/fixtures/repeater_node.py

@@ -19,8 +19,7 @@
 from rclpy.node import Node
 from rclpy.qos import qos_profile_system_default
 from rclpy.utilities import remove_ros_args
-
-from ros2topic.api import import_message_type
+from rosidl_runtime_py.utilities import get_message
 
 
 class RepeaterNode(Node):
@@ -37,14 +36,10 @@ def callback(self):
         self.pub.publish(self.message_type())
 
 
-def message_type(message_typename):
-    return import_message_type('~/output', message_typename)
-
-
 def parse_arguments(args=None):
     parser = argparse.ArgumentParser()
     parser.add_argument(
-        'message_type', type=message_type,
+        'message_type', type=get_message,
         help='Message type for the repeater to publish.'
     )
     return parser.parse_args(args=remove_ros_args(args))