ROS 2 Jazzy Python 패키지 만들기: rclpy Publisher·Subscriber 실습

ROS 2 Jazzy Python 패키지 만들기: rclpy Publisher·Subscriber 실습

ROS 2 Jazzy 실전 시리즈 3편 · Ubuntu 24.04 · Python 3 · 검토 기준 2026년 8월 25일

이번 글에서는 ament_python 패키지를 직접 만들고, 1초마다 문자열을 발행하는 Publisher와 이를 받는 Subscriber를 구현합니다. 단순히 예제가 실행되는 데서 끝내지 않고 Parameter, 명시적 QoS, entry point, launch 파일과 검증 명령까지 연결합니다.

목차

1. 워크스페이스 준비
2. 패키지 생성과 구조
3. Publisher 작성
4. Subscriber 작성
5. setup.py 등록
6. 빌드와 실행
7. launch 파일 추가
8. 센서 QoS로 바꾸기
9. 오류 해결

1. 워크스페이스 준비

이 글은 Ubuntu 24.04에 ROS 2 Jazzy가 설치되어 있다고 가정합니다. 터미널마다 underlay를 먼저 source합니다.

source /opt/ros/jazzy/setup.bash
mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src

용어: /opt/ros/jazzy는 underlay, 직접 빌드할 ~/ros2_ws는 overlay입니다. 실행 전에는 underlay 다음 overlay 순서로 source합니다.

2. ament_python 패키지 생성

ros2 pkg create --build-type ament_python \
  --license Apache-2.0 \
  --dependencies rclpy std_msgs \
  jazzy_py_pubsub

생성 직후 구조는 다음과 같습니다.

jazzy_py_pubsub/
├── jazzy_py_pubsub/
│   └── __init__.py
├── resource/
│   └── jazzy_py_pubsub
├── test/
├── package.xml
├── setup.cfg
└── setup.py

package.xml에는 rclpy와 std_msgs 실행 의존성이 생성됩니다. pure Python 노드만 포함하므로 ament_python이 적합합니다. custom message를 같은 패키지에서 생성하거나 C++를 섞는다면 ament_cmake 구성을 검토합니다.

3. Publisher 노드 작성

jazzy_py_pubsub/jazzy_py_pubsub/publisher_node.py를 만듭니다.

import rclpy
from rclpy.node import Node
from rclpy.qos import HistoryPolicy, QoSProfile, ReliabilityPolicy
from std_msgs.msg import String


class TextPublisher(Node):
    def __init__(self):
        super().__init__('text_publisher')

        self.declare_parameter('publish_period', 1.0)
        period = self.get_parameter('publish_period').value

        qos = QoSProfile(
            history=HistoryPolicy.KEEP_LAST,
            depth=10,
            reliability=ReliabilityPolicy.RELIABLE,
        )
        self.publisher = self.create_publisher(String, 'chatter', qos)
        self.timer = self.create_timer(period, self.publish_message)
        self.count = 0

        self.get_logger().info(
            f'text_publisher started: period={period:.2f}s')

    def publish_message(self):
        message = String()
        message.data = f'Hello Jazzy: {self.count}'
        self.publisher.publish(message)
        self.get_logger().info(f'Published: {message.data}')
        self.count += 1


def main(args=None):
    rclpy.init(args=args)
    node = TextPublisher()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

코드 흐름

1. Node를 상속하고 그래프 이름을 text_publisher로 지정합니다.
2. publish_period Parameter를 선언한 뒤 timer 주기로 사용합니다.
3. std_msgs/msg/String 타입의 chatter Publisher를 만듭니다.
4. executor가 timer callback을 실행하면 메시지를 publish합니다.
5. rclpy.spin()이 callback 실행을 기다립니다.

4. Subscriber 노드 작성

jazzy_py_pubsub/jazzy_py_pubsub/subscriber_node.py를 만듭니다. Publisher와 같은 Topic 이름·타입·호환 QoS를 사용해야 합니다.

import rclpy
from rclpy.node import Node
from rclpy.qos import HistoryPolicy, QoSProfile, ReliabilityPolicy
from std_msgs.msg import String


class TextSubscriber(Node):
    def __init__(self):
        super().__init__('text_subscriber')

        qos = QoSProfile(
            history=HistoryPolicy.KEEP_LAST,
            depth=10,
            reliability=ReliabilityPolicy.RELIABLE,
        )
        self.subscription = self.create_subscription(
            String,
            'chatter',
            self.message_callback,
            qos,
        )
        self.get_logger().info('text_subscriber started')

    def message_callback(self, message):
        self.get_logger().info(f'Received: {message.data}')


def main(args=None):
    rclpy.init(args=args)
    node = TextSubscriber()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

self.subscription을 멤버로 보관하는 이유는 subscription 객체가 수명 동안 유지되어야 하기 때문입니다. callback 안에서는 오래 걸리는 반복문이나 blocking I/O를 직접 실행하지 않는 편이 좋습니다.

5. setup.py에 실행 명령 등록

setup.py의 entry_points를 다음처럼 수정합니다.

entry_points={
    'console_scripts': [
        'publisher = jazzy_py_pubsub.publisher_node:main',
        'subscriber = jazzy_py_pubsub.subscriber_node:main',
    ],
},

setup.cfg의 기본 생성 내용은 실행 파일이 ROS 패키지 경로에 설치되도록 합니다.

[develop]
script_dir=$base/lib/jazzy_py_pubsub

[install]
install_scripts=$base/lib/jazzy_py_pubsub

6. 의존성 설치, 빌드, 실행

cd ~/ros2_ws
source /opt/ros/jazzy/setup.bash

rosdep install --from-paths src --ignore-src -y
colcon build --packages-select jazzy_py_pubsub --symlink-install
source install/setup.bash

첫 번째 터미널에서 Publisher를 실행합니다.

source /opt/ros/jazzy/setup.bash
source ~/ros2_ws/install/setup.bash
ros2 run jazzy_py_pubsub publisher

두 번째 터미널에서 Subscriber를 실행합니다.

source /opt/ros/jazzy/setup.bash
source ~/ros2_ws/install/setup.bash
ros2 run jazzy_py_pubsub subscriber

세 번째 터미널에서 그래프와 메시지를 확인합니다.

ros2 node list
ros2 topic list -t
ros2 topic info /chatter -v
ros2 topic echo /chatter --once
ros2 topic hz /chatter

Parameter override도 바로 시험할 수 있습니다.

ros2 run jazzy_py_pubsub publisher \
  --ros-args -p publish_period:=0.2

7. 두 노드를 launch 파일로 함께 실행

패키지 루트에 launch/pubsub.launch.py를 추가합니다.

from launch import LaunchDescription
from launch_ros.actions import Node


def generate_launch_description():
    return LaunchDescription([
        Node(
            package='jazzy_py_pubsub',
            executable='publisher',
            name='text_publisher',
            parameters=[{'publish_period': 0.5}],
            output='screen',
        ),
        Node(
            package='jazzy_py_pubsub',
            executable='subscriber',
            name='text_subscriber',
            output='screen',
        ),
    ])

setup.py 상단에 glob와 os를 import하고 data_files에 launch 파일 설치 규칙을 추가합니다.

import os
from glob import glob

# setup(...)의 data_files 목록 안에 추가
(os.path.join('share', package_name, 'launch'),
    glob(os.path.join('launch', '*launch.py'))),

cd ~/ros2_ws
colcon build --packages-select jazzy_py_pubsub --symlink-install
source install/setup.bash
ros2 launch jazzy_py_pubsub pubsub.launch.py

8. 카메라·LiDAR라면 센서 QoS 사용

문자열 상태나 명령에는 RELIABLE이 자연스럽지만, 고주기 센서 데이터는 오래된 메시지를 재전송하는 것보다 최신 값을 빨리 받는 것이 중요합니다. 이 경우 ROS 2가 제공하는 sensor data profile을 사용할 수 있습니다.

from rclpy.qos import qos_profile_sensor_data

self.subscription = self.create_subscription(
    LaserScan,
    'scan',
    self.scan_callback,
    qos_profile_sensor_data,
)

용도	권장 출발점	이유
카메라·LiDAR	BEST_EFFORT, KEEP_LAST, 작은 depth	유실보다 지연과 오래된 데이터 누적을 줄임
명령·상태	RELIABLE, KEEP_LAST	작은 메시지의 전달 신뢰성 중시
정적 지도	RELIABLE, TRANSIENT_LOCAL	늦게 들어온 subscriber도 마지막 값 수신

중요: BEST_EFFORT publisher는 RELIABLE subscriber의 요구를 만족하지 못합니다. 메시지가 안 오면 먼저 ros2 topic info /토픽 -v로 양쪽 QoS를 비교합니다.

9. 자주 발생하는 오류

증상	원인과 해결
Package not found	source ~/ros2_ws/install/setup.bash 누락 확인
No executable found	setup.py entry point 이름과 module 경로 확인 후 재빌드
Python 수정이 반영되지 않음	--symlink-install로 빌드하고 실행 중 노드 재시작
Topic은 보이지만 메시지 없음	이름, 타입, domain, QoS를 ros2 topic info -v로 확인
callback이 가끔 멈춤	callback 내부 blocking 작업을 worker나 별도 설계로 분리

• package.xml에 직접 import한 ROS 패키지 의존성을 선언했다.
• Topic 이름과 메시지 타입이 양쪽에서 같다.
• QoS를 의도에 맞게 명시했다.
• 빌드 후 overlay를 source했다.
• Ctrl+C 종료 시 노드와 rclpy가 정리된다.

마무리

rclpy 노드는 “Node 생성 → Parameter·QoS 설정 → publisher/subscription/timer 생성 → spin → 종료 정리”의 흐름으로 작성합니다. 다음 편에서는 같은 기능을 rclcpp와 C++17로 구현하고 CMake 설정까지 비교합니다.

참고 자료

• ROS 2 Jazzy: Writing a simple publisher and subscriber (Python)
• ROS 2 Jazzy: Creating a package
• ROS 2 Jazzy QoS settings
• robotics-agent-skills: ROS 2 development