ROS2 run command for fixed time

[r.malmvig]

Hi developers

I am learning about services in ROS2 and need to run a robot for 10 seconds and then stop then robot. How can I do this?

/Rasmus

[GasPatxo]

The simplest way is to use the time library:

import time

# Send your command to the robot

time.sleep(10)

[r.malmvig]

Thank you very much. I had such a hard time finding the information on the net. I know see it is because this is python related and not ROS2

[GasPatxo]

Glad I could be of help.

There are also ways using for example rclpy.timer.Timer.

PD: Remember to mark the solution to your posts

[r.malmvig]

I tried using that but failed… Could you give me an example of how that works?

[GasPatxo]

import rclpy
from rclpy.node import Node
from services_quiz_srv.srv import Turn
from geometry_msgs.msg import Twist


class TurningServer(Node):
    def __init__(self):
        super().__init__('turning_service_server_node')
        self.cmd = Twist()
        self.publisher = self.create_publisher(Twist, 'cmd_vel', 10)
        self.server = self.create_service(
            Turn, 'turn', self.turn_callback)

    def turn_callback(self, request, response):
        self.get_logger().info(
            f"I've been called for {request.time}s at {request.angular_velocity}rad/sec towards the {request.direction}")

        self.cmd.angular.z = request.angular_velocity if request.direction == 'left' else - \
            request.angular_velocity

        self.publisher.publish(self.cmd)  # Send the movement command

        timer = self.create_timer(
            request.time, lambda: self.publisher.publish(Twist()))  # Send a stop message after the requested time

        while not timer.is_ready():  # Wait for the timer action to be triggered, but you could do other processes in the meantime
            print(
                f'The robot will turn for {timer.time_until_next_call()} more nseconds')

        self.get_logger().info('robot stopperd')

        response.success = True
        return response


def main(args=None):
    rclpy.init(args=args)
    server = TurningServer()
    rclpy.spin(server)
    rclpy.shutdown()