[ROS2] Usage and Node Coding Guide

ROS2는 로봇 프로그램을 여러 node로 나누고, node들이 topic, service, action으로 통신하게 만드는 middleware이다. 드론, 로봇팔, 자율주행, Gazebo simulation을 다룰 때 ROS2를 쓰면 센서 처리, 제어, 시각화, logging, launch 구성을 표준화할 수 있다.

이 글은 ROS2를 어떻게 쓰는지, Python으로 node 코드는 어떻게 짜는지, package와 launch file은 어떻게 구성하는지 정리한다. 기준 환경은 Ubuntu 22.04 + ROS2 Humble이다.

ROS2 코드는 하나의 큰 프로그램으로 짜기보다, 역할별 node를 만들고 topic/service/action으로 연결하는 방식으로 설계한다.

ROS2에서 자주 쓰는 개념

대부분의 robot application은 다음 흐름으로 구성된다.

sensor node
  -> topic
  -> perception / state estimation node
  -> topic
  -> controller node
  -> command topic
  -> robot / simulator

PX4나 Gazebo를 붙여도 구조는 비슷하다. PX4 topic, Gazebo bridge topic, custom controller node가 ROS2 graph 안에서 연결된다.

1. 환경 설정

ROS2를 설치한 뒤 새 terminal을 열면 먼저 setup script를 source한다.

source /opt/ros/humble/setup.bash

매번 입력하기 싫으면 shell 설정에 추가한다.

echo "source /opt/ros/humble/setup.bash" >> ~/.bashrc

zsh를 쓴다면 ~/.zshrc에 넣는다.

echo "source /opt/ros/humble/setup.zsh" >> ~/.zshrc

ROS2 명령이 동작하는지 확인한다.

ros2 --help
ros2 node list
ros2 topic list

아무 node도 실행하지 않았다면 node list와 topic list가 비어 있거나 기본 daemon 관련 항목만 보일 수 있다.

2. Workspace 만들기

ROS2 package는 workspace 안에서 관리한다.

mkdir -p ~/ros2_ws/src
cd ~/ros2_ws

처음에는 빈 workspace를 build해 본다.

colcon build

build가 끝나면 workspace setup을 source한다.

source install/setup.bash

workspace 구조는 보통 다음과 같다.

ros2_ws/
  src/
    my_robot_tools/
      package.xml
      setup.py
      my_robot_tools/
        __init__.py
  build/
  install/
  log/

ROS2에서는 /opt/ros/humble을 먼저 source하고, 그 다음 workspace의 install/setup.bash를 source한다.

3. Python package 만들기

src로 이동한다.

cd ~/ros2_ws/src

Python package를 만든다.

ros2 pkg create my_robot_tools \
  --build-type ament_python \
  --dependencies rclpy std_msgs geometry_msgs sensor_msgs

생성된 package는 대략 다음 구조를 가진다.

my_robot_tools/
  package.xml
  setup.py
  setup.cfg
  resource/
    my_robot_tools
  my_robot_tools/
    __init__.py

여기에 node file을 추가한다.

my_robot_tools/
  my_robot_tools/
    __init__.py
    heartbeat_publisher.py
    heartbeat_listener.py
    velocity_commander.py
    imu_monitor.py

4. Publisher node 작성

가장 단순한 publisher는 timer callback에서 message를 주기적으로 발행한다.

my_robot_tools/heartbeat_publisher.py:

import rclpy
from rclpy.node import Node
from std_msgs.msg import String


class HeartbeatPublisher(Node):
    def __init__(self):
        super().__init__("heartbeat_publisher")
        self.publisher = self.create_publisher(String, "heartbeat", 10)
        self.count = 0
        self.timer = self.create_timer(1.0, self.publish_heartbeat)

    def publish_heartbeat(self):
        msg = String()
        msg.data = f"alive {self.count}"
        self.publisher.publish(msg)
        self.get_logger().info(f"published: {msg.data}")
        self.count += 1


def main(args=None):
    rclpy.init(args=args)
    node = HeartbeatPublisher()
    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

핵심은 네 가지다.

QoS depth 10은 subscriber가 잠깐 늦을 때 message queue를 몇 개까지 둘지 정하는 값이다. 처음에는 10으로 시작해도 충분하다.

5. Subscriber node 작성

subscriber는 topic message가 들어올 때 callback을 실행한다.

my_robot_tools/heartbeat_listener.py:

import rclpy
from rclpy.node import Node
from std_msgs.msg import String


class HeartbeatListener(Node):
    def __init__(self):
        super().__init__("heartbeat_listener")
        self.subscription = self.create_subscription(
            String,
            "heartbeat",
            self.on_heartbeat,
            10,
        )

    def on_heartbeat(self, msg):
        self.get_logger().info(f"received: {msg.data}")


def main(args=None):
    rclpy.init(args=args)
    node = HeartbeatListener()
    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

publisher와 subscriber를 둘 다 실행하면 heartbeat topic을 통해 문자열이 전달된다.

Publisher는 topic에 쓰고, subscriber는 topic을 읽는다. 두 node는 서로를 직접 호출하지 않는다.

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

Python node를 ros2 run으로 실행하려면 setup.py의 entry_points에 등록한다.

from setuptools import setup

package_name = "my_robot_tools"

setup(
    name=package_name,
    version="0.0.0",
    packages=[package_name],
    data_files=[
        ("share/ament_index/resource_index/packages", [f"resource/{package_name}"]),
        (f"share/{package_name}", ["package.xml"]),
    ],
    install_requires=["setuptools"],
    zip_safe=True,
    maintainer="br4c3",
    maintainer_email="br4c3@example.com",
    description="ROS2 Python node examples",
    license="MIT",
    tests_require=["pytest"],
    entry_points={
        "console_scripts": [
            "heartbeat_publisher = my_robot_tools.heartbeat_publisher:main",
            "heartbeat_listener = my_robot_tools.heartbeat_listener:main",
        ],
    },
)

workspace root로 돌아가 build한다.

cd ~/ros2_ws
colcon build --packages-select my_robot_tools
source install/setup.bash

publisher를 실행한다.

ros2 run my_robot_tools heartbeat_publisher

다른 terminal에서 subscriber를 실행한다.

source /opt/ros/humble/setup.bash
source ~/ros2_ws/install/setup.bash
ros2 run my_robot_tools heartbeat_listener

7. Topic 명령어로 디버깅

node를 만들 때는 CLI로 ROS graph를 계속 확인한다.

node 목록:

ros2 node list

topic 목록:

ros2 topic list

topic type 확인:

ros2 topic info /heartbeat

message 출력:

ros2 topic echo /heartbeat

발행 주기 확인:

ros2 topic hz /heartbeat

message type 구조 확인:

ros2 interface show std_msgs/msg/String
ros2 interface show geometry_msgs/msg/Twist

이 명령어들만 익숙해져도 node 연결 문제의 대부분을 좁힐 수 있다.

8. 드론/로봇 속도 명령 publisher

로봇 제어 예제에서 자주 쓰는 message가 geometry_msgs/msg/Twist이다. 선속도와 각속도를 담는다.

my_robot_tools/velocity_commander.py:

import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist


class VelocityCommander(Node):
    def __init__(self):
        super().__init__("velocity_commander")
        self.publisher = self.create_publisher(Twist, "cmd_vel", 10)
        self.declare_parameter("linear_x", 0.2)
        self.declare_parameter("angular_z", 0.0)
        self.timer = self.create_timer(0.1, self.publish_command)

    def publish_command(self):
        linear_x = self.get_parameter("linear_x").value
        angular_z = self.get_parameter("angular_z").value

        msg = Twist()
        msg.linear.x = float(linear_x)
        msg.angular.z = float(angular_z)
        self.publisher.publish(msg)


def main(args=None):
    rclpy.init(args=args)
    node = VelocityCommander()
    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

setup.py에 entry point를 추가한다.

"velocity_commander = my_robot_tools.velocity_commander:main",

build 후 실행한다.

colcon build --packages-select my_robot_tools
source install/setup.bash
ros2 run my_robot_tools velocity_commander

parameter를 실행 시점에 바꿀 수 있다.

ros2 run my_robot_tools velocity_commander \
  --ros-args \
  -p linear_x:=0.5 \
  -p angular_z:=0.2

주의할 점은 PX4 offboard control에서는 보통 /cmd_vel을 그대로 쓰지 않는다는 것이다. PX4는 px4_msgs의 TrajectorySetpoint, VehicleCommand, OffboardControlMode 같은 message를 사용한다. 그래도 Twist publisher 예제는 ROS2 publisher 구조를 이해하는 데 좋다.

9. Sensor subscriber 예제

IMU를 받는 node는 다음처럼 작성한다.

my_robot_tools/imu_monitor.py:

import math

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Imu


class ImuMonitor(Node):
    def __init__(self):
        super().__init__("imu_monitor")
        self.declare_parameter("imu_topic", "/imu")
        imu_topic = self.get_parameter("imu_topic").value
        self.subscription = self.create_subscription(
            Imu,
            imu_topic,
            self.on_imu,
            10,
        )
        self.get_logger().info(f"subscribing: {imu_topic}")

    def on_imu(self, msg):
        ax = msg.linear_acceleration.x
        ay = msg.linear_acceleration.y
        az = msg.linear_acceleration.z
        accel_norm = math.sqrt(ax * ax + ay * ay + az * az)
        self.get_logger().info(f"accel norm: {accel_norm:.3f} m/s^2")


def main(args=None):
    rclpy.init(args=args)
    node = ImuMonitor()
    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

Gazebo bridge로 /gazebo/imu 같은 topic이 ROS2에 들어온다면 parameter로 바꿔 실행한다.

ros2 run my_robot_tools imu_monitor \
  --ros-args \
  -p imu_topic:=/gazebo/imu

코드 안에 topic 이름을 고정하기보다 parameter로 빼두면 Gazebo, 실제 로봇, rosbag replay를 오갈 때 수정이 줄어든다.

10. Launch file 작성

publisher와 subscriber를 매번 terminal 여러 개에서 실행하면 번거롭다. launch file로 묶는다.

package root에 launch directory를 만든다.

mkdir -p ~/ros2_ws/src/my_robot_tools/launch

launch/heartbeat.launch.py:

from launch import LaunchDescription
from launch_ros.actions import Node


def generate_launch_description():
    return LaunchDescription([
        Node(
            package="my_robot_tools",
            executable="heartbeat_publisher",
            name="heartbeat_publisher",
            output="screen",
        ),
        Node(
            package="my_robot_tools",
            executable="heartbeat_listener",
            name="heartbeat_listener",
            output="screen",
        ),
    ])

Python package에서 launch file이 설치되도록 setup.py의 data_files에 추가한다.

import os
from glob import glob

data_files=[
    ("share/ament_index/resource_index/packages", [f"resource/{package_name}"]),
    (f"share/{package_name}", ["package.xml"]),
    (os.path.join("share", package_name, "launch"), glob("launch/*.launch.py")),
],

build 후 launch한다.

cd ~/ros2_ws
colcon build --packages-select my_robot_tools
source install/setup.bash
ros2 launch my_robot_tools heartbeat.launch.py

11. Parameter YAML 사용

parameter가 많아지면 command line보다 YAML이 낫다.

config/velocity.yaml:

velocity_commander:
  ros__parameters:
    linear_x: 0.5
    angular_z: 0.2

launch file에서 parameter file을 넘긴다.

import os

from ament_index_python.packages import get_package_share_directory
from launch import LaunchDescription
from launch_ros.actions import Node


def generate_launch_description():
    package_share = get_package_share_directory("my_robot_tools")
    params = os.path.join(package_share, "config", "velocity.yaml")

    return LaunchDescription([
        Node(
            package="my_robot_tools",
            executable="velocity_commander",
            name="velocity_commander",
            output="screen",
            parameters=[params],
        ),
    ])

setup.py에도 config 설치를 추가한다.

(os.path.join("share", package_name, "config"), glob("config/*.yaml")),

12. Service는 언제 쓰나

topic은 계속 흐르는 data stream에 적합하다. service는 "한 번 요청하고 응답을 받는" 작업에 적합하다.

예시:

처음 ROS2 코드를 짤 때는 topic부터 익히고, reset이나 mode change 같은 명령을 service로 분리하면 된다. navigation처럼 시간이 오래 걸리고 feedback이 필요한 작업은 action이 더 적합하다.

13. Gazebo, PX4와 연결할 때의 코드 구조

Gazebo + PX4 + ROS2 프로젝트에서는 node를 다음처럼 나누는 편이 좋다.

my_drone_stack/
  drone_state_node
    subscribes: /fmu/out/vehicle_odometry
  offboard_controller_node
    publishes: /fmu/in/trajectory_setpoint
    publishes: /fmu/in/offboard_control_mode
    publishes: /fmu/in/vehicle_command
  gazebo_sensor_node
    subscribes: /gazebo/imu, /camera/image, /scan
  mission_manager_node
    coordinates state and controller

역할을 섞으면 디버깅이 어려워진다. 예를 들어 Gazebo sensor 확인 node와 PX4 offboard control node는 분리하는 편이 낫다.

Gazebo topic
  -> ros_gz_bridge
  -> ROS2 sensor topic
  -> sensor processing node

PX4 uORB
  -> Micro XRCE-DDS Agent
  -> ROS2 /fmu/out/*
  -> controller node
  -> ROS2 /fmu/in/*
  -> PX4

14. 자주 하는 실수

workspace setup을 source하지 않음

build 후에도 새 executable을 못 찾으면 setup을 source했는지 확인한다.

source ~/ros2_ws/install/setup.bash

setup.py entry point를 빼먹음

file을 만들었는데 ros2 run에서 executable이 안 보이면 entry_points를 확인한다.

ros2 pkg executables my_robot_tools

topic 이름이 다름

상대 node가 실제로 어떤 topic을 쓰는지 확인한다.

ros2 node info /node_name
ros2 topic list

message type이 다름

topic 이름이 같아도 type이 다르면 통신이 안 된다.

ros2 topic info /topic_name

callback에서 너무 오래 걸리는 작업을 함

subscriber callback에서 heavy computation이나 blocking I/O를 오래 잡고 있으면 message 처리가 밀린다. 처음에는 단순하게 작성하고, 필요해지면 timer, queue, multi-thread executor 구조를 검토한다.

실전 workflow

ROS2 node를 만들 때는 다음 순서로 진행한다.

1. ros2 pkg create로 package를 만든다.
2. publisher 또는 subscriber node 하나만 먼저 작성한다.
3. setup.py entry point에 등록한다.
4. colcon build --packages-select <package>로 build한다.
5. source install/setup.bash를 한다.
6. ros2 run으로 node를 실행한다.
7. ros2 topic list, ros2 topic echo, ros2 node info로 연결을 확인한다.
8. parameter를 추가한다.
9. launch file로 여러 node를 묶는다.
10. Gazebo/PX4 bridge topic에 연결한다.

ROS2는 처음부터 큰 stack을 만들면 어렵다. 작은 publisher와 subscriber를 만든 뒤, topic type과 node graph를 CLI로 확인하면서 확장하는 방식이 가장 안정적이다.

참고 자료

• ROS2 Humble - Creating your first ROS2 package
• ROS2 Humble - Writing a simple publisher and subscriber, Python
• ROS2 Humble - Using parameters in a class, Python
• ROS2 Humble - Creating a launch file
• ROS2 Humble - About topic, service, and action interfaces