Launch와 Lifecycle — 재현 가능한 실행과 안전한 상태 관리

수동 실행                         Launch 실행
────────────────────────────────────────────────
명령을 사람에게 의존              구성을 파일로 기록
Terminal마다 Parameter 입력        YAML·Argument로 재현
종료 Process를 하나씩 찾음          전체 Process를 감시·종료
Robot별 이름 충돌                  Namespace로 분리
배선 변경에 Code 수정              Remapping으로 연결 변경

mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src
ros2 pkg create robot_bringup \
  --build-type ament_python \
  --dependencies launch launch_ros
mkdir -p robot_bringup/launch robot_bringup/config robot_bringup/rviz

robot_bringup/
├── launch/
│   ├── sensors.launch.py
│   ├── navigation.launch.py
│   └── bringup.launch.py
├── config/
│   ├── sensors.yaml
│   └── safety.yaml
├── rviz/
│   └── robot.rviz
├── package.xml
└── setup.py

<exec_depend>ros2launch</exec_depend>
<exec_depend>launch</exec_depend>
<exec_depend>launch_ros</exec_depend>

import os
from glob import glob
from setuptools import setup

package_name = 'robot_bringup'

setup(
    name=package_name,
    # 나머지 Metadata 생략
    data_files=[
        ('share/ament_index/resource_index/packages',
         ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
        (os.path.join('share', package_name, 'launch'),
         glob('launch/*.launch.py')),
        (os.path.join('share', package_name, 'config'),
         glob('config/*.yaml')),
        (os.path.join('share', package_name, 'rviz'),
         glob('rviz/*.rviz')),
    ],
)

install(DIRECTORY launch config rviz
  DESTINATION share/${PROJECT_NAME}
)

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

# launch/minimal.launch.py
from launch import LaunchDescription
from launch_ros.actions import Node


def generate_launch_description():
    talker = Node(
        package='demo_nodes_cpp',
        executable='talker',
        name='talker',
        output='screen',
    )
    listener = Node(
        package='demo_nodes_py',
        executable='listener',
        name='listener',
        output='screen',
    )
    return LaunchDescription([talker, listener])

ros2 launch robot_bringup minimal.launch.py
ros2 node list
ros2 topic info /chatter --verbose

Node(
    package='parameter_lab_py',
    executable='speed_governor',
    name='speed_governor',
    namespace='robot1',
    parameters=[{'max_speed_mps': 0.4}],
    remappings=[('requested_speed', 'cmd_vel_nav')],
    arguments=['--device', '/dev/ttyUSB0'],
    ros_arguments=['--log-level', 'info'],
    output='screen',
    emulate_tty=True,
)

from launch.actions import DeclareLaunchArgument
from launch.substitutions import LaunchConfiguration

use_sim_time = LaunchConfiguration('use_sim_time')
robot_name = LaunchConfiguration('robot_name')

return LaunchDescription([
    DeclareLaunchArgument(
        'use_sim_time',
        default_value='false',
        description='Simulation Clock 사용 여부',
    ),
    DeclareLaunchArgument(
        'robot_name',
        default_value='robot1',
        description='Robot Namespace',
    ),
    # Node Actions...
])

ros2 launch robot_bringup bringup.launch.py --show-args
ros2 launch robot_bringup bringup.launch.py \
  robot_name:=robot2 use_sim_time:=true

# 잘못된 예
if LaunchConfiguration('use_rviz') == 'true':
    ...

from launch.substitutions import PathJoinSubstitution
from launch_ros.substitutions import FindPackageShare

params_file = PathJoinSubstitution([
    FindPackageShare('robot_bringup'),
    'config',
    'safety.yaml',
])

safety_node = Node(
    package='robot_safety',
    executable='safety_filter',
    name='safety_filter',
    parameters=[
        params_file,
        {'use_sim_time': LaunchConfiguration('use_sim_time')},
    ],
    output='screen',
)

ros2 param list /robot1/safety_filter
ros2 param get /robot1/safety_filter stop_distance_m

from launch.actions import GroupAction
from launch_ros.actions import Node, PushRosNamespace


def robot_group(namespace):
    return GroupAction([
        PushRosNamespace(namespace),
        Node(
            package='robot_driver',
            executable='base_driver',
            name='base_driver',
            output='screen',
        ),
        Node(
            package='robot_safety',
            executable='safety_filter',
            name='safety_filter',
            remappings=[
                ('cmd_vel_in', 'cmd_vel_nav'),
                ('cmd_vel_out', 'cmd_vel'),
            ],
            output='screen',
        ),
    ])


def generate_launch_description():
    return LaunchDescription([
        robot_group('robot1'),
        robot_group('robot2'),
    ])

ros2 node list
ros2 topic list | grep cmd_vel

Node(
    package='robot_safety',
    executable='safety_filter',
    remappings=[
        ('cmd_vel_in', 'cmd_vel_muxed'),
        ('cmd_vel_out', 'cmd_vel_safe'),
        ('scan', 'front/scan'),
    ],
)

Navigation → cmd_vel_nav ┐
Joystick   → cmd_vel_joy ├→ Velocity Mux → cmd_vel_muxed
                         └→ Safety Filter → cmd_vel_safe → Driver

from launch.conditions import IfCondition, UnlessCondition

rviz_node = Node(
    package='rviz2',
    executable='rviz2',
    condition=IfCondition(LaunchConfiguration('use_rviz')),
    output='screen',
)

hardware_driver = Node(
    package='robot_driver',
    executable='base_driver',
    condition=UnlessCondition(LaunchConfiguration('use_sim_time')),
    output='screen',
)

ros2 launch robot_bringup bringup.launch.py \
  use_rviz:=false use_sim_time:=true

from launch.actions import IncludeLaunchDescription
from launch.launch_description_sources import PythonLaunchDescriptionSource

sensors_launch = IncludeLaunchDescription(
    PythonLaunchDescriptionSource(
        PathJoinSubstitution([
            FindPackageShare('robot_bringup'),
            'launch',
            'sensors.launch.py',
        ])
    ),
    launch_arguments={
        'use_sim_time': LaunchConfiguration('use_sim_time'),
        'namespace': LaunchConfiguration('namespace'),
    }.items(),
)

bringup.launch.py
├── description.launch.py
├── sensors.launch.py
├── localization.launch.py
├── navigation.launch.py
└── visualization.launch.py

from launch.actions import OpaqueFunction


def create_robot_nodes(context):
    count_text = LaunchConfiguration('robot_count').perform(context)
    count = int(count_text)
    nodes = []
    for index in range(count):
        nodes.append(Node(
            package='demo_nodes_cpp',
            executable='talker',
            namespace=f'robot{index + 1}',
            name='talker',
        ))
    return nodes


OpaqueFunction(function=create_robot_nodes)

from launch.actions import LogInfo, RegisterEventHandler, Shutdown
from launch.event_handlers import OnProcessExit, OnProcessStart

driver = Node(
    package='robot_driver',
    executable='base_driver',
    name='base_driver',
    output='screen',
)

driver_started = RegisterEventHandler(
    OnProcessStart(
        target_action=driver,
        on_start=[LogInfo(msg='Driver Process가 시작되었습니다.')],
    )
)

driver_exited = RegisterEventHandler(
    OnProcessExit(
        target_action=driver,
        on_exit=[
            LogInfo(msg='Driver가 종료되어 Bringup을 내립니다.'),
            Shutdown(reason='critical driver exited'),
        ],
    )
)

from launch.actions import ExecuteProcess, RegisterEventHandler
from launch.event_handlers import OnExecutionComplete

calibration = ExecuteProcess(
    cmd=['ros2', 'run', 'robot_tools', 'check_config'],
    output='screen',
)

start_driver_after_check = RegisterEventHandler(
    OnExecutionComplete(
        target_action=calibration,
        on_completion=[driver],
    )
)

from launch.actions import TimerAction

delayed_rviz = TimerAction(
    period=3.0,
    actions=[rviz_node],
)

Node(
    package='robot_monitor',
    executable='battery_monitor',
    respawn=True,
    respawn_delay=2.0,
    output='screen',
)

t=0.0  Driver Process 시작
t=0.1  Controller Process 시작
t=0.2  Controller가 Service 검색
t=1.8  Driver가 Serial 연결 완료

# launch/bringup.launch.py
from launch import LaunchDescription
from launch.actions import (
    DeclareLaunchArgument,
    GroupAction,
    IncludeLaunchDescription,
    RegisterEventHandler,
    Shutdown,
)
from launch.conditions import IfCondition
from launch.event_handlers import OnProcessExit
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch.substitutions import LaunchConfiguration, PathJoinSubstitution
from launch_ros.actions import Node, PushRosNamespace
from launch_ros.substitutions import FindPackageShare


def generate_launch_description():
    namespace = LaunchConfiguration('namespace')
    use_sim_time = LaunchConfiguration('use_sim_time')
    use_rviz = LaunchConfiguration('use_rviz')
    safety_params = PathJoinSubstitution([
        FindPackageShare('robot_bringup'), 'config', 'safety.yaml'])

    driver = Node(
        package='robot_driver', executable='base_driver',
        name='base_driver', output='screen', emulate_tty=True)

    safety = Node(
        package='robot_safety', executable='safety_filter',
        name='safety_filter',
        parameters=[safety_params, {'use_sim_time': use_sim_time}],
        remappings=[('cmd_vel_in', 'cmd_vel_muxed'),
                    ('cmd_vel_out', 'cmd_vel')],
        output='screen')

    rviz = Node(
        package='rviz2', executable='rviz2',
        condition=IfCondition(use_rviz), output='screen')

    return LaunchDescription([
        DeclareLaunchArgument('namespace', default_value='robot1'),
        DeclareLaunchArgument('use_sim_time', default_value='false'),
        DeclareLaunchArgument('use_rviz', default_value='true'),
        GroupAction([
            PushRosNamespace(namespace),
            driver,
            safety,
        ]),
        rviz,
        RegisterEventHandler(OnProcessExit(
            target_action=driver,
            on_exit=[Shutdown(reason='base driver stopped')],
        )),
    ])

Process 시작
    ↓
Unconfigured ──configure──> Inactive ──activate──> Active
      ↑                         │                    │
      └────────cleanup──────────┘<──deactivate──────┘

cd ~/ros2_ws/src
ros2 pkg create lifecycle_lab_py \
  --build-type ament_python \
  --dependencies rclpy lifecycle_msgs std_msgs sensor_msgs geometry_msgs

entry_points={
    'console_scripts': [
        'managed_sensor = lifecycle_lab_py.managed_sensor:main',
        'scan_guard = lifecycle_lab_py.scan_guard:main',
    ],
},

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

# lifecycle_lab_py/managed_sensor.py
import rclpy
from rclpy.lifecycle import LifecycleNode
from rclpy.lifecycle import State, TransitionCallbackReturn
from std_msgs.msg import String


class ManagedSensor(LifecycleNode):
    def __init__(self):
        super().__init__('managed_sensor')
        self.publisher = None
        self.timer = None

    def on_configure(self, state: State):
        self.get_logger().info('configure: 자원 생성')
        self.publisher = self.create_lifecycle_publisher(
            String, 'sensor_status', 10)
        self.timer = self.create_timer(1.0, self.tick)
        return TransitionCallbackReturn.SUCCESS

    def on_activate(self, state: State):
        self.get_logger().info('activate: 출력 활성화')
        return super().on_activate(state)

    def on_deactivate(self, state: State):
        self.get_logger().info('deactivate: 출력 비활성화')
        return super().on_deactivate(state)

    def on_cleanup(self, state: State):
        self.get_logger().info('cleanup: 자원 해제')
        if self.timer is not None:
            self.destroy_timer(self.timer)
            self.timer = None
        if self.publisher is not None:
            self.destroy_lifecycle_publisher(self.publisher)
            self.publisher = None
        return TransitionCallbackReturn.SUCCESS

    def tick(self):
        if self.publisher is None:
            return
        message = String()
        message.data = 'sensor ready'
        self.publisher.publish(message)


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


if __name__ == '__main__':
    main()

ros2 run lifecycle_lab_py managed_sensor

ros2 lifecycle nodes
ros2 lifecycle get /managed_sensor
ros2 lifecycle list /managed_sensor

ros2 lifecycle set /managed_sensor configure
ros2 lifecycle get /managed_sensor

ros2 lifecycle set /managed_sensor activate
ros2 topic echo /sensor_status

ros2 lifecycle set /managed_sensor deactivate
ros2 lifecycle set /managed_sensor cleanup
ros2 lifecycle set /managed_sensor shutdown

ros2 topic echo /managed_sensor/transition_event

# lifecycle_lab_py/scan_guard.py
import math

import rclpy
from geometry_msgs.msg import Twist
from rclpy.lifecycle import LifecycleNode
from rclpy.lifecycle import State, TransitionCallbackReturn
from rclpy.qos import qos_profile_sensor_data
from sensor_msgs.msg import LaserScan


class ScanGuard(LifecycleNode):
    def __init__(self):
        super().__init__('scan_guard')
        self.publisher = None
        self.subscription = None
        self.stop_distance = 0.0
        self.active = False

    def on_configure(self, state: State):
        self.declare_parameter('stop_distance_m', 0.4)
        value = self.get_parameter('stop_distance_m').value
        if not isinstance(value, float) or not math.isfinite(value):
            self.get_logger().error('stop_distance_m Type·값 오류')
            return TransitionCallbackReturn.FAILURE
        if not 0.1 <= value <= 3.0:
            self.get_logger().error('stop_distance_m 범위는 0.1~3.0m')
            return TransitionCallbackReturn.FAILURE

        self.stop_distance = value
        self.publisher = self.create_lifecycle_publisher(
            Twist, 'cmd_vel_safe', 1)
        self.subscription = self.create_subscription(
            LaserScan, 'scan', self.on_scan, qos_profile_sensor_data)
        return TransitionCallbackReturn.SUCCESS

    def on_activate(self, state: State):
        self.active = True
        return super().on_activate(state)

    def on_deactivate(self, state: State):
        # Publisher를 비활성화하기 전에 마지막 정지 명령 시도
        if self.publisher is not None:
            self.publisher.publish(Twist())
        self.active = False
        return super().on_deactivate(state)

    def on_cleanup(self, state: State):
        self.active = False
        if self.subscription is not None:
            self.destroy_subscription(self.subscription)
            self.subscription = None
        if self.publisher is not None:
            self.destroy_lifecycle_publisher(self.publisher)
            self.publisher = None
        return TransitionCallbackReturn.SUCCESS

    def on_scan(self, message):
        # 일반 Subscription은 Lifecycle이 자동 중지하지 않으므로 검사한다.
        if not self.active or self.publisher is None:
            return
        valid = [value for value in message.ranges
                 if math.isfinite(value)
                 and message.range_min <= value <= message.range_max]
        command = Twist()
        if valid and min(valid) >= self.stop_distance:
            command.linear.x = 0.2
        self.publisher.publish(command)


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

def on_configure(self, state):
    try:
        self.device = open_device(timeout_sec=2.0)
        if self.device is None:
            return TransitionCallbackReturn.FAILURE
        return TransitionCallbackReturn.SUCCESS
    except Exception as error:
        self.get_logger().exception(f'예상하지 못한 오류: {error}')
        return TransitionCallbackReturn.ERROR

def on_configure(self, state):
    try:
        self.camera = self.open_camera()
        self.publisher = self.create_lifecycle_publisher(Image, 'image', 10)
        return TransitionCallbackReturn.SUCCESS
    except Exception as error:
        if self.camera is not None:
            self.camera.close()
            self.camera = None
        if self.publisher is not None:
            self.destroy_lifecycle_publisher(self.publisher)
            self.publisher = None
        self.get_logger().error(f'configure 실패: {error}')
        return TransitionCallbackReturn.FAILURE

from launch_ros.actions import LifecycleNode

managed_sensor = LifecycleNode(
    package='lifecycle_lab_py',
    executable='managed_sensor',
    name='managed_sensor',
    namespace='robot1',
    output='screen',
    parameters=[{'use_sim_time': False}],
)

from launch.actions import EmitEvent, RegisterEventHandler
from launch.events import matches_action
from launch_ros.event_handlers import OnStateTransition
from launch_ros.events.lifecycle import ChangeState
from lifecycle_msgs.msg import Transition

configure_event = EmitEvent(event=ChangeState(
    lifecycle_node_matcher=matches_action(managed_sensor),
    transition_id=Transition.TRANSITION_CONFIGURE,
))

activate_after_inactive = RegisterEventHandler(
    OnStateTransition(
        target_lifecycle_node=managed_sensor,
        goal_state='inactive',
        entities=[EmitEvent(event=ChangeState(
            lifecycle_node_matcher=matches_action(managed_sensor),
            transition_id=Transition.TRANSITION_ACTIVATE,
        ))],
    )
)

시작 순서
1. Hardware Driver configure
2. Sensor Driver configure
3. Localization configure
4. Safety Controller configure
5. 모두 Inactive 확인
6. Driver → Sensor → Localization → Safety activate

정지 순서
1. Command Source deactivate
2. Safety·Controller deactivate
3. Sensor deactivate
4. Hardware Driver deactivate
5. 역순 cleanup

Process Alive     OS Process가 존재
Lifecycle Active  기능 실행이 허용됨
Application Ready 실제 Data·Hardware·품질 조건 충족

전원·E-stop 확인
      ↓
Driver Configure: Port Open, Firmware·ID 검증
      ↓
Sensor Configure: Calibration·Frame·주기 검증
      ↓
Localization Configure: Map·TF 준비
      ↓
모두 Inactive + Health OK
      ↓
Safety Filter Activate
      ↓
Driver Activate
      ↓
Command Source Activate

def on_deactivate(self, state):
    self.command_enabled = False
    self.publish_zero_velocity()
    if not self.wait_until_stopped(timeout_sec=1.0):
        self.get_logger().error('정지 확인 Timeout')
        return TransitionCallbackReturn.ERROR
    self.disable_motor_command_path()
    return super().on_deactivate(state)

ros2 launch robot_bringup bringup.launch.py --show-args
ros2 launch -d robot_bringup bringup.launch.py
ros2 lifecycle get /robot1/managed_sensor
ros2 lifecycle list /robot1/managed_sensor

ros2 launch robot_bringup bringup.launch.py --show-args
ros2 pkg executables robot_driver
ros2 pkg prefix robot_bringup
ros2 launch -d robot_bringup bringup.launch.py
ros2 node list
ros2 topic list -t
ros2 param list /robot1/safety_filter
ros2 topic info /robot1/cmd_vel --verbose

ros2 lifecycle nodes
ros2 lifecycle get /managed_sensor
ros2 lifecycle list /managed_sensor
ros2 service list | grep managed_sensor
ros2 topic echo /managed_sensor/transition_event
ros2 topic echo /rosout