Devin.KR
통합 검색 Ctrl K

노드 만들기 - rclpy 로 첫 노드

개발자KR 조회 7

이 장에서 배우는 것

앞 장에서 워크스페이스를 만들고 colcon으로 패키지를 빌드하는 과정까지 확인했다. 이번 장에서는 그 패키지 안에 실제로 동작하는 첫 프로그램을 채워 넣는다. rclpy의 Node 클래스를 상속해 두리(Duri) 로봇의 첫 노드를 만들고, 타이머로 주기적인 동작을 등록하고, 로그를 남기고, spin으로 그 노드를 계속 돌게 만드는 과정을 다룬다.

  • rclpy.node.Node를 상속해 두리의 첫 노드 클래스를 작성한다
  • create_timer로 일정 주기마다 실행되는 콜백을 등록한다
  • get_logger()로 노드 상태를 터미널에 남긴다
  • rclpy.spin()이 콜백을 계속 실행시키는 원리를 이해한다
  • 노드 이름과 네임스페이스로 같은 코드를 여러 인스턴스로 구분해 실행한다

문제 상황

두리 프로젝트 팀은 앞 장에서 duri_core라는 패키지 골격을 만들고 colcon으로 빌드까지 성공했다. 그런데 그 패키지 안에는 아직 실행할 수 있는 코드가 하나도 없다. 팀은 두리가 켜져 있는 동안 소프트웨어가 살아 있다는 것을 알리는 가장 기초적인 기능부터 만들기로 한다. 1초마다 "정상 동작 중"이라는 로그를 남기는 프로그램이다.

처음에는 평범한 파이썬 스크립트처럼 while True 루프와 time.sleep(1)로 이 기능을 짜려 했다. 하지만 이렇게 짠 코드는 나중에 센서 값을 받거나 서비스 요청에 응답하는 코드를 추가할 때마다 루프 안에 조건문을 계속 끼워 넣어야 한다. ROS 2는 이런 반복 작업을 노드와 콜백, spin이라는 구조로 대신 처리해준다. 또 창고에 두리를 두 대, 세 대 투입하는 시나리오도 있다. 같은 코드를 그대로 실행하면 두 로봇의 노드 이름이 똑같아져서 ROS 그래프에서 구분이 안 된다. 이 장에서는 이 두 문제, 즉 "반복 동작을 어떻게 등록할 것인가"와 "여러 대를 어떻게 구분할 것인가"를 rclpy의 기본기로 풀어본다.

Node 클래스와 초기화 흐름

rclpy로 만드는 노드는 항상 rclpy.node.Node를 상속하는 클래스로 작성한다. 생성자에서 super().__init__('노드_이름')을 호출하면, 그 시점에 이 객체가 ROS 그래프에 등록할 이름을 갖게 된다. 노드 이름은 영문자·숫자·밑줄만 쓸 수 있고 숫자로 시작할 수 없다. 공백이나 하이픈이 섞이면 실행 시 오류가 난다.

노드 객체가 생기기 전에는 반드시 rclpy.init()을 먼저 호출해야 한다. 이 호출이 프로세스 안에서 ROS 통신을 담당하는 컨텍스트를 준비한다. 순서를 바꿔서 Node부터 만들면 통신 컨텍스트가 없다는 오류가 발생한다. 이 초기화 순서는 이 장 전체에서 반복되는 패턴이라 이후 장에서 만드는 모든 노드도 똑같은 순서를 따른다.

노드 객체는 get_name()으로 자신의 이름을, get_namespace()로 네임스페이스를 돌려준다. 네임스페이스를 따로 지정하지 않으면 기본값은 /다. 이 두 값을 합친 것이 ROS 그래프 안에서 이 노드를 가리키는 전체 이름이 된다.

타이머 콜백과 spin

주기적으로 실행할 동작은 create_timer(주기_초, 콜백함수)로 등록한다. 이 호출은 콜백을 즉시 실행하지 않는다. "이 시간마다 이 함수를 실행 목록에 올려 달라"고 rclpy에 예약해 두는 것뿐이다. 실제로 콜백이 하나씩 실행되게 만드는 것은 rclpy.spin(node)다.

spin은 호출한 곳에서 프로그램 흐름을 멈추고, 등록된 콜백들을 조건이 될 때마다 반복해서 실행하는 루프를 돈다. Ctrl+C로 인터럽트가 오거나 rclpy.shutdown()이 호출될 때까지 이 루프는 끝나지 않는다. 다음 장에서 다룰 토픽 구독도, 이후 장의 서비스 응답도 모두 이 spin 루프 안에서 콜백으로 실행된다는 점에서, 지금 만드는 타이머 콜백은 가장 단순한 형태의 콜백이라고 볼 수 있다.

노드는 init로 시작해 spin으로 콜백을 반복하다가 destroy_node와 shutdown으로 끝난다

spin이 끝난 뒤에는 반드시 node.destroy_node()로 노드가 들고 있던 타이머·퍼블리셔 같은 자원을 정리하고, rclpy.shutdown()으로 통신 컨텍스트를 닫아야 한다. 이 두 호출을 생략해도 짧은 스크립트는 대체로 문제없이 끝나지만, 노드 개수가 늘어나는 뒤 장의 launch 파일 예제에서는 자원이 정리되지 않은 채 프로세스가 쌓이는 원인이 될 수 있다.

노드 이름과 네임스페이스, ros2 node list

코드에 적은 노드 이름은 고정값이 아니다. 실행할 때 --ros-args 뒤에 remap 인자를 붙이면 이름과 네임스페이스를 바꿀 수 있다. -r __node:=새이름은 노드 이름 자체를 바꾸고, -r __ns:=/네임스페이스는 그 노드를 특정 네임스페이스 아래에 둔다. 같은 코드로 두리 여러 대를 띄울 때는 코드를 복사하지 않고 이 remap 인자만 바꿔서 실행하면 된다.

실행 중인 노드들은 ros2 node list 명령으로 확인한다. 이 명령은 네임스페이스와 노드 이름을 합친 전체 이름을 한 줄씩 보여준다. 같은 heartbeat_node 코드라도 네임스페이스가 /duri1과 /duri2로 다르면, ROS 그래프에는 /duri1/heartbeat_node와 /duri2/heartbeat_node라는 서로 다른 이름으로 나타난다.

같은 이름의 노드라도 네임스페이스가 다르면 ros2 node list에서 서로 다른 이름으로 구분된다

완성 코드

duri_core/duri_core/heartbeat_node.py

import rclpy
from rclpy.node import Node


class HeartbeatNode(Node):
    """두리의 소프트웨어가 살아 있음을 주기적으로 알리는 노드."""

    def __init__(self):
        super().__init__('heartbeat_node')
        self._tick = 0
        self._timer = self.create_timer(1.0, self._on_timer)
        self.get_logger().info(
            f'{self.get_name()} 시작 (네임스페이스: {self.get_namespace()})'
        )

    def _on_timer(self):
        self._tick += 1
        self.get_logger().info(f'두리 정상 동작 중 (tick={self._tick})')


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


if __name__ == '__main__':
    main()

duri_core/setup.py

from setuptools import find_packages, setup

package_name = 'duri_core'

setup(
    name=package_name,
    version='0.0.1',
    packages=find_packages(exclude=['test']),
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='duri-team',
    maintainer_email='you@example.com',
    description='두리 로봇의 핵심 노드 모음',
    license='Apache-2.0',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'heartbeat_node = duri_core.heartbeat_node:main',
        ],
    },
)

heartbeat_sim.py (ROS 없이 개념만 확인하는 보조 예제)

class MiniNode:
    """rclpy 없이 노드의 이름·타이머 개념만 흉내 내는 연습용 클래스."""

    def __init__(self, name, namespace='/'):
        self.name = name
        self.namespace = namespace
        self._tick = 0

    def full_name(self):
        ns = '' if self.namespace == '/' else self.namespace
        return f'{ns}/{self.name}'

    def log(self, message):
        print(f'[{self.full_name()}] {message}')

    def on_timer(self):
        self._tick += 1
        self.log(f'정상 동작 중 (tick={self._tick})')

    def spin_ticks(self, count):
        self.log('노드 시작')
        for _ in range(count):
            self.on_timer()


if __name__ == '__main__':
    node = MiniNode('heartbeat_node', namespace='/duri1')
    node.spin_ticks(3)

줄별 해설

  • super().__init__('heartbeat_node') — 이 노드가 ROS 그래프에 등록될 이름을 정한다. 이 줄이 없으면 Node의 초기화가 끝나지 않아 이후 호출이 모두 실패한다.
  • self._timer = self.create_timer(1.0, self._on_timer) — 1.0초마다 _on_timer를 실행 목록에 올린다. 괄호 없이 함수 객체 자체를 넘긴 점에 주의한다. 반환값을 self에 저장해 두는 이유는 파이썬 가비지 컬렉터가 타이머 객체를 회수하지 않도록 붙잡아 두기 위해서다.
  • self.get_logger().info(...) — 이 노드 이름이 자동으로 붙는 로그를 터미널에 남긴다. 생성자 안의 호출은 노드가 시작될 때 한 번, _on_timer 안의 호출은 1초마다 반복 실행된다.
  • rclpy.spin(node) — 등록된 타이머 콜백을 계속 실행하는 루프를 돈다. Ctrl+C로 인터럽트가 들어올 때까지 이 줄에서 흐름이 멈춰 있는다.
  • finally: node.destroy_node(); rclpy.shutdown() — spin이 예외로 끝나든 정상으로 끝나든 자원 정리 코드가 항상 실행되도록 finally에 둔다.
  • MiniNode.spin_ticks(count) — 실제 spin과 달리 정해진 횟수만큼만 콜백을 실행하고 즉시 반환한다. 시간을 실제로 기다리지 않기 때문에 파이썬만으로 바로 실행해서 결과를 확인할 수 있다.

실행 결과

colcon build와 source는 앞 장에서 다룬 그대로 진행했다고 가정한다. 기본 이름으로 노드를 하나 실행하면 다음과 같이 보인다.

$ ros2 run duri_core heartbeat_node
[INFO] [1730000000.100000000] [heartbeat_node]: heartbeat_node 시작 (네임스페이스: /)
[INFO] [1730000001.100000000] [heartbeat_node]: 두리 정상 동작 중 (tick=1)
[INFO] [1730000002.100000000] [heartbeat_node]: 두리 정상 동작 중 (tick=2)
[INFO] [1730000003.100000000] [heartbeat_node]: 두리 정상 동작 중 (tick=3)
^C

같은 코드를 네임스페이스만 바꿔 두 번 실행하면, 세 번째 터미널에서 ros2 node list로 둘을 구분해 볼 수 있다.

# 터미널 1
$ ros2 run duri_core heartbeat_node --ros-args -r __ns:=/duri1

# 터미널 2
$ ros2 run duri_core heartbeat_node --ros-args -r __ns:=/duri2

# 터미널 3
$ ros2 node list
/duri1/heartbeat_node
/duri2/heartbeat_node

ROS 없이 개념만 확인하고 싶다면 보조 예제를 그냥 파이썬으로 실행하면 된다.

$ python3 heartbeat_sim.py
[/duri1/heartbeat_node] 노드 시작
[/duri1/heartbeat_node] 정상 동작 중 (tick=1)
[/duri1/heartbeat_node] 정상 동작 중 (tick=2)
[/duri1/heartbeat_node] 정상 동작 중 (tick=3)

실무에서 자주 틀리는 것

노드 이름에 공백이나 특수문자를 쓴다

super().__init__('heartbeat node')

공백이 들어간 이름은 노드 생성 시점에 오류를 낸다. 밑줄로 구분한다.

super().__init__('heartbeat_node')

spin을 호출하지 않고 노드를 바로 종료한다

def main():
    rclpy.init()
    node = HeartbeatNode()
    node.destroy_node()
    rclpy.shutdown()

이 코드는 타이머 콜백이 단 한 번도 실행되지 않은 채 프로그램이 끝난다. spin을 생성과 정리 사이에 반드시 넣는다.

def main():
    rclpy.init()
    node = HeartbeatNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

타이머 콜백 안에서 시간이 오래 걸리는 코드를 그대로 실행한다

def _on_timer(self):
    time.sleep(5)
    self.get_logger().info('작업 완료')

spin 루프는 한 콜백이 끝나야 다음 콜백을 처리한다. 콜백 안에서 5초를 그냥 잠들면 그동안 다른 타이머나 이후 장에서 추가할 토픽·서비스 콜백까지 함께 멈춘다. 콜백은 짧게 유지하고, 오래 걸리는 작업은 뒤 장에서 다루는 방식으로 분리한다.

def _on_timer(self):
    self._tick += 1
    self.get_logger().info(f'두리 정상 동작 중 (tick={self._tick})')

같은 이름의 노드를 네임스페이스 구분 없이 동시에 띄운다

$ ros2 run duri_core heartbeat_node
$ ros2 run duri_core heartbeat_node

두 프로세스 모두 기본 이름 /heartbeat_node를 쓰게 되어 어느 로그가 어느 로봇 것인지 구분할 수 없다. 실행할 때마다 네임스페이스를 다르게 지정한다.

$ ros2 run duri_core heartbeat_node --ros-args -r __ns:=/duri1
$ ros2 run duri_core heartbeat_node --ros-args -r __ns:=/duri2

한눈에 보기

노드 생성에 쓰는 rclpy 핵심 요소
요소역할비고
rclpy.init()프로세스에서 ROS 통신을 초기화한다main() 맨 앞에서 한 번만 호출
super().__init__(이름)노드 이름을 정하고 등록을 마친다공백·특수문자 불가
create_timer(주기, 콜백)주기(초)마다 콜백을 실행 목록에 등록한다등록 시점에 즉시 실행되지 않음
get_logger().info(메시지)노드 이름이 붙은 로그를 남긴다warn·error 등 다른 레벨도 있음
rclpy.spin(node)등록된 콜백을 반복 실행하는 루프를 돈다Ctrl+C 전까지 블로킹
destroy_node() / shutdown()노드 자원과 통신 컨텍스트를 정리한다spin 종료 후 호출
노드 이름을 정하는 세 시점
시점방법예시
코드 작성 시super().__init__('이름')super().__init__('heartbeat_node')
실행 시(이름)--ros-args -r __node:=이름ros2 run duri_core heartbeat_node --ros-args -r __node:=hb2
실행 시(네임스페이스)--ros-args -r __ns:=/이름ros2 run duri_core heartbeat_node --ros-args -r __ns:=/duri1

연습 문제

  1. heartbeat_node.py의 create_timer(1.0, self._on_timer) 호출에서 두 번째 인자를 self._on_timer()처럼 괄호를 붙여 넘기면 어떤 문제가 생기는지 설명하라.
  2. 두리 로봇 세 대를 동시에 띄우면서 각각 다른 이름으로 ros2 node list에 나타나게 하려면 어떤 실행 명령을 쓰면 되는지 한 가지 예를 적어라.
  3. heartbeat_sim.py의 MiniNode.spin_ticks(count)는 실제 rclpy.spin(node)와 어떤 점이 다른지 최소 한 가지를 설명하라.
  4. destroy_node()를 호출하지 않고 rclpy.shutdown()만 호출해도 짧은 스크립트는 대체로 동작한다. 그런데도 두 호출을 모두 finally 블록에 넣는 이유를 한 문장으로 적어라.

정답과 해설

  1. 괄호를 붙이면 create_timer가 호출되기 전에 _on_timer가 즉시 한 번 실행되고, 그 반환값인 None이 콜백으로 등록된다. 이후 타이머가 만료될 때 rclpy가 None을 실행하려다 오류를 낸다. 콜백 자리에는 반드시 함수 객체 자체를 넘겨야 한다.
  2. 예: ros2 run duri_core heartbeat_node --ros-args -r __ns:=/duri3. 세 프로세스 각각에 서로 다른 네임스페이스(또는 노드 이름)를 remap 인자로 넘기면 된다.
  3. spin_ticks는 정해진 횟수만큼 콜백을 실행한 뒤 곧바로 함수가 반환되지만, rclpy.spin(node)는 Ctrl+C 같은 외부 인터럽트가 올 때까지 횟수 제한 없이 계속 콜백을 실행하는 블로킹 루프다. 또한 spin_ticks는 실제로 1초씩 기다리지 않고 콜백을 곧바로 이어서 실행한다는 점도 다르다.
  4. 노드가 들고 있는 타이머·퍼블리셔 같은 내부 자원을 rclpy 컨텍스트가 종료되기 전에 명시적으로 정리해, 프로세스가 예상치 못하게 끝나더라도 자원이 정리되지 않은 채 남는 상황을 피하기 위해서다.

댓글 0

아직 댓글이 없습니다. 첫 댓글을 남겨 보세요.

댓글을 남기려면 로그인이 필요합니다.