[Physical AI W1D2] 4/6 — ROS 2 토픽 직접 만들기: Publisher·Subscriber

[Physical AI W1D2 · 4/6]

개념으로 잡은 토픽을 손으로 만든다. ROS 2 Humble 환경을 준비하고 패키지를 생성한 뒤, rclpy로 Publisher와 Subscriber 노드를 작성해 colcon으로 빌드하고 ros2 run으로 실행 — 메시지가 흐르는 걸 직접 확인한다.

이 글에서 직접 만드는 것

• ROS 2 Humble 실습 환경 준비(Python 3.10 + 패키지 설치)
• ros2 pkg create로 Python 패키지 생성
• ros2 node·ros2 topic CLI로 통신 들여다보기
• simple_publisher.py / simple_subscriber.py 작성 → colcon build → 실행

(3편의 개념을 코드로 구현하는 단계입니다. Day1에서 ROS 2를 한 번 깔아봤다면 빠르게 따라올 수 있습니다.)

---

1. 실습 환경 준비

ROS 2 Humble은 Python 3.10과 짝입니다(Day1에서 본 그것). Colab은 기본 python3가 3.12로 잡힐 수 있으니 먼저 맞춥니다.

# Python 3.10 적용
sudo ln -sf /usr/bin/python3.10 /usr/bin/python3
hash -r
python3 --version          # → Python 3.10.x

기본 도구와 ROS 2 저장소를 등록하고 설치합니다.

# 기본 도구
sudo apt update
sudo apt install -y curl gnupg lsb-release software-properties-common tree nano

# ROS 2 Humble 저장소 등록
sudo add-apt-repository universe -y
sudo curl -sSL https://raw.githubusercontent.com/ros/rosdistro/master/ros.key \
  -o /usr/share/keyrings/ros-archive-keyring.gpg
echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/ros-archive-keyring.gpg] http://packages.ros.org/ros2/ubuntu jammy main" \
| sudo tee /etc/apt/sources.list.d/ros2.list > /dev/null
sudo apt update

# ROS 2 + 실습 패키지 설치
sudo apt install -y \
  ros-humble-ros-base \
  ros-humble-demo-nodes-cpp \
  ros-humble-demo-nodes-py \
  ros-humble-example-interfaces \
  ros-humble-std-msgs \
  ros-humble-action-msgs \
  python3-colcon-common-extensions \
  python3-argcomplete

• example-interfaces·std-msgs·action-msgs — 토픽·서비스·액션 실습용 메시지 타입(5·6편에서 사용).
• colcon-common-extensions — colcon build 도구.

환경을 적용하고 확인합니다.

source /opt/ros/humble/setup.bash
python3 --version       # Python 3.10.x
which ros2              # /opt/ros/humble/bin/ros2
ros2 topic list         # /parameter_events, /rosout
colcon --help

💡 Colab은 런타임이 초기화될 수 있어, .bashrc에 고정하기보다 셀/터미널마다 source /opt/ros/humble/setup.bash 를 직접 실행하는 게 안전합니다.

talker/listener 사전 확인

본격 실습 전 ROS 2 기본 통신이 사는지 확인합니다.

source /opt/ros/humble/setup.bash
ros2 run demo_nodes_cpp talker > talker.log 2>&1 &   # 백그라운드
sleep 2
timeout 5 ros2 run demo_nodes_py listener
tail -n 10 talker.log

성공 기준: listener에 I heard: [Hello World: ...], talker 로그에 Publishing: 'Hello World: ...'.

---

2. 워크스페이스 & 패키지 생성

ROS 2 작업은 워크스페이스(패키지를 모아 빌드하는 작업 공간)에서 합니다.

source /opt/ros/humble/setup.bash
echo $ROS_DISTRO            # humble

# 워크스페이스 생성
mkdir -p ~/ros2_comm_ws/src
cd ~/ros2_comm_ws

# Python 패키지 생성
cd ~/ros2_comm_ws/src
ros2 pkg create ros2_comm_practice --build-type ament_python \
  --dependencies rclpy std_msgs example_interfaces action_msgs

# 구조 확인
cd ~/ros2_comm_ws
tree src/ros2_comm_practice

• --build-type ament_python — Python 패키지로 생성(6편에서 C++의 ament_cmake와 비교).
• --dependencies … — 이 패키지가 쓸 의존성을 미리 등록(package.xml에 반영).

---

3. 노드 CLI 실습 — "지금 뭐가 도나"

먼저 예제 노드로 노드/토픽 개념을 CLI로 확인합니다.

# 터미널 1 — talker (백그라운드)
source /opt/ros/humble/setup.bash
ros2 run demo_nodes_cpp talker > talker.log 2>&1 &

# 터미널 2 — listener
source /opt/ros/humble/setup.bash
ros2 run demo_nodes_py listener

# 실행 중 노드 목록
ros2 node list
# /listener
# /talker

# 특정 노드 정보
ros2 node info /talker
# Publishers:
#   /chatter: std_msgs/msg/String

→ /talker가 /chatter 토픽으로 문자열을 발행 중이라는 뜻입니다.

---

4. 토픽 CLI 실습 — 흐르는 데이터 들여다보기

ros2 topic list                 # /chatter, /parameter_events, /rosout
ros2 topic type /chatter        # std_msgs/msg/String
ros2 topic echo /chatter        # 실제 메시지를 화면에 흘려봄
# data: 'Hello World: 1'
# ---
ros2 topic hz /chatter          # 발행 주기(Hz)
ros2 topic info /chatter
# Type: std_msgs/msg/String
# Publisher count: 1
# Subscription count: 1

💡 ros2 topic echo는 토픽 디버깅의 1순위 도구 — "데이터가 진짜 흐르나?"를 눈으로 확인합니다(Day1의 ros2 topic echo /clock과 같은 용도).

---

5. 토픽 Publisher 코드 작성

이제 직접 발행 노드를 만듭니다.

cd ~/ros2_comm_ws/src/ros2_comm_practice/ros2_comm_practice
nano simple_publisher.py

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


class SimplePublisher(Node):
    def __init__(self):
        super().__init__('simple_publisher')

        self.publisher_ = self.create_publisher(
            String,
            'robot_status',
            10
        )

        self.count = 0

        self.timer = self.create_timer(
            1.0,
            self.publish_message
        )

    def publish_message(self):
        msg = String()
        msg.data = f'Robot is running. Count: {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 = SimplePublisher()

    rclpy.spin(node)

    node.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

simple_publisher 노드를 만들어 robot_status 토픽으로 1초마다 문자열을 발행합니다. 핵심 구조:

코드	의미
Node	ROS 2 노드 클래스(상속)
create_publisher(String, 'robot_status', 10)	토픽 발행자 생성(타입·이름·큐크기)
create_timer(1.0, …)	1초마다 콜백 실행
publish()	메시지 발행
rclpy.spin()	노드를 계속 살려둠

• 마지막 인자 10은 큐 크기(처리 못한 메시지를 몇 개까지 쌓을지) — Day1의 /clock 구독에서도 봤던 그 값입니다.

---

6. 토픽 Subscriber 코드 작성

cd ~/ros2_comm_ws/src/ros2_comm_practice/ros2_comm_practice
nano simple_subscriber.py

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


class SimpleSubscriber(Node):
    def __init__(self):
        super().__init__('simple_subscriber')

        self.subscription = self.create_subscription(
            String,
            'robot_status',
            self.listener_callback,
            10
        )

    def listener_callback(self, msg):
        self.get_logger().info(f'Received: {msg.data}')


def main(args=None):
    rclpy.init(args=args)

    node = SimpleSubscriber()

    rclpy.spin(node)

    node.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

robot_status 토픽을 구독하다가, 메시지가 오면 listener_callback()이 실행됩니다. Publisher와 Subscriber는 같은 토픽 이름 robot_status + 같은 타입 String 으로만 연결됩니다(3편의 "느슨한 연결").

---

7. setup.py에 실행 등록

작성한 Python 파일을 ros2 run으로 실행하려면 setup.py의 entry_points에 등록해야 합니다.

cd ~/ros2_comm_ws/src/ros2_comm_practice
nano setup.py

entry_points={
    'console_scripts': [
        'simple_publisher = ros2_comm_practice.simple_publisher:main',
        'simple_subscriber = ros2_comm_practice.simple_subscriber:main',
    ],
},

• 여기에 등록한 이름(simple_publisher)으로 ros2 run을 실행하게 됩니다. 패키지.파일:main 형식이 핵심.

---

8. 빌드 & 실행

cd ~/ros2_comm_ws
colcon build                    # 워크스페이스 전체 빌드
source install/setup.bash       # 빌드 결과를 현재 터미널에 반영

# 터미널 1 — Publisher
ros2 run ros2_comm_practice simple_publisher

# 터미널 2 — Subscriber
source /opt/ros/humble/setup.bash
cd ~/ros2_comm_ws
source install/setup.bash
ros2 run ros2_comm_practice simple_subscriber

💡 Colab은 터미널이 하나입니다. "터미널 1/2"는 개념 구분일 뿐 — Colab에선 발행자를 백그라운드(& + 로그)로 띄우고 구독자를 포그라운드로 실행하세요.

# Colab 단일 터미널 패턴
source /opt/ros/humble/setup.bash && cd ~/ros2_comm_ws && source install/setup.bash
ros2 run ros2_comm_practice simple_publisher > pub.log 2>&1 &   # 발행자: 백그라운드
ros2 run ros2_comm_practice simple_subscriber                    # 구독자: 포그라운드(Ctrl+C로 종료)
tail -n 5 pub.log         # 발행자 로그 확인
pkill -f simple_publisher # 끝나면 백그라운드 정리

Publisher 쪽:

[INFO] [simple_publisher]: Published: Robot is running. Count: 0
[INFO] [simple_publisher]: Published: Robot is running. Count: 1

Subscriber 쪽:

[INFO] [simple_subscriber]: Received: Robot is running. Count: 0
[INFO] [simple_subscriber]: Received: Robot is running. Count: 1

토픽이 생겼는지 확인:

ros2 topic list             # /robot_status
ros2 topic echo /robot_status

⚠️ 흔한 함정 — source install/setup.bash 누락 — 빌드 후 이 명령을 안 하면 새 노드를 못 찾습니다("Package not found"). 빌드 → source → 실행을 한 세트로 기억하세요. 새 터미널을 열 때마다 source /opt/ros/humble/setup.bash(ROS 기본) + source install/setup.bash(내 워크스페이스) 둘 다 필요합니다.

---

4편 정리

• 환경: ROS 2 Humble + Python 3.10, 매 터미널 source /opt/ros/humble/setup.bash.
• 흐름: 워크스페이스 → ros2 pkg create(ament_python) → 코드 작성 → setup.py 등록 → colcon build → source install/setup.bash → ros2 run.
• 토픽 코드 핵심: create_publisher / create_subscription, 같은 이름+타입으로 연결, rclpy.spin()으로 유지.
• CLI: ros2 node list/info, ros2 topic list/type/echo/hz/info.

이 빌드·실행 사이클은 이후 모든 ROS 2 실습에서 반복됩니다.

다음 편 예고

5편에서는 서비스(두 정수를 더하는 add_two_ints Server/Client)와 액션(피보나치 수열을 피드백과 함께 계산하는 fibonacci Server/Client)을 같은 방식으로 구현하고 CLI로 호출합니다.

📚 Week1 Day2 전체 목차 (총 6편)

• 1/6 리눅스 기초 — 쉘·파일시스템·핵심 명령어
• 2/6 리눅스 실전 8 시나리오
• 3/6 ROS 2 통신 4종 개념 — 노드·토픽·서비스·액션
• 4/6 ROS 2 토픽 Pub/Sub 직접 만들기 — 이번 글
• 5/6 ROS 2 서비스·액션 구현
• 6/6 ROS 2 패키지 빌드 — Python·C++·colcon