1. 액션(Action)이란?
액션 개념도
- 액션에는 서버와 클라이언트가 존재 (서비스와 비슷)
- 클라이언트: 서버에 어떠한 연산을 해달라고 요청을 보냄
- 서버: 클라이언트의 요청을 받아 어떤 연산을 수행하고, 그 결과를 돌려보냄
- 다만, 액션에서는 서비스와 달리 서버가 작업하는 중간에 그 상태를 피드백으로 클라이언트에게 알려 줄 수 있고, 중간에 작업의 취소도 가능함
- 따라서 시간이 오래걸리는 작업을 할 때 알맞음
- 동작 순서 예시
- 클라이언트가 서버에 요청을 보냄
- 서버가 요청을 받으면 콜백함수가 동작하여 특정 작업을 수행함
- 작업 중간마다 그 상태를 서버에서 클라이언트로 피드백을 보냄
- 작업이 완료되면 서버에서 클라이언트로 그 결과가 송신됨
- 클라이언트에서 작업 결과를 받아 활용
2. 액션 서버(Server), 클라이언트(Client) 소스코드 작성
-
여기서는 이전 글에서 생성한 패키지에 액션 인터페이스를 추가함으로써, 이미 생성된 패키지를 업데이트하는 방법을 실습함
-
해당 패키지로 이동
$ cd ~/my_ws/src/my_custom_interface - 액션 인터페이스 파일을 넣을 디랙토리 생성
$ mkdir action
$ cd action- 액션 인터페이스 파일 작성
- 이때, 인터페이스 파일에서 Goal와 Result, Feedback 부분은 "---"로 구분한다
- 나중에 액션 노드 소스코드에서
goal=Fibonacci.Goal(),result=Fibonacci.Result(),feedback=Fibonacci.Feedback()와 같이 자료형처럼 선언하여 사용한다.
$ code Fibonacci.action# Goal
int32 order
---
# Result
int32[] sequence
---
# Feedback
int32[] partial_sequence- package.xml 수정
<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
<name>my_custom_interface</name>
<version>0.0.0</version>
<description>TODO: Package description</description>
<maintainer email="koyeongmin@todo.todo">koyeongmin</maintainer>
<license>TODO: License declaration</license>
<buildtool_depend>ament_cmake</buildtool_depend>
<test_depend>ament_lint_auto</test_depend>
<test_depend>ament_lint_common</test_depend>
<build_depend>rosidl_default_generators</build_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>
<export>
<build_type>ament_cmake</build_type>
</export>
</package>- CMakeLists.txt 수정
- 서비스의 예시에 action 파트만 간단히 추가하면 됨!
cmake_minimum_required(VERSION 3.8)
project(my_custom_interface)
if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
add_compile_options(-Wall -Wextra -Wpedantic)
endif()
# find dependencies
find_package(ament_cmake REQUIRED)
find_package(rosidl_default_generators REQUIRED)
# uncomment the following section in order to fill in
# further dependencies manually.
# find_package(<dependency> REQUIRED)
################################################################################
# Declare ROS messages, services and actions
################################################################################
set(srv_files
"srv/PlusService.srv"
)
set(action_files
"action/Fibonacci.action"
)
rosidl_generate_interfaces(${PROJECT_NAME}
${srv_files}
${action_files}
DEPENDENCIES builtin_interfaces
)
################################################################################
# Macro for ament package
################################################################################
ament_export_dependencies(rosidl_default_runtime)
ament_package()액션 노드 소스코드 작성
- 해당 패키지로 이동
$ cd ~/my_ws/src/my_test_package/my_test_package - 액션 서버 소스코드 작성
$ code action_server_test.pyimport time
import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node
from my_custom_interface.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): # 액션 요청이 들어왔을 때의 콜백함수
self.get_logger().info('Executing goal...')
feedback_msg = Fibonacci.Feedback() # 정의한 액션 인터페이스의 피드백 자료형 선언
feedback_msg.partial_sequence = [0, 1]
for i in range(1, goal_handle.request.order):
feedback_msg.partial_sequence.append( # 피드백에 데이터 삽입
feedback_msg.partial_sequence[i] + feedback_msg.partial_sequence[i-1])
self.get_logger().info('Feedback: {0}'.format(feedback_msg.partial_sequence))
goal_handle.publish_feedback(feedback_msg) # 피드백 보내기
time.sleep(1)
goal_handle.succeed() # 작업 완료 시 상태를 성공으로 변경
result = Fibonacci.Result() # 정의한 액션 인터페이스의 result 자료형 선언
result.sequence = feedback_msg.partial_sequence # result에 데이터 삽입
return result # 결과 리턴 및 client로 보내기
def main(args=None):
rclpy.init(args=args)
fibonacci_action_server = FibonacciActionServer()
rclpy.spin(fibonacci_action_server)
if __name__ == '__main__':
main()- 액션 클라이언트 소스코드 작성
$ code action_client_test.pyimport rclpy
from rclpy.action import ActionClient
from rclpy.node import Node
from action_msgs.msg import GoalStatus
from my_custom_interface.action import Fibonacci
class FibonacciActionClient(Node):
def __init__(self):
super().__init__('fibonacci_action_client')
self._action_client = ActionClient(self, Fibonacci, 'fibonacci') # 액션 클라이언트 정의
def send_goal(self, order):
goal_msg = Fibonacci.Goal() # 정의한 액션 인터페이스의 goal 자료형 선언
goal_msg.order = order # 데이터 삽입
self._action_client.wait_for_server()
# 클라이언드 정의, 피드백이 들어올 때의 콜백함수 지정
self._send_goal_future = self._action_client.send_goal_async(goal_msg, feedback_callback=self.feedback_callback)
# 작업이 완료되어 상태가 변했을 때의 콜백함수 추가
self._send_goal_future.add_done_callback(self.goal_response_callback)
def goal_response_callback(self, future): # 처음 요청을 보냈을 때 accept에 대한 콜백함수
goal_handle = future.result()
if not goal_handle.accepted:
self.get_logger().info('Goal rejected :(')
return
self.get_logger().info('Goal accepted :)')
self._get_result_future = goal_handle.get_result_async() # 결과를 받아옴
self._get_result_future.add_done_callback(self.get_result_callback) # 결과가 정상적으로 나왔을 때의 콜백함수 추가
def get_result_callback(self, future): # 결과를 받았을 때의 콜백함수
action_status = future.result().status
result = future.result().result
if action_status == GoalStatus.STATUS_SUCCEEDED: # 작업이 정상적으로 끝나 succeed되었을 때만 출력 (goal_handle.succeed())
self.get_logger().info('Action succeeded!')
self.get_logger().info('Result: {0}'.format(result.sequence))
rclpy.shutdown()
def feedback_callback(self, feedback_msg): # 피드백을 받았을 때의 콜백함수
feedback = feedback_msg.feedback
self.get_logger().info('Received feedback: {0}'.format(feedback.partial_sequence))
def main(args=None):
rclpy.init(args=args)
action_client = FibonacciActionClient()
action_client.send_goal(10)
rclpy.spin(action_client)
if __name__ == '__main__':
main()- setup.py 수정
- 앞 글의 서비스 setup.py에 entry_points 부분 만 추가된 것에 주목!
$ cd ~/my_ws/src/my_test_package
$ code setup.pyfrom setuptools import find_packages, setup
package_name = 'my_test_package'
setup(
name=package_name,
version='0.0.0',
packages=find_packages(exclude=['test']),
data_files=[
('share/ament_index/resource_index/packages',
['resource/' + package_name]),
('share/' + package_name, ['package.xml']),
],
install_requires=['setuptools'],
zip_safe=True,
maintainer='koyeongmin',
maintainer_email='koyeongmin@todo.todo',
description='TODO: Package description',
license='TODO: License declaration',
tests_require=['pytest'],
entry_points={
'console_scripts': [
'service_server_test = my_test_package.service_server_test:main',
'service_client_test = my_test_package.service_client_test:main',
'action_server_test = my_test_package.action_server_test:main',
'action_client_test = my_test_package.action_client_test:main',
],
},
)빌드 및 실행
- 빌드
$ cd ~/my_ws
$ colcon build --symlink-install- 실행
Terminal 1: $ ros2 run my_test_package action_server_test
Terminal 2: $ ros2 run my_test_package action _client_test
실행 예시
3. 유용한 명령어
- ros2 action send_goal <action 이름> <action 구조에 맞는 값>: 커맨드 라인에서 액션 요청
- ros2 action send_goal --feedback <action 이름> <action 구조에 맞는 값>: 피드백 옵션을 넣으면 피드백까지 출력
