이번 실습에서는 ROS 2에서 Topic 통신을 Python으로 구현하는 방법을 실습했다.
Topic Publisher와 Subscriber를 확인하고, argument 노드에서 데이터를 발행하고 calculator 노드에서 이를 구독하는 과정을 직접 실행했다.
1. Topic Publisher / Subscriber
ROS 2의 Topic은 Publisher가 메시지를 발행하면 Subscriber가 해당 메시지를 구독하는 방식으로 데이터를 전달한다.
이번 실습에서는 다음과 같은 구조로 구성되어 있다.
argument→ Topic Publishercalculator→ Topic Subscriber- Topic 이름 →
arithmetic_argument - 메시지 타입 →
ArithmeticArgument
argument 노드는 현재 시간과 argument_a, argument_b 값을 생성하여 arithmetic_argument 토픽으로 발행한다.
calculator 노드는 해당 토픽을 구독하여 전달받은 값을 저장하고 출력한다.
2. Publisher 코드 확인
먼저 argument.py 파일을 확인했다.
nano ~/robot_ws/src/ros2-seminar-examples/topic_service_action_rclpy_example/topic_service_action_rclpy_example/arithmetic/argument.pyArgument 클래스는 ROS 2의 Node를 상속하며 노드 이름은 argument로 설정되어 있다.
class Argument(Node):
def __init__(self):
super().__init__('argument')또한 QoS 설정을 사용한다.
QOS_RKL10V = QoSProfile(
reliability=QoSReliabilityPolicy.RELIABLE,
history=QoSHistoryPolicy.KEEP_LAST,
depth=qos_depth,
durability=QoSDurabilityPolicy.VOLATILE)Publisher는 다음과 같이 생성된다.
self.arithmetic_argument_publisher = self.create_publisher(
ArithmeticArgument,
'arithmetic_argument',
QOS_RKL10V)그리고 타이머를 이용해 1초마다 데이터를 발행한다.
self.timer = self.create_timer(
1.0,
self.publish_random_arithmetic_arguments)3. 랜덤 데이터 발행
publish_random_arithmetic_arguments() 함수에서는 ArithmeticArgument 메시지를 생성한다.
msg = ArithmeticArgument()
msg.stamp = self.get_clock().now().to_msg()
msg.argument_a = float(
random.randint(self.min_random_num, self.max_random_num))
msg.argument_b = float(
random.randint(self.min_random_num, self.max_random_num))생성한 메시지는 publish()를 통해 Topic으로 전달한다.
self.arithmetic_argument_publisher.publish(msg)따라서 argument 노드는 1초마다 argument_a, argument_b 값을 랜덤하게 생성하여 arithmetic_argument 토픽으로 발행한다.
4. Subscriber 코드 확인
다음으로 calculator.py에서 Subscriber 부분을 확인했다.
sed -n '20,80p' ~/robot_ws/src/ros2-seminar-examples/topic_service_action_rclpy_example/topic_service_action_rclpy_example/calculator/calculator.pycalculator 노드에서는 다음과 같이 Subscription을 생성한다.
self.arithmetic_argument_subscriber = self.create_subscription(
ArithmeticArgument,
'arithmetic_argument',
self.get_arithmetic_argument,
QOS_RKL10V,
callback_group=self.callback_group)Publisher와 동일하게 ArithmeticArgument 메시지와 arithmetic_argument 토픽을 사용한다.
메시지를 전달받으면 get_arithmetic_argument() 콜백 함수가 실행된다.
5. Subscriber 콜백 함수
전달받은 메시지는 다음과 같이 처리한다.
def get_arithmetic_argument(self, msg):
self.argument_a = msg.argument_a
self.argument_b = msg.argument_b
self.get_logger().info(
'Timestamp of the message: {0}'.format(msg.stamp))
self.get_logger().info(
'Subscribed argument a: {0}'.format(self.argument_a))
self.get_logger().info(
'Subscribed argument b: {0}'.format(self.argument_b))즉, Publisher가 보낸 argument_a, argument_b 값을 Subscriber에서 받아 저장하고 로그로 출력한다.
6. MultiThreadedExecutor 확인
calculator의 main.py에서는 MultiThreadedExecutor를 사용하고 있다.
sed -n '1,80p' ~/robot_ws/src/ros2-seminar-examples/topic_service_action_rclpy_example/topic_service_action_rclpy_example/calculator/main.pyexecutor = MultiThreadedExecutor(num_threads=4)
executor.add_node(calculator)
executor.spin()MultiThreadedExecutor를 사용하면 여러 Callback을 병렬로 처리할 수 있도록 구성할 수 있다.
7. Calculator 노드 실행
먼저 새로운 터미널에서 calculator 노드를 실행했다.
source ~/robot_ws/install/setup.zsh
ros2 run topic_service_action_rclpy_example calculatorcalculator 노드는 Subscriber로 동작하면서 arithmetic_argument 토픽의 메시지를 기다린다.
8. Argument 노드 실행
다른 터미널에서 argument 노드를 실행했다.
source ~/robot_ws/install/setup.zsh
ros2 run topic_service_action_rclpy_example argument실행하면 argument 노드에서 랜덤한 argument_a, argument_b 값이 생성되어 Topic으로 발행된다.
9. Topic 확인
새로운 터미널에서 현재 실행 중인 Topic을 확인했다.
source ~/robot_ws/install/setup.zsh
ros2 topic list실행 결과에서
/arithmetic_argument토픽이 생성된 것을 확인했다.
이를 통해 argument 노드가 생성한 Publisher와 calculator 노드의 Subscriber가 사용하는 Topic이 정상적으로 존재하는 것을 확인했다.
10. Topic 메시지 확인
마지막으로 실제로 Topic을 통해 어떤 메시지가 전달되는지 확인했다.
ros2 topic echo /arithmetic_argument실행 결과 stamp, argument_a, argument_b 값이 계속 출력되는 것을 확인했다.
이를 통해 argument 노드에서 생성한 데이터가 /arithmetic_argument 토픽을 통해 실제로 전달되고 있음을 확인했다.
마무리
이번 실습을 통해 ROS 2에서 Python을 이용한 Topic Publisher와 Subscriber의 동작 과정을 확인했다.
argument
│
│ Publisher
▼
/arithmetic_argument
│
│ Subscriber
▼
calculatorargument 노드에서 랜덤하게 생성한 argument_a, argument_b 값이 /arithmetic_argument 토픽을 통해 calculator 노드로 전달되는 것을 확인했으며, ros2 topic echo를 이용해 실제 메시지 데이터까지 확인했다.
