1.6액션 (Action)

# Fibonacci.action (action_tutorials_interfaces)
int32 order
---
int32[] sequence
---
int32[] partial_sequence

ros2 action list                # 그래프 안의 액션 이름 목록
ros2 action list -t             # 액션 타입까지 표시
ros2 action info /turtle1/rotate_absolute      # 클라이언트/서버 수와 노드 이름
ros2 interface show turtlesim/action/RotateAbsolute   # 목표/결과/피드백 구조

# 목표 보내기: ros2 action send_goal <액션 이름> <액션 타입> <YAML 값>
ros2 action send_goal /turtle1/rotate_absolute turtlesim/action/RotateAbsolute "{theta: 1.57}"

# --feedback을 붙이면 진행 중 피드백도 출력
ros2 action send_goal /turtle1/rotate_absolute turtlesim/action/RotateAbsolute "{theta: -1.57}" --feedback

import time

import rclpy
from rclpy.action import ActionServer, CancelResponse, GoalResponse
from rclpy.node import Node

from action_tutorials_interfaces.action import Fibonacci

class FibonacciActionServer(Node):
    def __init__(self):
        super().__init__('fibonacci_action_server')
        self._action_server = ActionServer(
            self,
            Fibonacci,
            'fibonacci',                           # 액션 이름
            execute_callback=self.execute_callback,
            goal_callback=self.goal_callback,      # 생략하면 모든 목표를 수락
            cancel_callback=self.cancel_callback)  # 생략하면 모든 취소를 거절

    def goal_callback(self, goal_request):
        if goal_request.order < 0:
            return GoalResponse.REJECT
        return GoalResponse.ACCEPT

    def cancel_callback(self, goal_handle):
        return CancelResponse.ACCEPT

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

        for i in range(1, goal_handle.request.order):
            if goal_handle.is_cancel_requested:    # 취소 요청이 왔는지 확인
                goal_handle.canceled()
                result = Fibonacci.Result()
                result.sequence = feedback.partial_sequence
                return result

            feedback.partial_sequence.append(
                feedback.partial_sequence[i] + feedback.partial_sequence[i - 1])
            goal_handle.publish_feedback(feedback)  # 진행 상황 전달
            time.sleep(1)

        goal_handle.succeed()                       # 성공 상태로 표시
        result = Fibonacci.Result()
        result.sequence = feedback.partial_sequence
        return result                               # 결과 메시지는 반드시 반환

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

if __name__ == '__main__':
    main()

import rclpy
from rclpy.action import ActionClient
from rclpy.node import Node

from action_tutorials_interfaces.action import Fibonacci

class FibonacciActionClient(Node):
    def __init__(self):
        super().__init__('fibonacci_action_client')
        self._client = ActionClient(self, Fibonacci, 'fibonacci')

    def send_goal(self, order):
        goal = Fibonacci.Goal()
        goal.order = order

        self._client.wait_for_server()
        future = self._client.send_goal_async(
            goal, feedback_callback=self.on_feedback)
        future.add_done_callback(self.on_goal_response)

    def on_goal_response(self, future):
        self._goal_handle = future.result()
        if not self._goal_handle.accepted:
            self.get_logger().info('목표가 거절되었습니다')
            return
        self.get_logger().info('목표가 수락되었습니다')
        result_future = self._goal_handle.get_result_async()
        result_future.add_done_callback(self.on_result)

    def on_feedback(self, feedback_msg):
        partial = feedback_msg.feedback.partial_sequence
        self.get_logger().info(f'피드백: {list(partial)}')

    def on_result(self, future):
        result = future.result().result
        self.get_logger().info(f'결과: {list(result.sequence)}')
        rclpy.shutdown()

    # 취소하려면 self._goal_handle.cancel_goal_async() 를 호출합니다.

def main(args=None):
    rclpy.init(args=args)
    node = FibonacciActionClient()
    node.send_goal(10)
    rclpy.spin(node)

if __name__ == '__main__':
    main()