Code 스튜디오 — ROS 2 코드를 빠르게 읽고 설명하는 법

1. Code 스튜디오의 목표 — 번역이 아니라 동작을 설명한다

코드를 설명할 때 create_publisher는 Publisher를 만든다처럼 문장을 한국어로 바꾸는 데 그치면 실제 동작을 이해하기 어렵습니다. 좋은 해석은 다음 네 질문에 답합니다.

1. 한 줄 목적 — 이 Node는 무엇을 받아 무엇을 만드는가?
2. ROS Graph 계약 — Topic·Service·Action·Parameter의 이름, Type과 QoS는 무엇인가?
3. 실행 흐름 — 어느 Event가 어떤 Callback을 부르고 상태가 어떻게 바뀌는가?
4. 실패와 안전 — Data가 없거나 늦고, 값이 비정상이고, Process가 종료되면 어떻게 되는가?

예를 들어 SafeDriver를 빠르게 설명하면 다음과 같습니다.

/scan의 LaserScan을 받아 가장 가까운 장애물을 계산하고, 10 Hz Timer가 /cmd_vel의 Twist를 발행한다. 최근 Scan이 0.5초 이상 없거나 장애물이 stop_distance보다 가까우면 선속도를 0으로 만든다. 종료할 때도 빈 Twist를 한 번 발행한다.

이 한 문장에는 입력, 계산, 출력, 주기, Parameter와 안전 동작이 모두 들어 있습니다. 세부 문법은 이 지도를 만든 뒤 읽습니다.

30초 빠른 해석 순서

코드 설명 카드 표준

앞으로 예제마다 다음 형식으로 해석하면 처음 보는 코드도 빠뜨리는 부분이 줄어듭니다.

[한 줄 목적]
무엇을 입력받아 어떤 판단을 한 뒤 무엇을 출력하는가?

[Graph 계약]
입력 /scan : sensor_msgs/msg/LaserScan
출력 /cmd_vel : geometry_msgs/msg/Twist
Parameter stop_distance : double, 단위 m

[Trigger와 흐름]
Scan 도착 → on_scan → nearest 갱신
0.1초 Timer → tick → 안전 조건 검사 → Twist 발행

[상태]
nearest, last_scan_time

[안전과 실패]
첫 Scan 전 정지, 0.5초 Timeout 정지, 종료 시 정지

[검증]
ros2 node info, ros2 topic echo, ros2 topic hz, Parameter 변경

문법·ROS 의미·Robot 의미를 분리해서 설명한다

같은 한 줄도 세 층으로 설명해야 합니다.

self.timer = self.create_timer(0.1, self.tick)

• Python 문법: self.tick은 함수를 실행한 결과가 아니라 호출 가능한 Method 자체입니다. 따라서 괄호를 붙이지 않습니다.
• ROS 2 의미: Executor가 처리할 0.1초 주기의 Timer Entity와 Callback을 등록합니다.
• Robot 의미: 목표 속도를 약 10 Hz로 다시 계산합니다. 그러나 OS Scheduling 때문에 정확한 Hard Real-time 10 Hz를 보장하지는 않습니다.

이 구분을 적용하면 “코드가 무엇을 쓰는가”뿐 아니라 “왜 이 코드가 Robot에서 필요한가”까지 설명할 수 있습니다.

2. 모든 rclpy 노드는 같은 뼈대를 가진다

남이 짠 ROS 2 노드를 처음 열면 낯설어 보이지만, 사실 거의 모든 노드가 같은 여섯 부분으로 이루어져 있습니다. 이 구조를 외워 두면 처음 보는 코드도 30초 만에 지도를 그릴 수 있습니다.

① import — rclpy, Node, 그리고 쓰는 메시지 타입들. 여기만 봐도 이 노드가 어떤 데이터를 다루는지 절반은 알 수 있습니다. LaserScan이 보이면 라이다를, Twist가 보이면 주행 명령을 다루는 노드입니다.

② 클래스 선언 — class MyNode(Node): 노드는 관례적으로 Node를 상속한 클래스로 만듭니다. 함수만으로도 쓸 수 있지만, 상태를 들고 있어야 하는 순간이 반드시 오므로 처음부터 클래스로 쓰세요.

③ __init__의 첫 줄 — super().__init__("이름") 이 이름이 ros2 node list에 뜨는 이름입니다. 파일 이름도 클래스 이름도 아닙니다. 초보자가 "내 노드가 안 보인다"고 할 때 여기부터 확인하면 됩니다.

④ 통신 구성 — 파라미터 선언, 발행자, 구독자, 서비스, 타이머 생성. 노드의 입출력 명세가 전부 여기 있습니다. 코드를 읽을 때 가장 먼저 볼 곳입니다.

⑤ 콜백 함수들 — 실제 로직. 구독 콜백, 타이머 콜백, 서비스 콜백. 각각 짧아야 합니다.

이 뼈대는 C++(rclcpp)에서도 똑같습니다. 언어가 달라도 구조는 같습니다.

rclpy 노드의 표준 뼈대. 여섯 부분의 위치와 역할을 주석으로 표시했습니다.

#!/usr/bin/env python3
# rclpy 노드의 표준 뼈대. 여섯 부분이 어디인지 주석으로 표시했다.

# ---- ① import -------------------------------------------------
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
from sensor_msgs.msg import LaserScan


# ---- ② 클래스 선언 --------------------------------------------
class SafeDriver(Node):

    def __init__(self):
        # ---- ③ 노드 이름. ros2 node list 에 뜨는 것이 바로 이 문자열이다 ----
        super().__init__("safe_driver")

        # ---- ④ 통신 구성. 이 노드의 입출력 명세가 전부 여기 있다 ----
        self.declare_parameter("stop_distance", 0.4)
        self.declare_parameter("cruise_speed", 0.25)
        self.stop_distance = self.get_parameter("stop_distance").value
        self.cruise_speed = self.get_parameter("cruise_speed").value

        self.pub = self.create_publisher(Twist, "cmd_vel", 10)
        self.sub = self.create_subscription(
            LaserScan, "scan", self.on_scan, 10)
        self.timer = self.create_timer(0.1, self.tick)   # 괄호를 붙이지 않는다

        self.nearest = float("inf")
        self.last_scan_time = None

        self.get_logger().info(
            f"safe_driver 시작 — stop_distance={self.stop_distance} m")

    # ---- ⑤ 콜백들. 각각 짧게 유지한다 ----------------------------
    def on_scan(self, msg: LaserScan):
        valid = [r for r in msg.ranges if r > 0.0]
        self.nearest = min(valid) if valid else float("inf")
        self.last_scan_time = self.get_clock().now()

    def tick(self):
        cmd = Twist()

        # 센서가 끊긴 지 오래되었으면 무조건 정지 (deadman timeout)
        if self.last_scan_time is None:
            cmd.linear.x = 0.0
        else:
            age = (self.get_clock().now() - self.last_scan_time).nanoseconds / 1e9
            if age > 0.5:
                cmd.linear.x = 0.0
                self.get_logger().warn(
                    f"스캔이 {age:.1f} 초째 없습니다. 정지합니다.",
                    throttle_duration_sec=2.0)
            elif self.nearest < self.stop_distance:
                cmd.linear.x = 0.0
            else:
                cmd.linear.x = self.cruise_speed

        self.pub.publish(cmd)


# ---- ⑥ main. 이 형태를 외워 두면 매번 고민할 필요가 없다 --------
def main(args=None):
    rclpy.init(args=args)
    node = SafeDriver()
    try:
        rclpy.spin(node)               # 여기서 콜백들이 돌기 시작한다
    except KeyboardInterrupt:
        pass
    finally:
        node.pub.publish(Twist())      # 마지막으로 정지 명령을 보낸다
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

SafeDriver 코드 빠른 해석

한 줄 목적: Scan에서 가장 가까운 유효 거리를 기억하고, 주기적으로 안전 조건을 검사해 전진 또는 정지 명령을 발행하는 Node입니다.

생성부터 첫 명령까지의 실행 흐름

1. main()이 ROS Context를 초기화합니다.
2. SafeDriver() 생성 중 Parameter, Publisher, Subscription과 Timer가 등록됩니다.
3. spin()이 Executor Event Loop를 시작합니다.
4. 첫 Scan 전에는 last_scan_time is None이므로 Timer가 빈 Twist, 즉 정지를 발행합니다.
5. Scan이 오면 on_scan()이 유효 거리 중 최솟값과 수신 시각을 저장합니다.
6. 다음 tick()은 Data Age와 nearest를 검사해 정지 또는 cruise_speed를 결정합니다.
7. Ctrl+C가 들어오면 finally가 마지막 정지 명령을 발행하고 Node와 Context를 정리합니다.

중요한 설계는 Sensor 수신과 Command 발행을 분리했다는 점입니다. Scan이 순간적으로 100개 몰려와도 그 횟수만큼 속도 명령을 발행하지 않고 Timer 주기에 맞춰 최신 상태만 사용합니다.

핵심 줄 해설

valid = [r for r in msg.ranges if r > 0.0]
self.nearest = min(valid) if valid else float("inf")

첫 줄은 양수만 남깁니다. 두 번째 줄은 목록이 비어 있을 때 min([]) 예외를 피합니다. 하지만 실제 LaserScan에서는 NaN, 양의 무한대와 range_min·range_max 밖의 값도 고려하는 편이 안전합니다.

valid = [
    r for r in msg.ranges
    if math.isfinite(r) and msg.range_min <= r <= msg.range_max
]

float("inf")를 “장애물 없음”으로 쓰면 Sensor가 전부 고장 난 경우에도 전진할 수 있습니다. 실제 Robot에서는 유효 Sample이 하나도 없을 때 별도의 scan_valid = False 상태를 두고 정지하는 Fail-safe 정책이 더 적절합니다.

age = (self.get_clock().now() - self.last_scan_time).nanoseconds / 1e9

두 ROS Time 객체의 차이는 Duration이고, nanoseconds / 1e9로 초 단위 실수가 됩니다. Node가 use_sim_time=true라면 이 Clock도 /clock을 따르므로 Bag 재생 Pause 중에는 Age가 증가하지 않을 수 있습니다. 현장 Watchdog가 Wall Clock을 요구하는지 ROS Time을 요구하는지 설계 단계에서 정해야 합니다.

실행 전에 예측할 수 있는 결과

• Scan이 한 번도 오지 않음 → /cmd_vel.linear.x = 0.0
• 최근접 거리 0.25 m, 정지 거리 0.4 m → 정지
• 최근접 거리 1.2 m, Scan Age 0.1 s → 0.25 m/s 전진
• 최근접 거리 1.2 m, Scan Age 0.8 s → Sensor Timeout으로 정지
• stop_distance:=1.5로 시작 → 같은 1.2 m 입력에서도 정지

CLI로 설명을 검증한다

# Graph 계약 확인
ros2 node info /safe_driver

# 장애물이 먼 가상 Scan을 한 번 발행
ros2 topic pub --once /scan sensor_msgs/msg/LaserScan \
  "{range_min: 0.1, range_max: 8.0, ranges: [1.2, 2.0, 0.9]}"

# 결과와 주기 확인
ros2 topic echo /cmd_vel
ros2 topic hz /cmd_vel

# 정책 값을 바꾼 뒤 같은 입력으로 결과 비교
ros2 param set /safe_driver stop_distance 1.5

이 예제는 Parameter 값을 __init__에서 한 번만 읽기 때문에 실행 중 param set을 해도 self.stop_distance가 자동 갱신되지 않습니다. 실시간 변경을 지원하려면 Parameter Callback에서 검증하고 내부 상태를 갱신하거나, 사용할 때마다 Parameter를 읽어야 합니다. 이처럼 CLI 명령이 성공했다는 사실과 Algorithm에 반영됐다는 사실은 다릅니다.

개념도 · stack

• 단계/참여자
• import — 어떤 데이터를 다루는가
• 클래스 선언 — Node를 상속한다
• super().init — ros2 node list에 뜨는 이름
• 통신 구성 — 이 노드의 입출력 명세 전부
• 콜백들 — 실제 로직
• main — init, spin, 정리
• 연결/행
• 타입만 봐도 절반은 안다
• ``
• 가장 자주 헷갈리는 곳
• 먼저 볼 곳
• 짧게 유지
• 형태를 외운다

위에서 아래로 읽으면 노드의 정체가 드러납니다. ④가 가장 정보가 많은 곳입니다.

3. main 함수의 정석 — 정리까지가 코드다

main은 짧지만 지켜야 할 것이 있습니다.

rclpy.init()은 노드를 만들기 전에. 초기화 전에 Node()를 만들면 오류가 납니다. 순서가 있습니다.

rclpy.spin(node)가 실행의 중심입니다. 이 줄에 도달하기 전까지는 콜백이 하나도 돌지 않습니다. __init__에서 구독자를 만들었다고 데이터가 들어오는 것이 아닙니다. spin이 Executor를 돌리기 시작해야 비로소 콜백이 불립니다. 이것을 모르면 "구독자를 만들었는데 콜백이 안 불린다"에서 막힙니다.

try / finally로 감쌉니다. Ctrl+C를 누르면 KeyboardInterrupt가 발생합니다. 이것을 잡지 않으면 터미널에 긴 역추적이 쏟아지고, 더 중요하게는 정리 코드가 실행되지 않습니다.

정리 순서가 중요합니다. ① 안전 조치 — 로봇이라면 마지막으로 정지 명령을 발행합니다. 이것을 빠뜨리면 프로그램이 죽어도 로봇은 마지막 속도로 계속 갑니다. ② node.destroy_node() — 발행자·구독자·타이머를 정리합니다 ③ rclpy.shutdown() — 컨텍스트를 닫습니다

main(args=None)으로 받는 이유: launch나 ros2 run이 --ros-args 뒤의 인자를 전달하기 때문입니다. 이것을 rclpy.init(args=args)로 넘겨야 네임스페이스, 리매핑, 파라미터가 적용됩니다. 빠뜨리면 --ros-args -p로 준 파라미터가 무시됩니다.

요점

• rclpy.init()은 반드시 Node 생성보다 먼저 부른다.
• spin에 도달해야 콜백이 돌기 시작한다. __init__만으로는 아무 일도 일어나지 않는다.
• try/finally로 감싸야 Ctrl+C에서도 정리 코드가 실행된다.
• 로봇이라면 정리 단계에서 반드시 정지 명령을 한 번 발행한다.
• main(args=None)을 rclpy.init(args=args)로 넘겨야 --ros-args가 적용된다.

주의 try/finally 없이 짠 노드는 Ctrl+C를 누른 순간 정지 명령을 보내지 못하고 죽습니다. 프로그램은 끝났는데 로봇은 마지막으로 받은 속도로 계속 굴러갑니다. 이것은 이론이 아니라 실제로 자주 일어나는 사고입니다.

4. 처음 보는 노드를 읽는 일곱 단계

남의 코드를 만났을 때 위에서부터 한 줄씩 읽으면 시간이 오래 걸립니다. 순서를 정해 두면 훨씬 빠릅니다.

① super().__init__("...")의 이름을 찾는다. 이 노드가 그래프에서 무엇으로 불리는지 확인합니다.

② create_subscription을 전부 찾는다. — 입력 이 노드가 무엇을 받는지, 어떤 타입인지, 어떤 콜백으로 가는지 목록을 만듭니다.

③ create_publisher를 전부 찾는다. — 출력 무엇을 내보내는지 봅니다. ②와 ③만 알면 이 노드가 그래프에서 무슨 역할인지 파악됩니다.

④ create_timer를 찾는다. — 주기적 동작 주기가 얼마인지 봅니다. 0.1이면 10 Hz입니다. 제어 노드라면 여기가 심장입니다.

⑤ declare_parameter를 전부 찾는다. — 조정 가능한 것 무엇을 바꿔 볼 수 있는지 알 수 있습니다. 실험할 때 여기부터 만집니다.

⑥ 콜백 본문을 읽는다. — 로직 여기서 처음으로 알고리즘을 읽습니다. 앞의 다섯 단계로 맥락을 갖춘 뒤라 훨씬 잘 읽힙니다.

⑦ main을 본다. — 특이 사항 MultiThreadedExecutor를 쓰는지, 여러 노드를 함께 띄우는지 확인합니다.

코드 없이 밖에서 보는 방법도 있습니다. ros2 node info /노드이름을 쓰면 그 노드의 구독·발행·서비스 목록이 그대로 나옵니다. ②③⑤를 코드를 열지 않고 알 수 있는 셈입니다. 남의 패키지를 처음 만났을 때 특히 유용합니다.

코드를 열지 않고 노드를 파악하는 명령들과, 고친 코드를 시험하는 4단계.

# ── 코드를 열지 않고 노드를 파악하는 법 ─────────────────

# 이 노드가 무엇을 구독하고 발행하는가 (읽기 단계 ②③⑤에 해당)
ros2 node info /safe_driver
#   Subscribers:  /scan: sensor_msgs/msg/LaserScan
#   Publishers:   /cmd_vel: geometry_msgs/msg/Twist
#   Service Servers: /safe_driver/get_parameters ...

# 조정 가능한 것은 무엇인가 (읽기 단계 ⑤)
ros2 param list /safe_driver
ros2 param describe /safe_driver stop_distance

# 실제로 얼마나 자주 도는가 (읽기 단계 ④를 실측으로 확인)
ros2 topic hz /cmd_vel

# 그래프에서 누구와 연결되어 있는가
ros2 topic info /cmd_vel --verbose

# 전체 그림
rqt_graph


# ── 고친 코드를 시험하는 순서 ───────────────────────────
# (1) 문법과 타입 확인
python3 -m py_compile src/my_pkg/my_pkg/safe_driver.py

# (2) 재빌드. Python 은 --symlink-install 을 쓰면 다시 빌드하지 않아도 된다
colcon build --packages-select my_pkg --symlink-install
source install/setup.bash

# (3) 파라미터를 바꿔 가며 동작 확인
ros2 run my_pkg safe_driver --ros-args -p stop_distance:=0.6

# (4) 기록해 두고 나중에 비교
ros2 bag record -o tune_06 /scan /cmd_vel

5. 자주 나오는 버그 여섯 가지

아래 여섯 가지는 초보자뿐 아니라 경험자도 반복해서 만드는 것들입니다. 눈으로 잡아낼 수 있게 패턴을 외워 두세요.

① 콜백 등록에 괄호를 붙였다 self.create_timer(1.0, self.tick()) — 잘못입니다. 괄호를 붙이면 지금 즉시 한 번 실행되고, 그 반환값(대개 None)이 콜백으로 등록됩니다. 타이머는 돌지만 아무 일도 하지 않습니다. 괄호를 빼야 함수 자체가 넘어갑니다.

② 자기가 발행하는 토픽을 자기가 구독한다 같은 이름으로 create_publisher와 create_subscription을 만들면 자기 메시지를 자기가 받습니다. 콜백에서 다시 발행하면 무한 피드백 루프가 되어 CPU가 100 %로 치솟습니다. 리매핑을 잘못했을 때도 이렇게 됩니다.

③ 콜백 안에서 time.sleep()을 쓴다 단일 스레드 Executor에서 그 시간 동안 노드 전체가 멈춥니다. "잠깐 기다렸다가"가 필요하면 타이머를 쓰거나 상태 변수로 단계를 나눕니다.

④ 예외를 조용히 삼킨다 except Exception: pass는 최악입니다. 콜백이 매번 실패하는데 아무도 모릅니다. 최소한 self.get_logger().error(...)로 남기세요.

⑤ spin을 콜백 안에서 부른다 rclpy.spin()이나 spin_until_future_complete()를 콜백 안에서 부르면 앞에서 배운 그대로 데드락입니다.

⑥ 정리를 하지 않는다 destroy_node()와 shutdown()이 없으면 프로세스가 지저분하게 끝납니다. 더 중요한 것은 정지 명령을 못 보낸다는 점입니다.

C++를 쓴다면 하나 더: rclcpp에서는 구독자·타이머의 핸들을 반드시 멤버 변수로 보관해야 합니다. 지역 변수에 담으면 함수가 끝나는 순간 소멸되어 조용히 사라집니다. Python은 노드가 내부적으로 참조를 들고 있어 이 문제가 없습니다.

여섯 가지 버그가 모두 들어 있는 나쁜 코드와, 같은 기능을 올바르게 쓴 코드의 대조.

#!/usr/bin/env python3
# ============================================================
# 나쁜 코드 — 여섯 가지 버그가 모두 들어 있다
# ============================================================
import time
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist


class BadNode(Node):
    def __init__(self):
        super().__init__("bad_node")
        # ② 같은 토픽을 발행하고 구독한다 -> 무한 피드백 루프
        self.pub = self.create_publisher(Twist, "cmd_vel", 10)
        self.create_subscription(Twist, "cmd_vel", self.on_cmd, 10)
        # ① 괄호를 붙였다 -> 지금 한 번 실행되고 None 이 등록된다
        self.create_timer(1.0, self.tick())

    def on_cmd(self, msg):
        try:
            time.sleep(2.0)             # ③ 노드 전체가 2초 멈춘다
            self.pub.publish(msg)       # 그리고 자기에게 다시 보낸다
        except Exception:
            pass                        # ④ 조용히 삼킨다

    def tick(self):
        self.get_logger().info("tick")


def main():
    rclpy.init()
    rclpy.spin(BadNode())               # ⑥ 정리가 없다


# ============================================================
# 고친 코드
# ============================================================
class GoodNode(Node):
    def __init__(self):
        super().__init__("good_node")

        # ② 입력과 출력 이름을 분리한다. 리매핑은 launch 에서 한다.
        self.pub = self.create_publisher(Twist, "cmd_vel_out", 10)
        self.sub = self.create_subscription(
            Twist, "cmd_vel_in", self.on_cmd, 10)

        # ① 괄호를 빼서 함수 자체를 넘긴다
        self.timer = self.create_timer(1.0, self.tick)

        self.latest = None
        self.pending_since = None

    def on_cmd(self, msg):
        # ③ sleep 대신 상태만 기록한다. 콜백은 즉시 끝난다.
        try:
            self.latest = msg
            self.pending_since = self.get_clock().now()
        except Exception as error:
            # ④ 삼키지 않고 남긴다. 그리고 안전한 값으로 되돌린다.
            self.get_logger().error(f"명령 처리 실패: {error}")
            self.latest = None

    def tick(self):
        # 기다리는 일은 타이머가 대신한다.
        if self.latest is None or self.pending_since is None:
            return
        waited = (self.get_clock().now() - self.pending_since).nanoseconds / 1e9
        if waited >= 2.0:
            self.pub.publish(self.latest)
            self.latest = None


def good_main(args=None):
    # ⑥ init -> spin -> 안전 조치 -> destroy -> shutdown
    rclpy.init(args=args)
    node = GoodNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.pub.publish(Twist())      # 마지막 정지 명령
        node.destroy_node()
        rclpy.shutdown()

6. 나쁜 코드와 고친 코드의 실행 결과를 비교한다

앞의 BadNode는 단순히 “문법이 좋지 않은 코드”가 아닙니다. 실제 실행 순서를 따라가면 왜 위험한지 분명해집니다.

BadNode 생성 순간

1. BadNode()가 생성됩니다.
2. self.tick()의 괄호 때문에 Timer를 만들기 전에 tick이 즉시 실행되어 Log가 한 번 출력됩니다.
3. tick()의 반환값은 None이므로 create_timer()에 올바른 Callback이 전달되지 않습니다. ROS 2 배포판에 따라 생성 단계에서 Type Error가 발생할 수도 있습니다.
4. 생성에 성공하더라도 cmd_vel에 외부 Message가 한 개 들어오면 on_cmd()가 Executor Thread를 2초 막습니다.
5. 같은 Message를 같은 Topic으로 재발행해 자기 Callback이 다시 예약됩니다. Queue가 쌓이고 다른 Callback은 제때 실행되지 않습니다.

따라서 “Timer가 안 돈다”, “명령이 계속 반복된다”, “Node 응답이 느리다”라는 세 증상이 하나의 코드에서 동시에 나타날 수 있습니다.

GoodNode가 개선한 것

• cmd_vel_in과 cmd_vel_out을 분리해 Data 흐름의 방향을 이름으로 드러냅니다.
• Subscription과 Timer Handle을 멤버에 보관해 수명과 의도를 명확히 합니다.
• Subscription Callback은 최신 Message와 시각만 저장하고 즉시 반환합니다.
• Timer가 경과 시간을 검사하므로 Executor를 막지 않습니다.
• 예외를 Log로 남기고 잘못된 상태를 None으로 되돌립니다.
• finally가 정상 종료와 Ctrl+C 모두에서 정리 동작을 수행합니다.

다만 이 예제도 Production Code로 사용하려면 보완이 필요합니다. 입력 Message를 그대로 2초 후 발행하면 오래된 속도 명령이 될 수 있으므로 Command에는 최대 Age를 두고, Timeout이면 빈 Twist를 발행해야 합니다. 또한 예상하지 못한 예외가 발생했을 때도 Actuator Driver의 독립 Watchdog가 Robot을 정지시켜야 합니다.

수정 전후를 수치로 확인하는 명령

# Process와 CPU 사용률 확인
ps -C python3 -o pid,pcpu,pmem,cmd

# Topic 발행 빈도가 계속 증가하거나 비정상적으로 높은지 확인
ros2 topic hz /cmd_vel

# Publisher와 Subscriber가 자기 자신인지 확인
ros2 topic info /cmd_vel --verbose

# Callback 지연의 간접 증거: 출력 Timestamp와 주기 기록
ros2 bag record -o callback_check /cmd_vel_in /cmd_vel_out
ros2 bag info callback_check

7. 작은 Publisher를 줄 단위로 해석한다

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


class Counter(Node):
    def __init__(self):
        super().__init__('counter')
        self.publisher = self.create_publisher(Int32, 'count', 10)
        self.value = 0
        self.timer = self.create_timer(0.5, self.publish_count)

    def publish_count(self):
        message = Int32()
        message.data = self.value
        self.publisher.publish(message)
        self.get_logger().info(f'count={self.value}')
        self.value += 1


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

빠른 해석: counter Node가 0.5초마다 현재 정수를 /count에 발행하고 값을 1 증가시킵니다. 입력은 없고, 상태는 value 하나이며, 약 2 Hz로 동작합니다.

다음 명령으로 해석이 맞는지 검증합니다.

ros2 node info /counter
ros2 topic type /count
ros2 interface show std_msgs/msg/Int32
ros2 topic echo /count
ros2 topic hz /count

topic echo에서는 data: 0, data: 1, data: 2가 차례로 보이고, topic hz는 약 2 Hz를 보여야 합니다. 이처럼 코드 설명은 예상 가능한 관측 결과까지 포함해야 검증 가능한 설명이 됩니다.

8. Callback을 읽을 때 Data·상태·부작용을 표시한다

Callback 한 개를 해석할 때는 각 줄을 다음 세 부류로 표시하면 빠릅니다.

• Data 입력·계산: Message Field를 읽고 새 값을 계산합니다.
• 상태 변경: self.*에 저장해 다음 Callback에 영향을 줍니다.
• 부작용: Publish, Log, Service 호출, File·Hardware I/O를 수행합니다.

def on_temperature(self, msg):
    celsius = msg.temperature                         # Data
    self.maximum = max(self.maximum, celsius)         # 상태
    if celsius > self.limit:                          # Data + 정책
        self.alarm_pub.publish(Bool(data=True))       # 부작용
        self.get_logger().warn(f'hot: {celsius:.1f}') # 부작용

이 Callback의 한 줄 설명은 “온도를 받아 최고값을 기억하고 Limit를 넘으면 Alarm과 Warning Log를 발생시킨다”입니다. Code Review에서는 다음을 추가로 묻습니다.

• NaN이나 Sensor 오류 값은 걸러지는가?
• limit의 단위와 허용 범위가 명확한가?
• 온도가 내려왔을 때 False Alarm도 발행하는가?
• 매 Message마다 Warning을 남겨 Log 폭주가 발생하지 않는가?
• 상태를 여러 Callback Thread가 동시에 수정할 가능성이 있는가?

9. Code Review 체크리스트

Interface

• Node·Topic·Service·Action 이름이 책임을 드러내는가?
• Message Type, Frame, Timestamp와 단위가 명확한가?
• QoS가 상대 Endpoint와 호환되고 Data 성질에 맞는가?
• Parameter에 Type, 범위, 기본값과 변경 정책이 있는가?

Execution

• Callback을 부르는 Trigger와 목표 주기는 무엇인가?
• Callback 안에 sleep, 동기 Service, 긴 File·Network I/O가 없는가?
• 상태 변수의 초기값과 갱신 순서가 안전한가?
• MultiThreadedExecutor라면 Callback Group과 Lock이 필요한가?

Failure and safety

• 첫 Data가 오기 전 출력은 안전한가?
• Sensor·Command Timeout이 있는가?
• NaN, 무한대, 빈 배열과 범위 밖 값이 처리되는가?
• 예외가 Log와 진단 정보로 남고 안전 상태로 전환되는가?
• Process 종료와 Hang에도 Driver Watchdog·E-stop이 작동하는가?

Verification

• 예상 입력과 예상 출력이 예제로 정의되어 있는가?
• 정상·경계·오류 Case를 Unit Test할 수 있는가?
• node info, topic hz, topic delay, Bag으로 동작을 관측할 수 있는가?
• 변경 전후 Metric을 같은 입력으로 비교했는가?

10. Python이냐 C++이냐

초보 단계에서는 Python(rclpy)으로 시작하세요. 이유는 분명합니다. 빌드가 빠르고, 문법이 짧고, 오류 메시지가 읽기 쉽습니다. 배우는 동안에는 알고리즘에 집중하는 것이 맞습니다.

그런데 실무에서 C++(rclcpp)로 넘어가야 하는 순간이 옵니다. 기준은 감이 아니라 숫자와 성질입니다.

C++가 필요한 경우 • 주기가 100 Hz를 넘는 제어 루프. Python은 GIL과 인터프리터 오버헤드 때문에 고주파 실시간 제어에 불리합니다 • 포인트 클라우드나 영상을 픽셀 단위로 처리. 큰 배열을 직접 훑는 작업은 차이가 큽니다 • 지연 변동(jitter)이 중요한 안전 기능. GC가 언제 도는지 알 수 없는 것이 문제입니다 • Component로 묶어 intra-process 통신을 쓰고 싶을 때. 같은 프로세스 안에서 메시지를 복사 없이 전달하는 최적화는 C++에서 훨씬 성숙합니다

Python이 충분한 경우 • 상위 로직, 상태 기계, 임무 관리 • 설정과 실험, 데이터 분석 • 10~50 Hz 정도의 제어 루프 • 프로토타이핑 전반

중요한 점: 한 시스템 안에서 섞어 쓰는 것이 정상입니다. 드라이버와 고속 제어는 C++, 임무 관리와 도구는 Python. ROS 2는 애초에 그렇게 쓰라고 만들어졌습니다. 토픽으로 대화하므로 언어가 달라도 아무 문제가 없습니다.

그리고 순서: 성능 문제가 생기기 전에 미리 C++로 쓰는 것은 대개 손해입니다. 먼저 Python으로 만들어 동작을 확정하고, ros2 topic hz로 실제로 밀리는지 측정한 뒤, 병목인 노드만 옮기세요.

개념도 · compare

• 단계/참여자
• Python (rclpy)
• C++ (rclcpp)
• 연결/행
• 빌드가 빠르고 고치기 쉽다|빌드가 느리지만 실행이 빠르다
• 오류 메시지가 읽기 쉽다|템플릿 오류는 길고 어렵다
• 10~50 Hz 제어에 충분|100 Hz 이상에 적합
• 분석과 도구 생태계가 강하다|intra-process 최적화가 성숙하다
• 먼저 여기서 만든다|측정 후 병목만 옮긴다

한쪽을 고르는 문제가 아니라, 어느 노드를 어느 쪽에 둘지의 문제입니다.

11. 확인 퀴즈 15문항

1. ros2 node list에 표시되는 노드 이름은 어디서 정해지는가?

• (1) 파일 이름
• (2) super().init()에 넘긴 문자열 정답
• (3) 클래스 이름
• (4) setup.py의 entry_points

해설: 노드 이름은 super().init("이름")에 넘긴 문자열입니다. 파일 이름이나 클래스 이름과는 아무 관계가 없습니다. "내 노드가 목록에 안 보인다"고 할 때 이 줄부터 확인하면 대부분 해결되며, 리매핑이나 네임스페이스가 붙으면 앞에 접두사가 더해집니다.

2. 노드 코드를 읽을 때 그 노드의 역할을 가장 빨리 알 수 있는 곳은?

• (1) import 목록
• (2) create_subscription과 create_publisher 호출들 정답
• (3) main 함수
• (4) 클래스 이름

해설: 무엇을 받아 무엇을 내보내는지가 그래프에서의 역할 그 자체입니다. 구독과 발행 목록만 만들어도 이 노드가 필터인지 변환기인지 제어기인지 파악됩니다. 알고리즘 본문은 그 맥락을 갖춘 뒤에 읽어야 훨씬 잘 읽힙니다.

3. self.create_timer(1.0, self.tick()) 처럼 괄호를 붙이면?

• (1) 1초마다 정상적으로 호출된다
• (2) tick이 즉시 한 번 실행되고 그 반환값(None)이 콜백으로 등록된다 정답
• (3) 컴파일 오류가 난다
• (4) 타이머가 두 배 빠르게 돈다

해설: 괄호를 붙이면 함수가 그 자리에서 호출되고 결과값이 인자로 넘어갑니다. 대개 None이 등록되므로 타이머는 돌지만 아무 일도 하지 않습니다. 오류도 나지 않아 찾기 어려우므로, 콜백을 넘길 때는 괄호 없이 함수 자체를 넘겨야 합니다.

4. 같은 토픽 이름으로 발행자와 구독자를 만들고 콜백에서 다시 발행하면?

• (1) ROS 2가 자동으로 차단한다
• (2) 자기 메시지를 자기가 받아 무한 피드백 루프가 되어 CPU가 치솟는다 정답
• (3) 메시지가 두 배로 느려진다
• (4) 아무 일도 일어나지 않는다

해설: ROS 2는 자기 자신의 발행을 걸러 주지 않습니다. 콜백에서 같은 토픽에 다시 발행하면 그 메시지가 다시 자기 콜백을 부르며 무한히 반복됩니다. 입력과 출력 토픽 이름을 분리하고 배선은 launch의 리매핑으로 처리하는 것이 안전합니다.

5. rclpy.spin(node)의 역할을 가장 정확히 설명한 것은?

• (1) 노드를 생성한다
• (2) Executor를 돌려 등록된 콜백들이 실행되기 시작한다 정답
• (3) 토픽을 발행한다
• (4) 파라미터를 읽는다

해설: spin에 도달하기 전까지는 콜백이 하나도 실행되지 않습니다. __init__에서 구독자를 만들었다고 해서 데이터가 들어오는 것이 아니라, Executor가 돌기 시작해야 비로소 콜백이 불립니다. "구독자를 만들었는데 콜백이 안 불린다"의 흔한 원인입니다.

6. main을 try/finally로 감싸야 하는 가장 중요한 이유는?

• (1) 코드가 깔끔해 보이기 때문
• (2) Ctrl+C로 종료할 때도 정지 명령과 정리 코드가 실행되게 하려고 정답
• (3) 예외를 숨기기 위해서
• (4) 성능이 좋아지기 때문

해설: try/finally가 없으면 Ctrl+C 순간 프로그램이 그냥 죽고 정리 코드가 실행되지 않습니다. 그러면 마지막 정지 명령이 나가지 못해 프로그램은 끝났는데 로봇은 마지막 속도로 계속 굴러갑니다. 실제로 자주 일어나는 사고입니다.

7. main(args=None)을 rclpy.init(args=args)로 넘기지 않으면?

• (1) 노드가 아예 실행되지 않는다
• (2) --ros-args로 준 리매핑, 네임스페이스, 파라미터가 적용되지 않는다 정답
• (3) 로그가 출력되지 않는다
• (4) QoS가 초기화된다

해설: ros2 run이나 launch는 --ros-args 뒤의 인자를 프로그램에 전달합니다. 이것을 rclpy.init에 넘겨야 리매핑과 파라미터가 반영됩니다. 빠뜨리면 "-p 로 파라미터를 줬는데 기본값으로 뜬다"는 현상이 생기며 원인을 찾기 어렵습니다.

8. 콜백 안에서 time.sleep(2.0)을 쓰면 어떤 일이 생기는가?

• (1) 그 콜백만 2초 늦어진다
• (2) 단일 스레드 Executor에서는 노드 전체가 2초 동안 멈춘다 정답
• (3) 자동으로 다른 스레드에서 처리된다
• (4) 경고만 출력된다

해설: 기본 Executor는 단일 스레드이므로 콜백이 sleep하는 동안 그 노드의 타이머와 다른 구독 콜백이 전부 멈춥니다. 기다림이 필요하면 타이머로 넘기거나 상태 변수로 단계를 나누어 콜백 자체는 즉시 끝나게 만들어야 합니다.

9. except Exception: pass 가 특히 위험한 이유는?

• (1) 성능이 떨어지기 때문
• (2) 콜백이 매번 실패하는데 아무도 그 사실을 알 수 없기 때문 정답
• (3) 메모리를 많이 쓰기 때문
• (4) QoS가 깨지기 때문

해설: 예외를 삼키면 노드는 살아 있고 로그도 조용하지만 실제로는 아무 일도 하지 않는 상태가 됩니다. 로봇에서 이런 조용한 실패는 찾아내기 가장 어려운 종류입니다. 최소한 error 로그를 남기고, 가능하면 안전한 기본값으로 되돌려야 합니다.

10. rclcpp(C++)에는 있고 rclpy(Python)에는 없는 함정은?

• (1) 노드 이름을 지정해야 한다
• (2) 구독자와 타이머 핸들을 멤버로 보관하지 않으면 소멸되어 사라진다 정답
• (3) 콜백에 괄호를 붙이면 안 된다
• (4) spin을 불러야 콜백이 돈다

해설: C++에서는 구독자나 타이머를 지역 변수에 담으면 함수가 끝나는 순간 소멸되어 조용히 동작을 멈춥니다. 반드시 멤버 변수로 보관해야 합니다. Python은 노드가 내부적으로 참조를 유지하므로 이 문제가 없지만, 나머지 함정은 두 언어에 공통입니다.

11. 코드를 열지 않고 노드의 구독과 발행 목록을 확인하는 명령은?

• (1) ros2 topic list
• (2) ros2 node info /노드이름 정답
• (3) ros2 param list
• (4) ros2 pkg list

해설: ros2 node info는 그 노드의 구독자, 발행자, 서비스 목록을 그대로 보여 줍니다. 남의 패키지를 처음 만났을 때 소스를 찾아 헤매지 않고 역할을 파악할 수 있어 매우 유용하며, ros2 param list와 함께 쓰면 조정 가능한 항목까지 알 수 있습니다.

12. Python 노드를 수정한 뒤 매번 다시 빌드하지 않아도 되게 하는 빌드 옵션은?

• (1) --packages-select
• (2) --symlink-install 정답
• (3) --parallel-workers
• (4) --continue-on-error

해설: --symlink-install은 설치본을 원본 파일에 대한 심볼릭 링크로 만들어 주므로, 이미 등록된 Python 파일의 내용을 고친 경우 다시 빌드하지 않아도 반영됩니다. 다만 새 파일을 추가하거나 entry_points를 바꾸면 다시 빌드해야 합니다.

13. 100 Hz 이상으로 도는 저수준 제어 루프에 C++를 권하는 주된 이유는?

• (1) C++가 문법이 더 쉽기 때문
• (2) GIL과 인터프리터 오버헤드, 그리고 GC로 인한 지연 변동을 피할 수 있기 때문 정답
• (3) ROS 2가 Python 고속 제어를 지원하지 않기 때문
• (4) C++만 토픽을 발행할 수 있기 때문

해설: Python은 인터프리터 오버헤드와 GIL 때문에 고주파 루프에서 불리하고, 가비지 컬렉션이 언제 도는지 알 수 없어 지연 변동이 생깁니다. 안전 직결 제어에서는 평균 속도보다 이 변동이 더 큰 문제이므로 C++가 유리합니다.

14. 한 시스템에서 Python 노드와 C++ 노드를 섞어 쓰는 것에 대해 옳은 설명은?

• (1) 섞으면 통신이 되지 않는다
• (2) 토픽으로 대화하므로 언어가 달라도 아무 문제가 없으며 오히려 정상적인 구성이다 정답
• (3) 반드시 하나로 통일해야 한다
• (4) 브리지 노드를 따로 만들어야 한다

해설: ROS 2는 애초에 언어를 섞어 쓰도록 설계되었습니다. 메시지 타입이 같으면 미들웨어가 알아서 전달하므로 드라이버와 고속 제어는 C++, 임무 관리와 도구는 Python처럼 나누는 구성이 실무의 표준입니다.

15. 성능 최적화를 위해 언어를 바꾸는 올바른 순서는?

• (1) 처음부터 전부 C++로 쓴다
• (2) Python으로 동작을 확정하고 실제로 밀리는지 측정한 뒤 병목 노드만 옮긴다 정답
• (3) 느려 보이면 바로 전부 옮긴다
• (4) 언어는 성능과 무관하므로 바꾸지 않는다

해설: 성능 문제가 실제로 생기기 전에 미리 C++로 쓰는 것은 대개 개발 속도만 잃는 손해입니다. 먼저 Python으로 동작을 확정하고, ros2 topic hz 같은 도구로 어느 노드가 목표 주기를 못 지키는지 측정한 뒤 그 노드만 옮기는 것이 합리적입니다.

연습 문제

1. ROS 2 Node Code를 30초 안에 읽는 순서를 설명하세요.
2. Code 설명에서 Python 문법, ROS 2 의미와 Robot 의미를 분리해야 하는 이유는 무엇인가요?
3. SafeDriver의 입력·출력·Parameter·상태와 Trigger를 표로 정리하세요.
4. SafeDriver가 첫 Scan 전과 Scan Timeout 후에 정지해야 하는 이유를 설명하세요.
5. create_timer(0.1, self.tick)에서 0.1과 self.tick의 의미를 각각 설명하세요.
6. LaserScan.ranges에서 양수만 고르는 방식이 충분하지 않은 이유와 개선 방법을 쓰세요.
7. use_sim_time이 Sensor Age와 Watchdog 계산에 미칠 수 있는 영향을 설명하세요.
8. 실행 중 ros2 param set이 성공해도 SafeDriver 동작이 바뀌지 않을 수 있는 이유는 무엇인가요?
9. BadNode에서 Timer Callback에 괄호를 붙였을 때 실제 실행 순서를 설명하세요.
10. 자기 발행 Topic을 다시 구독하는 Feedback Loop를 CLI로 찾는 방법을 설명하세요.
11. GoodNode가 time.sleep()을 Timer와 상태로 바꾼 이유는 무엇인가요?
12. Counter 예제에서 Publisher의 10이 뜻하는 것과 뜻하지 않는 것을 설명하세요.
13. Callback을 Data·상태·부작용으로 나누어 읽으면 어떤 문제가 잘 보이나요?
14. 처음 보는 Node의 코드를 열지 않고 Interface와 실행 주기를 확인하는 명령을 쓰세요.
15. Python Node를 C++로 옮기기 전에 측정해야 할 항목과 판단 기준을 설명하세요.

참고 자료

• Writing a simple publisher and subscriber — Python
• Using parameters in a class — Python
• Executors
• rclpy API
• ROS 2 command line tools