bag 기록과 재생 — 현장 데이터 재현과 회귀 시험

1. 현장 실패는 한 번뿐이다

Robot은 조명, 바닥 마찰, 사람의 위치, Battery 전압, Sensor 지연과 Network 손실이 겹친 물리 환경에서 동작합니다. 실패 장면을 정확히 다시 만드는 일은 매우 어렵습니다.

rosbag2는 ROS 2 통신 Data와 기록 시각을 Storage에 저장하고 다시 발행합니다.

현장 Robot
  │ Sensor·State·Command·TF
  ↓
rosbag2 record ──> Bag Storage
                      │
          ┌───────────┼───────────┐
          ↓           ↓           ↓
       재생·RViz   오프라인 분석   회귀 시험

주요 활용:

• 한 번 발생한 충돌·정지·Localization 실패 재현
• 같은 입력으로 알고리즘 수정 전후 비교
• Camera·LiDAR·Odometry Dataset 수집
• Parameter 조합별 결과 비교
• 현장과 개발팀 사이의 문제 Data 전달
• Simulation에서 재생 기반 통합 시험

Bag은 물리 세계 전체를 저장하지 않습니다. Message로 발행되지 않은 바닥 상태, 실제 Motor 전류와 외부 사건은 남지 않으므로 필요한 관측 항목을 미리 설계해야 합니다.

2. Bag에 저장되는 것과 저장되지 않는 것

일반적인 Topic 기록에는 다음이 포함됩니다.

• Topic 이름과 Type
• 직렬화된 Message Data
• rosbag2가 Message를 받은 기록 Timestamp
• Topic Metadata와 Offered QoS 정보
• Storage와 기록 Session Metadata

자동으로 보장되지 않는 것:

• 모든 Node의 Code Version
• Parameter 현재값
• Environment Variable과 RMW 설정
• Sensor Hardware 설정과 Firmware
• Operator 행동의 이유
• Message Header의 측정 Timestamp 정확성
• 기록 밖의 실제 Robot 물리 상태

재현 가능한 Dataset은 Bag뿐 아니라 다음 Manifest를 함께 남깁니다.

run_id: warehouse_2026_09_02_001
robot_id: robot_03
git_commit: a1b2c3d
ros_distro: kilted
rmw: rmw_cyclonedds_cpp
map_version: warehouse_v7
parameter_profile: production_v4
operator: team_a
purpose: aisle_4_obstacle_failure

3. 기록 전 확인할 것

ros2 topic list -t
ros2 topic info /scan --verbose
ros2 topic hz /scan
ros2 topic bw /scan
df -h

기록 계획 질문:

1. 어떤 사건을 재현하려는가?
2. 입력뿐 아니라 알고리즘 출력도 필요한가?
3. TF와 Clock 없이 공간·시간을 복원할 수 있는가?
4. 초당 Data 양과 예상 기록 시간은 얼마인가?
5. 디스크 지속 쓰기 속도와 남은 공간은 충분한가?
6. CPU 압축 부하가 제어 Loop에 영향을 주지 않는가?
7. 개인정보와 민감한 영상·음성이 포함되는가?

4. 기록 용량 계산

Payload 대역폭 ≈ 평균 Message 크기 × 발행률
예상 용량 ≈ 전체 대역폭 × 기록 시간

640×480 RGB8 Image 한 장의 순수 Pixel 크기:

640 × 480 × 3 = 921,600 byte
30Hz → 27,648,000 byte/s ≈ 27.65MB/s
10분 → 약 16.6GB

실제 저장량은 Header, 직렬화, Storage Index, Compression과 Image 내용에 따라 달라집니다. ros2 topic bw와 짧은 시험 기록을 사용해 실측합니다.

5. 가장 기본적인 기록

특정 Topic을 선택하여 기록합니다.

ros2 bag record \
  -o run_001 \
  /scan /odom /tf /tf_static /cmd_vel /diagnostics

기록을 정상 종료하려면 Ctrl+C를 한 번 누르고 Metadata와 Storage가 마무리될 시간을 줍니다. 전원을 강제로 끄면 Index나 Metadata가 완성되지 않을 수 있습니다.

ls -lah run_001
ros2 bag info run_001

배포판마다 단축 Option이 달라질 수 있으므로 실제 환경에서 확인합니다.

ros2 bag record --help
ros2 bag play --help
ros2 bag info --help

6. 모든 Topic 기록과 제외 Regex

# 모든 Topic
ros2 bag record --all -o full_run

# Camera와 Debug 영상 제외
ros2 bag record --all -o no_camera \
  --exclude '/camera/.+|/debug_image.+'

# Regex에 맞는 Topic만
ros2 bag record --regex '/(scan|odom|tf|tf_static|diagnostics)' \
  -o navigation_inputs

Option 이름과 Regex Matching 범위는 배포판 도움말을 확인합니다. --all은 짧은 진단에서 편하지만 Camera, Point Cloud와 숨은 System Topic이 많은 Robot에서는 디스크·CPU 부하가 급증할 수 있습니다.

7. 무엇을 함께 기록해야 하는가

Navigation 문제를 분석하는 추천 묶음:

ros2 bag record -o nav_failure \
  /scan /odom /tf /tf_static \
  /map /amcl_pose \
  /cmd_vel_nav /cmd_vel_safe \
  /plan /local_plan \
  /diagnostics /rosout

Manipulator 문제 묶음:

ros2 bag record -o arm_failure \
  /joint_states \
  /joint_trajectory_controller/joint_trajectory \
  /joint_trajectory_controller/controller_state \
  /tf /tf_static /diagnostics

Sensor Fusion 문제 묶음:

ros2 bag record -o fusion_failure \
  /imu/data /wheel/odometry /gps/fix \
  /odometry/filtered /tf /tf_static /diagnostics

입력만 기록하면 알고리즘이 당시 어떤 출력을 냈는지 비교하기 어렵습니다. 원인 입력, 중간 상태, 최종 Command와 Health를 함께 기록합니다.

8. Hidden Topic·Service·Action 기록

최근 rosbag2는 배포판과 Option에 따라 Hidden Topic, Service Event와 Action 관련 Data를 기록·재생하는 기능을 제공합니다.

ros2 bag record --help | grep -E 'hidden|service|action'

Service와 Action은 Request·Response·Goal·Feedback·Result의 의미가 있고 Introspection 설정과 배포판 지원이 관련됩니다. --all만으로 모든 Service 호출 의미가 완전히 보존된다고 가정하지 마십시오.

필요한 경우:

1. 해당 Service·Action Introspection 지원을 확인합니다.
2. Event Topic이 실제 Graph에 나타나는지 확인합니다.
3. 짧은 호출을 기록한 뒤 bag info에서 Message 수를 검증합니다.
4. 재생이 실제 Server에 부작용을 만들지 격리 환경에서 시험합니다.

9. MCAP과 sqlite3 Storage

rosbag2는 Storage Plugin 구조를 사용합니다. 흔한 형식은 MCAP과 sqlite3입니다.

ros2 bag record --storage mcap -o mcap_run /scan /odom
ros2 bag record --storage sqlite3 -o db3_run /scan /odom
ros2 bag info mcap_run

배포판의 기본 Storage를 외우지 말고 bag info의 Storage ID와 설치된 Plugin을 확인합니다.

MCAP 장점:

• Chunk와 Index 기반의 효율적인 접근
• Compression, CRC와 여러 도구 생태계
• ROS를 실행하지 않고도 지원 Viewer에서 확인 가능

sqlite3 장점:

• 널리 사용된 기존 ROS 2 Bag 형식
• SQL 기반 Storage의 익숙한 도구

프로젝트의 ROS 배포판, Viewer, 장기 보관과 처리 Pipeline 호환성을 기준으로 선택합니다.

10. 긴 기록은 분할한다

# 최대 파일 크기로 분할: 값 단위와 Option은 --help 확인
ros2 bag record --all -o long_run \
  --max-bag-size 1073741824

# 최대 지속시간으로 분할
ros2 bag record --all -o long_run \
  --max-bag-duration 60

분할 장점:

• 한 파일이 지나치게 커지는 것을 방지
• 전송·보관·삭제가 쉬움
• 일부 Chunk 손상 영향 축소
• 필요한 시간 구간만 처리 가능

분할 파일은 하나의 Bag Directory와 Metadata로 관리될 수 있으므로 임의로 일부 파일만 이동하기 전에 bag info와 Storage 구조를 확인합니다.

11. Compression 전략

일반 rosbag2 Compression Option 예시입니다.

ros2 bag record -o compressed_run /scan /odom \
  --compression-mode file \
  --compression-format zstd

Compression은 디스크 용량과 쓰기 대역폭을 줄일 수 있지만 CPU와 지연 비용이 있습니다. 이미 CPU가 포화된 Robot에서는 기록 Drop이나 제어 지연이 늘 수 있습니다.

MCAP은 Storage Config와 Preset Profile로 Chunk Compression을 구성할 수 있습니다.

ros2 bag record --storage mcap --all -o run_mcap \
  --storage-preset-profile zstd_fast

지원 Profile은 설치된 Version 문서를 확인합니다. 고속 기록용 Profile이 Index나 CRC를 줄일 수 있으므로 장기 보관 전 후처리가 필요한지 검토합니다.

12. MCAP Writer 설정 예제

# mcap_writer_options.yaml
noChunking: false
noMessageIndex: false
noSummary: false
chunkSize: 786432
compression: Zstd
compressionLevel: Fast
forceCompression: false

ros2 bag record --storage mcap --all -o configured_run \
  --storage-config-file mcap_writer_options.yaml

CRC를 끄면 처리량은 좋아질 수 있지만 손상 검출 능력이 줄어듭니다. 현장 고속 수집과 장기 Archive의 요구를 분리하십시오.

13. Cache와 기록 Drop

Recorder는 Memory Cache로 순간적인 디스크 지연을 흡수할 수 있습니다.

ros2 bag record --all -o cached_run \
  --max-cache-size 268435456

Cache를 크게 하면 Burst에는 도움이 되지만 다음 한계가 있습니다.

• 디스크가 지속적으로 느리면 결국 Cache가 참
• 전원 손실 시 아직 쓰지 않은 Data가 사라질 수 있음
• Memory Pressure로 다른 Node가 영향받을 수 있음

근본 대책은 Topic·해상도·주기 축소, 빠른 Storage, Compression Profiling과 Recorder 전용 Computer 분리입니다.

14. 기록 상태를 운영 중 확인

ros2 node list | grep rosbag
ros2 topic hz /scan
df -h
iostat -xz 1

확인할 지표:

• Recorder Drop 경고
• Disk 사용률과 남은 용량
• CPU·Memory 사용량
• Sensor Hz 변화
• 제어 Loop Deadline 변화
• Bag Directory 크기 증가

기록 자체가 Robot 성능을 바꾸면 수집한 Data도 원래 사건을 대표하지 못합니다. 기록 On/Off 비교 시험을 수행합니다.

15. ros2 bag info 읽기

ros2 bag info run_001

주요 항목:

• Storage ID
• Duration과 시작·종료 시각
• 전체 Message 수
• Topic 이름과 Type
• Topic별 Message 수
• Serialization Format
• Offered QoS Profiles
• 분할 파일과 크기

검증 질문:

예상 10Hz Scan을 60초 기록 → 약 600개인가?
TF와 TF Static이 존재하는가?
Command 입력과 안전 출력이 모두 있는가?
Bag Duration이 현장 시험 시간과 맞는가?

기록을 마친 즉시 info를 확인하면 현장을 떠나기 전에 재수집할 수 있습니다.

16. 기본 재생

ros2 bag play run_001

다른 Terminal에서 확인합니다.

ros2 topic list -t
ros2 topic hz /scan
ros2 topic echo /odom --once
ros2 topic info /scan --verbose

재생기는 기록 Timestamp 간격을 기준으로 Message를 발행합니다. Message 내부 header.stamp는 일반적으로 원래 기록 Data를 유지합니다.

17. 속도·반복·시작 위치·일시정지

# 절반 속도
ros2 bag play run_001 --rate 0.5

# 두 배 속도
ros2 bag play run_001 --rate 2.0

# 반복
ros2 bag play run_001 --loop

# 특정 Offset부터 시작
ros2 bag play run_001 --start-offset 42.0

# 일시정지 상태로 시작
ros2 bag play run_001 --start-paused

Keyboard Control은 배포판별 도움말을 확인합니다. 일반적으로 Space로 Pause·Resume, 방향키로 속도 변경이나 한 Message 진행 기능을 제공할 수 있습니다.

고속 재생에서는 Consumer가 처리율을 따라가지 못해 Queue Drop과 지연이 달라질 수 있으므로 “같은 입력”이더라도 실행 Timing이 실제 기록과 같지 않습니다.

18. 일부 Topic만 재생

ros2 bag play run_001 \
  --topics /scan /odom /tf /tf_static

알고리즘 입력만 재생하고 기록 당시 출력을 제외하면 새 알고리즘 출력과 충돌을 피할 수 있습니다.

재생: /scan, /odom, /tf
제외: 기록 당시 /cmd_vel_new, /detections_new
실행: 수정한 Algorithm Node
비교: 새 출력과 Baseline Metric

19. Remap으로 안전한 재생

ros2 bag play run_001 --ros-args \
  -r /cmd_vel:=/replay/cmd_vel \
  -r /scan:=/replay/scan \
  -r /odom:=/replay/odom

배포판의 ros2 bag play --help에서 Remap 전달 문법을 확인합니다.

실제 Robot이 연결된 상태에서 Bag의 /cmd_vel, Joint Trajectory, Motor Enable과 Mode Service Event를 원래 이름으로 재생하면 물리 동작이 발생할 수 있습니다.

안전 원칙:

1. Motor Driver 전원 또는 Enable을 차단합니다.
2. Command Topic을 안전한 Namespace로 Remap합니다.
3. Hardware Node를 실행하지 않습니다.
4. Simulation·Digital Twin에서 먼저 검증합니다.
5. E-stop을 준비하고 물리 구역을 격리합니다.

20. 기록 Timestamp·Header Stamp·ROS Clock

세 시간을 구분합니다.

Network·Queue 지연이 있으면 Record Timestamp와 Header Stamp가 다릅니다.

Sensor 측정        Network 전송        Recorder 수신
header.stamp=t0 ──────────────────────> record_time=t0+delay

Sensor Fusion과 TF 변환은 측정 Stamp가 중요하며, Bag 재생 순서는 설정에 따라 Received Timestamp 같은 기록 기준을 사용할 수 있습니다.

21. /clock과 use_sim_time

Bag 재생 중 기록 시간축을 사용하려면 Player가 /clock을 발행하고 Consumer Node가 Simulation Time을 사용하도록 구성합니다.

ros2 bag play run_001 --clock

# replay_params.yaml
/**:
  ros__parameters:
    use_sim_time: true

ros2 topic echo /clock --once
ros2 param get /algorithm_node use_sim_time

모든 Node에 무조건 use_sim_time을 켜야 한다는 뜻은 아닙니다. 재생 Pipeline에서 시간 기반 계산, Timer, TF와 Timeout을 기록 시간축으로 재현해야 하는 Node들을 일관되게 구성합니다. 일부는 Wall Time, 일부는 Bag Time을 쓰면서 상호 비교하면 큰 시간 오차가 생깁니다.

22. Pause와 Clock 정지의 영향

Bag을 Pause하면 /clock도 진행하지 않을 수 있습니다. Simulation Time을 쓰는 Node의 ROS Timer와 Timeout 역시 멈춥니다.

Wall Timer 관점: 30초 Pause
ROS Time 관점:   0초 경과

이것은 재현에는 유용하지만 다음 차이를 만듭니다.

• Wall Clock 기반 Network Timeout은 계속 흐를 수 있음
• ROS Timer는 멈출 수 있음
• 외부 Hardware는 실제 시간대로 계속 움직임
• Thread·OS Timeout은 Clock Source가 다를 수 있음

따라서 Bag Time을 사용하는 Offline Pipeline에 실제 Hardware를 섞지 않습니다.

23. TF 재생 체크리스트

ros2 bag info run_001 | grep -E '/tf|/tf_static'
ros2 bag play run_001 --clock
ros2 run tf2_tools view_frames
ros2 run tf2_ros tf2_echo map base_link

TF가 실패하는 원인:

• /tf 또는 /tf_static 미기록
• 필요한 Sensor Frame 누락
• Consumer의 use_sim_time 불일치
• Bag 중간부터 시작해 필요한 Dynamic TF가 아직 안 나옴
• 두 Publisher가 같은 Transform을 발행
• 기록 당시부터 Stamp가 잘못됨

/tf_static은 Transient Local Endpoint입니다. rosbag2가 Offered QoS를 적응할 수 있지만 배포판과 Publisher 구성에 따라 기록 결과를 반드시 확인하십시오. 가장 안전한 운영 절차는 Recorder를 먼저 시작하고 Bringup을 시작한 뒤 bag info와 재생 Test로 Static TF 포함을 검증하는 것입니다.

24. QoS Override

# qos_overrides.yaml
/scan:
  reliability: best_effort
  durability: volatile
  history: keep_last
  depth: 10

/map:
  reliability: reliable
  durability: transient_local
  history: keep_last
  depth: 1

/tf_static:
  reliability: reliable
  durability: transient_local
  history: keep_last
  depth: 100

기록:

ros2 bag record --all -o qos_run \
  --qos-profile-overrides-path qos_overrides.yaml

재생:

ros2 bag play qos_run \
  --qos-profile-overrides-path qos_overrides.yaml

Override는 무조건 붙이는 Template이 아닙니다. topic info --verbose로 Publisher Offered와 Recorder/Consumer Requested 요구를 조사하고 필요한 Topic에만 적용합니다.

25. 여러 Publisher QoS 문제

같은 Topic에 Best Effort와 Reliable Publisher가 동시에 있으면 Recorder가 모든 Endpoint와 만족스러운 단일 Subscription QoS를 고르기 어려운 경우가 있습니다.

ros2 topic info /detections --verbose

해결 방법:

• Source별 Topic으로 분리
• Publisher QoS 계약 통일
• 목적별 Recorder를 별도 실행
• QoS Override를 시험하고 Message 수 비교

기록된 Topic Message 수가 실제 두 Source 합계와 맞는지 Sequence와 Source ID로 검증합니다.

26. Bag 복구와 Reindex

비정상 종료로 Metadata나 Index가 완성되지 않았을 때 Storage가 남아 있다면 reindex가 복구에 도움을 줄 수 있습니다.

ros2 bag reindex damaged_bag
ros2 bag info damaged_bag

Reindex가 손상된 Message Payload 자체를 복구하는 것은 아닙니다. 원본을 복사해 보존한 뒤 복구 작업을 수행하고 Duration·Topic별 Count·재생 가능 구간을 검증합니다.

27. Bag Convert로 형식과 Profile 변경

ros2 bag convert는 입력 Bag을 다른 Storage나 설정으로 변환할 수 있습니다. 세부 YAML Schema는 배포판 문서를 따릅니다.

# convert.yaml 개념 예시
output_bags:
  - uri: converted_mcap
    storage_id: mcap
    storage_preset_profile: zstd_small

ros2 bag convert -i input_bag -o convert.yaml
ros2 bag info converted_mcap

변환 후 검증:

• Topic·Type 목록
• Topic별 Message 수
• 시작·종료와 Duration
• Header Stamp 범위
• Compression·Storage ID
• 실제 재생과 Random Seek

28. Python으로 Bag 읽기

# bag_tools/bag_summary.py
import sys

import rosbag2_py
from rclpy.serialization import deserialize_message
from rosidl_runtime_py.utilities import get_message


def open_reader(uri):
    reader = rosbag2_py.SequentialReader()
    reader.open(
        rosbag2_py.StorageOptions(uri=uri, storage_id=''),
        rosbag2_py.ConverterOptions('', ''),
    )
    topic_types = {
        item.name: item.type
        for item in reader.get_all_topics_and_types()
    }
    return reader, topic_types


def summarize(uri):
    reader, topic_types = open_reader(uri)
    message_classes = {}
    counts = {}
    first_ns = None
    last_ns = None

    while reader.has_next():
        topic, serialized, timestamp_ns = reader.read_next()
        counts[topic] = counts.get(topic, 0) + 1
        first_ns = timestamp_ns if first_ns is None else min(first_ns, timestamp_ns)
        last_ns = timestamp_ns if last_ns is None else max(last_ns, timestamp_ns)

        if topic not in message_classes:
            message_classes[topic] = get_message(topic_types[topic])
        # 필요할 때만 역직렬화한다.
        _message = deserialize_message(serialized, message_classes[topic])

    duration = 0.0 if first_ns is None else (last_ns - first_ns) / 1e9
    print(f'bag={uri}, duration={duration:.3f}s')
    for topic, count in sorted(counts.items()):
        rate = count / duration if duration > 0 else 0.0
        print(f'{topic}: {count} messages, average={rate:.2f}Hz')


if __name__ == '__main__':
    summarize(sys.argv[1])

ConverterOptions와 자동 Storage 감지 동작은 배포판 API에 맞게 확인합니다. Serialization Format은 bag info와 Metadata를 기준으로 지정할 수 있습니다.

29. Topic Filter로 빠르게 분석

reader, topic_types = open_reader(uri)
reader.set_filter(rosbag2_py.StorageFilter(
    topics=['/scan', '/cmd_vel_safe']))

대용량 Camera가 있는 Bag에서 관심 Topic만 Filter하면 Disk Read와 Deserialization 비용을 크게 줄일 수 있습니다.

Type Class도 Message마다 찾지 말고 Topic별로 Cache합니다.

message_classes = {
    topic: get_message(type_name)
    for topic, type_name in topic_types.items()
}

Custom Message를 읽으려면 해당 Interface Package가 현재 Environment에 설치되고 Source되어 있어야 합니다.

30. LaserScan 안전 지표 분석기

import math
import rosbag2_py
from rclpy.serialization import deserialize_message
from sensor_msgs.msg import LaserScan


def analyze_scan(uri, danger_distance=0.3):
    reader = rosbag2_py.SequentialReader()
    reader.open(
        rosbag2_py.StorageOptions(uri=uri, storage_id=''),
        rosbag2_py.ConverterOptions('', ''),
    )
    reader.set_filter(rosbag2_py.StorageFilter(topics=['/scan']))

    nearest_all = float('inf')
    danger_frames = 0
    total_frames = 0

    while reader.has_next():
        topic, data, timestamp = reader.read_next()
        message = deserialize_message(data, LaserScan)
        valid = [value for value in message.ranges
                 if math.isfinite(value)
                 and message.range_min <= value <= message.range_max]
        if not valid:
            continue
        nearest = min(valid)
        nearest_all = min(nearest_all, nearest)
        danger_frames += int(nearest < danger_distance)
        total_frames += 1

    print(f'valid frames={total_frames}')
    print(f'nearest={nearest_all:.3f}m')
    print(f'danger frames={danger_frames}')
    return danger_frames

Frame 개수만 세면 10Hz와 40Hz Sensor 결과를 공정하게 비교하기 어렵습니다. 사건 지속시간과 연속 구간도 함께 계산합니다.

31. Command 간격 회귀 시험

def command_gap_metrics(stamps_ns):
    gaps_ms = [
        (current - previous) / 1e6
        for previous, current in zip(stamps_ns, stamps_ns[1:])
    ]
    if not gaps_ms:
        return {'count': 0, 'max_ms': None, 'over_100ms': 0}
    return {
        'count': len(gaps_ms),
        'mean_ms': sum(gaps_ms) / len(gaps_ms),
        'max_ms': max(gaps_ms),
        'over_100ms': sum(gap > 100.0 for gap in gaps_ms),
    }

CI에서 기준을 판정합니다.

metrics = command_gap_metrics(stamps_ns)
if metrics['over_100ms'] > 0:
    print(f'FAIL: {metrics}')
    raise SystemExit(1)
print(f'PASS: {metrics}')

Bag Record Timestamp 간격과 Message Header Stamp 간격은 다른 의미이므로 어떤 시간을 Metric으로 사용하는지 명시합니다.

32. Odometry 주행 거리 계산

import math


def path_length(poses):
    total = 0.0
    for previous, current in zip(poses, poses[1:]):
        dx = current.position.x - previous.position.x
        dy = current.position.y - previous.position.y
        total += math.hypot(dx, dy)
    return total

Odometry가 순간 Jump하거나 Reset되면 단순 합산이 과대 계산됩니다.

step = math.hypot(dx, dy)
if step > 1.0:
    print('Odometry jump 후보 제외')
else:
    total += step

Threshold는 Robot 최대 속도, Sample 간격과 Reset 규칙으로 정합니다.

33. Offline Algorithm 회귀 Pipeline

고정 입력 Bag
   ↓
새 Algorithm 실행
   ↓
새 출력 Bag 기록
   ↓
Metric 추출
   ↓
Baseline과 비교 → PASS / FAIL

비교 지표 예:

• Collision 위험 구간 수와 지속시간
• Localization Error·Covariance
• Goal 성공률과 완료시간
• Command Jerk·최대 속도
• CPU와 처리 지연
• Detection Precision·Recall

Bag을 같은 Rate로 재생해도 Executor Scheduling과 GPU Algorithm이 완전히 결정적이지 않을 수 있습니다. 허용 오차, Random Seed, Hardware와 Software Version을 함께 기록합니다.

34. Dataset Export 시 시간 동기

Camera Frame과 가장 가까운 LiDAR·Odometry를 짝짓는 예입니다.

from bisect import bisect_left


def nearest_index(sorted_stamps, target):
    index = bisect_left(sorted_stamps, target)
    candidates = []
    if index < len(sorted_stamps):
        candidates.append(index)
    if index > 0:
        candidates.append(index - 1)
    return min(candidates, key=lambda i: abs(sorted_stamps[i] - target))

최대 허용 시간 차이를 검사합니다.

index = nearest_index(lidar_stamps, camera_stamp)
delta_ms = abs(lidar_stamps[index] - camera_stamp) / 1e6
if delta_ms > 50.0:
    print('동기 Pair 제외')

Recorder 수신 Timestamp 대신 Sensor Header Stamp를 사용할지 Sensor Clock 특성에 따라 결정합니다.

35. 개인정보와 보안

Bag에는 얼굴, 음성, 위치, 시설 지도, Network 정보와 Robot 동작이 포함될 수 있습니다.

운영 원칙:

• 목적에 필요한 Topic만 기록
• 보존기간과 삭제 절차 정의
• 접근 권한과 저장 위치 제한
• 전송·보관 암호화
• 외부 공유 전 얼굴·음성·위치 비식별화
• Secret와 Token을 Topic·Parameter로 발행하지 않음
• Dataset License와 동의 조건 기록

Bag 파일 확장자만 바꾸거나 압축하는 것은 암호화가 아닙니다.

36. 기록·재생 안전 체크리스트

기록 전:

• [ ] 사건 목적과 필수 Topic을 정했다.
• [ ] 예상 용량과 Disk 쓰기 속도를 확인했다.
• [ ] /tf, /tf_static, Health와 Command를 포함했다.
• [ ] 개인정보와 보존 정책을 검토했다.
• [ ] 기록 부하가 Robot 제어에 미치는 영향을 시험했다.

재생 전:

• [ ] Hardware Driver를 종료하거나 Motor를 Disable했다.
• [ ] Command Topic을 Replay Namespace로 Remap했다.
• [ ] /clock과 use_sim_time 적용 범위를 정했다.
• [ ] 기록 당시 출력 Topic과 새 Algorithm 출력 충돌을 제거했다.
• [ ] QoS 호환성을 확인했다.
• [ ] Simulation에서 먼저 시험했다.

37. 표준 진단 순서

1. bag info로 Topic·Type·Count 확인
        ↓
2. play 후 topic list·hz 확인
        ↓
3. topic info --verbose로 QoS 확인
        ↓
4. /clock·use_sim_time·Header Stamp 확인
        ↓
5. TF Tree와 Static TF 확인
        ↓
6. Remap·Namespace·Consumer Callback 확인

ros2 bag info run_001
ros2 bag play run_001 --clock
ros2 topic hz /scan
ros2 topic info /scan --verbose
ros2 topic echo /clock --once
ros2 param get /algorithm use_sim_time
ros2 run tf2_tools view_frames

38. 증상별 해결 방법

39. 흔한 실수와 교정

1. 항상 --all로 기록한다. 목적별 Topic과 용량을 먼저 정합니다.
2. 입력 Sensor만 기록한다. TF, Health, 중간 상태와 최종 Command도 기록합니다.
3. 배포판의 기본 Storage를 고정 사실로 외운다. bag info로 확인합니다.
4. Compression은 무조건 이득이라고 생각한다. CPU·Drop·제어 지연을 측정합니다.
5. Cache를 키우면 지속적인 Disk 부족이 해결된다고 믿는다. Data Rate와 쓰기율을 맞춥니다.
6. Bag 존재만 확인하고 현장을 떠난다. Topic별 Count와 짧은 재생을 확인합니다.
7. Header Stamp와 Record Timestamp를 혼동한다. Metric의 시간 기준을 명시합니다.
8. 재생 때 일부 Node만 Bag Clock을 쓴다. 상호작용 Node의 Clock을 일관되게 합니다.
9. QoS Override를 모든 Bag에 복사한다. 실제 Endpoint 요구를 조사합니다.
10. 기록 당시 출력과 새 출력 모두 재생한다. 입력·Baseline·새 출력 Namespace를 분리합니다.
11. 실제 Robot에 Command를 그대로 재생한다. Driver 차단과 Remap을 먼저 합니다.
12. Bag만 있으면 완전 재현된다고 생각한다. Version·Parameter·환경 Manifest를 함께 보관합니다.

40. 정리

rosbag2는 Robot Topic을 시간과 함께 저장하는 기록 도구이자, 현장 실패 재현·Dataset·오프라인 분석·회귀 시험의 기반입니다. 좋은 기록은 모든 Topic을 무작정 담는 것이 아니라 사건 원인, 공간·시간 복원, 출력과 Health에 필요한 Data를 선택합니다.

기록에서는 대역폭·Disk·CPU·Storage·분할·Compression과 QoS를 검증합니다. 재생에서는 Command 격리, Topic 선택, Remap, /clock, use_sim_time, TF와 QoS를 확인합니다.

Python rosbag2_py로 Bag을 직접 읽으면 실시간 재생을 기다리지 않고 Message 수, Sensor 위험 구간, Command 간격과 주행 거리를 계산할 수 있습니다. 같은 Bag과 기준 Metric을 지속적으로 사용하면 Robot을 매번 굴리지 않고도 알고리즘 퇴행을 발견할 수 있습니다.

연습 문제

1. Robot 디버깅에서 rosbag2가 일반 Log보다 중요한 이유를 설명하세요.
2. Bag에 저장되는 Metadata와 자동으로 저장되지 않는 실행 환경 정보를 구분하세요.
3. Camera 한 대의 예상 기록 용량을 계산하는 절차를 설명하세요.
4. --all, Topic 목록, Include Regex와 Exclude Regex를 어떤 상황에서 사용하나요?
5. Navigation 실패를 재현하기 위해 Sensor 외에 함께 기록할 Topic을 쓰세요.
6. MCAP과 sqlite3를 선택할 때 고려할 항목은 무엇인가요?
7. 파일 분할, Compression과 Cache의 장점과 위험을 설명하세요.
8. ros2 bag info에서 현장을 떠나기 전에 확인할 항목은 무엇인가요?
9. Record Timestamp, Header Stamp와 /clock의 차이를 설명하세요.
10. Bag 재생에서 --clock과 use_sim_time이 필요한 상황은 언제인가요?
11. /tf_static이 없거나 TF Extrapolation이 발생할 때 진단 순서를 설명하세요.
12. QoS Override가 필요한 기록·재생 상황을 각각 설명하세요.
13. 실제 Robot에서 /cmd_vel Bag 재생이 위험한 이유와 방지책을 쓰세요.
14. SequentialReader, get_message, deserialize_message의 역할을 설명하세요.
15. 고정 Bag을 이용한 Algorithm 회귀 시험 Pipeline과 비결정성 처리 방법을 설명하세요.

참고 자료

• Recording and playing back data
• rosbag2 README and documentation
• rosbag2 QoS Overrides
• Reading from a bag file with Python
• Writing a bag file with Python
• rosbag2_storage_mcap
• Visualizing ROS 2 data with Foxglove