This package provides a wrapper that mimics the interface of the Timer class in the Python threading package, but implements the timer functionality using a provided ROS2 Timer.

A secondary goal of this package is to gather documentation about ROS2 clocks and timers.

This package should generally be checked-out and built as part of a ROS2 workspace.

Perhaps the best way to introduce the package and task is via an example. This example is provided in doctest format, and can be run via the following code:¹

import doctest
doctest.testfile('README.md', module_relative=False)

It is assumed that the ROS2 environment and the ROS2 overlay have both been properly initialized, before running the example.

This example illustrates how the ros_threading_timer package can be used to treat a ROS2 Timer like a threading.Timer. It is assumed that an initialized ROS2 node exists, and that we wish to plug a timer derived from it into code tat expects a threading style of interface. Initialize a ROS2 interface, a ROS2 node, and a ROS2 timer.

>>> import rclpy
>>> import rclpy.node
>>> rclpy.init()
>>> node = rclpy.node.Node('test')
>>> ros_timer = node.create_timer(timer_period_sec=1, 
...                               callback=lambda: None)

Define a timeout interval and a tolerance. For this example, we will use an interval of 20 milliseconds. The tolerance is the maximum amount of deviation from an expected interval that we will tolerate before raising an error. In general, this should be quite small. Here, we will use a tolerance of one millisecond.

>>> timer_period_s = 0.020
>>> tolerance_s = 0.001

Wrap the ROS2 timer with a threading-style timer class, and create a new timer interface.

>>> from ros_threading_timer import TimerWrapper
>>> Timer = TimerWrapper(ros_timer, node=node)
>>> timer = Timer(function=lambda: print('Timeout'),
...               interval=timer_period_s)
>>> timer.is_alive()
False

The callback is called once -- and only once -- after the timer is activated.

>>> timer.start()
>>> timer.is_alive()
True
>>> timer.join(timeout=timer_period_s + tolerance_s)
Timeout
>>> timer.is_alive()
False
>>> timer.join(timeout=timer_period_s + tolerance_s) # No callback.

As required by the threading package, the timer cannot be started a second time.

>>> try: timer.start()
... except RuntimeError as e: print(e)
Timer has already been started.

A new timer can easily be created and run.

>>> timer = Timer(function=lambda: print('New timeout'),
...               interval=timer_period_s)
>>> timer.start()
>>> timer.join(timeout=timer_period_s + tolerance_s)
New timeout

The callback will not be called for a timer that is cancelled before the timeout interval expires.

>>> timer = Timer(function=lambda: print('Timeout'),
...               interval=timer_period_s)
>>> timer.start()
>>> timer.join(timeout=round(timer_period_s / 2))
>>> timer.cancel()
>>> timer.is_alive()
False

Once cancelled, a timer cannot be restarted. A new timer must be created.

>>> try: timer.start()
... except RuntimeError as e: print(e)
Timer has already been started.

Cleanup by destroying the node and shutting down the ROS2 interface.

>>> node.destroy_node()
>>> rclpy.shutdown()

This package also contains an implementation of a wrapper that mimics the interface of ROS2 Timer but acts as a one-shot (i.e., non-repeating, non-periodic) timer. See one_shot.py for further information and documentation.

• rclcpp::TimerBase
• threading.Thread
• threading.Timer
• rcl/timer.h

Copyright 2022 Neuromechatronics Lab, Carnegie Mellon University

Created by: a. whit. ([email protected])

This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0. If a copy of the MPL was not distributed with this file, You can obtain one at https://mozilla.org/MPL/2.0/.