Writing a simple publisher and subscriber(Python)
Background
node를 만들어, 각 topic에 대해 string message를 통해 정보를 주고받는다.
이는 간단한 'talker', 'listener' system으로, 하나의 node가 data를 publish하면 다른 node가 data를 subcribe한다.
Tasks
Create a package
# ros2_ws/src
ros2 pkg create --build-type ament_python --license Apache-2.0 py_pubsub
Write the publisher node
# ros2_ws/src/py_pubsub/py_pubsub
wget https://raw.githubusercontent.com/ros2/examples/humble/rclpy/topics/minimal_publisher/examples_rclpy_minimal_publisher/publisher_member_function.py# ros2_ws/src/py_pubsub/py_sub/publisher_member_function.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
class MinimalPublisher(Node):
def __init__(self):
super().__init__('minimal_publisher')
self.publisher_ = self.create_publisher(String, 'topic', 10)
timer_period = 0.5 # seconds
self.timer = self.create_timer(timer_period, self.timer_callback)
self.i = 0
def timer_callback(self):
msg = String()
msg.data = 'Hello World: %d' % self.i
self.publisher_.publish(msg)
self.get_logger().info('Publishing: "%s"' % msg.data)
self.i += 1
def main(args=None):
rclpy.init(args=args)
minimal_publisher = MinimalPublisher()
rclpy.spin(minimal_publisher)
# Destroy the node explicitly
# (optional - otherwise it will be done automatically
# when the garbage collector destroys the node object)
minimal_publisher.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
Examine the code
코드를 하나씩 분석해보자.
rclpy는 ROS2 Python library로 Node class를 위해 필요하다.
import rclpy
from rclpy.node import Node
topic을 통해 data를 전달하기 위해 string message type 필요하다.
이는 노드의 dependency를 나타내며, package.xml에 추가되어야 한다.
from std_msgs.msg import String
Node를 상속받는 MinimalPublisher class를 생성한다.
class MinimalPublisher(Node):
create_publisher은 String type의 message를 'topic'이라는 이름의 topic으로 publish하는 node를 정의한다.
이 때 queue size은 10이며, queue size는 subscriber가 message를 빠르게 받지 못할 때 대기중인 message의 양을 제한하기 위해 QoS(quality of service) setting에 필요하다.
timer는 0.5초마다 callback되며, self.i는 callback에 사용되는 counter이다.
def __init__(self):
super().__init__('minimal_publisher')
self.publisher_ = self.create_publisher(String, 'topic', 10)
timer_period = 0.5 # seconds
self.timer = self.create_timer(timer_period, self.timer_callback)
self.i = 0
timer_callback은 counter value를 포함하는 message를 생성하고 'get_logger().info'를 사용하여 console로 publish한다.
def timer_callback(self):
msg = String()
msg.data = 'Hello World: %d' % self.i
self.publisher_.publish(msg)
self.get_logger().info('Publishing: "%s"' % msg.data)
self.i += 1
마지막으로 main function을 정의한다.
rclpy libray를 초기화하고 node를 생성한다.
이는 spin되어 callback함수를 호출한다.
(spin은 callback을 호출하는 기능을 한다. 즉, 노드가 종료될 때까지 return되지 않고 반복해서 callback함수를 호출한다.)
def main(args=None):
rclpy.init(args=args)
minimal_publisher = MinimalPublisher()
rclpy.spin(minimal_publisher)
# Destroy the node explicitly
# (optional - otherwise it will be done automatically
# when the garbage collector destroys the node object)
minimal_publisher.destroy_node()
rclpy.shutdown()
Add dependencies
ros2_ws/src/py_pubsub directory로 가 package.xml을 열어보자.
지난번에 언급했던 description, maintainer, license tag를 채워야 한다.
<description>Examples of minimal publisher/subscriber using rclpy</description>
<maintainer email="you@email.com">Your Name</maintainer>
<license>Apache License 2.0</license>
그 후 node가 import하는 dependency를 추가한다.
이는 package가 rclpy, std_msgs를 필요로 함을 선언한다.
<exec_depend>rclpy</exec_depend>
<exec_depend>std_msgs</exec_depend>
Add an entry point
setup.py를 열어 maintainer, maintainer_email, description, license를 package.xml과 매칭시킨다.
maintainer='YourName',
maintainer_email='you@email.com',
description='Examples of minimal publisher/subscriber using rclpy',
license='Apache License 2.0',
entry_point안에 있는 console_scripts에 다음을 추가한다.
entry_points={
'console_scripts': [
'talker = py_pubsub.publisher_member_function:main',
],
},
Check setup.cfg
setup.cfg는 자동으로 수정되어 있을 것이다.
ros2 run을 할 때 해당 위치에서 파일을 찾기 때문에 setuptool에게 lib directory에 실행파일을 배치하라는 것이다.
Write the subscriber node
ros2_ws/src/py_pubsub/py_pubsub로 돌아와 다음 node를 만들어보자.
wget https://raw.githubusercontent.com/ros2/examples/humble/rclpy/topics/minimal_subscriber/examples_rclpy_minimal_subscriber/subscriber_member_function.py
directory에 아래의 파일이 있어야 한다.
Examine the code
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
class MinimalSubscriber(Node):
def __init__(self):
super().__init__('minimal_subscriber')
self.subscription = self.create_subscription(
String,
'topic',
self.listener_callback,
10)
self.subscription # prevent unused variable warning
def listener_callback(self, msg):
self.get_logger().info('I heard: "%s"' % msg.data)
def main(args=None):
rclpy.init(args=args)
minimal_subscriber = MinimalSubscriber()
rclpy.spin(minimal_subscriber)
# Destroy the node explicitly
# (optional - otherwise it will be done automatically
# when the garbage collector destroys the node object)
minimal_subscriber.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
subscriber node는 publisher node와 흡사하다.
topic name과 message type은 publisher, subscriber가 communicate하기 위해 같아야 한다.
self.subscription = self.create_subscription(
String,
'topic',
self.listener_callback,
10)
subscriber의 constructor와 callback은 timer가 없다. message를 받자마자 callback하기 때문에 필요가 없다.
callback은 단순히 받은 data와 info message를 출력한다.
publisher가 정의한 msg.data = 'Hello World: %d' % self.i를 기억하자.
def listener_callback(self, msg):
self.get_logger().info('I heard: "%s"' % msg.data)
main정의는 거의 동일하다.
minimal_subscriber = MinimalSubscriber()
rclpy.spin(minimal_subscriber)
dependency는 publisher와 동일하기 때문에 package.xml에 추가할 것은 없고 setup.cfg 또한 동일하다.
Add an entry point
setup.py를 열어 entry point에 subscriber node를 추가하자.
entry_points={
'console_scripts': [
'talker = py_pubsub.publisher_member_function:main',
'listener = py_pubsub.subscriber_member_function:main',
],
},
Build and run
build하기 전에 놓친 dependency를 확인하기 위해 rosdep을 실행하는 것도 좋은 방법이다.
# ros2_ws
rosdep install -i --from-path src --rosdistro humble -y
이제 package를 build 하자.
# ros2_ws/src
colcon build --packages-select py_pubsub
sourcing 후 talker와 listener node를 실행하면 message들이 출력되는 것을 확인할 수 있다.
source install/setup.bashros2 run py_pubsub talker
ros2 run py_pubsub listener
ROS2_Humble Documentation : https://docs.ros.org/en/humble/Tutorials/Beginner-Client-Libraries.html