Action server implementation in ROS

Introduction

An action server lets your robot do tasks that take time and can give updates while working. It helps your robot respond to commands and tell you progress or if it finished.

When your robot needs to move to a place and you want to know how far it has gone.

When you start a task like picking up an object and want updates until done.

When a task might take a while and you want to cancel it if needed.

When you want your robot to report progress during a long operation.

When you want to handle commands that can succeed, fail, or be stopped.

Syntax

ROS

import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from example_interfaces.action import Fibonacci

class FibonacciActionServer(Node):
    def __init__(self):
        super().__init__('fibonacci_action_server')
        self._action_server = ActionServer(
            self,
            Fibonacci,
            'fibonacci',
            self.execute_callback)

    def execute_callback(self, goal_handle):
        feedback_msg = Fibonacci.Feedback()
        feedback_msg.sequence = [0, 1]

        for i in range(1, goal_handle.request.order):
            feedback_msg.sequence.append(
                feedback_msg.sequence[i] + feedback_msg.sequence[i - 1])
            goal_handle.publish_feedback(feedback_msg)
            self.get_logger().info(f'Feedback: {feedback_msg.sequence}')

        goal_handle.succeed()
        result = Fibonacci.Result()
        result.sequence = feedback_msg.sequence
        return result

The ActionServer class creates the server that listens for goals.

The execute_callback runs when a goal is received and sends feedback and result.

Examples

This line creates an action server named 'fibonacci' that uses the Fibonacci action type and calls execute_callback when a goal arrives.

ROS

self._action_server = ActionServer(self, Fibonacci, 'fibonacci', self.execute_callback)

This function handles the goal, sends progress updates, and returns the final result.

ROS

def execute_callback(self, goal_handle):
    # process goal and send feedback
    goal_handle.publish_feedback(feedback_msg)
    # finish and return result
    goal_handle.succeed()
    return result

Sample Program

This program creates a ROS 2 action server that calculates Fibonacci numbers. It sends feedback as it calculates and returns the full sequence when done.

ROS

import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from example_interfaces.action import Fibonacci

class FibonacciActionServer(Node):
    def __init__(self):
        super().__init__('fibonacci_action_server')
        self._action_server = ActionServer(
            self,
            Fibonacci,
            'fibonacci',
            self.execute_callback)

    def execute_callback(self, goal_handle):
        feedback_msg = Fibonacci.Feedback()
        feedback_msg.sequence = [0, 1]

        for i in range(1, goal_handle.request.order):
            feedback_msg.sequence.append(
                feedback_msg.sequence[i] + feedback_msg.sequence[i - 1])
            goal_handle.publish_feedback(feedback_msg)
            self.get_logger().info(f'Feedback: {feedback_msg.sequence}')

        goal_handle.succeed()
        result = Fibonacci.Result()
        result.sequence = feedback_msg.sequence
        return result


def main(args=None):
    rclpy.init(args=args)
    action_server = FibonacciActionServer()
    rclpy.spin(action_server)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

Important Notes

Remember to match the action type with your .action file definition.

Use feedback to keep clients informed during long tasks.

Always call succeed() or abort() on the goal handle to finish the action.

Summary

An action server handles long tasks and sends progress updates.

Use execute_callback to process goals and send feedback.

Finish by returning a result and marking the goal succeeded or aborted.

Practice

1. What is the main role of an action server in ROS?

easy

A. To handle long-running tasks and provide feedback during execution

B. To send simple messages without feedback

C. To only receive goals without processing them

D. To replace the ROS parameter server

Solution

1. Step 1: Understand the purpose of an action server

   An action server is designed to manage tasks that take time and can provide updates while working.

2. Step 2: Compare options with the role

   Only To handle long-running tasks and provide feedback during execution correctly describes handling long tasks and sending feedback, unlike others which describe unrelated roles.

3. Final Answer:

   To handle long-running tasks and provide feedback during execution -> Option A

4. Quick Check:

   Action server = long tasks + feedback [OK]

Hint: Remember: action servers manage long tasks with feedback [OK]

Common Mistakes:

• Confusing action server with simple message publishers
• Thinking action server only receives goals without processing
• Mixing action server with parameter server functionality

2. Which of the following is the correct way to define the execute callback function in a ROS action server using Python?

easy

A. def execute(goal, self):

B. def executeCallback(goal):

C. def callback_execute(goal, self):

D. def execute_callback(self, goal):

Solution

1. Step 1: Recall the standard signature for execute callback

   In ROS Python action servers, the execute callback is a method with 'self' and 'goal' parameters.

2. Step 2: Match options to this signature

   Only def execute_callback(self, goal): correctly uses 'def execute_callback(self, goal):' matching the expected pattern.

3. Final Answer:

   def execute_callback(self, goal): -> Option D

4. Quick Check:

   Method signature = self first, then goal [OK]

Hint: Method callbacks in classes always have self first [OK]

Common Mistakes:

• Omitting 'self' parameter in method definition
• Swapping parameter order incorrectly
• Using camelCase instead of snake_case for method name

3. Given this snippet inside an action server's execute callback:

for i in range(3):
    feedback.percent_complete = (i + 1) * 33
    self._action_server.publish_feedback(feedback)
    rospy.sleep(1)
result.success = True
self._action_server.set_succeeded(result)

What will the client receive?

medium

A. No feedback updates, only success result

B. Three feedback updates at 33%, 66%, and 99%, then success result

C. Only one feedback update at 99%, then success result

D. Feedback updates but the result will be failure

Solution

1. Step 1: Analyze the loop sending feedback

   The loop runs 3 times, each time publishing feedback with percent_complete increasing by 33.

2. Step 2: Check the result setting after loop

   After feedback, the result.success is set to True and the goal is marked succeeded.

3. Final Answer:

   Three feedback updates at 33%, 66%, and 99%, then success result -> Option B

4. Quick Check:

   Loop feedback + success result = Three feedback updates at 33%, 66%, and 99%, then success result [OK]

Hint: Loop sends multiple feedbacks before final success [OK]

Common Mistakes:

• Assuming feedback is sent only once after loop
• Confusing success with failure result
• Ignoring the sleep delay between feedbacks

4. Identify the error in this action server execute callback snippet:

def execute_callback(self, goal):
    if goal.order < 0:
        self._action_server.set_aborted()
    result = Result()
    result.output = goal.order * 2
    self._action_server.set_succeeded(result)

medium

A. set_aborted() is called without a result argument

B. Result() is not imported or defined

C. The goal attribute 'order' does not exist

D. set_succeeded() should be called before set_aborted()

Solution

1. Step 1: Check set_aborted() usage

   set_aborted() requires a result argument to inform the client about failure details.

2. Step 2: Verify other parts

   Result() usage is typical; goal.order is assumed valid; set_succeeded() after set_aborted() is not called sequentially here.

3. Final Answer:

   set_aborted() is called without a result argument -> Option A

4. Quick Check:

   set_aborted needs result argument [OK]

Hint: Always pass a result when aborting goals [OK]

Common Mistakes:

• Calling set_aborted() without result causes runtime error
• Assuming set_aborted() can be called without parameters
• Ignoring the need to inform client on abort

5. You want to implement an action server that cancels the goal if a certain condition is met during execution. Which approach correctly handles goal cancellation in the execute callback?

hard

A. Call set_aborted() immediately when the goal is received without checking cancellation

B. Ignore cancellation requests and always call set_succeeded() at the end

C. Check self._action_server.is_preempt_requested() inside the loop, then call self._action_server.set_preempted() and return

D. Use rospy.sleep() to delay cancellation handling until after completion

Solution

1. Step 1: Understand cancellation handling in action servers

   Cancellation requests are detected by checking is_preempt_requested() during execution.

2. Step 2: Proper response to cancellation

   When cancellation is requested, call set_preempted() to notify client and stop processing.

3. Final Answer:

   Check self._action_server.is_preempt_requested() inside the loop, then call self._action_server.set_preempted() and return -> Option C

4. Quick Check:

   Cancellation check + set_preempted = Check self._action_server.is_preempt_requested() inside the loop, then call self._action_server.set_preempted() and return [OK]

Hint: Use is_preempt_requested() to detect and handle cancellations [OK]

Common Mistakes:

• Ignoring cancellation requests during execution
• Calling set_aborted() instead of set_preempted() on cancel
• Delaying cancellation handling until after task finishes