ROS Actions
A collection of my personal notes.
Conceptually
Under the hood, a ROS 2 action is built from three services (send_goal, cancel_goal, get_result) and two topics (feedback, status),
Custom Action
# Goal. Access this part by using <YourActionType>.Goal
geometry_msgs/PoseStamped target_pose
---
# Result. Access this part by using <YourActionType>.Result
bool success
---
# Feedback. Access this part by using <YourActionType>.Feedback
float32 distance_remainingMinimal Toy Action Server
import math
import time
import rclpy
from rclpy.node import Node
from rclpy.action import ActionServer
from my_nav_interfaces.action import NavigateToPose
class NavigateServer(Node):
def __init__(self):
super().__init__('navigate_server')
self.server = ActionServer(
self,
NavigateToPose,
'navigate_to_pose',
self.execute
)
self.get_logger().info('NavigateToPose server ready')
def execute(self, goal_handle):
# -----------------------------------------
# Read target from client's goal
# -----------------------------------------
target_x = goal_handle.request.target_pose.pose.position.x
target_y = goal_handle.request.target_pose.pose.position.y
self.get_logger().info(
f'Going to ({target_x:.1f}, {target_y:.1f})'
)
# Fake robot starts at the origin
start_x = 0.0
start_y = 0.0
steps = 10
# -----------------------------------------
# Pretend to move
# -----------------------------------------
for step in range(1, steps + 1):
progress = step / steps
current_x = start_x + progress * (target_x - start_x)
current_y = start_y + progress * (target_y - start_y)
dx = target_x - current_x
dy = target_y - current_y
distance = math.sqrt(dx**2 + dy**2)
# -------------------------------------
# Publish feedback
# -------------------------------------
feedback = NavigateToPose.Feedback()
feedback.distance_remaining = distance
goal_handle.publish_feedback(feedback)
self.get_logger().info(
f'{distance:.2f} m remaining'
)
time.sleep(0.5)
# -----------------------------------------
# Finish action
# -----------------------------------------
goal_handle.succeed()
result = NavigateToPose.Result()
result.success = True
return result
def main():
rclpy.init()
node = NavigateServer()
try:
rclpy.spin(node)
except KeyboardInterrupt:
pass
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()Action Server
rclpy.action.server.ActionServer(
node=<Node>, # the ROS 2 node that owns this action server
action_type=<YourAction>, # generated action class, e.g. Fibonacci
action_name="<action_name>", # base name of the action, e.g. "fibonacci"
execute_callback=<callable_or_None>, # called to actually execute an accepted goal
# signature:
# (goal_handle: ServerGoalHandle) -> <YourAction>.Result
# usually the main "work" function of the server
*, # everything below this must be passed by keyword
callback_group=<CallbackGroup_or_None>, # callback group used by the action server
# None -> node.default_callback_group
goal_callback=<callable>, # decides whether a new goal is accepted or rejected
# signature:
# (goal_request: <YourAction>.Goal) -> GoalResponse
# default: accept all goals
handle_accepted_callback=<callable>, # runs after a goal has already been accepted
# signature:
# (goal_handle: ServerGoalHandle) -> None
# default: goal_handle.execute()
# use this if you want to queue/defer/schedule goals yourself
cancel_callback=<callable>, # decides whether a cancel request is accepted or rejected
# public signature:
# (cancel_request: CancelGoal.Request) -> CancelResponse
# default: reject all cancellations
goal_service_qos_profile=<QoSProfile>, # QoS for the hidden "send_goal" service
# default: qos_profile_services_default
result_service_qos_profile=<QoSProfile>, # QoS for the hidden "get_result" service
# default: qos_profile_services_default
cancel_service_qos_profile=<QoSProfile>, # QoS for the hidden "cancel_goal" service
# default: qos_profile_services_default
feedback_pub_qos_profile=<QoSProfile>, # QoS for the hidden feedback publisher
# current default: QoSProfile(depth=10)
status_pub_qos_profile=<QoSProfile>, # QoS for the hidden status publisher
# default: qos_profile_action_status_default
result_timeout=10 # seconds to keep a finished goal result available
# after terminal state (succeeded/aborted/canceled)
)Minimal Toy Action Client
import rclpy
from rclpy.node import Node
from rclpy.action import ActionClient
from my_nav_interfaces.action import NavigateToPose
class NavigateClient(Node):
def __init__(self):
super().__init__('navigate_client')
self.client = ActionClient(
self,
NavigateToPose,
'navigate_to_pose'
)
def send_goal(self, x, y):
# -----------------------------------------
# Build goal
# -----------------------------------------
goal = NavigateToPose.Goal()
goal.pose.header.frame_id = 'map'
goal.pose.header.stamp = self.get_clock().now().to_msg()
goal.pose.pose.position.x = x
goal.pose.pose.position.y = y
# Valid "no rotation" quaternion
goal.pose.pose.orientation.w = 1.0
# -----------------------------------------
# Send goal
# -----------------------------------------
self.client.wait_for_server()
future = self.client.send_goal_async(
goal,
feedback_callback=self.feedback_callback
)
rclpy.spin_until_future_complete(self, future)
goal_handle = future.result()
if not goal_handle.accepted:
self.get_logger().info('Goal rejected')
return
self.get_logger().info('Goal accepted')
# -----------------------------------------
# Wait for final result
# -----------------------------------------
result_future = goal_handle.get_result_async()
rclpy.spin_until_future_complete(
self,
result_future
)
result = result_future.result().result
self.get_logger().info(
f'Success: {result.success}'
)
def feedback_callback(self, msg):
distance = msg.feedback.distance_remaining
self.get_logger().info(
f'{distance:.2f} m remaining'
)
def main():
rclpy.init()
node = NavigateClient()
node.send_goal(3.0, 4.0)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()Action Client API
from rclpy.action import ActionClient
from rclpy.qos import QoSProfile, qos_profile_services_default, qos_profile_action_status_default
ActionClient(
node=<Node>, # the ROS 2 node that owns this action client
action_type=<YourAction>, # generated action class, e.g. Fibonacci
action_name="<action_name>", # base name of the action, e.g. "fibonacci"
*, # everything below this must be passed by keyword
callback_group=<CallbackGroup_or_None>, # callback group used by the action client
# None -> node.default_callback_group
goal_service_qos_profile=qos_profile_services_default, # QoS for hidden "send_goal" service
result_service_qos_profile=qos_profile_services_default, # QoS for hidden "get_result" service
cancel_service_qos_profile=qos_profile_services_default, # QoS for hidden "cancel_goal" service
feedback_sub_qos_profile=QoSProfile(depth=10), # QoS for hidden feedback subscriber
status_sub_qos_profile=qos_profile_action_status_default # QoS for hidden status subscriber
)