[ROS2] Action 통신 - Client & Server 생성

Action

• 노드 간 통신 방식 중 하나로, Service와 유사하지만 더 오래 걸리는 작업에 적합
  ㄴ Action Server: 클라이언트의 요청(goal)을 받아 작업을 수행하고 결과(result)를 반환
  ㄴ Action Client: 목표(goal)를 보내고, 피드백과 결과를 비동기적으로 수신
• 수행 중인 작업의 진행 상황을 알려주는 feedback 메시지가 존재함

명령어

1. 액션 목록

• 존재하는 액션 리스트를 볼 수 있음
  ros2 action list
  

2. 액션 정보 확인

• 액션 정보 확인
  ros2 action info [action_name]
  ㄴ 옵션
  -t : 인터페이스 정보 출력

3. 인터페이스 목록

ros2 interface list
ㄴ 필요 시 사용자 정의 메시지로 확장하는 것이 일반적

4. 인터페이스 데이터 구조

• .action 파일은 다음과 같은 세 가지 영역으로 구성됨

# goal 영역
---
# result 영역
---
# feedback 영역

5. 액션 서버에 요청 보내기

• 액션 서버에 목표(goal)를 보내고 결과 수신
  ros2 action send_goal [action_name] [action_type] '{goal_field: value}'
  ㄴ 옵션
  --feedback : 처리 중 피드백 출력
  --once : 피드백 한 번만 출력

명령어 요약

목적	명령어 예시
실행 중 액션 보기	ros2 action list
액션 타입 확인	ros2 action type /my_action
액션 인터페이스 보기	ros2 interface show my_pkg/action/MyAction
액션 goal 보내기	ros2 action send_goal /my_action my_pkg/action/MyAction '{goal: ...}'
액션 상태 확인	ros2 action status /my_action
액션 취소	ros2 action cancel /my_action

Server & Client 생성

• 클라이언트가 서버에 로봇을 좌/우로 회전 시키는 요청 발송
• 서버가 로봇을 좌우로 회전시키고 클라이어트에 응답 회신
  ㄴ 1초에 30도씩 회전
  ㄴ 중간 피드백 전송

1. 인터페이스 생성

복잡한 인터페이스 생성 작업에는 ament_cmake 기반이 더 쉽고 표준적
ament_cmake 기반 패키지를 생성하여 인터페이스 관리

• 인터페이스 관리를 위한 패키지 생성
  ros2 pkg create --build-type ament_cmake interface_package --dependencies rclcpp std_msgs

• 액션 인터페이스를 추가하기 위한 디렉터리 생성
  mkdir action

• action/RotateTiago.action 추가

  # Goal
  float64 angle
  ---
  # Result
  bool success
  ---
  # Feedback
  float64 remaining_angle

• CMakeLists.txt 및 package.xml 수정

CMakeLists.txt

# 이 부분 추가
find_package(rosidl_default_generators REQUIRED)

# 아래는 패키지 생성시 --dependencies std_msgs 시 자동 생성
# find_package(std_msgs REQUIRED)

set(interface_files "action/RotateTiago.action")
### 여러개 추가할 때
### set(interface_files "action/RotateTiago.action" "action/RotateTiago_2.action" "action/RotateTiago_3.action" )

rosidl_generate_interfaces(${PROJECT_NAME}
  ${interface_files}
  DEPENDENCIES std_msgs
)

package.xml

# 아래 내용 추가
# 빌드할 때 메시지를 컴파일하기 위해 rosidl_default_generators가 필요하다고 선언
<build_depend>rosidl_default_generators</build_depend>

# 실행 시간(runtime)에 메시지를 해석하거나 사용하는 데 필요한 의존성
<exec_depend>rosidl_default_runtime</exec_depend>

# 인터페이스 패키지 그룹(메시지나 서비스 정의하는 패키지)에 속함을 선언
# ROS2에서 메시지 정의 패키지로 자동 인식되기 위해 필요
<member_of_group>rosidl_interface_packages</member_of_group>

• 패키지 빌드 및 생성 확인
  colcon build --symlink-install --packages-select interface_package
ros2 interface show interface_package/action/RotateTiago
  

3. Server & Client 노드 작성

• server

import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from interface_package.action import RotateTiago
from geometry_msgs.msg import Twist
import time

class RotateActionServer(Node):
    def __init__(self):
        super().__init__('rotate_action_server')
        # 액션 서버 오픈
        self.action_server = ActionServer(
            self,
            RotateTiago,
            'rotate_tiago',
            self.execute_callback
        )

        # 로봇 제어 publisher
        self.publisher_ = self.create_publisher(Twist, 'cmd_vel', 10)
        self.get_logger().info(f"rotate_tiage server is ready")
        
    def execute_callback(self, goal):
        self.get_logger().info(f"execute goal to twist : {goal.request.angle}")
        
        remain_ang = goal.request.angle
        twist = Twist()
        twist.angular.z = 0.5 if remain_ang > 0 else -0.5

        # 피드백 메시지 초기화
        feedback = RotateTiago.Feedback()

        # 목표 각도에 도달할 때까지 반복
        while abs(remain_ang) > 0.35:
            # 로봇 움직임
            self.publisher_.publish(twist)
            remain_ang -= twist.angular.z

            # 피드백 메시지 업데이트 및 publish
            feedback.remaining_angle = remain_ang
            goal.publish_feedback(feedback)

            time.sleep(1.5)

        # 정지
        twist.angular.z = 0.0
        self.publisher_.publish(twist)

        # 목표 도달 및 응답
        # 성공적으로 처리됐음을 액션 서버 내부 상태에 표시
        goal.succeed()
        result = RotateTiago.Result()
        result.success = True 
        
        self.get_logger().info('goal compelete')
        return result

def main(args=None):
    rclpy.init(args=args)
    node = RotateActionServer()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

• 서버 실행 후 동작 확인
  서버 실행 : ros2 run tiago_rotation_action rotate_server
  서버 요청 : ros2 action send_goal -f /rotate_tiago interface_package/action/RotateTiago "{angle: -1.57}"
  ㄴ 움직임 확인 가능

• client

import rclpy
from rclpy.node import Node
from rclpy.action import ActionClient
from interface_package.action import RotateTiago
import sys

class RotateActionClient(Node):
    def __init__(self):
        super().__init__('rotate_action_client')

        # 액션 클라이언트 생성
        self.action_client = ActionClient(
            self,
            RotateTiago,
            '/rotate_tiago',
        )
        self.get_logger().info(f"rotate_tiage client start")

    def send_goal(self, angle):
        self.get_logger().info('Waiting for action server...')
        self.action_client.wait_for_server()

        goal_msg = RotateTiago.Goal()
        goal_msg.angle = angle

        self.get_logger().info(f'Sending goal: {angle}')

        # 1. 여기는 goal을 요청, 요청에 대한 처리결과(수락/거절)만 받음
        self.response = self.action_client.send_goal_async(
            goal_msg,
            feedback_callback=self.feedback_callback
        )
        # goal에 대한 처리결과(수락/거절)에 대한 콜백 함수
        self.response.add_done_callback(self.goal_response_callback)

    def goal_response_callback(self, res):   
        # 2. 처리결과(수락/거절) 확인
        if not res.result().accepted:
            self.get_logger().error('Goal rejected')
            return
        
        self.get_logger().info('Goal accepted')

        # 3. 수락시 goal의 결과를 여기서 요청 및 콜백 함수로 처리
        self.res_result = res.result().get_result_async()
        self.res_result.add_done_callback(self.get_result_callback)


    def get_result_callback(self, goal_result):
        result_msg = goal_result.result()
        if result_msg.result.success:
            self.get_logger().info('completed successfully!')
        else:
            self.get_logger().error(f'failed')

        self.destroy_node()
        rclpy.shutdown()
        

    def feedback_callback(self, feedback_msg):
        self.get_logger().info(f'remain angle : {feedback_msg.feedback.remaining_angle}')

def main(args=None):
    rclpy.init(args=args)
    action_client = RotateActionClient()

    # 명령줄 인자 처리
    angle = 1.57  # 기본값

    if len(sys.argv) > 1:
        angle = float(sys.argv[1])
    try:
        action_client.send_goal(angle)
        rclpy.spin(action_client)
    except KeyboardInterrupt:
        action_client.get_logger().info('Keyboard interrupt received, shutting down...')
    finally:
        if rclpy.ok():
            rclpy.shutdown()

if __name__ == '__main__':
    main()

• setup.py 추가

entry_points={
    'console_scripts': [
        'rotate_server = tiago_rotation_action.rotate_server:main',
        'rotate_client = tiago_rotation_action.rotate_client:main',
    ],
},

• 노드 실행
  서버 : ros2 run tiago_rotation_action rotate_server
  클라이언트 : ros2 run tiago_rotation_action rotate_client -2

geometry_msgs/msg/Twist 이해하기

• 

• Vector3 linear # 직진 속도 (m/s)
  ㄴ 📦 linear (직진 방향)
  ㄴ x: 앞으로(+)/뒤로(-) 직진
  ㄴ y: 왼쪽(+)/오른쪽(-) 옆걸음 (holonomic robot일 때만!)
  ㄴ z: 위(+)/아래(-) 이동 (드론이나 3D 이동 가능한 로봇만 해당)

---

• Vector3 angular # 회전 속도 (rad/s) : 초당 몇 라디안으로 회전할지를 의미
  ㄴ 🔁 angular (회전 방향)
  ㄴ x: 좌우로 몸 기울이기 (pitch) – 일반적으로 사용 안 함
  ㄴ y: 앞뒤로 몸 흔들기 (roll) – 일반적으로 사용 안 함
  ㄴ z: 좌우 회전 (yaw) – 이것만 잘 쓰면 회전 끝