### ROS2 Python订阅Topic消息的时间戳测量 在ROS2中，可以通过`rclpy`库来实现Python节点的创建和运行。为了测量订阅消息的延迟是否超过100毫秒，可以利用接收到的消息中的时间戳字段与当前接收时刻进行比较。 以下是具体的实现方法： #### 实现思路 1. 创建一个ROS2节点。 2. 使用`Subscription`对象订阅目标话题。 3. 接收每条消息时，获取其头部的时间戳（通常位于`header.stamp.sec`和`header.stamp.nanosec`）。 4. 将此时间戳转换为秒级表示形式，并与当前系统时间对比。 5. 如果差值大于等于100毫秒，则打印警告信息。 #### 示例代码 以下是一个完整的示例代码片段，用于检测订阅到的消息是否有超过100毫秒的延迟： ```python import rclpy from rclpy.node import Node from std_msgs.msg import String import time class DelayMonitorNode(Node): def __init__(self): super().__init__('delay_monitor_node') self.subscription_ = self.create_subscription( String, 'chatter', self.listener_callback, 10) self.subscription_ # prevent unused variable warning def listener_callback(self, msg): current_time = time.time() # 当前系统时间 (秒) # 提取消息中的时间戳 timestamp_seconds = msg.header.stamp.sec + msg.header.stamp.nanosec / 1e9 delay_ms = (current_time - timestamp_seconds) * 1000 # 转换为毫秒 if delay_ms >= 100: self.get_logger().warn(f'Delay detected! Message delayed by {delay_ms:.2f} ms.') else: self.get_logger().info(f'Message received with a delay of {delay_ms:.2f} ms.') def main(args=None): rclpy.init(args=args) node = DelayMonitorNode() rclpy.spin(node) node.destroy_node() rclpy.shutdown() if __name__ == '__main__': main() ``` #### 关键点说明 - **时间戳解析**：通过`msg.header.stamp.sec`和`msg.header.stamp.nanosec`提取发送方的时间戳[^1]。 - **延迟计算**：将时间戳转换为浮点型秒数并与当前时间做减法运算得到延迟值。 - **日志记录**：如果延迟超出阈值（即100毫秒），则触发警告日志；否则仅记录正常延迟信息。 #### 注意事项 - 需要确保发布者发布的消息确实带有有效的时间戳字段 `header.stamp`。如果没有设置该字段，则无法完成精确的延迟测量。 - 上述例子假设使用的标准消息类型为`std_msgs/String`，实际应用可能涉及其他自定义消息类型，请根据具体需求调整相应部分。 --- ###