ros2

name	ros2
description	ROS2 Humble development assistant - nodes, topics, services, actions, launch files, colcon builds, QoS, RViz2, Nav2, MoveIt, Gazebo, dev/deploy workflow on Ubuntu 22.04
triggers	["ros2","ros 2","rclpy","rclcpp","colcon","launch file","publisher","subscriber","topic","service server","action server","nav2","moveit","gazebo","rviz","rviz2","rosbag","tf2","urdf","xacro","qos","quality of service","symlink-install","marker","visualization","robot_description","lifecycle node","callback group","executor","composable node","slam_toolbox","cartographer","ros2_control"]
argument-hint	[task or question about ROS2]

ROS2 Humble 개발 스킬

환경 정보 (ROS2 Humble 기준)

• Distro: ROS2 Humble (Ubuntu 22.04 LTS)
• 경로: /opt/ros/humble/
• Python: 3.10 / C++: g++ 11.4 / CMake: 3.22
• DDS: FastRTPS (기본), CycloneDDS 지원
• 주요 스택: Nav2, MoveIt2, Gazebo, cartographer, rosbridge, tf2

환경 소싱

source /opt/ros/humble/setup.bash
# 또는 ~/.bashrc에 추가:
echo "source /opt/ros/humble/setup.bash" >> ~/.bashrc

---

워크스페이스 생성

mkdir -p ~/ros2_ws/src && cd ~/ros2_ws
colcon build --symlink-install
source install/setup.bash

package.xml (Python 패키지)

<?xml version="1.0"?>
<package format="3">
  <name>my_pkg</name>
  <version>0.0.1</version>
  <description>My ROS2 package</description>
  <maintainer email="[e-mail]">User</maintainer>
  <license>Apache-2.0</license>
  <depend>rclpy</depend>
  <depend>std_msgs</depend>
  <depend>geometry_msgs</depend>
  <build_depend>ament_cmake</build_depend>
  <buildtool_depend>ament_cmake</buildtool_depend>
  <exec_depend>ament_cmake_python</exec_depend>
  <test_depend>ament_lint_auto</test_depend>
  <test_depend>ament_lint_common</test_depend>
  <export>
    <build_type>ament_python</build_type>
  </export>
</package>

setup.py (Python 패키지)

from setuptools import setup

package_name = 'my_pkg'
setup(
    name=package_name,
    version='0.0.1',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index/packages', ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
        ('share/' + package_name + '/launch', ['launch/my_launch.py']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    entry_points={
        'console_scripts': [
            'my_node = my_pkg.my_node:main',
        ],
    },
)

---

Python 노드 패턴

Publisher + Timer

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

class MyPublisher(Node):
    def __init__(self):
        super().__init__('my_publisher')
        self.pub = self.create_publisher(String, 'topic', 10)
        self.timer = self.create_timer(0.5, self.timer_callback)
        self.i = 0

    def timer_callback(self):
        msg = String()
        msg.data = f'Hello {self.i}'
        self.pub.publish(msg)
        self.get_logger().info(f'Published: {msg.data}')
        self.i += 1

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

Subscriber

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

class MySubscriber(Node):
    def __init__(self):
        super().__init__('my_subscriber')
        self.sub = self.create_subscription(String, 'topic', self.callback, 10)

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

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

Service Server

from rclpy.node import Node
from std_srvs.srv import SetBool
import rclpy

class MyServiceServer(Node):
    def __init__(self):
        super().__init__('my_service_server')
        self.srv = self.create_service(SetBool, 'my_service', self.handle_request)

    def handle_request(self, request, response):
        self.get_logger().info(f'Request: {request.data}')
        response.success = True
        response.message = 'OK'
        return response

Service Client (async)

import rclpy
from rclpy.node import Node
from std_srvs.srv import SetBool

class MyServiceClient(Node):
    def __init__(self):
        super().__init__('my_service_client')
        self.client = self.create_client(SetBool, 'my_service')
        while not self.client.wait_for_service(timeout_sec=1.0):
            self.get_logger().warn('Service not available...')

    def send_request(self, value: bool):
        req = SetBool.Request()
        req.data = value
        future = self.client.call_async(req)
        rclpy.spin_until_future_complete(self, future)
        return future.result()

Action Server

import rclpy
from rclpy.node import Node
from rclpy.action import ActionServer
from action_msgs.msg import GoalStatus
# from my_interfaces.action import MyAction  # 커스텀 액션

class MyActionServer(Node):
    def __init__(self):
        super().__init__('my_action_server')
        self._action_server = ActionServer(
            self, MyAction, 'my_action', self.execute_callback)

    async def execute_callback(self, goal_handle):
        self.get_logger().info('Executing goal...')
        feedback = MyAction.Feedback()
        for i in range(10):
            feedback.progress = float(i) / 10.0
            goal_handle.publish_feedback(feedback)
            await asyncio.sleep(0.1)
        goal_handle.succeed()
        result = MyAction.Result()
        result.success = True
        return result

Parameter 사용

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node')
        self.declare_parameter('my_param', 'default_value')
        self.declare_parameter('speed', 1.0)

        param = self.get_parameter('my_param').get_parameter_value().string_value
        speed = self.get_parameter('speed').get_parameter_value().double_value

        # 파라미터 변경 콜백
        self.add_on_set_parameters_callback(self.param_callback)

    def param_callback(self, params):
        from rcl_interfaces.msg import SetParametersResult
        return SetParametersResult(successful=True)

---

C++ 노드 패턴

CMakeLists.txt

cmake_minimum_required(VERSION 3.8)
project(my_cpp_pkg)

find_package(ament_cmake REQUIRED)
find_package(rclcpp REQUIRED)
find_package(std_msgs REQUIRED)
find_package(geometry_msgs REQUIRED)

add_executable(my_node src/my_node.cpp)
ament_target_dependencies(my_node rclcpp std_msgs geometry_msgs)

install(TARGETS my_node DESTINATION lib/${PROJECT_NAME})
install(DIRECTORY launch DESTINATION share/${PROJECT_NAME})

ament_package()

Publisher (C++)

#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"

class MyPublisher : public rclcpp::Node {
public:
    MyPublisher() : Node("my_publisher"), count_(0) {
        pub_ = this->create_publisher<std_msgs::msg::String>("topic", 10);
        timer_ = this->create_wall_timer(
            std::chrono::milliseconds(500),
            std::bind(&MyPublisher::timer_callback, this));
    }

private:
    void timer_callback() {
        auto msg = std_msgs::msg::String();
        msg.data = "Hello " + std::to_string(count_++);
        pub_->publish(msg);
        RCLCPP_INFO(this->get_logger(), "Published: '%s'", msg.data.c_str());
    }
    rclcpp::Publisher<std_msgs::msg::String>::SharedPtr pub_;
    rclcpp::TimerBase::SharedPtr timer_;
    size_t count_;
};

int main(int argc, char* argv[]) {
    rclcpp::init(argc, argv);
    rclcpp::spin(std::make_shared<MyPublisher>());
    rclcpp::shutdown();
    return 0;
}

---

Launch 파일 (Python)

from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, IncludeLaunchDescription
from launch.substitutions import LaunchConfiguration, PathJoinSubstitution
from launch_ros.actions import Node
from launch_ros.substitutions import FindPackageShare

def generate_launch_description():
    use_sim_time = LaunchConfiguration('use_sim_time', default='false')

    return LaunchDescription([
        DeclareLaunchArgument('use_sim_time', default_value='false'),

        Node(
            package='my_pkg',
            executable='my_node',
            name='my_node',
            output='screen',
            parameters=[{
                'use_sim_time': use_sim_time,
                'speed': 1.5,
            }],
            remappings=[('/old_topic', '/new_topic')],
        ),

        IncludeLaunchDescription(
            PathJoinSubstitution([
                FindPackageShare('other_pkg'), 'launch', 'other.launch.py'
            ]),
            launch_arguments={'param': 'value'}.items(),
        ),
    ])

---

colcon 빌드 워크플로우

cd ~/ros2_ws

# 전체 빌드
colcon build --symlink-install

# 특정 패키지만
colcon build --symlink-install --packages-select my_pkg

# 의존성 포함
colcon build --symlink-install --packages-up-to my_pkg

# 병렬 빌드 (CPU 수 조절)
colcon build --symlink-install --parallel-workers 4

# 릴리즈 빌드
colcon build --cmake-args -DCMAKE_BUILD_TYPE=Release

# 빌드 후 소싱
source install/setup.bash

# 테스트 실행
colcon test --packages-select my_pkg
colcon test-result --verbose

빌드 캐시 정리

rm -rf build/ install/ log/
colcon build --symlink-install

---

디버깅 CLI 도구

# 토픽
ros2 topic list
ros2 topic echo /topic_name
ros2 topic hz /topic_name       # 주파수
ros2 topic bw /topic_name       # 대역폭
ros2 topic info /topic_name     # 타입 및 구독/퍼블리셔 정보
ros2 topic pub /topic std_msgs/msg/String "data: 'hello'"

# 노드
ros2 node list
ros2 node info /node_name

# 서비스
ros2 service list
ros2 service type /service_name
ros2 service call /my_service std_srvs/srv/SetBool "{data: true}"

# 액션
ros2 action list
ros2 action info /action_name
ros2 action send_goal /action nav2_msgs/action/NavigateToPose "{}"

# 파라미터
ros2 param list
ros2 param get /node_name param_name
ros2 param set /node_name param_name value
ros2 param dump /node_name

# 인터페이스 확인
ros2 interface show std_msgs/msg/String
ros2 interface show nav_msgs/msg/OccupancyGrid

# 그래프 시각화
rqt_graph
rqt

# rosbag
ros2 bag record -a                          # 전체 토픽 녹화
ros2 bag record /topic1 /topic2             # 선택 토픽
ros2 bag play my_bag/                       # 재생
ros2 bag info my_bag/                       # 정보 확인

---

TF2 패턴

import rclpy
from rclpy.node import Node
from tf2_ros import TransformBroadcaster, Buffer, TransformListener
from geometry_msgs.msg import TransformStamped
import tf2_ros
import tf2_geometry_msgs  # noqa

class TFNode(Node):
    def __init__(self):
        super().__init__('tf_node')

        # Broadcaster
        self.br = TransformBroadcaster(self)

        # Listener
        self.tf_buffer = Buffer()
        self.tf_listener = TransformListener(self.tf_buffer, self)

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

    def broadcast_tf(self):
        t = TransformStamped()
        t.header.stamp = self.get_clock().now().to_msg()
        t.header.frame_id = 'world'
        t.child_frame_id = 'robot'
        t.transform.translation.x = 1.0
        t.transform.translation.y = 0.0
        t.transform.translation.z = 0.0
        t.transform.rotation.w = 1.0
        self.br.sendTransform(t)

    def lookup_transform(self, target, source):
        try:
            return self.tf_buffer.lookup_transform(
                target, source, rclpy.time.Time())
        except tf2_ros.LookupException as e:
            self.get_logger().error(f'TF error: {e}')
            return None

---

QoS 프로파일

from rclpy.qos import QoSProfile, QoSDurabilityPolicy, QoSReliabilityPolicy, QoSHistoryPolicy

# 센서 데이터 (best effort, volatile)
sensor_qos = QoSProfile(
    reliability=QoSReliabilityPolicy.BEST_EFFORT,
    history=QoSHistoryPolicy.KEEP_LAST,
    depth=1,
    durability=QoSDurabilityPolicy.VOLATILE,
)

# 상태 데이터 (reliable, transient local)
state_qos = QoSProfile(
    reliability=QoSReliabilityPolicy.RELIABLE,
    history=QoSHistoryPolicy.KEEP_LAST,
    depth=1,
    durability=QoSDurabilityPolicy.TRANSIENT_LOCAL,
)

# 사용
self.pub = self.create_publisher(Image, '/camera/image', sensor_qos)

---

Nav2 통합

from nav2_simple_commander.robot_navigator import BasicNavigator, TaskResult
from geometry_msgs.msg import PoseStamped
import rclpy

def main():
    rclpy.init()
    navigator = BasicNavigator()
    navigator.waitUntilNav2Active()

    # 목표 포즈 설정
    goal = PoseStamped()
    goal.header.frame_id = 'map'
    goal.header.stamp = navigator.get_clock().now().to_msg()
    goal.pose.position.x = 2.0
    goal.pose.position.y = 1.0
    goal.pose.orientation.w = 1.0

    navigator.goToPose(goal)

    while not navigator.isTaskComplete():
        feedback = navigator.getFeedback()
        # 피드백 처리

    result = navigator.getResult()
    if result == TaskResult.SUCCEEDED:
        print('목표 도달!')
    elif result == TaskResult.FAILED:
        print('네비게이션 실패')

Nav2 실행

# 시뮬레이션
ros2 launch nav2_bringup tb3_simulation_launch.py

# 실제 로봇
ros2 launch nav2_bringup bringup_launch.py map:=/path/to/map.yaml

---

Gazebo 시뮬레이션

# Gazebo + ROS2 브릿지
ros2 launch gazebo_ros gazebo.launch.py

# 로봇 스폰
ros2 run gazebo_ros spawn_entity.py \
  -entity my_robot \
  -file /path/to/robot.urdf

# gz_ros2_control 사용시
ros2 launch my_pkg robot_sim.launch.py

URDF/Xacro 로봇 기술

<?xml version="1.0"?>
<robot name="my_robot" xmlns:xacro="http://www.ros.org/wiki/xacro">
  <xacro:property name="base_radius" value="0.2"/>

  <link name="base_link">
    <visual>
      <geometry><cylinder radius="${base_radius}" length="0.1"/></geometry>
    </visual>
    <collision>
      <geometry><cylinder radius="${base_radius}" length="0.1"/></geometry>
    </collision>
    <inertial>
      <mass value="5.0"/>
      <inertia ixx="0.1" iyy="0.1" izz="0.1" ixy="0" ixz="0" iyz="0"/>
    </inertial>
  </link>

  <!-- Gazebo 플러그인 -->
  <gazebo>
    <plugin name="diff_drive" filename="libgazebo_ros_diff_drive.so">
      <ros><namespace>/my_robot</namespace></ros>
      <left_joint>left_wheel_joint</left_joint>
      <right_joint>right_wheel_joint</right_joint>
      <wheel_separation>0.4</wheel_separation>
      <wheel_diameter>0.1</wheel_diameter>
    </plugin>
  </gazebo>
</robot>

---

커스텀 메시지/서비스/액션

패키지 구조

my_interfaces/
├── msg/
│   └── MyMsg.msg
├── srv/
│   └── MySrv.srv
├── action/
│   └── MyAction.action
├── CMakeLists.txt
└── package.xml

메시지 정의

# MyMsg.msg
std_msgs/Header header
float64 value
string label
int32[] data_array

서비스 정의

# MySrv.srv
float64 input_x
float64 input_y
---
float64 result
bool success
string message

액션 정의

# MyAction.action
# Goal
float64 target
---
# Result
bool success
float64 final_value
---
# Feedback
float32 progress

CMakeLists.txt (인터페이스 패키지)

find_package(rosidl_default_generators REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/MyMsg.msg"
  "srv/MySrv.srv"
  "action/MyAction.action"
  DEPENDENCIES std_msgs
)

---

라이프사이클 노드

import rclpy
from rclpy.lifecycle import LifecycleNode, State, TransitionCallbackReturn

class MyLifecycleNode(LifecycleNode):
    def __init__(self):
        super().__init__('my_lifecycle_node')

    def on_configure(self, state: State) -> TransitionCallbackReturn:
        self.get_logger().info('Configuring...')
        # 리소스 초기화
        return TransitionCallbackReturn.SUCCESS

    def on_activate(self, state: State) -> TransitionCallbackReturn:
        self.get_logger().info('Activating...')
        # 퍼블리셔/타이머 활성화
        return TransitionCallbackReturn.SUCCESS

    def on_deactivate(self, state: State) -> TransitionCallbackReturn:
        return TransitionCallbackReturn.SUCCESS

    def on_cleanup(self, state: State) -> TransitionCallbackReturn:
        return TransitionCallbackReturn.SUCCESS

    def on_shutdown(self, state: State) -> TransitionCallbackReturn:
        return TransitionCallbackReturn.SUCCESS

---

테스팅

# test_my_node.py
import pytest
import rclpy
from rclpy.node import Node
from std_msgs.msg import String

@pytest.fixture(autouse=True)
def ros_init():
    rclpy.init()
    yield
    rclpy.shutdown()

def test_publisher():
    node = Node('test_node')
    pub = node.create_publisher(String, 'test_topic', 10)
    assert pub is not None
    msg = String()
    msg.data = 'test'
    pub.publish(msg)
    node.destroy_node()

# launch 통합 테스트
from launch import LaunchDescription
from launch_ros.actions import Node as LaunchNode
import launch_testing

@pytest.mark.launch_test
def generate_test_description():
    return LaunchDescription([
        LaunchNode(package='my_pkg', executable='my_node', name='test_node'),
        launch_testing.actions.ReadyToTest(),
    ])

테스트 실행

cd ~/ros2_ws
colcon test --packages-select my_pkg
colcon test-result --verbose
# 개별 테스트
python3 -m pytest src/my_pkg/test/ -v

---

rosdep 의존성 설치

cd ~/ros2_ws
rosdep install --from-paths src --ignore-src -r -y

---

자주 쓰는 메시지 타입

용도	메시지 타입
문자열	std_msgs/msg/String
정수/실수	std_msgs/msg/Int32, Float64
2D 포즈	geometry_msgs/msg/Pose2D
3D 포즈	geometry_msgs/msg/PoseStamped
속도 명령	geometry_msgs/msg/Twist
이미지	sensor_msgs/msg/Image
포인트클라우드	sensor_msgs/msg/PointCloud2
IMU	sensor_msgs/msg/Imu
레이저스캔	sensor_msgs/msg/LaserScan
지도	nav_msgs/msg/OccupancyGrid
경로	nav_msgs/msg/Path
오도메트리	nav_msgs/msg/Odometry
조인트 상태	sensor_msgs/msg/JointState
진단	diagnostic_msgs/msg/DiagnosticArray

---

DDS 환경변수

# CycloneDDS 사용
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp

# FastRTPS (기본)
export RMW_IMPLEMENTATION=rmw_fastrtps_cpp

# 같은 호스트 내 통신만 (멀티캐스트 비활성화)
export ROS_LOCALHOST_ONLY=1

# 도메인 ID 설정 (기본 0)
export ROS_DOMAIN_ID=42

---

트러블슈팅

증상	원인	해결
ros2: command not found	환경 미소싱	source /opt/ros/humble/setup.bash
토픽 안 보임	도메인 ID 불일치	ROS_DOMAIN_ID 확인
빌드 실패	의존성 누락	rosdep install --from-paths src
QoS 호환성 경고	Publisher/Subscriber QoS 불일치	Reliability, Durability 맞추기
TF 에러	타임스탬프 불일치	use_sim_time 파라미터 확인
노드 안 죽음	Ctrl+C 미처리	rclpy.spin() 감싸기 with try/except
colcon 빌드 느림	병렬 빌드 안됨	--parallel-workers N

---

빠른 참조 명령

# 환경 소싱
source /opt/ros/humble/setup.bash && source ~/ros2_ws/install/setup.bash

# 새 패키지 생성
ros2 pkg create --build-type ament_python my_pkg --dependencies rclpy std_msgs
ros2 pkg create --build-type ament_cmake my_cpp_pkg --dependencies rclcpp std_msgs

# 빌드 & 실행
colcon build --symlink-install --packages-select my_pkg && source install/setup.bash
ros2 run my_pkg my_node

# 전체 시스템 상태
ros2 node list && ros2 topic list && ros2 service list

---

고급 패턴 (20개 에이전트 연구 결과)

Python 고급 패턴

ROS2 Humble / Python 3.10. 기본 튜토리얼에 없는 실전 패턴만 수록.

1. MultiThreadedExecutor + CallbackGroup

from rclpy.executors import MultiThreadedExecutor
from rclpy.callback_groups import MutuallyExclusiveCallbackGroup, ReentrantCallbackGroup

class ConcurrentNode(Node):
    def __init__(self):
        super().__init__('concurrent_node')
        srv_group  = MutuallyExclusiveCallbackGroup()  # 서비스: 동시 실행 방지
        scan_group = ReentrantCallbackGroup()           # 센서: 동시 처리 허용
        ctrl_group = MutuallyExclusiveCallbackGroup()  # 제어루프: 독립 전용

        self.create_service(SetBool, 'enable', self._enable_cb, callback_group=srv_group)
        self.create_subscription(LaserScan, '/scan', self._scan_cb, 10, callback_group=scan_group)
        self.create_timer(0.02, self._control_loop, callback_group=ctrl_group)

# ReentrantCallbackGroup은 MultiThreadedExecutor 없이 병렬 실행 안 됨
executor = MultiThreadedExecutor(num_threads=4)
executor.add_node(node); executor.spin()

2. async 서비스 호출 (콜백 내부)

콜백 안에서 spin_until_future_complete() → 데드락. async def + await 사용.

class AsyncServiceNode(Node):
    def __init__(self):
        super().__init__('async_svc')
        g = ReentrantCallbackGroup()
        self.cli = self.create_client(SetBool, 'upstream', callback_group=g)
        self.create_service(Trigger, 'do_work', self._handle, callback_group=g)

    async def _handle(self, req, res):
        future = self.cli.call_async(SetBool.Request(data=True))
        await future                    # executor가 다른 콜백 계속 처리
        res.success = bool(future.result() and future.result().success)
        return res
# 반드시 MultiThreadedExecutor + ReentrantCallbackGroup 조합

3. ApproximateTimeSynchronizer

header.stamp가 있는 N개 토픽을 시간 오차 범위 내에서 동기화.

import message_filters
from sensor_msgs.msg import Image, LaserScan, Imu

class FusionNode(Node):
    def __init__(self):
        super().__init__('fusion_node')
        subs = [
            message_filters.Subscriber(self, Image,     '/camera/image_raw'),
            message_filters.Subscriber(self, LaserScan, '/scan'),
            message_filters.Subscriber(self, Imu,       '/imu/data'),
        ]
        # slop: 허용 시간차(초) — 느린 토픽 주기의 절반 이하 권장
        self.ts = message_filters.ApproximateTimeSynchronizer(subs, queue_size=20, slop=0.05)
        self.ts.registerCallback(self._fused_cb)

    def _fused_cb(self, img: Image, scan: LaserScan, imu: Imu):
        self.get_logger().info('synced')  # slop 이내일 때만 호출; 안 되면 queue_size↑/slop↑

4. sim_time 처리

from rclpy.parameter import Parameter

class SimAwareNode(Node):
    def __init__(self):
        super().__init__('sim_node', parameter_overrides=[
            Parameter('use_sim_time', Parameter.Type.BOOL, True)
        ])
        self.create_timer(0.5, self._wait_clock)

    def _wait_clock(self):
        if self.get_clock().now().nanoseconds == 0:
            self.get_logger().warn('Waiting for /clock...'); return
        self.get_logger().info('Clock ready')

ros2 run my_pkg my_node --ros-args -p use_sim_time:=true
ros2 bag play my.bag --clock   # --clock 없으면 /clock 미게시 → 타이머 전부 정지

5. Transient Local QoS (ROS1 latched 대체)

from rclpy.qos import QoSProfile, DurabilityPolicy, ReliabilityPolicy, HistoryPolicy

def latched_qos(depth=1) -> QoSProfile:
    return QoSProfile(depth=depth, durability=DurabilityPolicy.TRANSIENT_LOCAL,
                      reliability=ReliabilityPolicy.RELIABLE, history=HistoryPolicy.KEEP_LAST)

# 퍼블리셔 한 번만 게시 → 늦게 접속한 구독자도 즉시 수신
pub = node.create_publisher(String, '/robot_description', latched_qos())
sub = node.create_subscription(String, '/robot_description', cb, latched_qos())
# 양쪽 durability 불일치 시 토픽 연결 자체가 성립하지 않음

Quick Reference

패턴	핵심	주의
병렬 콜백	MultiThreadedExecutor + ReentrantCallbackGroup	executor 없이는 직렬
콜백 직렬화	MutuallyExclusiveCallbackGroup	서비스/상태 변경용
async 서비스 호출	async def + await future	SingleThreaded에서 데드락
타임스탬프 동기화	ApproximateTimeSynchronizer(subs, queue_size, slop)	header.stamp 필수
sim_time 활성화	Parameter('use_sim_time', ..., True)	/clock 없으면 타이머 정지
rosbag 시계 재생	ros2 bag play --clock	use_sim_time과 함께 사용
보존 토픽	DurabilityPolicy.TRANSIENT_LOCAL	양쪽 QoS 일치 필수

---

C++ 고급 패턴

1. rclcpp Components — 런타임 로드 가능한 노드

// my_component.cpp
#include "my_pkg/my_component.hpp"
#include "rclcpp_components/register_node_macro.hpp"

namespace my_pkg {
MyComponent::MyComponent(const rclcpp::NodeOptions & options)
: Node("my_component", options)
{
  pub_ = this->create_publisher<std_msgs::msg::String>("chatter", 10);
  timer_ = this->create_timer(std::chrono::seconds(1), [this]() {
    pub_->publish(std_msgs::msg::String().set__data("hello"));
  });
}
}
RCLCPP_COMPONENTS_REGISTER_NODE(my_pkg::MyComponent)  // .cpp에 한 번만

# 런타임 로드
ros2 run rclcpp_components component_container &
ros2 component load /ComponentManager my_pkg my_pkg::MyComponent

2. MultiThreadedExecutor + CallbackGroup

// 생성자 내부
reentrant_cb_ = this->create_callback_group(rclcpp::CallbackGroupType::Reentrant);
exclusive_cb_ = this->create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive);

rclcpp::SubscriptionOptions opts;
opts.callback_group = reentrant_cb_;
sub_ = this->create_subscription<Msg>("topic", 10, callback, opts);
timer_ = this->create_timer(100ms, timer_cb, exclusive_cb_); // 별도 그룹

// main: num_threads=0 → hardware_concurrency() 자동 사용
rclcpp::executors::MultiThreadedExecutor exec(rclcpp::ExecutorOptions(), 4);
exec.add_node(node); exec.spin();

3. ParameterEventHandler — 원격 파라미터 감시

#include "rclcpp/parameter_event_handler.hpp"
// 생성자 내부 (/parameter_events 토픽 구독)
param_handler_ = std::make_shared<rclcpp::ParameterEventHandler>(this);

// 자신 노드 파라미터 감시
handle1_ = param_handler_->add_parameter_callback("speed",
  [this](const rclcpp::Parameter & p) {
    RCLCPP_INFO(get_logger(), "speed = %s", p.value_to_string().c_str());
  });
// 원격 노드 감시 (3번째 인자: FQN) — handle 소멸 시 자동 해제 (RAII)
handle2_ = param_handler_->add_parameter_callback("speed", cb, "/driver_node");

4. LifecycleNode C++

#include "rclcpp_lifecycle/lifecycle_node.hpp"
#include "rclcpp_lifecycle/lifecycle_publisher.hpp"
using CallbackReturn =
  rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn;

class Driver : public rclcpp_lifecycle::LifecycleNode {
public:
  Driver() : rclcpp_lifecycle::LifecycleNode("driver") {}
  CallbackReturn on_configure(const rclcpp_lifecycle::State &) override {
    pub_ = this->create_publisher<std_msgs::msg::String>("out", 10);
    return CallbackReturn::SUCCESS;
  }
  CallbackReturn on_activate(const rclcpp_lifecycle::State &) override {
    pub_->on_activate();  // 명시적 활성화 필수
    timer_ = this->create_timer(500ms, [this]() {
      if (pub_->is_activated()) pub_->publish(std_msgs::msg::String());
    });
    return CallbackReturn::SUCCESS;
  }
  CallbackReturn on_deactivate(const rclcpp_lifecycle::State &) override {
    timer_.reset(); pub_->on_deactivate(); return CallbackReturn::SUCCESS;
  }
  CallbackReturn on_cleanup(const rclcpp_lifecycle::State &) override {
    pub_.reset(); return CallbackReturn::SUCCESS;
  }
private:
  rclcpp_lifecycle::LifecyclePublisher<std_msgs::msg::String>::SharedPtr pub_;
  rclcpp::TimerBase::SharedPtr timer_;
};
// ros2 lifecycle set /driver configure && ros2 lifecycle set /driver activate

5. NodeOptions — 제로카피 + 파라미터 자동 선언

// NodeOptions는 fluent API — 체이닝 가능
auto node = std::make_shared<MyComponent>(rclcpp::NodeOptions{}
  .use_intra_process_comms(true)                         // 프로세스 내 zero-copy
  .automatically_declare_parameters_from_overrides(true) // YAML 자동 선언
  .allow_undeclared_parameters(true));

CMakeLists.txt — Component 빌드

find_package(rclcpp_components REQUIRED)
add_library(my_component SHARED src/my_component.cpp)
ament_target_dependencies(my_component rclcpp rclcpp_components std_msgs)
# 플러그인 등록 + 단독 실행 바이너리(my_component_node) 동시 생성
rclcpp_components_register_node(my_component
  PLUGIN "my_pkg::MyComponent"
  EXECUTABLE my_component_node)
install(TARGETS my_component my_component_node
  ARCHIVE DESTINATION lib
  LIBRARY DESTINATION lib
  RUNTIME DESTINATION lib/${PROJECT_NAME})

---

Service/Action 고급 패턴

Humble 제한사항: configure_introspection() (서비스 인트로스펙션)은 Iron+에서만 지원. Humble 대안: ros2 service list -t / ros2 service find <type> CLI 사용.

GoalStatus 상수 빠른 참조

값	상수	의미
0	STATUS_UNKNOWN	상태 미설정
1	STATUS_ACCEPTED	수락, 실행 대기
2	STATUS_EXECUTING	실행 중
3	STATUS_CANCELING	취소 수락, 정리 중
4	STATUS_SUCCEEDED	성공 완료
5	STATUS_CANCELED	외부 요청으로 취소됨
6	STATUS_ABORTED	서버 자체 종료

from action_msgs.msg import GoalStatus          # GoalStatus.STATUS_SUCCEEDED 등
from rclpy.action.server import GoalResponse, CancelResponse  # .ACCEPT / .REJECT

Cancel 처리 — execute_callback 패턴

루프마다 is_cancel_requested 폴링. 반환 전 터미널 메서드 반드시 호출.

async def execute_callback(self, goal_handle):
    try:
        for step in range(goal_handle.request.total_steps):
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()        # → CANCELED
                return MyAction.Result()
            goal_handle.publish_feedback(feedback)
            await asyncio.sleep(0.01)
        goal_handle.succeed()                 # → SUCCEEDED
        return result
    except Exception:
        goal_handle.abort()                   # → ABORTED
        return MyAction.Result()

Preempt 패턴 (built-in 없음, 직접 구현)

def handle_accepted_callback(self, goal_handle):
    if self._current_goal is not None and self._current_goal.is_active:
        self._current_goal.abort()       # 기존 goal 강제 종료
    self._current_goal = goal_handle
    goal_handle.execute()

ServerGoalHandle 속성: is_active, is_cancel_requested, request, goal_id

비동기 서비스 호출

future = self.cli.call_async(SetBool.Request(data=data))
future.add_done_callback(lambda f: self.get_logger().info(
    f'success={f.result().success}, msg={f.result().message}'))

데드락: SingleThreadedExecutor + 콜백 내 cli.call() = 데드락. 동기 호출 필요 시 MultiThreadedExecutor + ReentrantCallbackGroup 필수.

검증된 인터페이스:

std_srvs/SetBool:  bool data  →  bool success, string message
std_srvs/Trigger:  (empty)    →  bool success, string message

Multi-Goal: 동시 실행

ReentrantCallbackGroup + MultiThreadedExecutor 필수. 순차 큐는 handle_accepted_callback에서 큐잉 후 완료 시 queue[0].execute().

class MultiGoalServer(Node):
    def __init__(self):
        self._goals, self._lock = {}, threading.Lock()
        self._server = ActionServer(
            self, MyAction, 'my_action',
            execute_callback=self.execute_cb,
            goal_callback=lambda req: GoalResponse.REJECT
                if len(self._goals) >= 5 else GoalResponse.ACCEPT,
            handle_accepted_callback=lambda gh: (
                self._goals.__setitem__(str(gh.goal_id), gh) or gh.execute()),
            cancel_callback=lambda gh: CancelResponse.ACCEPT,
            callback_group=ReentrantCallbackGroup(),
        )

    async def execute_cb(self, gh):
        gid = str(gh.goal_id)
        try:
            for _ in range(gh.request.steps):
                if gh.is_cancel_requested:
                    gh.canceled(); return MyAction.Result()
                await asyncio.sleep(0.1)
            gh.succeed()
            return MyAction.Result()
        finally:
            with self._lock: self._goals.pop(gid, None)

# main(): MultiThreadedExecutor(num_threads=8).add_node(node); executor.spin()

Action Client: 비동기 흐름

# 전송 → 수락 확인 → 결과 대기
send_f = self._client.send_goal_async(goal_msg, feedback_callback=self.fb_cb)
send_f.add_done_callback(lambda f: (
    f.result().get_result_async().add_done_callback(self.result_cb)
    if f.result().accepted else None))

def result_cb(self, future):
    status = future.result().status   # GoalStatus int
    result = future.result().result   # action Result 객체

# 취소: goals_canceling 비어 있으면 서버가 거부
self._goal_handle.cancel_goal_async()

---

Launch 고급 패턴

OpaqueFunction — 런타임 분기

대체 표현식으로 처리 불가한 로직(경로 계산, 환경 변수 분기)에 사용. 서명 func(context, *args, **kwargs), 반환값은 반드시 list.

def launch_setup(context, *args, **kwargs):
    robot_name = LaunchConfiguration('robot_name').perform(context)
    return [Node(
        package='robot_state_publisher', executable='robot_state_publisher',
        parameters=[{'robot_description': open(f'/robots/{robot_name}/model.urdf').read()}],
    )]

def generate_launch_description():
    return LaunchDescription([
        DeclareLaunchArgument('robot_name', default_value='my_robot'),
        OpaqueFunction(function=launch_setup),
    ])

ComposableNodeContainer — 제로 복사 인트라 프로세스

동일 컨테이너 내 노드는 직렬화 없이 포인터로 메시지 전달. 양쪽 노드 모두 extra_arguments=[{'use_intra_process_comms': True}] 필수.

ComposableNodeContainer(
    name='image_proc_container',
    namespace='',
    package='rclcpp_components',
    executable='component_container_mt',   # mt=멀티스레드, isolated=노드별 스레드
    composable_node_descriptions=[
        ComposableNode(
            package='image_proc', plugin='image_proc::DebayerNode',
            name='debayer', remappings=[('image_raw', 'camera/image_raw')],
            extra_arguments=[{'use_intra_process_comms': True}],
        ),
        ComposableNode(
            package='image_proc', plugin='image_proc::RectifyColorNode',
            name='rectify',
            extra_arguments=[{'use_intra_process_comms': True}],
        ),
    ],
    output='screen',
)

이미 실행 중인 컨테이너에 추가할 때는 LoadComposableNodes(target_container=..., ...) 사용. LaunchConfigurationEquals/NotEquals로 생성/로드를 분기하는 것이 표준 패턴.

GroupAction + PushRosNamespace

PushRosNamespace는 GroupAction 범위 내 모든 노드에 네임스페이스 자동 적용. IfCondition과 결합해 조건부 그룹 구성.

GroupAction(
    condition=IfCondition(LaunchConfiguration('launch_camera')),
    actions=[
        PushRosNamespace('camera'),
        Node(package='usb_cam', executable='usb_cam_node_exe'),
        Node(package='image_proc', executable='image_proc'),
        # 두 노드 모두 /camera/ 네임스페이스 자동 상속
    ],
)

IfCondition / UnlessCondition

'true'/'false'/'1'/'0'으로 resolve되는 Substitution을 받음. AndSubstitution, OrSubstitution, NotSubstitution으로 복합 조건 구성.

# use_rviz AND simulation 모두 true일 때만 실행
Node(package='rviz2', executable='rviz2',
     condition=IfCondition(AndSubstitution(
         LaunchConfiguration('use_rviz'), LaunchConfiguration('simulation'))))

# headless=true 이면 실행 안 함
Node(package='my_pkg', executable='visualizer',
     condition=UnlessCondition(LaunchConfiguration('headless')))

TimerAction — 지연 실행

노드 기동 순서 의존성 해소. period는 초 단위 float.

LaunchDescription([
    TimerAction(period=0.0, actions=[GroupAction([secondary_include])]),
    TimerAction(period=2.0, actions=[GroupAction([primary_include])]),
])

파라미터 YAML 로딩

파일 경로와 인라인 dict를 parameters= 혼합 가능. 동적 경로는 OpaqueFunction 내 perform(context)로 resolve.

pkg_share = get_package_share_directory('my_robot_bringup')

Node(
    package='my_robot', executable='my_node',
    parameters=[
        os.path.join(pkg_share, 'config', 'params.yaml'),  # 정적 경로
        {'use_sim_time': True},                              # 인라인 오버라이드
    ],
)

YAML 구조:

my_node:
  ros__parameters:
    update_rate: 50.0
    sensor_frame: base_link

/**:                        # 와일드카드: 모든 노드에 적용
  ros__parameters:
    use_sim_time: true

---

Nav2 심층 통합

BasicNavigator 전체 API

메서드	설명
setInitialPose(pose)	AMCL 초기 위치 설정 (PoseStamped)
waitUntilNav2Active(navigator, localizer)	스택 활성화 대기
goToPose(goal, behavior_tree=None)	단일 목표 탐색
goThroughPoses(poses, behavior_tree=None)	중간 경유지 순차 통과
followWaypoints(poses)	웨이포인트 리스트 순차 실행
followPath(path, controller_id, goal_checker_id)	사전 계산 경로 실행
spin(spin_dist, time_allowance)	제자리 회전 (라디안)
backup(backup_dist, backup_speed, time_allowance)	후진 이동
cancelTask()	현재 액션 취소
isTaskComplete()	완료 여부 폴링 (100ms 타임아웃)
getFeedback()	최신 피드백 메시지 반환 또는 None
getResult()	TaskResult 열거형 반환
getPath(start, goal, planner_id, use_start)	경로 계산 (미실행)
smoothPath(path, smoother_id, max_duration)	경로 스무딩
clearAllCostmaps()	로컬+글로벌 costmap 초기화
clearLocalCostmap() / clearGlobalCostmap()	개별 초기화
getGlobalCostmap() / getLocalCostmap()	Costmap 반환
changeMap(map_filepath)	정적 지도 핫스왑
lifecycleStartup() / lifecycleShutdown()	수동 시작/종료

최소 nav2_params.yaml 구조

lifecycle_manager:
  ros__parameters:
    use_sim_time: True
    autostart: True
    node_names: [map_server, amcl, planner_server, controller_server,
                 bt_navigator, behavior_server, waypoint_follower]
planner_server:
  ros__parameters:
    planner_plugins: ["GridBased"]
    GridBased:
      plugin: "nav2_navfn_planner/NavfnPlanner"
      tolerance: 0.5
      use_astar: false
controller_server:
  ros__parameters:
    controller_frequency: 20.0
    controller_plugins: ["FollowPath"]
    FollowPath:
      plugin: "dwb_core::DWBLocalPlanner"
      max_vel_x: 0.26
      max_vel_theta: 1.0
global_costmap:
  global_costmap:
    ros__parameters:
      resolution: 0.05
      robot_radius: 0.22
      plugins: ["static_layer", "obstacle_layer", "inflation_layer"]
      static_layer:
        plugin: "nav2_costmap_2d::StaticLayer"
        map_subscribe_transient_local: True
      obstacle_layer:
        plugin: "nav2_costmap_2d::ObstacleLayer"
        observation_sources: scan
        scan: {topic: /scan, data_type: LaserScan, clearing: True, marking: True}
      inflation_layer: {plugin: "nav2_costmap_2d::InflationLayer", cost_scaling_factor: 3.0, inflation_radius: 0.55}
local_costmap:
  local_costmap:
    ros__parameters:
      rolling_window: true
      width: 3
      height: 3
      resolution: 0.05
      global_frame: odom
      plugins: ["voxel_layer", "inflation_layer"]
      voxel_layer:
        plugin: "nav2_costmap_2d::VoxelLayer"
        observation_sources: scan
        scan: {topic: /scan, data_type: LaserScan, clearing: True, marking: True}
      inflation_layer: {plugin: "nav2_costmap_2d::InflationLayer", cost_scaling_factor: 3.0, inflation_radius: 0.55}

비용 값: 0=FREE, 1-252=Inflated, 253=LETHAL_OBSTACLE, 255=NO_INFORMATION

웨이포인트 순찰 패턴

import rclpy
from nav2_simple_commander.robot_navigator import BasicNavigator, TaskResult
from geometry_msgs.msg import PoseStamped

def make_pose(nav, x, y):
    p = PoseStamped()
    p.header.frame_id = 'map'
    p.header.stamp = nav.get_clock().now().to_msg()
    p.pose.position.x = x; p.pose.position.y = y
    p.pose.orientation.w = 1.0
    return p
rclpy.init()
nav = BasicNavigator()
nav.waitUntilNav2Active()
waypoints = [make_pose(nav,x,y) for x,y in [(0,0),(2,0),(2,2),(0,2)]]  # 순찰 경로

while rclpy.ok():
    nav.followWaypoints(waypoints)
    while not nav.isTaskComplete():
        fb = nav.getFeedback()
        if fb: print(f'Waypoint {fb.current_waypoint + 1}/{len(waypoints)}')
    if nav.getResult() != TaskResult.SUCCEEDED:
        nav.clearAllCostmaps()  # 오래된 장애물 제거 후 재시도

slam_toolbox 런치 스니펫

# launch/slam_nav2.launch.py — AMCL 대신 slam_toolbox 사용
from launch_ros.actions import Node

slam_node = Node(
    package='slam_toolbox',
    executable='async_slam_toolbox_node',
    name='slam_toolbox',
    parameters=['/path/to/mapper_params_online_async.yaml', {'use_sim_time': True}]
)
# yaml 핵심: mode: mapping|localization, scan_topic: /scan, map_update_interval: 5.0,
#            resolution: 0.05, max_laser_range: 20.0
# setInitialPose() 불필요 — slam_toolbox가 map->odom TF 퍼블리시
nav.waitUntilNav2Active(navigator='bt_navigator', localizer='slam_toolbox')

# SLAM 중 지도 저장 → localization 모드 전환 시 map_file_name에 지정
ros2 run nav2_map_server map_saver_cli -f ~/my_map --ros-args -p save_map_timeout:=5.0

---

MoveIt2 통합

MoveIt2는 ROS2의 표준 모션 플래닝 프레임워크다. MoveGroupInterface로 플래닝·실행을 제어하고, PlanningSceneInterface로 충돌 객체를 관리한다. 설치: sudo apt install ros-humble-moveit ros-humble-moveit-servo

MoveGroupInterface 핵심 메서드 (C++)

메서드	설명
MoveGroupInterface(node, "group")	초기화 (백그라운드 executor 필수)
setPoseTarget(pose)	목표 포즈 지정
setNamedTarget("ready")	SRDF 이름 상태로 이동
setJointValueTarget(values)	관절 목표값 지정
plan(plan)	경로 플래닝 (반환: MoveItErrorCode)
execute(plan) / move()	실행 / 플래닝+실행 한 번에
computeCartesianPath(wps, eef_step, jump, traj)	Cartesian 경로 (반환: 달성률)
getCurrentPose("link")	현재 엔드이펙터 포즈
setMaxVelocityScalingFactor(0.5)	최대 속도 비율 설정
setPathConstraints(c) / clearPathConstraints()	경로 제약 설정·해제
getPlanningFrame() / getEndEffectorLink()	프레임·링크 이름 조회

// MoveGroupInterface는 백그라운드 spin 필요
auto node = std::make_shared<rclcpp::Node>("moveit_node");
auto exec = std::make_shared<rclcpp::executors::SingleThreadedExecutor>();
exec->add_node(node); std::thread([&exec]() { exec->spin(); }).detach();
moveit::planning_interface::MoveGroupInterface mg(node, "panda_arm");
moveit::planning_interface::PlanningSceneInterface psi;

// 포즈 목표 플래닝·실행
geometry_msgs::msg::Pose target;
target.orientation.w = 1.0; target.position.x = 0.28; target.position.z = 0.5;
mg.setPoseTarget(target);
mg.setMaxVelocityScalingFactor(0.5);
moveit::planning_interface::MoveGroupInterface::Plan plan;
if (mg.plan(plan) == moveit::core::MoveItErrorCode::SUCCESS) mg.execute(plan);

// Cartesian 경로 (eef_step=1cm, jump_threshold=0 비활성)
std::vector<geometry_msgs::msg::Pose> wps = {mg.getCurrentPose().pose};
wps.back().position.z += 0.2;
moveit_msgs::msg::RobotTrajectory traj;
if (mg.computeCartesianPath(wps, 0.01, 0.0, traj) > 0.95) mg.execute(traj);

Planning Scene: 추가·제거·부착

// 박스 추가
moveit_msgs::msg::CollisionObject obj;
obj.header.frame_id = mg.getPlanningFrame(); obj.id = "box1";
shape_msgs::msg::SolidPrimitive prim;
prim.type = prim.BOX; prim.dimensions = {0.1, 0.1, 0.1};
geometry_msgs::msg::Pose p; p.orientation.w = 1.0;
p.position.x = 0.48; p.position.z = 0.25;
obj.primitives.push_back(prim); obj.primitive_poses.push_back(p);
obj.operation = moveit_msgs::msg::CollisionObject::ADD;
psi.addCollisionObjects({obj});

// 파지 후 부착 → 내려놓고 제거
mg.attachObject("box1", "panda_hand", {"panda_leftfinger", "panda_rightfinger"});
mg.detachObject("box1");
psi.removeCollisionObjects({"box1"});

MoveItConfigsBuilder 런치 패턴

# launch/move_group.launch.py
from launch import LaunchDescription
from launch_ros.actions import Node
from moveit_configs_utils import MoveItConfigsBuilder

def generate_launch_description():
    moveit_config = (
        MoveItConfigsBuilder("panda", package_name="panda_moveit_config")
        .robot_description(file_path="config/panda.urdf.xacro")
        .robot_description_semantic(file_path="config/panda.srdf")
        .trajectory_execution(file_path="config/moveit_controllers.yaml")
        .planning_pipelines(pipelines=["ompl", "pilz_industrial_motion_planner"])
        .to_moveit_configs()
    )
    return LaunchDescription([
        Node(
            package="moveit_ros_move_group",
            executable="move_group",
            output="screen",
            parameters=[moveit_config.to_dict()],  # 설정 전체를 dict로 전달
        )
    ])

Servo 실시간 Cartesian 제어

MoveIt Servo는 TwistStamped 명령을 ~100 Hz로 수신해 관절 명령으로 변환한다. 특이점·관절 한계 접근 시 자동 감속/정지한다.

토픽 / 서비스	타입	용도
/servo_node/delta_twist_cmds	TwistStamped	Cartesian 속도 명령
/servo_node/delta_joint_cmds	JointJog	관절 속도 명령
/servo_node/start_servo	Trigger (srv)	Servo 시작
/servo_node/stop_servo	Trigger (srv)	Servo 정지

from geometry_msgs.msg import TwistStamped
from std_srvs.srv import Trigger

node.create_client(Trigger, '/servo_node/start_servo').call_async(Trigger.Request())

pub = node.create_publisher(TwistStamped, '/servo_node/delta_twist_cmds', 10)
cmd = TwistStamped()
cmd.header.stamp = node.get_clock().now().to_msg()
cmd.header.frame_id = "panda_link0"
cmd.twist.linear.x = 0.1    # m/s
cmd.twist.angular.z = 0.2   # rad/s
pub.publish(cmd)  # timeout 내 반복 발행 필요

node.create_client(Trigger, '/servo_node/stop_servo').call_async(Trigger.Request())

servo_params.yaml 핵심 항목:

moveit_servo:
  move_group_name: panda_arm
  ee_frame_name: panda_link8
  robot_link_command_frame: panda_link0
  incoming_command_timeout: 0.1       # 명령 없으면 자동 정지
  lower_singularity_threshold: 17.0
  hard_stop_singularity_threshold: 30.0
  publish_period: 0.034               # ~30 Hz 출력

---

Gazebo 시뮬레이션 고급

Gazebo Classic 11 + ROS2 Humble. 핵심 패키지: gazebo_ros_pkgs.

시스템 플러그인 (world SDF 필수)

<plugin name="gazebo_ros_init"    filename="libgazebo_ros_init.so"/>
<plugin name="gazebo_ros_factory" filename="libgazebo_ros_factory.so"/>
<plugin name="gazebo_ros_state"   filename="libgazebo_ros_state.so"/>

gazebo_ros.launch.py 사용 시 자동 로드. init→/clock, factory→spawn/delete 서비스, state→model_states.

spawn_entity.py

ros2 run gazebo_ros spawn_entity.py \
  -topic /robot_description -entity my_robot -x 0.0 -y 0.0 -z 0.1
ros2 run gazebo_ros spawn_entity.py \
  -file robot.sdf -entity robot1 -robot_namespace /robot1 -x 1.0

주요 인자: -topic/-file/-database (소스 택1), -entity (필수), -x/-y/-z, -R/-P/-Y, -robot_namespace, -unpause, -timeout.

diff_drive 플러그인

<gazebo>
  <plugin name="diff_drive" filename="libgazebo_ros_diff_drive.so">
    <left_joint>left_wheel_joint</left_joint>
    <right_joint>right_wheel_joint</right_joint>
    <wheel_separation>0.3</wheel_separation>  <!-- 중심-중심 (m) -->
    <wheel_diameter>0.1</wheel_diameter>
    <max_wheel_torque>20</max_wheel_torque>
    <max_wheel_acceleration>1.0</max_wheel_acceleration>
    <command_topic>cmd_vel</command_topic>
    <odometry_topic>odom</odometry_topic>
    <odometry_frame>odom</odometry_frame>
    <robot_base_frame>base_footprint</robot_base_frame>
    <odometry_source>0</odometry_source>  <!-- 0=엔코더, 1=ground truth -->
    <publish_odom>true</publish_odom>
    <publish_odom_tf>true</publish_odom_tf>
    <update_rate>50</update_rate>
    <ros><namespace>/</namespace></ros>
  </plugin>
</gazebo>

카메라 + IMU 센서 플러그인

<!-- libgazebo_ros_camera.so -->
<sensor name="camera_sensor" type="camera">
  <update_rate>30.0</update_rate>
  <camera name="front_camera">
    <horizontal_fov>1.3962634</horizontal_fov>
    <image><width>640</width><height>480</height><format>R8G8B8</format></image>
    <clip><near>0.02</near><far>300</far></clip>
  </camera>
  <plugin name="camera_plugin" filename="libgazebo_ros_camera.so">
    <ros><namespace>/camera</namespace></ros>
    <camera_name>front_camera</camera_name>
    <frame_name>camera_link_optical</frame_name>
  </plugin>
</sensor>
<!-- libgazebo_ros_imu_sensor.so -->
<sensor name="imu_sensor" type="imu">
  <update_rate>200</update_rate>
  <plugin name="imu_plugin" filename="libgazebo_ros_imu_sensor.so">
    <ros><namespace>/</namespace><remapping>~/out:=imu</remapping></ros>
    <initial_orientation_as_reference>false</initial_orientation_as_reference>
    <frame_name>imu_link</frame_name>
  </plugin>
</sensor>

Lidar: libgazebo_ros_ray_sensor.so + <output_type>sensor_msgs/LaserScan</output_type>. 퍼블리시: camera/image_raw, imu.

gazebo_ros2_control

<ros2_control name="GazeboSystem" type="system">
  <hardware><plugin>gazebo_ros2_control/GazeboSystem</plugin></hardware>
  <joint name="left_wheel_joint">
    <command_interface name="velocity"><param name="min">-10</param><param name="max">10</param></command_interface>
    <state_interface name="position"/><state_interface name="velocity"/>
  </joint>
  <joint name="right_wheel_joint">
    <command_interface name="velocity"><param name="min">-10</param><param name="max">10</param></command_interface>
    <state_interface name="position"/><state_interface name="velocity"/>
  </joint>
</ros2_control>
<gazebo>
  <plugin filename="libgazebo_ros2_control.so" name="gazebo_ros2_control">
    <robot_param>robot_description</robot_param>
    <robot_param_node>robot_state_publisher</robot_param_node>
    <parameters>$(find my_robot)/config/controllers.yaml</parameters>
  </plugin>
</gazebo>

controller_manager:
  ros__parameters:
    update_rate: 100
    diff_drive_controller:
      type: diff_drive_controller/DiffDriveController
    joint_state_broadcaster:
      type: joint_state_broadcaster/JointStateBroadcaster
diff_drive_controller:
  ros__parameters:
    left_wheel_names: ["left_wheel_joint"]
    right_wheel_names: ["right_wheel_joint"]
    wheel_separation: 0.3
    wheel_radius: 0.05
    odom_frame_id: odom
    base_frame_id: base_link
    enable_odom_tf: true
    cmd_vel_timeout: 0.5

컨트롤러 로드: ros2 run controller_manager spawner joint_state_broadcaster diff_drive_controller

검증된 주요 플러그인 (/opt/ros/humble/lib/)

libgazebo_ros_init.so · libgazebo_ros_factory.so · libgazebo_ros_state.so (시스템 3종 필수) libgazebo_ros_diff_drive.so · libgazebo_ros2_control.so · libgazebo_ros_camera.so libgazebo_ros_ray_sensor.so · libgazebo_ros_imu_sensor.so · libgazebo_ros_p3d.so · libgazebo_ros_joint_state_publisher.so · libgazebo_ros_gps_sensor.so · libgazebo_ros_ft_sensor.so

모든 노드에 use_sim_time:=true 필수. gazebo_ros_init 가 /clock 을 자동 퍼블리시함.

---

디버깅 고급 도구

별도 설치 필요: ros2 doctor → ros-humble-ros2doctor, rqt_* → ros-humble-rqt-graph ros-humble-rqt-console ros-humble-rqt-plot ros-humble-rqt-topic ros-humble-rqt-service-caller

---

rqt GUI 도구

rqt --standalone rqt_graph          # 노드/토픽 publish-subscribe 그래프
ros2 run rqt_console rqt_console    # /rosout 로그 스트림 (severity/regex 필터)
ros2 run rqt_plot rqt_plot /cmd_vel/linear/x /odom/pose/pose/position/x
ros2 run rqt_topic rqt_topic        # 토픽 목록 + Hz/BW 표시
ros2 run rqt_service_caller rqt_service_caller  # GUI 서비스 호출

# 헤드리스 서버: Xvfb 필요
export DISPLAY=:99 && Xvfb :99 -screen 0 1024x768x24 &

---

ros2 topic / node / service 고급 명령

# 메시지 타입으로 토픽 검색
ros2 topic find sensor_msgs/msg/LaserScan
ros2 topic find geometry_msgs/msg/Twist

# Header.stamp 기반 end-to-end 지연 측정 (지연 증가 → 하류 처리 병목)
ros2 topic delay /camera/image_raw --window 100

# 노드 상세 정보 (publishers/subscribers/services/actions)
ros2 node info /my_node
ros2 node list --all   # 숨겨진 노드 포함

# 서비스 타입 확인 및 직접 호출
ros2 service type /my_service
ros2 service call /set_bool std_srvs/srv/SetBool "{data: true}"

---

RCUTILS 로깅 환경 변수

변수	용도
RCUTILS_LOGGING_SEVERITY_THRESHOLD	전체 최소 레벨: DEBUG INFO WARN ERROR FATAL
RCUTILS_CONSOLE_OUTPUT_FORMAT	포맷 토큰: {severity} {time} {name} {message} {line_number}
RCUTILS_COLORIZED_OUTPUT	ANSI 색상 (1 = 활성화)

RCUTILS_LOGGING_SEVERITY_THRESHOLD=DEBUG ros2 run my_pkg my_node
ros2 run my_pkg my_node --ros-args --log-level my_node:=DEBUG

---

rosbag2 필터링 / 압축

# 토픽 선택 및 정규식 필터
ros2 bag record /scan /odom /cmd_vel -o nav_bag
ros2 bag record -a -e "/camera.*" -x "/camera/.*/compressed" -o cam_bag

# zstd 메시지 레벨 압축
ros2 bag record -a --compression-mode message --compression-format zstd \
  --compression-threads 4 -o compressed_bag

# 재생: 속도 조정 + 토픽 필터 + 시뮬레이션 클럭
ros2 bag play nav_bag/ -r 0.5 --topics /scan /odom --clock 200

# sqlite3 → mcap 변환 (convert_options.yaml: storage_id: mcap)
ros2 bag convert -i my_bag/ -o convert_options.yaml

---

GDB로 ROS2 노드 디버깅

# 인터랙티브 디버깅
ros2 run my_pkg my_node --prefix 'gdb -ex run --args'

# 크래시 시 backtrace만 출력 후 종료 (CI 자동화)
ros2 run my_pkg my_node --prefix 'gdb -batch -ex run -ex bt --args'

# 디버그 심볼 빌드
colcon build --cmake-args -DCMAKE_BUILD_TYPE=Debug           # 최적화 제거
colcon build --cmake-args -DCMAKE_BUILD_TYPE=RelWithDebInfo  # 최적화 + 심볼 (운영 환경 권장)

# 실행 중인 노드에 attach
gdb -p $(pgrep my_node)
# (gdb) thread apply all bt   → 전체 스레드 backtrace 동시 출력

---

테스팅 고급 패턴

launch_testing 완전 패턴

active test (노드 실행 중)와 post-shutdown test (종료 후) 두 단계로 구성된다.

# test/test_node.launch.py
import unittest, pytest, launch, launch_ros.actions
import launch_testing, launch_testing.actions

@pytest.mark.launch_test
def generate_test_description():
    node = launch_ros.actions.Node(
        package='my_pkg', executable='my_node', name='my_node')
    return launch.LaunchDescription([
        node, launch_testing.actions.ReadyToTest(),  # 필수
    ]), {'my_node': node}

class TestActive(unittest.TestCase):
    def test_startup(self, my_node, proc_output):
        proc_output.assertWaitFor('Node ready', process=my_node, timeout=10)

@launch_testing.post_shutdown_test()
class TestAfterShutdown(unittest.TestCase):
    def test_exit_codes(self, proc_info):
        launch_testing.asserts.assertExitCodes(proc_info)

@launch_testing.markers.keep_alive: 프로세스 종료 후에도 launch 유지.

MockPublisherNode 패턴

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

class MockPublisherNode(Node):
    def __init__(self, topic='/test'):
        super().__init__('mock_publisher')
        self.pub = self.create_publisher(Int32, topic, 10)
    def publish(self, data):
        self.pub.publish(Int32(data=data))

class MockSubscriberNode(Node):
    def __init__(self, topic='/test'):
        super().__init__('mock_subscriber')
        self.received = []
        self.create_subscription(Int32, topic, self.received.append, 10)

def test_pub_sub(rclpy_init):
    pub, sub = MockPublisherNode(), MockSubscriberNode()
    pub.publish(42)
    rclpy.spin_once(sub, timeout_sec=1.0)
    assert sub.received[0].data == 42
    pub.destroy_node(); sub.destroy_node()

colcon test 핵심 옵션

colcon test --packages-select my_pkg --return-code-on-test-failure
colcon test --event-handlers console_direct+   # 실시간 출력
colcon test --executor sequential              # DDS 격리
colcon test --pytest-args -v -k test_my_case   # pytest 인자 전달
colcon test-result            # 실패만 출력
colcon test-result --all      # 전체 결과
colcon test-result --verbose  # 상세 (실패 로그 포함)

ament 린팅

ament_copyright src/ test/   # 저작권 헤더
ament_flake8 src/ test/      # Python PEP8 + pyflakes
ament_cpplint include/ src/  # C++ Google style

CMakeLists.txt (colcon test 시 lint 자동 실행):

if(BUILD_TESTING)
  find_package(ament_lint_auto REQUIRED)
  ament_lint_auto_find_test_dependencies()  # copyright, flake8, cpplint 등록
endif()

package.xml: <test_depend>ament_lint_auto</test_depend> + <test_depend>ament_lint_common</test_depend>.

GitHub Actions — ros-tooling/action-ros-ci

name: ROS2 CI
on:
  push:         { branches: [main] }
  pull_request: { branches: [main] }
jobs:
  test:
    runs-on: ubuntu-22.04
    env:
      ROS_DOMAIN_ID: ${{ github.run_number % 101 + 100 }}  # DDS 충돌 방지
    steps:
      - uses: actions/checkout@v4
      - uses: ros-tooling/setup-ros@v0.7
        with:
          required-ros-distributions: humble
      - uses: ros-tooling/action-ros-ci@v0.3
        with:
          package-name: my_pkg
          target-ros2-distro: humble
          colcon-defaults: |
            {
              "build": {"cmake-args": ["-DCMAKE_BUILD_TYPE=Release"]},
              "test":  {"pytest-with-coverage": true}
            }
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: test-results
          path: ros_ws/build/*/test_results/**/*.xml

action-ros-ci: rosdep 설치, colcon build/test, 결과 수집 자동 처리. skip-tests: true로 빌드 전용 실행 가능.

---

TF2 / QoS / 라이프사이클 고급

TF2 C++ 조회 패턴

#include "tf2_ros/buffer.hpp"
#include "tf2_ros/transform_listener.hpp"
#include "tf2_geometry_msgs/tf2_geometry_msgs.hpp"

// 생성자 (spin_thread=true 기본값 → 블로킹 lookup 가능)
tf_buffer_ = std::make_unique<tf2_ros::Buffer>(this->get_clock());
tf_listener_ = std::make_shared<tf2_ros::TransformListener>(*tf_buffer_, this);

// timeout 포함 lookup
try {
  auto tf = tf_buffer_->lookupTransform(
    "map", "base_link",
    tf2::TimePointZero,          // 최신 transform
    tf2::durationFromSec(1.0));  // 최대 1 초 대기
  geometry_msgs::msg::PointStamped out;
  tf2::doTransform(point_in, out, tf); // tf2_geometry_msgs 링크 필요
} catch (const tf2::TransformException & ex) {
  RCLCPP_WARN(get_logger(), "TF: %s", ex.what());
}

TF2 예외 (tf2::TransformException 부모): LookupException · ConnectivityException · ExtrapolationException · InvalidArgumentException

TF2 Python 조회 패턴

import tf2_ros, tf2_geometry_msgs  # do_transform_* 등록 필수
self.tf_buf = tf2_ros.Buffer()
self.tf_listener = tf2_ros.TransformListener(self.tf_buf, self)

# 고수준 API (권장)
transformed = self.tf_buf.transform(pose_stamped, 'map')
# 저수준: lookup 후 명시적 변환
tf = self.tf_buf.lookup_transform('map', 'base_link', rclpy.time.Time())
out = tf2_geometry_msgs.do_transform_pose(pose_stamped, tf)

/tf_static → TRANSIENT_LOCAL (늦게 참여해도 전체 수신) /tf → VOLATILE

---

QoS 프로파일 비교표 (Humble 검증값)

프로파일	History	Depth	Reliability	Durability
sensor_data	KEEP_LAST	5	BEST_EFFORT	VOLATILE
parameters	KEEP_LAST	1000	RELIABLE	VOLATILE
parameter_events	KEEP_LAST	1000	RELIABLE	VOLATILE
services_default	KEEP_LAST	10	RELIABLE	VOLATILE
system_default	SYSTEM_DEFAULT	0	SYSTEM_DEFAULT	SYSTEM_DEFAULT

호환성: 구독자 정책이 발행자보다 엄격하면 연결 불가 (무음 실패). BEST_EFFORT 발행 → RELIABLE 구독: 불가 / VOLATILE 발행 → TRANSIENT_LOCAL 구독: 불가.

from rclpy.qos import QoSProfile, QoSReliabilityPolicy, QoSDurabilityPolicy, qos_profile_sensor_data
self.create_subscription(Imu, '/imu', self.cb, qos_profile_sensor_data)
custom = QoSProfile(reliability=QoSReliabilityPolicy.RELIABLE,
                    durability=QoSDurabilityPolicy.TRANSIENT_LOCAL, depth=10)

rclcpp::SensorDataQoS()  // best_effort, volatile, keep_last(5)
rclcpp::ServicesQoS()    // reliable,    volatile, keep_last(10)
auto qos = rclcpp::QoS(rclcpp::KeepLast(10)).reliable().transient_local();

---

LifecycleNode 상태/전환 테이블

ID	Primary State	진입 콜백	설명
1	UNCONFIGURED	—	초기 상태, 리소스 없음
2	INACTIVE	on_configure()	리소스 할당, 미처리
3	ACTIVE	on_activate()	정상 동작, 퍼블리시
4	FINALIZED	on_shutdown()	종료 직전

반환값: SUCCESS (정상) · FAILURE (이전 상태 복귀) · ERROR → on_error(), 실패 시 FINALIZED

CallbackReturn on_configure(const rclcpp_lifecycle::State &) override {
  pub_ = this->create_publisher<Msg>("out", 10); // LifecyclePublisher 생성
  return CallbackReturn::SUCCESS;
}
CallbackReturn on_activate(const rclcpp_lifecycle::State &) override {
  pub_->on_activate();   // 반드시 명시적 활성화
  return CallbackReturn::SUCCESS;
}
CallbackReturn on_deactivate(const rclcpp_lifecycle::State &) override {
  pub_->on_deactivate();
  return CallbackReturn::SUCCESS;
}

CLI:

ros2 lifecycle set /my_node configure    # UNCONFIGURED → INACTIVE
ros2 lifecycle set /my_node activate     # INACTIVE → ACTIVE
ros2 lifecycle set /my_node deactivate   # ACTIVE → INACTIVE
ros2 lifecycle set /my_node cleanup      # INACTIVE → UNCONFIGURED
ros2 lifecycle set /my_node shutdown     # any → FINALIZED

---

LifecycleManager YAML (Nav2)

lifecycle_manager_navigation:
  ros__parameters:
    autostart: true          # 시작 즉시 configure → activate
    node_names:              # 순서대로 configure/activate, 역순으로 deactivate/cleanup
      - map_server
      - amcl
      - controller_server
      - planner_server
      - bt_navigator
    bond_timeout: 4.0        # 초; 하트비트 소실 시 전체 재시작

---

---

QoS 심층 가이드 (이슈 다발 영역)

ROS2 현장에서 가장 많은 무음 실패(silent failure)를 유발하는 영역. 토픽 연결은 됐는데 메시지가 안 오면 QoS 불일치 먼저 의심.

빠른 진단

# Publisher/Subscriber QoS 전체 확인
ros2 topic info /topic_name --verbose

# 예시 출력 해석:
# Publisher: Reliability=BEST_EFFORT, Durability=VOLATILE
# Subscriber: Reliability=RELIABLE   ← 불일치 → 메시지 수신 안 됨 (오류 없음!)

---

호환성 매트릭스 (완전판)

구독자 정책이 발행자보다 엄격하면 무음 실패 (연결은 되지만 메시지 0개).

Reliability

Publisher	Subscriber	결과
RELIABLE	RELIABLE	✅
RELIABLE	BEST_EFFORT	✅ (구독자가 덜 엄격)
BEST_EFFORT	RELIABLE	❌ 무음 실패
BEST_EFFORT	BEST_EFFORT	✅

Durability

Publisher	Subscriber	결과
TRANSIENT_LOCAL	TRANSIENT_LOCAL	✅ (늦게 참여해도 수신)
TRANSIENT_LOCAL	VOLATILE	✅
VOLATILE	VOLATILE	✅
VOLATILE	TRANSIENT_LOCAL	❌ 무음 실패

---

자주 발생하는 QoS 이슈 패턴

이슈 1: 카메라/센서 토픽 수신 안 됨

# 원인: 드라이버는 BEST_EFFORT, 내 구독자는 기본값 RELIABLE
# 진단
ros2 topic info /camera/image_raw --verbose | grep -A3 "Publisher\|Subscriber"

# 해결: sensor_data QoS 사용
from rclpy.qos import qos_profile_sensor_data
self.create_subscription(Image, '/camera/image_raw', cb, qos_profile_sensor_data)

이슈 2: /robot_description 늦게 시작한 노드가 못 받음

# 원인: robot_state_publisher는 TRANSIENT_LOCAL로 발행
# 내 구독자가 VOLATILE이면 이미 보낸 메시지 못 받음
# 해결: TRANSIENT_LOCAL QoS로 구독

from rclpy.qos import QoSProfile, QoSDurabilityPolicy, QoSReliabilityPolicy
latched = QoSProfile(
    depth=1,
    durability=QoSDurabilityPolicy.TRANSIENT_LOCAL,
    reliability=QoSReliabilityPolicy.RELIABLE,
)
self.create_subscription(String, '/robot_description', cb, latched)

이슈 3: Nav2 / 지도 서버 메시지 못 받음

# /map 토픽은 TRANSIENT_LOCAL RELIABLE
ros2 topic info /map --verbose
# 구독 시 반드시 맞춰야 함

map_qos = QoSProfile(
    depth=1,
    durability=QoSDurabilityPolicy.TRANSIENT_LOCAL,
    reliability=QoSReliabilityPolicy.RELIABLE,
    history=QoSHistoryPolicy.KEEP_LAST,
)

이슈 4: rosbag 재생 시 메시지 수신 안 됨

# rosbag는 녹화 당시 QoS로 재생
# 확인: ros2 bag info my_bag/ → offered QoS profiles 항목
# 해결: --qos-profile-overrides-path 옵션
ros2 bag play my_bag/ --qos-profile-overrides-path qos_override.yaml

# qos_override.yaml
/camera/image_raw:
  reliability: best_effort
  durability: volatile
  history: keep_last
  depth: 10

이슈 5: 같은 머신인데 DDS 도메인 불일치

# 기본 ROS_DOMAIN_ID=0, 다른 터미널에서 다른 값 설정 시 안 보임
echo $ROS_DOMAIN_ID
export ROS_DOMAIN_ID=0  # 동일하게 맞추기
# 또는 로컬 통신만 허용
export ROS_LOCALHOST_ONLY=1

---

고급 QoS 정책

Deadline (주기 보장)

Publisher가 지정 주기 내 게시 안 하거나 Subscriber가 지정 주기 내 못 받으면 이벤트 발생.

from rclpy.qos import QoSProfile
from rclpy.qos_event import PublisherEventCallbacks, SubscriptionEventCallbacks
import rclpy.duration

deadline_qos = QoSProfile(depth=10)
deadline_qos.deadline = rclpy.duration.Duration(seconds=0, nanoseconds=100_000_000)  # 100ms

callbacks = SubscriptionEventCallbacks(
    deadline=lambda event: node.get_logger().warn('Deadline missed!')
)
sub = node.create_subscription(Twist, '/cmd_vel', cb, deadline_qos,
                                event_callbacks=callbacks)

Liveliness (생존 확인)

from rclpy.qos import QoSLivelinessPolicy
import rclpy.duration

qos = QoSProfile(depth=10)
qos.liveliness = QoSLivelinessPolicy.MANUAL_BY_TOPIC
qos.liveliness_lease_duration = rclpy.duration.Duration(seconds=1)

# Publisher가 1초 내 assert_liveliness() 안 하면 Subscriber에 이벤트
pub = node.create_publisher(String, 'topic', qos)
pub.assert_liveliness()  # 주기적으로 호출

Lifespan (메시지 유효기간)

# 게시 후 500ms 지나면 구독자에게 전달 안 됨
qos = QoSProfile(depth=10)
qos.lifespan = rclpy.duration.Duration(seconds=0, nanoseconds=500_000_000)

---

QoS 이슈 디버깅 체크리스트

□ ros2 topic info /topic --verbose 로 양쪽 QoS 확인
□ Reliability 불일치? → 구독자를 BEST_EFFORT로 낮추거나 발행자를 RELIABLE로 높이기
□ Durability 불일치? → /robot_description, /map 등은 TRANSIENT_LOCAL 필요
□ ROS_DOMAIN_ID 동일한지 확인
□ ROS_LOCALHOST_ONLY 설정 확인
□ rosbag 재생 시 QoS override 필요 여부 확인
□ 같은 타입인데 연결 안 되면 ros2 topic find <type> 로 이름 오타 확인
□ FastRTPS vs CycloneDDS 혼용 시 DDS 레벨 비호환 가능 → RMW_IMPLEMENTATION 통일

주요 토픽별 기본 QoS 정리

토픽	Reliability	Durability	비고
/scan, /imu, /camera/*	BEST_EFFORT	VOLATILE	sensor_data
/map	RELIABLE	TRANSIENT_LOCAL	늦게 구독해도 수신
/robot_description	RELIABLE	TRANSIENT_LOCAL	늦게 구독해도 수신
/tf	RELIABLE	VOLATILE	KEEP_LAST 100
/tf_static	RELIABLE	TRANSIENT_LOCAL	늦게 구독해도 수신
/cmd_vel	RELIABLE	VOLATILE	기본값
/odom	RELIABLE	VOLATILE	기본값
ROS2 파라미터	RELIABLE	VOLATILE	parameters QoS
ROS2 서비스	RELIABLE	VOLATILE	services_default

---

RViz2 설정 가이드

RViz2는 ROS2의 표준 3D 시각화 도구. 설정 파일(.rviz)로 레이아웃 저장/재사용.

기본 실행

# 기본 실행
rviz2

# 설정 파일 로드
rviz2 -d /path/to/config.rviz
rviz2 -d $(ros2 pkg prefix my_pkg)/share/my_pkg/rviz/default.rviz

# 특정 Fixed Frame으로 실행
rviz2 -d config.rviz --ros-args -p fixed_frame:=map

# 헤드리스 (화면 없는 서버)
export DISPLAY=:99 && Xvfb :99 -screen 0 1024x768x24 &
rviz2 -d config.rviz

---

Launch 파일에서 RViz2 실행

from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument
from launch.conditions import IfCondition
from launch.substitutions import LaunchConfiguration, PathJoinSubstitution
from launch_ros.actions import Node
from launch_ros.substitutions import FindPackageShare

def generate_launch_description():
    use_rviz = LaunchConfiguration('use_rviz', default='true')
    rviz_config = PathJoinSubstitution([
        FindPackageShare('my_pkg'), 'rviz', 'default.rviz'
    ])

    return LaunchDescription([
        DeclareLaunchArgument('use_rviz', default_value='true'),

        Node(
            package='rviz2',
            executable='rviz2',
            name='rviz2',
            arguments=['-d', rviz_config],
            parameters=[{'use_sim_time': True}],
            output='screen',
            condition=IfCondition(use_rviz),
        ),
    ])

---

.rviz 설정 파일 구조

# config/default.rviz
Panels:
  - Class: rviz_common/Displays
    Name: Displays
  - Class: rviz_common/Views
    Name: Views

Visualization Manager:
  Class: ""
  Displays:
    - Alpha: 0.5
      Class: rviz_default_plugins/Grid
      Name: Grid
      Cell Size: 1
      Color: 160; 160; 164
      Enabled: true
      Plane: XY

    - Class: rviz_default_plugins/RobotModel
      Name: RobotModel
      Enabled: true
      Description Topic:
        Depth: 5
        Durability Policy: Transient Local   # /robot_description 필수
        Value: /robot_description
      Visual Enabled: true
      Collision Enabled: false

    - Class: rviz_default_plugins/LaserScan
      Name: LaserScan
      Topic:
        Value: /scan
        Depth: 5
        Reliability Policy: Best Effort      # sensor QoS
        Durability Policy: Volatile
      Style: Points
      Size (m): 0.03
      Color Transformer: AxisColor
      Enabled: true

    - Class: rviz_default_plugins/PointCloud2
      Name: PointCloud2
      Topic:
        Value: /points
        Depth: 5
        Reliability Policy: Best Effort
        Durability Policy: Volatile
      Style: Points
      Size (m): 0.01
      Color Transformer: RGB8
      Enabled: true

    - Class: rviz_default_plugins/Image
      Name: Camera
      Topic:
        Value: /camera/image_raw
        Depth: 5
        Reliability Policy: Best Effort
        Durability Policy: Volatile
      Enabled: true

    - Class: rviz_default_plugins/Map
      Name: Map
      Topic:
        Value: /map
        Depth: 1
        Reliability Policy: Reliable
        Durability Policy: Transient Local   # /map은 TRANSIENT_LOCAL
      Color Scheme: map
      Enabled: true

    - Class: rviz_default_plugins/Path
      Name: Global Path
      Topic:
        Value: /plan
        Depth: 5
      Color: 0; 128; 0
      Enabled: true

    - Class: rviz_default_plugins/Odometry
      Name: Odometry
      Topic:
        Value: /odom
        Depth: 10
      Shape: Arrow
      Enabled: true

    - Class: rviz_default_plugins/Axes
      Name: Axes
      Enabled: true
      Length: 0.5

    - Class: rviz_default_plugins/TF
      Name: TF
      Enabled: false       # 노드 많으면 성능 저하 → 필요 시만 켜기
      Marker Scale: 0.5
      Show Names: true
      Show Axes: true

    - Class: rviz_default_plugins/Marker
      Name: Markers
      Topic:
        Value: /visualization_marker
        Depth: 100
      Enabled: true

    - Class: rviz_default_plugins/MarkerArray
      Name: MarkerArray
      Topic:
        Value: /visualization_marker_array
      Enabled: true

    - Class: rviz_default_plugins/PoseArray
      Name: Particles
      Topic:
        Value: /particlecloud
      Color: 255; 25; 0
      Enabled: true

  Fixed Frame: map           # 기준 프레임 (map / odom / base_link)
  Background Color: 48; 48; 48

  Tools:
    - Class: rviz_default_plugins/Interact
    - Class: rviz_default_plugins/MoveCamera
    - Class: rviz_default_plugins/Select
    - Class: rviz_default_plugins/FocusCamera
    - Class: rviz_default_plugins/Measure
    - Class: rviz_default_plugins/SetInitialPose  # 2D Pose Estimate
      Topic:
        Value: /initialpose
    - Class: rviz_default_plugins/SetGoal         # 2D Nav Goal
      Topic:
        Value: /goal_pose

  Views:
    Current:
      Class: rviz_default_plugins/Orbit
      Target Frame: base_link
      Distance: 5.0
      Pitch: 0.785398   # 45도
      Yaw: 0.0

---

Python에서 Marker 발행 (RViz2 시각화)

from visualization_msgs.msg import Marker, MarkerArray
from geometry_msgs.msg import Point
import rclpy
from rclpy.node import Node

class MarkerNode(Node):
    def __init__(self):
        super().__init__('marker_node')
        self.pub = self.create_publisher(MarkerArray, '/visualization_marker_array', 10)
        self.create_timer(1.0, self.publish_markers)

    def publish_markers(self):
        arr = MarkerArray()

        # 구체 마커
        m = Marker()
        m.header.frame_id = 'map'
        m.header.stamp = self.get_clock().now().to_msg()
        m.ns = 'obstacles'; m.id = 0
        m.type = Marker.SPHERE
        m.action = Marker.ADD
        m.pose.position.x = 1.0; m.pose.position.y = 0.5
        m.pose.orientation.w = 1.0
        m.scale.x = m.scale.y = m.scale.z = 0.3
        m.color.r = 1.0; m.color.a = 1.0   # 빨강, 불투명
        m.lifetime.sec = 0   # 0 = 영구
        arr.markers.append(m)

        # 선(LINE_STRIP) 마커
        line = Marker()
        line.header.frame_id = 'map'
        line.header.stamp = self.get_clock().now().to_msg()
        line.ns = 'path'; line.id = 1
        line.type = Marker.LINE_STRIP
        line.action = Marker.ADD
        line.scale.x = 0.05   # 선 굵기
        line.color.g = 1.0; line.color.a = 1.0
        for x, y in [(0,0),(1,0),(1,1),(0,1)]:
            p = Point(); p.x = float(x); p.y = float(y)
            line.points.append(p)
        arr.markers.append(line)

        # 텍스트 마커
        txt = Marker()
        txt.header.frame_id = 'map'
        txt.header.stamp = self.get_clock().now().to_msg()
        txt.ns = 'labels'; txt.id = 2
        txt.type = Marker.TEXT_VIEW_FACING
        txt.action = Marker.ADD
        txt.pose.position.x = 0.5; txt.pose.position.z = 0.5
        txt.pose.orientation.w = 1.0
        txt.scale.z = 0.3    # 텍스트 높이
        txt.color.r = txt.color.g = txt.color.b = txt.color.a = 1.0
        txt.text = 'Hello RViz2'
        arr.markers.append(txt)

        # 마커 삭제
        del_m = Marker(); del_m.action = Marker.DELETEALL  # 전체 삭제
        # del_m.action = Marker.DELETE; del_m.ns='obs'; del_m.id=0  # 개별 삭제

        self.pub.publish(arr)

Marker 타입 빠른 참조

타입	상수	용도
ARROW	0	방향 벡터
CUBE	1	박스
SPHERE	2	구체
CYLINDER	3	원통
LINE_STRIP	4	연속선
LINE_LIST	5	선 쌍
CUBE_LIST	6	박스 배열
SPHERE_LIST	7	구체 배열
POINTS	8	점 구름
TEXT_VIEW_FACING	9	화면 향 텍스트
MESH_RESOURCE	10	3D 메시 파일

---

패키지에 RViz 설정 포함시키기

my_pkg/
├── rviz/
│   └── default.rviz
├── setup.py        ← data_files에 추가 필요
└── package.xml

# setup.py data_files에 추가
data_files=[
    ('share/' + package_name + '/rviz', ['rviz/default.rviz']),
],

---

RViz2 트러블슈팅

증상	원인	해결
Fixed Frame 에러	TF에 해당 프레임 없음	ros2 run tf2_tools view_frames
RobotModel 안 보임	/robot_description QoS 불일치	Durability: Transient Local 설정
LaserScan 안 보임	QoS 불일치 (BEST_EFFORT 필요)	Reliability: Best Effort 설정
Map 안 보임	/map QoS 불일치	Durability: Transient Local 설정
TF 화살표 느림	TF display 활성화 시 부하	TF display 비활성 또는 필터링
RViz2 멈춤	대용량 PointCloud2	depth↓, Decimate 플러그인 사용

# TF 트리 확인 (RViz Fixed Frame 에러 진단)
ros2 run tf2_tools view_frames   # frames.pdf 생성
ros2 run tf2_ros tf2_echo map base_link   # 실시간 TF 확인

---

개발 vs 배포 Launch 워크플로우

핵심 원칙: 개발 중에는 src/ 직접 참조, 배포 시에는 install/ 사용.

개발 환경 (src 폴더 직접 사용)

# --symlink-install: install/에 심볼릭 링크 생성 → src/ 파일 수정이 즉시 반영
cd ~/ros2_ws
colcon build --symlink-install

# 소싱 (install/의 심링크가 src/를 가리킴)
source install/setup.bash

# Python 노드: src/ 수정 → 재빌드 없이 즉시 반영
# C++ 노드: src/ 수정 → 재빌드 필요 (심링크는 실행파일에 미적용)

개발 환경 파일 참조 구조:
ros2_ws/
├── src/my_pkg/
│   ├── my_pkg/my_node.py     ← 실제 파일 (여기서 편집)
│   ├── launch/my.launch.py   ← 실제 파일
│   └── config/params.yaml    ← 실제 파일
└── install/my_pkg/
    ├── lib/my_pkg/my_node    → symlink → src/my_pkg/my_pkg/my_node.py
    ├── share/my_pkg/launch/  → symlink → src/my_pkg/launch/
    └── share/my_pkg/config/  → symlink → src/my_pkg/config/

개발 시 Launch 파일에서 src 경로 참조 금지:

# ❌ 잘못된 방법 (하드코딩, 배포 시 깨짐)
config_path = '/home/user/ros2_ws/src/my_pkg/config/params.yaml'

# ✅ 올바른 방법 (install/ 기준, 심링크로 src/ 연결됨)
from ament_index_python.packages import get_package_share_directory
import os
config_path = os.path.join(get_package_share_directory('my_pkg'), 'config', 'params.yaml')

---

배포 환경 (install 폴더 사용)

# 배포용 빌드: --symlink-install 없이 (실제 파일 복사)
colcon build --cmake-args -DCMAKE_BUILD_TYPE=Release

# install/ 폴더만 배포 대상 (src/ 불필요)
source install/setup.bash
ros2 launch my_pkg my.launch.py

배포 환경 파일 구조:
install/
├── setup.bash              ← 반드시 source
├── my_pkg/
│   ├── lib/my_pkg/my_node  ← 실제 실행파일 복사본
│   └── share/my_pkg/
│       ├── launch/         ← 실제 launch 파일 복사본
│       ├── config/         ← 실제 설정 파일 복사본
│       └── rviz/           ← 실제 rviz 설정 복사본
└── local_setup.bash        ← 이 패키지만 소싱

---

setup.py - 배포 파일 등록 (반드시 포함)

# setup.py: install/에 복사할 파일 목록 - 누락 시 배포 환경에서 FileNotFoundError
from setuptools import setup, find_packages
import os
from glob import glob

package_name = 'my_pkg'

setup(
    name=package_name,
    version='0.0.1',
    packages=find_packages(exclude=['test']),
    data_files=[
        # ament index 등록 (필수)
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        # package.xml (필수)
        ('share/' + package_name, ['package.xml']),
        # Launch 파일 (반드시 등록)
        ('share/' + package_name + '/launch',
            glob('launch/*.launch.py')),
        # 설정 파일
        ('share/' + package_name + '/config',
            glob('config/*.yaml')),
        # RViz 설정
        ('share/' + package_name + '/rviz',
            glob('rviz/*.rviz')),
        # URDF/Xacro
        ('share/' + package_name + '/urdf',
            glob('urdf/*.urdf') + glob('urdf/*.xacro')),
        # Maps
        ('share/' + package_name + '/maps',
            glob('maps/*')),
        # 중첩 폴더 (glob 재귀 사용)
        *[('share/' + package_name + '/' + os.path.dirname(f),
           [f]) for f in glob('config/**/*.yaml', recursive=True)],
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    entry_points={
        'console_scripts': [
            'my_node = my_pkg.my_node:main',
        ],
    },
)

---

CMakeLists.txt - C++ 패키지 배포 파일 등록

# launch, config, rviz, urdf 폴더 install/에 복사
install(
  DIRECTORY launch config rviz urdf maps
  DESTINATION share/${PROJECT_NAME}
)

# 실행파일
install(
  TARGETS my_node my_component
  ARCHIVE DESTINATION lib
  LIBRARY DESTINATION lib
  RUNTIME DESTINATION lib/${PROJECT_NAME}
)

---

개발↔배포 전환 체크리스트

개발 시:
□ colcon build --symlink-install 사용
□ source install/setup.bash
□ Python 수정: 재빌드 불필요 (심링크)
□ C++ 수정: colcon build --packages-select my_pkg 필요
□ Launch/config/yaml 수정: 재빌드 불필요 (심링크)
□ 새 파일 추가: 반드시 재빌드 (setup.py/CMake에 등록 후)

배포 시:
□ colcon build --cmake-args -DCMAKE_BUILD_TYPE=Release (--symlink-install 제거)
□ install/ 폴더만 타겟 서버에 복사 (src/ 불필요)
□ 타겟 서버에서 source install/setup.bash
□ setup.py data_files에 모든 리소스 등록 확인
□ get_package_share_directory() 사용 확인 (하드코딩 경로 없음)
□ rosdep install --from-paths src --ignore-src -r -y (의존성 설치)

---

자주 발생하는 개발/배포 이슈

증상	원인	해결
배포 환경에서 FileNotFoundError (yaml/launch)	setup.py data_files 미등록	data_files에 해당 폴더 추가 후 재빌드
Python 수정이 반영 안 됨	--symlink-install 없이 빌드	colcon build --symlink-install 재실행
C++ 수정이 반영 안 됨	재빌드 안 함	colcon build --packages-select pkg
src/ 경로 하드코딩 오류	절대경로 사용	get_package_share_directory() 사용
install/ 없는 파일 참조	새 파일 추가 후 미등록	setup.py 등록 + 재빌드
타겟 서버에서 패키지 못 찾음	ROS2 미소싱	source /opt/ros/humble/setup.bash + source install/setup.bash

---

환경별 소싱 순서

# 개발 환경 (전체)
source /opt/ros/humble/setup.bash           # 1. ROS2 기본
source ~/ros2_ws/install/setup.bash         # 2. 워크스페이스

# 다중 워크스페이스 (의존성 순서)
source /opt/ros/humble/setup.bash
source ~/base_ws/install/setup.bash         # 기반 패키지
source ~/my_ws/install/setup.bash           # 개발 패키지 (위에 overlay)

# 배포 환경
source /opt/ros/humble/setup.bash
source /opt/my_robot/install/setup.bash     # 설치된 경로

---

Launch 파일 생성 가이드

상황별 launch 파일을 처음부터 만드는 템플릿 모음.

패키지에 launch 폴더 추가

mkdir -p ~/ros2_ws/src/my_pkg/launch
touch ~/ros2_ws/src/my_pkg/launch/my_robot.launch.py

setup.py data_files에 반드시 등록:

('share/' + package_name + '/launch', glob('launch/*.launch.py')),

---

템플릿 1: 단일 노드 실행 (가장 기본)

# launch/my_node.launch.py
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument
from launch.substitutions import LaunchConfiguration
from launch_ros.actions import Node

def generate_launch_description():
    return LaunchDescription([
        DeclareLaunchArgument('use_sim_time', default_value='false'),
        DeclareLaunchArgument('log_level',    default_value='info'),

        Node(
            package='my_pkg',
            executable='my_node',
            name='my_node',
            output='screen',
            emulate_tty=True,
            parameters=[{
                'use_sim_time': LaunchConfiguration('use_sim_time'),
            }],
            arguments=['--ros-args', '--log-level',
                       LaunchConfiguration('log_level')],
        ),
    ])

ros2 launch my_pkg my_node.launch.py use_sim_time:=true log_level:=debug

---

템플릿 2: 로봇 풀스택 (robot_state_publisher + 노드들)

# launch/robot.launch.py
import os
from ament_index_python.packages import get_package_share_directory
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, IncludeLaunchDescription
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch.substitutions import LaunchConfiguration, PathJoinSubstitution
from launch_ros.actions import Node
from launch_ros.substitutions import FindPackageShare
import xacro

def generate_launch_description():
    pkg_share = get_package_share_directory('my_pkg')
    use_sim_time = LaunchConfiguration('use_sim_time', default='false')

    # URDF/Xacro 처리
    xacro_file = os.path.join(pkg_share, 'urdf', 'robot.urdf.xacro')
    robot_description = xacro.process_file(xacro_file).toxml()

    return LaunchDescription([
        DeclareLaunchArgument('use_sim_time', default_value='false'),

        # 1. robot_state_publisher
        Node(
            package='robot_state_publisher',
            executable='robot_state_publisher',
            name='robot_state_publisher',
            output='screen',
            parameters=[{
                'robot_description': robot_description,
                'use_sim_time': use_sim_time,
            }],
        ),

        # 2. joint_state_publisher (실제 로봇 없을 때 GUI 슬라이더)
        Node(
            package='joint_state_publisher_gui',
            executable='joint_state_publisher_gui',
            name='joint_state_publisher_gui',
            output='screen',
        ),

        # 3. 다른 launch 파일 포함
        IncludeLaunchDescription(
            PythonLaunchDescriptionSource([
                PathJoinSubstitution([
                    FindPackageShare('my_pkg'), 'launch', 'sensors.launch.py'
                ])
            ]),
            launch_arguments={'use_sim_time': use_sim_time}.items(),
        ),

        # 4. RViz2
        Node(
            package='rviz2',
            executable='rviz2',
            name='rviz2',
            arguments=['-d', os.path.join(pkg_share, 'rviz', 'robot.rviz')],
            parameters=[{'use_sim_time': use_sim_time}],
            output='screen',
        ),
    ])

---

템플릿 3: Gazebo 시뮬레이션

# launch/gazebo_sim.launch.py
import os
from ament_index_python.packages import get_package_share_directory
from launch import LaunchDescription
from launch.actions import (DeclareLaunchArgument, IncludeLaunchDescription,