fix

docs/libraries.rst

@@ -1261,11 +1261,6 @@ Check the latency of the specified topic (in system time). If the check with ``c
      - ``string``
      - 
      - Class of message type, only required when 'wait_for_first_message' is set to false (e.g. ``std_msgs.msg.String``)
-   * - ``type_check``
-     - ``bool``
-     - ``true``
-     - If False, the action will not verify that the topic type matches the specified type.
-
 
 ``bag_play()``
 ^^^^^^^^^^^^^^^

scenario_execution_ros/scenario_execution_ros/actions/assert_topic_latency.py

@@ -18,19 +18,18 @@
 from collections import deque
 import py_trees  # pylint: disable=import-error
 from rclpy.node import Node
-import importlib
 import time
 from scenario_execution_ros.actions.conversions import get_comparison_operator, get_qos_preset_profile
 from scenario_execution.actions.base_action import BaseAction, ActionError
+from rosidl_runtime_py.utilities import get_message
 
 
 class AssertTopicLatency(BaseAction):
 
-    def __init__(self, topic_name: str, topic_type: str, wait_for_first_message: bool, type_check: bool):
+    def __init__(self, topic_name: str, topic_type: str, wait_for_first_message: bool):
         super().__init__()
         self.topic_name = topic_name
         self.topic_type = topic_type
-        self.type_check = type_check
         self.latency = None
         self.comparison_operator_feedback = None
         self.comparison_operator = None
@@ -129,12 +128,11 @@ def check_topic(self):
                     if not topic_types:
                         result = False
                         break
-                    current_topic_type = topic_types[0].replace('/', '.')
                     if not self.topic_type:
-                        self.topic_type = current_topic_type  # 'topic_type' is not specified, the process will continue with the found one
+                        self.topic_type = topic_types[0]  # 'topic_type' is not specified, the process will continue with the found one
                         self.call_subscriber()
                         self.is_topic = True
-                    elif self.type_check and self.topic_type != current_topic_type:
+                    elif self.topic_type != topic_types[0]:
                         result = False
                     else:
                         self.call_subscriber()
@@ -145,17 +143,16 @@ def check_topic(self):
         return result
 
     def call_subscriber(self):
-        datatype_in_list = self.topic_type.split(".")
-        topic_type = getattr(
-            importlib.import_module(".".join(datatype_in_list[:-1])),
-            datatype_in_list[-1]
-        )
-
+        topic_type = self.topic_type.replace('.', '/')
+        message_class = get_message(topic_type)
+        if message_class is None:
+            raise ValueError(f"Message type '{topic_type}' could not be found.")
         self.subscription = self.node.create_subscription(
-            msg_type=topic_type,
+            msg_type=message_class,
             topic=self.topic_name,
             callback=self._callback,
-            qos_profile=get_qos_preset_profile(['sensor_data']))
+            qos_profile=get_qos_preset_profile(['sensor_data'])
+        )
 
     def _callback(self, msg):
         self.first_message_received = True

scenario_execution_ros/scenario_execution_ros/lib_osc/ros.osc

@@ -70,7 +70,6 @@ action assert_topic_latency:
     rolling_average_count: int = 1       # the check is done aganist the rolling average over x elements
     wait_for_first_message: bool = true  # start measuring with the first received message
     topic_type: string                   # class of message type, only required when wait_for_first_message is set to false (e.g. std_msgs.msg.String)
-    type_check: bool = true              # If False, the action will not verify that the topic type matches the specified type.
 
 action bag_play:
     # play back a ros bag