15.ROS2 time related API

1. Introduction to Time-Related APIs

ROS2's time-related APIs include Rate, Time, Duration, and operations on Time and Duration. These are explained below.

• First, create a function package to store the relevant program files.

xxxxxxxxxx
ros2 pkg create learning_time --build-type ament_python --dependencies rclpy

2. create_rate

ROS2 also provides the create_rate function, a tool for controlling loop execution frequency. Its core function is to periodically execute a section of code at a fixed frequency. Rate ensures the stability of the loop execution frequency by controlling the "sleep time" of the loop. Remember that Rate should generally not be used directly in the main thread, as doing so will permanently block callback events. It is generally only used in programs with multi-threaded callbacks or in child threads.

How it works:

1. Records the start time of each loop;
2. Executes the code within the loop;
3. Calculates the difference between the current loop's actual duration and the target interval;
4. Automatically sleeps for the corresponding difference, ensuring that the interval from the start of one loop to the start of the next strictly equals the target interval (e.g., 100 milliseconds).

While both Rate and Timer can implement periodic execution, their application scenarios differ:

• The following is a basic usage of Rate to implement a loop that executes twice per second:
• Create a new file in the function package called rate_demo.py

xxxxxxxxxx
import rclpy
from rclpy.node import Node
import threading
​
class RateExampleNode(Node):
    def __init__(self):
        super().__init__("rate_example_node")
        self.get_logger().info("Rate 示例节点启动")
​
    def run_loop(self):
        # Use the node's create_rate() to create a 2Hz Rate
        rate = self.create_rate(2)
        
        count = 0
        try:
            while rclpy.ok():
                self.get_logger().info(f"循环执行 {count} 次")
                count += 1
                rate.sleep()  # Sleep until the next cycle (0.5 seconds)
        except KeyboardInterrupt:
            self.get_logger().info("循环被中断")
​
def main(args=None):
    rclpy.init(args=args)
    node = RateExampleNode()
    
    # Create a thread run loop (to avoid blocking the main thread)
    loop_thread = threading.Thread(target=node.run_loop)
    loop_thread.start()
    
    # The main thread executes spin to keep the ROS 2 node running
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        loop_thread.join()  # Waiting for the thread to end
        node.destroy_node()
        rclpy.shutdown()
​
if __name__ == "__main__":
    main()
    

• Configure the corresponding setup.py configuration file and add it in console_scripts

xxxxxxxxxx
'rate_demo=learning_time.rate_demo:main'

• Compile feature package

xxxxxxxxxx
colcon build --packages-select learning_time

• Refresh the workspace environment and run the node

xxxxxxxxxx
 source ./install/setup.bash

xxxxxxxxxx
ros2 run learning_time rate_demo

3. Timer Application

• Timer is used to create a timer that triggers periodic tasks.
• Example: Create two timers: one that prints the execution count every 1 second, and one that prints the current time every 0.5 seconds.
• Create a new program file called Timer_demo.py

xxxxxxxxxx
import rclpy
from rclpy.node import Node
​
class TimerDemoNode(Node):
    def __init__(self):
        super().__init__('timer_demo_node')
        
        # Counter, used to demonstrate the number of timer executions
        self.counter = 0
        
        # Create a timer: execute the callback function every 1 second
        self.timer = self.create_timer(1.0, self.timer_callback)
        
        # Create a faster timer: execute every 0.5 seconds
        self.fast_timer = self.create_timer(0.5, self.fast_timer_callback)
        
        self.get_logger().info("定时器节点已启动")
​
    def timer_callback(self):
        """1 second timer callback function"""
        self.counter += 1
        current_time = self.get_clock().now()
        
        # Print the current time and counter value
        self.get_logger().info(
            f"[1秒定时器] 第 {self.counter} 次执行，当前时间: {current_time.seconds_nanoseconds()}"
        )
​
    def fast_timer_callback(self):
        """0.5 second timer callback function"""
 ROS2 DDS
ROS2 common command tools