diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml new file mode 100644 index 0000000..b9b7043 --- /dev/null +++ b/.github/workflows/CI.yml @@ -0,0 +1,104 @@ +name: CI + +on: + push: + branches: [main] + tags: ['v*'] + pull_request: + branches: [main] + workflow_dispatch: + +permissions: + contents: read + +jobs: + linux: + runs-on: ubuntu-latest + strategy: + matrix: + target: [x86_64, aarch64] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + - name: Build wheels + uses: PyO3/maturin-action@v1 + with: + target: ${{ matrix.target }} + args: --release --out dist + manylinux: auto + - uses: actions/upload-artifact@v4 + with: + name: wheels-linux-${{ matrix.target }} + path: dist + + macos: + runs-on: macos-latest + strategy: + matrix: + target: [x86_64, aarch64] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + - name: Build wheels + uses: PyO3/maturin-action@v1 + with: + target: ${{ matrix.target }} + args: --release --out dist + - uses: actions/upload-artifact@v4 + with: + name: wheels-macos-${{ matrix.target }} + path: dist + + windows: + runs-on: windows-latest + strategy: + matrix: + target: [x64] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + architecture: ${{ matrix.target }} + - name: Build wheels + uses: PyO3/maturin-action@v1 + with: + args: --release --out dist + - uses: actions/upload-artifact@v4 + with: + name: wheels-windows-${{ matrix.target }} + path: dist + + sdist: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Build sdist + uses: PyO3/maturin-action@v1 + with: + command: sdist + args: --out dist + - uses: actions/upload-artifact@v4 + with: + name: wheels-sdist + path: dist + + release: + name: Release to PyPI + runs-on: ubuntu-latest + if: startsWith(github.ref, 'refs/tags/v') + needs: [linux, macos, windows, sdist] + permissions: + id-token: write + steps: + - uses: actions/download-artifact@v4 + with: + pattern: wheels-* + merge-multiple: true + path: dist + - name: Publish to PyPI + uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/Cargo.lock b/Cargo.lock index 95c82d1..a2c68ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -351,18 +351,6 @@ dependencies = [ "zstd", ] -[[package]] -name = "mcap-player" -version = "0.1.0" -dependencies = [ - "mcap", - "memmap2", - "parking_lot", - "pyo3", - "tempfile", - "thiserror 2.0.18", -] - [[package]] name = "memchr" version = "2.8.0" @@ -669,6 +657,18 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "tachy-mcap-reader" +version = "0.1.0" +dependencies = [ + "mcap", + "memmap2", + "parking_lot", + "pyo3", + "tempfile", + "thiserror 2.0.18", +] + [[package]] name = "target-lexicon" version = "0.12.16" diff --git a/Cargo.toml b/Cargo.toml index c334687..9f518cb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ [package] -name = "mcap-player" +name = "tachy-mcap-reader" version = "0.1.0" edition = "2021" license = "Apache-2.0" description = "Zero-service MCAP playback library for ROS2" [lib] -name = "_mcap_player_core" +name = "_core" crate-type = ["cdylib"] [dependencies] diff --git a/README.md b/README.md new file mode 100644 index 0000000..737e424 --- /dev/null +++ b/README.md @@ -0,0 +1,105 @@ +# tachy-reader + +Zero-service MCAP playback library for ROS2. Rust core (via PyO3) with a pure-Python API. + +**No DDS services are registered** — all playback control happens via in-process method calls, keeping the ROS2 graph clean. + +## Features + +- **Rust-powered core**: memory-mapped MCAP reading with zero-copy message access +- **Zero ROS2 services**: no parameter/logger/type-description services polluting the graph +- **Simple API**: context-manager pattern, blocking or async playback +- **Playback control**: play, pause, resume, stop, seek, rate adjustment +- **Topic filtering**: include/exclude specific topics +- **QoS overrides**: per-topic QoS profile configuration +- **Callbacks**: `on_publish` and `on_complete` hooks + +## Installation + +### From source (requires Rust toolchain) + +```bash +pip install git+https://github.com/tachycode/tachy-reader.git@dev +``` + +### Local development + +```bash +git clone https://github.com/tachycode/tachy-reader.git +cd tachy-reader +pip install -e . +``` + +> **Prerequisites**: Rust toolchain (`rustup`), Python ≥ 3.8, maturin ≥ 1.5 + +## Quick Start + +```python +from mcap_player import McapPlayer + +with McapPlayer("/path/to/recording.mcap") as player: + # Blocking playback at 2x speed + player.play(rate=2.0) +``` + +### Non-blocking playback + +```python +with McapPlayer("recording.mcap") as player: + player.play_async(rate=1.0) + + while player.state != "finished": + print(f"Progress: {player.progress:.1%}") + time.sleep(1.0) +``` + +### Topic filtering + +```python +player = McapPlayer( + "recording.mcap", + topics=["/camera/image_raw", "/imu/data"], + topic_prefix="/replay", +) +``` + +### Playback control + +```python +player.play_async() +player.pause() +player.seek(10.5) # jump to 10.5 seconds +player.set_rate(0.5) # half speed +player.resume() +player.stop() +``` + +## API Reference + +### `McapPlayer` + +| Parameter | Type | Description | +|-----------|------|-------------| +| `mcap_path` | `str` | Path to `.mcap` file | +| `node_name` | `str` | ROS2 node name (default: `"mcap_player"`) | +| `topic_prefix` | `str` | Prefix for published topics | +| `topics` | `list[str]` | Whitelist of topics to publish | +| `exclude_topics` | `list[str]` | Topics to skip | +| `qos_overrides` | `dict` | Per-topic QoS profiles | +| `on_publish` | `Callable` | Called after each message publish | +| `on_complete` | `Callable` | Called when playback finishes | +| `node` | `Node` | Existing rclpy Node (optional) | + +### Properties + +| Property | Type | Description | +|----------|------|-------------| +| `current_time` | `float` | Current position in seconds | +| `progress` | `float` | Playback progress (0.0 – 1.0) | +| `state` | `str` | `"idle"`, `"playing"`, `"paused"`, `"finished"` | +| `duration` | `float` | Total file duration in seconds | +| `topics_info` | `dict` | Topic metadata `{name: (msg_type, encoding)}` | + +## License + +Apache-2.0 diff --git a/docs/devlog/DEVLOG-002-ccg-raw-reader-api-design.md b/docs/devlog/DEVLOG-002-ccg-raw-reader-api-design.md new file mode 100644 index 0000000..f8cb66d --- /dev/null +++ b/docs/devlog/DEVLOG-002-ccg-raw-reader-api-design.md @@ -0,0 +1,53 @@ +--- +id: DEVLOG-002 +title: CCG 토론 — McapRawReader API 설계 분석 +task_type: feature +status: completed +complexity: high +created: 2026-03-19 +duration_estimate: 2h +tags: [ccg, codex, api-design, rosbag2_py, architecture, mmap, zero-copy] +--- + +## 목표 (Goal) +- rosbag2_py SequentialReader를 tachy-reader로 대체 가능한지 분석 +- Codex/Gemini CCG 토론을 통해 API 설계 방향 결정 +- 성능 이득 여부, 메모리 이슈, 테스트 전략 수립 + +## 접근 과정 (Approach Log) + +### 1차 시도 — omc ask 명령 +- **방법**: `omc ask codex/gemini` 스킬 호출 +- **결과**: 실패 — `omc` 명령어 미설치 +- **원인**: oh-my-claudecode CLI 미설치 환경 + +### 2차 시도 — codex/gemini CLI 직접 호출 +- **방법**: `codex exec --full-auto`, `gemini -p` 직접 실행 +- **결과**: Codex 성공, Gemini 실패 (API key 미설정) +- **원인**: Codex는 `--full-auto` 플래그로 비대화형 실행 가능. Gemini는 `GEMINI_API_KEY` 미설정 + +### 3차 시도 — Codex + Claude 듀얼 분석 +- **방법**: Codex(GPT-5.4) 분석 결과 + Claude(Opus 4.6) 독립 분석 합성 +- **결과**: 성공 — 4개 주제 모두 합의 도출 + +## 최종 해결 (Final Solution) +- **API 설계**: Option B (별도 McapRawReader pyclass) 만장일치 + - Option A (McapCore에 추가): 두 소비 모델 혼재 위험 + - Option C (rate=inf): semantic hack, Scheduler condvar 불필요하게 통과 +- **성능**: 현재 코드는 "mmap+zero-copy"가 아닌 "mmap+eager full-copy+per-message clone" + - reader.rs:134에서 `msg.data.to_vec()` — 전체 payload 복사 + - reader.rs:169에서 다시 clone — per-message 이중 복사 +- **메모리**: 10GB+ 파일에서 OOM 위험 → Phase 2 lazy chunk decoding 필요 +- **버그 2건 발견**: + 1. scheduler.rs:88 — double ns 변환 (lib.rs:66에서 이미 변환된 값을 재변환) + 2. scheduler.rs:204 — pause/seek 시 이미 consume된 메시지 유실 + +## 교훈 (Lessons Learned) +- Codex CLI: `codex exec --full-auto "prompt"` 형태로 비대화형 실행 +- Gemini CLI: `gemini -p "prompt"` — `GEMINI_API_KEY` 환경변수 필수 +- CCG 패턴: 한 모델 불가 시 나머지 + Claude 합성으로 충분히 유의미한 분석 가능 +- tachy-reader의 "zero-copy" 마케팅과 실제 구현 사이에 갭 존재 — open() 시 전체 파일 메모리 적재 + +## 변경 파일 (Changed Files) +- `.omc/artifacts/ask/codex-raw-reader-api.md` — Codex 분석 결과 +- `.omc/artifacts/ask/claude-raw-reader-synthesis.md` — 종합 분석 및 액션 체크리스트 diff --git a/docs/devlog/DEVLOG-003-implement-mcap-raw-reader.md b/docs/devlog/DEVLOG-003-implement-mcap-raw-reader.md new file mode 100644 index 0000000..cf06977 --- /dev/null +++ b/docs/devlog/DEVLOG-003-implement-mcap-raw-reader.md @@ -0,0 +1,57 @@ +--- +id: DEVLOG-003 +title: McapRawReader pyclass 구현 및 scheduler 버그 수정 +task_type: feature +status: completed +complexity: medium +created: 2026-03-19 +duration_estimate: 1h +tags: [mcap-raw-reader, pyo3, rust, iterator, scheduler-bug, maturin] +--- + +## 목표 (Goal) +- Scheduler를 우회하는 McapRawReader pyclass 구현 (타이밍 없는 순차 읽기) +- scheduler.rs double ns 변환 버그 수정 +- Python export 및 type stubs 업데이트 +- 정확성 테스트 및 벤치마크 스크립트 작성 + +## 접근 과정 (Approach Log) + +### 1차 시도 — McapRawReader 구현 + 빌드 +- **방법**: lib.rs에 McapRawReader pyclass 추가, parking_lot::Mutex로 McapReader 래핑, `__iter__`/`__next__` 구현 +- **결과**: 코드 작성 성공, 빌드 환경 문제 발생 +- **원인**: Rust 툴체인 미설치, python3-venv 미설치, sudo 불가 + +### 2차 시도 — 환경 구축 +- **방법**: rustup 설치 → maturin pip 설치 → maturin develop 시도 +- **결과**: `maturin develop`는 venv 필요하여 실패 +- **원인**: python3-venv 패키지 미설치 + sudo 권한 없음 + +### 3차 시도 — maturin build + pip install wheel +- **방법**: `maturin build --release` → wheel 생성 → `pip3 install --break-system-packages --no-deps` wheel +- **결과**: 성공 — 빌드, 설치, 테스트 실행 가능 +- **원인**: `maturin build`는 venv 불필요, `--no-deps`로 rosidl_runtime_py 의존성 우회 + +### 버그 수정 — scheduler.rs:88 +- **방법**: `file_start + (start_ns as f64 * 1_000_000_000.0)` → `file_start + start_ns` +- **결과**: 성공 — lib.rs:66에서 이미 초→ns 변환 완료된 값이므로 재변환 제거 + +## 최종 해결 (Final Solution) +- McapRawReader: McapReader를 Mutex로 직접 감싸서 Scheduler 완전 우회 +- Python iterator protocol 지원: `for topic, msg_type, data, ts in McapRawReader("file.mcap")` +- topics(), duration_ns(), message_count(), reset(), seek() 메서드 제공 +- 10개 테스트 작성, 벤치마크 스크립트 작성 + +## 교훈 (Lessons Learned) +- **Rust 환경 부트스트래핑**: `curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y` 로 즉시 설치 +- **venv 없는 환경에서 maturin**: `maturin build` → `pip install wheel` 패턴이 유효 +- **`--break-system-packages`**: PEP 668 제한을 우회하나, 프로덕션에서는 비권장 +- **ROS2 의존성 우회**: `--no-deps`로 rosidl_runtime_py 없이 설치 가능 (McapRawReader는 ROS2 불필요) + +## 변경 파일 (Changed Files) +- `src/lib.rs` — McapRawReader pyclass 추가, _mcap_player_core module 등록 +- `src/scheduler.rs` — line 88: double ns 변환 버그 수정 +- `python/mcap_player/__init__.py` — McapRawReader export 추가 +- `python/mcap_player/_core.pyi` — McapRawReader type stubs 추가 +- `tests/test_raw_reader.py` — 10개 정확성 테스트 +- `tests/bench_raw_reader.py` — rosbag2_py 대비 성능 벤치마크 diff --git a/docs/devlog/DEVLOG-004-fix-pyo3-vec-u8-to-bytes.md b/docs/devlog/DEVLOG-004-fix-pyo3-vec-u8-to-bytes.md new file mode 100644 index 0000000..e0f767e --- /dev/null +++ b/docs/devlog/DEVLOG-004-fix-pyo3-vec-u8-to-bytes.md @@ -0,0 +1,55 @@ +--- +id: DEVLOG-004 +title: PyO3 Vec → Python bytes 타입 변환 수정 +task_type: bugfix +status: completed +complexity: medium +created: 2026-03-19 +duration_estimate: 30m +tags: [pyo3, pyobject, vec-u8, pybytes, type-conversion, rust] +--- + +## 목표 (Goal) +- McapRawReader.next_raw_message()가 `bytes`가 아닌 `list[int]`를 반환하는 문제 수정 +- downstream 코드(deserialize_message 등)에서 `bytes` 타입 기대 + +## 접근 과정 (Approach Log) + +### 1차 시도 — PyResult 래핑 +- **방법**: 반환 타입을 `Option<(String, String, Vec, u64)>` → `PyResult>`로 변경 +- **결과**: 실패 — 여전히 `list[int]` 반환 +- **원인**: PyO3에서 `Vec` → Python 변환은 항상 `list[int]`. `PyResult` 래핑과 무관. McapCore.next_message()도 동일하게 `list[int]` 반환하고 있었음 (미인지) + +### 2차 시도 — PyBytes::new() +- **방법**: `pyo3::types::PyBytes::new(py, &m.data)` 사용 +- **결과**: 컴파일 실패 — `no function or associated item named 'new'` +- **원인**: PyO3 0.22에서 Bound API 전환으로 `new` → `new_bound`로 API 변경됨 + +### 3차 시도 — PyBytes::new_bound() + unbind() +- **방법**: `pyo3::types::PyBytes::new_bound(py, &m.data).unbind()` → `Py` 반환 +- **결과**: 성공 — Python에서 `bytes` 타입으로 정상 반환, 10/10 테스트 통과 + +## 최종 해결 (Final Solution) +```rust +fn next_raw_message<'py>(&self, py: Python<'py>) -> PyResult, u64)>> { + match self.reader.lock().next_raw_message() { + Some(m) => { + let bytes = pyo3::types::PyBytes::new_bound(py, &m.data).unbind(); + Ok(Some((m.topic, m.msg_type, bytes, m.log_time_ns))) + } + None => Ok(None), + } +} +``` + +핵심: `__next__`도 동일하게 `Python<'py>` 파라미터를 받아야 함. + +## 교훈 (Lessons Learned) +- **PyO3 `Vec` ≠ Python `bytes`**: PyO3의 자동 변환은 `Vec` → `list[int]`. Python `bytes`를 원하면 반드시 `PyBytes` 명시적 사용 필요 +- **PyO3 0.22 Bound API**: `PyBytes::new()` 제거됨 → `PyBytes::new_bound()` 사용. `unbind()`로 `Py` (GIL-independent handle) 생성 +- **McapCore도 동일 이슈**: `next_message()`도 `Vec` 반환 → `list[int]` 되고 있음. 향후 수정 대상 +- **테스트 비교 주의**: `list[int]`와 `bytes`는 `==`로 비교 불가 → `bytes(list_val) == bytes_val`로 정규화 필요 + +## 변경 파일 (Changed Files) +- `src/lib.rs` — `next_raw_message()`, `__next__()` 반환 타입을 `Py`로 변경 +- `tests/test_raw_reader.py` — consistency 테스트에서 `bytes()` 변환 비교 추가 diff --git a/docs/devlog/DEVLOG-005-lazy-chunk-decoding-benchmark.md b/docs/devlog/DEVLOG-005-lazy-chunk-decoding-benchmark.md new file mode 100644 index 0000000..9bf3abe --- /dev/null +++ b/docs/devlog/DEVLOG-005-lazy-chunk-decoding-benchmark.md @@ -0,0 +1,60 @@ +--- +id: DEVLOG-005 +title: Lazy chunk decoding + pytest-benchmark comparison +task_type: feature +status: completed +complexity: high +created: 2026-03-19 +duration_estimate: 4h +tags: [lazy-decoding, chunk, benchmark, pytest-benchmark, mcap, consensus, deep-interview] +--- + +## 목표 (Goal) +- McapRawReader에 lazy chunk decoding 구현 (Phase 2): open() 시 전체 메시지 로드 대신 chunk 단위 on-demand 로드 +- pytest-benchmark 기반 rosbag2_py 비교 벤치마크 작성 + +## 접근 과정 (Approach Log) + +### Deep Interview (7 rounds, ambiguity 18%) +- **방법**: Socratic 질문으로 요구사항 명확화 (독립 작업 확인, throughput 우선, 최소 버전 범위, pytest-benchmark 선택) +- **결과**: 성공 - 두 작업의 scope, 성공 기준, 제약 조건 확정 +- **핵심 발견**: Lazy decoding 목표가 OOM 방지가 아닌 throughput 최적화 (Contrarian mode에서 확인) + +### 1차 Plan: Summary API 직접 활용 (REVISE) +- **방법**: `Summary::read()` + `stream_chunk()` API를 직접 활용하는 Option A 계획 +- **결과**: Critic REVISE 판정 - Critical 이슈 2건 +- **원인 1**: `Summary<'a>`가 mmap을 borrow하므로 McapReader에 저장하면 self-referential struct. `stream_chunk()`은 `&self` (Summary) 메서드라 Summary가 살아있어야 함 +- **원인 2**: 현재 reader.rs의 synthetic channel ID (0,1,2...) vs mcap crate의 실제 channel ID 충돌. Lazy 모드에서 ChunkReader가 반환하는 records::MessageHeader.channel_id는 실제 ID + +### 2차 Plan: Summary drop + ChunkReader 직접 사용 (APPROVED) +- **방법**: Summary에서 owned 데이터만 추출 후 drop, `LinearReader::sans_magic()` + `ChunkReader::new()`로 직접 chunk 읽기 +- **결과**: Architect APPROVE, Critic ACCEPT +- **핵심 변경**: stream_chunk() 포기, ChunkReader 직접 사용, Lazy에서 mcap 실제 channel ID 사용 + +### 구현 + Security Review +- **방법**: 두 Task 병렬 구현 (executor agents) -> Phase 4 validation +- **결과**: Code reviewer가 CRITICAL 1건 + HIGH 1건, Security reviewer가 HIGH 1건 + MEDIUM 2건 발견 +- **수정 사항**: chunk offset overflow (checked_add+try_into), topics.get()?.->continue, f64 validation, GIL release + +## 최종 해결 (Final Solution) +- `ReadMode` enum (Eager/Lazy)으로 reader.rs 구조 변경 +- `load_chunk_static()` 독립 함수 - borrow checker의 `&mut self.mode` + `&self.mmap` 충돌 해결 +- Summary에서 chunk_indexes (owned) + channels->TopicInfo 추출 후 즉시 drop +- chunk 소진 시 다음 chunk 자동 로드, 이전 데이터 drop (O(chunk_size) 메모리) +- Summary 없는 파일은 기존 MessageStream Eager 방식으로 fallback + +## 교훈 (Lessons Learned) +- **mcap crate의 `Summary<'a>` lifetime 주의**: high-level API(`stream_chunk`, `MessageStream`)는 편리하지만 lifetime에 묶임. Self-referential을 피하려면 owned 데이터 추출 + low-level API(LinearReader, ChunkReader) 직접 조합 +- **Consensus planning의 가치**: 1차 plan의 self-referential 문제를 Architect/Critic이 사전 발견. 코딩 전에 발견하여 수시간 낭비 방지 +- **Rust disjoint field borrow**: `match &mut self.mode { Lazy { chunk_indices, .. } => { ... &self.mmap ... } }` - Rust 2021에서 같은 struct의 다른 필드는 독립 borrow 가능. 단, 복잡하면 standalone function으로 분리가 깔끔 +- **Security review 필수**: `as usize` 캐스트는 32-bit에서 truncation 위험. `checked_add` + `try_into` 패턴 사용 +- **mcap 0.9 버그 방어**: `Summary::read()`가 `summary_offset_start == 0`일 때 integer underflow panic. footer를 사전 체크하여 방어 + +## 변경 파일 (Changed Files) +- `src/reader.rs` - ReadMode enum, load_chunk_static(), lazy/eager open, seek, reset 전면 개편 +- `src/lib.rs` - seek f64 validation, GIL release (py.allow_threads), play() 음수 방어 +- `tests/bench_test.py` - pytest-benchmark 기반 5개 벤치마크 (신규) +- `tests/conftest.py` - --mcap-file pytest fixture (신규) +- `tests/create_test_mcap.py` - create_multi_chunk_mcap(), create_no_summary_mcap() 추가 +- `tests/test_raw_reader.py` - 3개 lazy 테스트 추가 (multi_chunk, seek_across, fallback) +- `pyproject.toml` - [project.optional-dependencies] dev 섹션 추가 diff --git a/docs/devlog/README.md b/docs/devlog/README.md index 79462fa..979d905 100644 --- a/docs/devlog/README.md +++ b/docs/devlog/README.md @@ -4,7 +4,17 @@ | ID | Status | Type | Complexity | Title | Date | Keywords | |----|--------|------|------------|-------|------|----------| | DEVLOG-001 | completed | feature | high | Implement mcap-player core (Rust + PyO3 + rclpy) | 2026-03-17 | rust, pyo3, mcap, ros2, rosbag2, maturin, zero-service, rclpy, condvar | +| DEVLOG-002 | completed | feature | high | CCG 토론 — McapRawReader API 설계 분석 | 2026-03-19 | ccg, codex, api-design, rosbag2_py, architecture, mmap, zero-copy | +| DEVLOG-003 | completed | feature | medium | McapRawReader pyclass 구현 및 scheduler 버그 수정 | 2026-03-19 | mcap-raw-reader, pyo3, rust, iterator, scheduler-bug, maturin | +| DEVLOG-004 | completed | bugfix | medium | PyO3 Vec → Python bytes 타입 변환 수정 | 2026-03-19 | pyo3, pyobject, vec-u8, pybytes, type-conversion, rust | +| DEVLOG-005 | completed | feature | high | Lazy chunk decoding + pytest-benchmark comparison | 2026-03-19 | lazy-decoding, chunk, benchmark, pytest-benchmark, mcap, consensus, deep-interview | ## By Type ### feature - [DEVLOG-001](DEVLOG-001-implement-mcap-player-core.md) - Implement mcap-player core (Rust + PyO3 + rclpy) +- [DEVLOG-002](DEVLOG-002-ccg-raw-reader-api-design.md) - CCG 토론 — McapRawReader API 설계 분석 +- [DEVLOG-003](DEVLOG-003-implement-mcap-raw-reader.md) - McapRawReader pyclass 구현 및 scheduler 버그 수정 +- [DEVLOG-005](DEVLOG-005-lazy-chunk-decoding-benchmark.md) - Lazy chunk decoding + pytest-benchmark comparison + +### bugfix +- [DEVLOG-004](DEVLOG-004-fix-pyo3-vec-u8-to-bytes.md) - PyO3 Vec → Python bytes 타입 변환 수정 diff --git a/pyproject.toml b/pyproject.toml index 208526a..6118bca 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,14 +3,17 @@ requires = ["maturin>=1.5,<2"] build-backend = "maturin" [project] -name = "mcap-player" +name = "tachy-mcap-reader" version = "0.1.0" requires-python = ">=3.8" license = { text = "Apache-2.0" } description = "Zero-service MCAP playback library for ROS2" -dependencies = ["rosidl_runtime_py"] +dependencies = [] + +[project.optional-dependencies] +dev = ["pytest", "pytest-benchmark", "mcap"] [tool.maturin] features = ["pyo3/extension-module"] python-source = "python" -module-name = "mcap_player._mcap_player_core" +module-name = "tachy_mcap_reader._core" diff --git a/python/mcap_player/__init__.py b/python/mcap_player/__init__.py deleted file mode 100644 index 53d28f7..0000000 --- a/python/mcap_player/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -"""mcap-player: Zero-service MCAP playback library for ROS2.""" - -from mcap_player.player import McapPlayer - -__all__ = ["McapPlayer"] -__version__ = "0.1.0" diff --git a/python/tachy_mcap_reader/__init__.py b/python/tachy_mcap_reader/__init__.py new file mode 100644 index 0000000..4fdb279 --- /dev/null +++ b/python/tachy_mcap_reader/__init__.py @@ -0,0 +1,7 @@ +"""tachy-mcap-reader: Zero-service MCAP playback library for ROS2.""" + +from tachy_mcap_reader.player import McapPlayer +from tachy_mcap_reader._core import McapRawReader + +__all__ = ["McapPlayer", "McapRawReader"] +__version__ = "0.1.0" diff --git a/python/mcap_player/_core.pyi b/python/tachy_mcap_reader/_core.pyi similarity index 56% rename from python/mcap_player/_core.pyi rename to python/tachy_mcap_reader/_core.pyi index afda77e..258b6d1 100644 --- a/python/mcap_player/_core.pyi +++ b/python/tachy_mcap_reader/_core.pyi @@ -1,3 +1,5 @@ +from typing import Iterator + class McapCore: def __init__( self, @@ -19,3 +21,19 @@ class McapCore: def progress(self) -> float: ... def is_playing(self) -> bool: ... def is_paused(self) -> bool: ... + +class McapRawReader: + def __init__( + self, + path: str, + topics: list[str] | None = None, + exclude_topics: list[str] | None = None, + ) -> None: ... + def topics(self) -> dict[str, tuple[str, str]]: ... + def duration_ns(self) -> int: ... + def message_count(self) -> int: ... + def reset(self) -> None: ... + def seek(self, time_secs: float) -> None: ... + def next_raw_message(self) -> tuple[str, str, bytes, int] | None: ... + def __iter__(self) -> Iterator[tuple[str, str, bytes, int]]: ... + def __next__(self) -> tuple[str, str, bytes, int]: ... diff --git a/python/mcap_player/player.py b/python/tachy_mcap_reader/player.py similarity index 99% rename from python/mcap_player/player.py rename to python/tachy_mcap_reader/player.py index 9511ec6..f006f8e 100644 --- a/python/mcap_player/player.py +++ b/python/tachy_mcap_reader/player.py @@ -39,7 +39,7 @@ def __init__( on_complete: Optional[Callable] = None, node=None, ): - from mcap_player._mcap_player_core import McapCore + from tachy_mcap_reader._core import McapCore self._core = McapCore( mcap_path, diff --git a/python/mcap_player/py.typed b/python/tachy_mcap_reader/py.typed similarity index 100% rename from python/mcap_player/py.typed rename to python/tachy_mcap_reader/py.typed diff --git a/src/lib.rs b/src/lib.rs index b3ee501..dda0555 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -63,7 +63,7 @@ impl McapCore { /// Start playback. rate must be > 0.0. start_time is in seconds from file start. #[pyo3(signature = (rate=1.0, start_time=None))] fn play(&self, rate: f64, start_time: Option) -> PyResult<()> { - let start_ns = start_time.map(|t| (t * 1_000_000_000.0) as u64); + let start_ns = start_time.map(|t| (t.max(0.0) * 1_000_000_000.0) as u64); // For play, we pass the raw ns offset, not absolute time // The scheduler's play method handles adding file_start self.scheduler @@ -148,9 +148,110 @@ impl McapCore { } } +/// Zero-overhead sequential reader for batch processing (no timing delays). +/// Bypasses the Scheduler entirely — reads messages as fast as possible. +#[pyclass] +pub struct McapRawReader { + reader: parking_lot::Mutex, + topics_meta: HashMap, + cached_duration_ns: u64, + cached_message_count: u64, +} + +#[pymethods] +impl McapRawReader { + #[new] + #[pyo3(signature = (path, topics=None, exclude_topics=None))] + fn new( + path: &str, + topics: Option>, + exclude_topics: Option>, + ) -> PyResult { + let reader = McapReader::open(path, topics.as_deref(), exclude_topics.as_deref()) + .map_err(|e| pyo3::exceptions::PyIOError::new_err(e.to_string()))?; + + let mut topics_meta = HashMap::new(); + for info in reader.topics().values() { + topics_meta.insert( + info.name.clone(), + (info.msg_type.clone(), info.encoding.clone()), + ); + } + + let cached_duration_ns = reader.duration_ns(); + let cached_message_count = reader.message_count(); + + Ok(McapRawReader { + reader: parking_lot::Mutex::new(reader), + topics_meta, + cached_duration_ns, + cached_message_count, + }) + } + + /// Get all topics: {topic_name: (msg_type, encoding)} + fn topics(&self) -> HashMap { + self.topics_meta.clone() + } + + /// Total file duration in nanoseconds. + fn duration_ns(&self) -> u64 { + self.cached_duration_ns + } + + /// Total number of messages. + fn message_count(&self) -> u64 { + self.cached_message_count + } + + /// Reset to beginning of file. + fn reset(&self) { + self.reader.lock().reset(); + } + + /// Seek to timestamp (seconds from file start). Must be finite and non-negative. + fn seek(&self, time_secs: f64) -> PyResult<()> { + if !time_secs.is_finite() || time_secs < 0.0 { + return Err(pyo3::exceptions::PyValueError::new_err( + "time_secs must be a finite non-negative number", + )); + } + let mut reader = self.reader.lock(); + let offset_ns = (time_secs * 1_000_000_000.0) as u64; + let target = reader.start_time_ns().saturating_add(offset_ns).min(reader.end_time_ns()); + reader + .seek_to(target) + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) + } + + /// Read next raw message without timing delay (GIL released during chunk decompression). + /// Returns (topic, msg_type, data_bytes, log_time_ns) or None. + fn next_raw_message<'py>(&self, py: Python<'py>) -> PyResult, u64)>> { + let result = py.allow_threads(|| self.reader.lock().next_raw_message()); + match result { + Some(m) => { + let bytes = pyo3::types::PyBytes::new_bound(py, &m.data).unbind(); + Ok(Some((m.topic, m.msg_type, bytes, m.log_time_ns))) + } + None => Ok(None), + } + } + + /// Python iterator protocol: __iter__ + fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> { + slf + } + + /// Python iterator protocol: __next__ + fn __next__<'py>(&self, py: Python<'py>) -> PyResult, u64)>> { + self.next_raw_message(py) + } +} + /// Python module definition. #[pymodule] -fn _mcap_player_core(m: &Bound<'_, PyModule>) -> PyResult<()> { +fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; + m.add_class::()?; Ok(()) } diff --git a/src/reader.rs b/src/reader.rs index c825f08..0d4d9c0 100644 --- a/src/reader.rs +++ b/src/reader.rs @@ -3,6 +3,9 @@ use std::collections::HashMap; use std::fs::File; use std::path::Path; +use mcap::read::{ChunkReader, LinearReader, Summary}; +use mcap::records; + use crate::error::McapError; /// Topic metadata extracted from MCAP channels/schemas. @@ -24,15 +27,18 @@ pub struct RawMessage { pub log_time_ns: u64, } -/// MCAP chunk index entry for O(log n) seek. -#[derive(Debug, Clone)] -pub struct ChunkIndexEntry { - pub message_start_time: u64, - pub message_end_time: u64, - pub chunk_start_offset: u64, - pub chunk_length: u64, - pub compressed_size: u64, - pub compression: String, +/// Internal read mode: eager loads all messages up front; lazy decodes chunk by chunk. +enum ReadMode { + Eager { + messages: Vec<(u64, u16, Vec)>, + current_pos: usize, + }, + Lazy { + chunk_indices: Vec, + current_chunk_idx: usize, + current_messages: Vec<(u64, u16, Vec)>, + current_pos: usize, + }, } /// Offset-based MCAP reader using mmap. @@ -41,13 +47,7 @@ pub struct McapReader { mmap: Mmap, /// channel_id -> TopicInfo topics: HashMap, - /// Sorted by message_start_time for binary search - chunk_indices: Vec, - /// All messages sorted by log_time (built on open from linear scan) - /// Each entry: (log_time_ns, channel_id, data as Vec) - messages: Vec<(u64, u16, Vec)>, - /// Current position in messages - current_pos: usize, + mode: ReadMode, /// File duration start_time_ns: u64, end_time_ns: u64, @@ -65,15 +65,123 @@ impl McapReader { let file = File::open(p)?; let mmap = unsafe { Mmap::map(&file)? }; - // Parse all messages using mcap crate to build our index + // Try lazy path first: read summary to get chunk indexes without scanning all messages. + // Pre-check the footer to avoid a panic in mcap 0.9 Summary::read() when + // summary_offset_start == 0 (integer underflow bug in the crate). + let has_summary_offsets = mcap::read::footer(&mmap) + .map(|f| f.summary_start != 0 && f.summary_offset_start != 0) + .unwrap_or(false); + let summary_result = if has_summary_offsets { + Summary::read(&mmap) + } else { + Ok(None) + }; + match summary_result { + Ok(Some(summary)) => { + // Extract owned channel info from Summary before it goes out of scope. + // Summary<'a> borrows from mmap; we must not store it in McapReader. + let mut topics: HashMap = HashMap::new(); + + for (&chan_id, channel) in &summary.channels { + let schema_name = channel.schema + .as_ref() + .map(|s| s.name.clone()) + .unwrap_or_default(); + + let topic_info = TopicInfo { + channel_id: chan_id, + name: channel.topic.clone(), + msg_type: schema_name, + encoding: channel.message_encoding.clone(), + }; + + // Apply topic filters + if let Some(include) = topics_filter { + if !include.iter().any(|t| t == &topic_info.name) { + continue; + } + } + if let Some(exclude) = exclude_topics { + if exclude.iter().any(|t| t == &topic_info.name) { + continue; + } + } + + topics.insert(chan_id, topic_info); + } + + // Extract stats before dropping summary + let (start_time_ns, end_time_ns, message_count) = if let Some(ref stats) = summary.stats { + (stats.message_start_time, stats.message_end_time, stats.message_count) + } else { + (0u64, 0u64, 0u64) + }; + + // Clone chunk indexes (owned, no borrow from mmap) + let mut chunk_indices: Vec = summary.chunk_indexes.clone(); + // Drop summary now — no more borrows from mmap + drop(summary); + + // Sort by message_start_time for binary-search seek + chunk_indices.sort_by_key(|ci| ci.message_start_time); + + // Load first chunk eagerly so next_raw_message() works immediately + let first_messages = if chunk_indices.is_empty() { + Vec::new() + } else { + load_chunk_static(&mmap, &topics, &chunk_indices, 0) + .unwrap_or_default() + }; + + // If stats were missing or zero, scan first chunk for timing bounds + let (start_time_ns, end_time_ns) = if start_time_ns == 0 && !first_messages.is_empty() { + let start = first_messages.first().map(|(t, _, _)| *t).unwrap_or(0); + let end = chunk_indices.last().map(|ci| ci.message_end_time).unwrap_or(start); + (start, end) + } else { + (start_time_ns, end_time_ns) + }; + + let message_count = if message_count == 0 { + // Count filtered messages: iterate all chunks (slow fallback) + // For now, leave as 0 and let callers handle it + // A better approach: sum message counts from chunk index + chunk_indices.iter().map(|ci| { + ci.message_index_offsets.len() as u64 + }).sum() + } else { + message_count + }; + + Ok(McapReader { + mmap, + topics, + mode: ReadMode::Lazy { + chunk_indices, + current_chunk_idx: 0, + current_messages: first_messages, + current_pos: 0, + }, + start_time_ns, + end_time_ns, + message_count, + }) + } + // No summary or error reading summary — fall back to eager linear scan + _ => Self::open_eager(mmap, topics_filter, exclude_topics), + } + } + + fn open_eager( + mmap: Mmap, + topics_filter: Option<&[String]>, + exclude_topics: Option<&[String]>, + ) -> Result { let mut topics: HashMap = HashMap::new(); let mut messages: Vec<(u64, u16, Vec)> = Vec::new(); let mut start_time_ns = u64::MAX; let mut end_time_ns = 0u64; - // Use mcap's streaming reader to parse the file - // mcap 0.9: Channel has topic, schema, message_encoding, metadata (no id field) - // We use the topic name as the key instead of channel_id let mut topic_to_id: HashMap = HashMap::new(); let mut next_id: u16 = 0; @@ -82,7 +190,6 @@ impl McapReader { let topic_name = msg.channel.topic.clone(); - // Assign a synthetic channel_id based on topic name let channel_id = if let Some(&id) = topic_to_id.get(&topic_name) { id } else { @@ -92,7 +199,6 @@ impl McapReader { id }; - // Register topic if not seen if !topics.contains_key(&channel_id) { let schema_name = msg.channel.schema .as_ref() @@ -106,7 +212,6 @@ impl McapReader { encoding: msg.channel.message_encoding.clone(), }; - // Apply topic filters if let Some(include) = topics_filter { if !include.iter().any(|t| t == &topic_info.name) { continue; @@ -121,7 +226,6 @@ impl McapReader { topics.insert(channel_id, topic_info); } - // Only index messages for included topics if !topics.contains_key(&channel_id) { continue; } @@ -130,13 +234,10 @@ impl McapReader { start_time_ns = start_time_ns.min(log_time); end_time_ns = end_time_ns.max(log_time); - // Store message data directly (mcap crate may decompress into separate buffer) messages.push((log_time, channel_id, msg.data.to_vec())); } - // Sort by log_time (should already be sorted, but ensure) messages.sort_by_key(|(t, _, _)| *t); - let message_count = messages.len() as u64; if start_time_ns == u64::MAX { @@ -146,9 +247,10 @@ impl McapReader { Ok(McapReader { mmap, topics, - chunk_indices: Vec::new(), // TODO: parse from summary for seek optimization - messages, - current_pos: 0, + mode: ReadMode::Eager { + messages, + current_pos: 0, + }, start_time_ns, end_time_ns, message_count, @@ -162,45 +264,138 @@ impl McapReader { /// Read the next raw message at current position. pub fn next_raw_message(&mut self) -> Option { - if self.current_pos >= self.messages.len() { - return None; + loop { + match &mut self.mode { + ReadMode::Eager { messages, current_pos } => { + if *current_pos >= messages.len() { + return None; + } + let (log_time, channel_id, ref data) = messages[*current_pos]; + *current_pos += 1; + // Skip messages with unknown channel_id (defensive — should not happen in eager mode) + let ti = match self.topics.get(&channel_id) { + Some(ti) => ti, + None => continue, + }; + return Some(RawMessage { + channel_id, + topic: ti.name.clone(), + msg_type: ti.msg_type.clone(), + data: data.clone(), + log_time_ns: log_time, + }); + } + ReadMode::Lazy { chunk_indices, current_chunk_idx, current_messages, current_pos } => { + if *current_pos < current_messages.len() { + let (log_time, channel_id, ref data) = current_messages[*current_pos]; + *current_pos += 1; + // Skip messages with unknown channel_id instead of aborting iteration + let ti = match self.topics.get(&channel_id) { + Some(ti) => ti, + None => continue, + }; + return Some(RawMessage { + channel_id, + topic: ti.name.clone(), + msg_type: ti.msg_type.clone(), + data: data.clone(), + log_time_ns: log_time, + }); + } + // Current chunk exhausted — advance to next + *current_chunk_idx += 1; + if *current_chunk_idx >= chunk_indices.len() { + return None; + } + // Disjoint field borrows: self.mode fields vs self.mmap / self.topics + // Rust 2021 edition allows this in match arms + let new_msgs = load_chunk_static( + &self.mmap, + &self.topics, + chunk_indices, + *current_chunk_idx, + ).ok()?; + *current_messages = new_msgs; + *current_pos = 0; + // Loop back to read from newly loaded chunk + } + } } - - let (log_time, channel_id, ref data) = self.messages[self.current_pos]; - self.current_pos += 1; - - let topic_info = self.topics.get(&channel_id)?; - - Some(RawMessage { - channel_id, - topic: topic_info.name.clone(), - msg_type: topic_info.msg_type.clone(), - data: data.clone(), - log_time_ns: log_time, - }) } /// Seek to the nearest message at or after the given timestamp. pub fn seek_to(&mut self, target_ns: u64) -> Result<(), McapError> { - let pos = self.messages - .partition_point(|(t, _, _)| *t < target_ns); - self.current_pos = pos; - Ok(()) + match &mut self.mode { + ReadMode::Eager { messages, current_pos } => { + *current_pos = messages.partition_point(|(t, _, _)| *t < target_ns); + Ok(()) + } + ReadMode::Lazy { chunk_indices, current_chunk_idx, current_messages, current_pos } => { + // Binary search: find the last chunk whose start_time <= target_ns + let ci_idx = chunk_indices.partition_point(|ci| ci.message_start_time < target_ns); + let ci_idx = if ci_idx > 0 { ci_idx - 1 } else { 0 }; + + *current_chunk_idx = ci_idx; + let new_msgs = load_chunk_static( + &self.mmap, + &self.topics, + chunk_indices, + ci_idx, + )?; + *current_messages = new_msgs; + *current_pos = current_messages.partition_point(|(t, _, _)| *t < target_ns); + Ok(()) + } + } } /// Reset to the beginning. pub fn reset(&mut self) { - self.current_pos = 0; + match &mut self.mode { + ReadMode::Eager { current_pos, .. } => { + *current_pos = 0; + } + ReadMode::Lazy { chunk_indices, current_chunk_idx, current_messages, current_pos } => { + if chunk_indices.is_empty() { + *current_chunk_idx = 0; + *current_pos = 0; + return; + } + *current_chunk_idx = 0; + // Reload first chunk + let new_msgs = load_chunk_static( + &self.mmap, + &self.topics, + chunk_indices, + 0, + ).unwrap_or_default(); + *current_messages = new_msgs; + *current_pos = 0; + } + } } /// Current position as log_time_ns. pub fn position_ns(&self) -> u64 { - if self.current_pos > 0 && self.current_pos <= self.messages.len() { - self.messages[self.current_pos - 1].0 - } else if !self.messages.is_empty() { - self.messages[0].0 - } else { - 0 + match &self.mode { + ReadMode::Eager { messages, current_pos } => { + if *current_pos > 0 && *current_pos <= messages.len() { + messages[*current_pos - 1].0 + } else if !messages.is_empty() { + messages[0].0 + } else { + 0 + } + } + ReadMode::Lazy { current_messages, current_pos, .. } => { + if *current_pos > 0 && *current_pos <= current_messages.len() { + current_messages[*current_pos - 1].0 + } else if !current_messages.is_empty() { + current_messages[0].0 + } else { + self.start_time_ns + } + } } } @@ -225,6 +420,65 @@ impl McapReader { } pub fn is_exhausted(&self) -> bool { - self.current_pos >= self.messages.len() + match &self.mode { + ReadMode::Eager { messages, current_pos } => *current_pos >= messages.len(), + ReadMode::Lazy { chunk_indices, current_chunk_idx, current_messages, current_pos } => { + *current_pos >= current_messages.len() + && *current_chunk_idx + 1 >= chunk_indices.len() + } + } + } +} + +/// Load and decompress messages from one chunk by index. +/// Standalone function (not &self method) to allow disjoint field borrows in callers. +fn load_chunk_static( + mmap: &Mmap, + topics: &HashMap, + chunk_indices: &[records::ChunkIndex], + chunk_idx: usize, +) -> Result)>, McapError> { + let ci = chunk_indices.get(chunk_idx) + .ok_or_else(|| McapError::Mcap("chunk index out of bounds".into()))?; + let end_u64 = ci.chunk_start_offset + .checked_add(ci.chunk_length) + .ok_or_else(|| McapError::Mcap("chunk offset overflow".into()))?; + let start: usize = ci.chunk_start_offset + .try_into() + .map_err(|_| McapError::Mcap("chunk offset exceeds address space".into()))?; + let end: usize = end_u64 + .try_into() + .map_err(|_| McapError::Mcap("chunk end exceeds address space".into()))?; + + if end > mmap.len() { + return Err(McapError::Mcap("chunk offset out of bounds".into())); + } + + // Use LinearReader::sans_magic to read the chunk record at this offset + let slice = &mmap[start..end]; + let mut linear = LinearReader::sans_magic(slice); + + let (header, data) = match linear.next() { + Some(Ok(records::Record::Chunk { header, data })) => (header, data), + Some(Ok(_)) => return Err(McapError::Mcap("expected chunk record".into())), + Some(Err(e)) => return Err(McapError::Mcap(e.to_string())), + None => return Err(McapError::Mcap("empty chunk slice".into())), + }; + + // Decompress and iterate messages via ChunkReader + let chunk_reader = ChunkReader::new(header, &data) + .map_err(|e| McapError::Mcap(e.to_string()))?; + + let mut messages = Vec::new(); + for record in chunk_reader { + let record = record.map_err(|e| McapError::Mcap(e.to_string()))?; + if let records::Record::Message { header: msg_hdr, data: msg_data } = record { + if !topics.contains_key(&msg_hdr.channel_id) { + continue; // topic filter + } + messages.push((msg_hdr.log_time, msg_hdr.channel_id, msg_data.into_owned())); + } } + messages.sort_by_key(|(t, _, _)| *t); + Ok(messages) } diff --git a/src/scheduler.rs b/src/scheduler.rs index a7e8dc7..0593c90 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -85,7 +85,7 @@ impl Scheduler { // If start_time specified, seek to it if let Some(start_ns) = start_time_ns { let file_start = guard.reader.start_time_ns(); - let target = file_start + (start_ns as f64 * 1_000_000_000.0) as u64; + let target = file_start + start_ns; guard.reader.seek_to(target)?; guard.playback_start_log = target; } else if guard.state == PlaybackState::Idle || guard.state == PlaybackState::Finished { diff --git a/tests/bench_raw_reader.py b/tests/bench_raw_reader.py new file mode 100644 index 0000000..82924e6 --- /dev/null +++ b/tests/bench_raw_reader.py @@ -0,0 +1,144 @@ +"""Benchmark: McapRawReader vs rosbag2_py SequentialReader. + +Usage: + python tests/bench_raw_reader.py + python tests/bench_raw_reader.py tests/test_data.mcap +""" + +import sys +import time +import os + + +def bench_tachy_raw(path: str, runs: int = 3) -> dict: + """Benchmark McapRawReader sequential read.""" + from mcap_player._mcap_player_core import McapRawReader + + results = [] + for i in range(runs): + t0 = time.perf_counter() + reader = McapRawReader(path) + t_open = time.perf_counter() + + count = 0 + total_bytes = 0 + t_first = None + + for topic, msg_type, data, log_time_ns in reader: + if t_first is None: + t_first = time.perf_counter() + count += 1 + total_bytes += len(data) + + t_done = time.perf_counter() + + results.append({ + "open_ms": (t_open - t0) * 1000, + "first_msg_ms": (t_first - t0) * 1000 if t_first else None, + "read_ms": (t_done - t_open) * 1000, + "total_ms": (t_done - t0) * 1000, + "count": count, + "total_bytes": total_bytes, + "mbps": total_bytes / max(t_done - t_open, 1e-9) / 1e6, + }) + + return _summarize(results, "McapRawReader") + + +def bench_rosbag2_py(path: str, runs: int = 3) -> dict: + """Benchmark rosbag2_py SequentialReader.""" + try: + import rosbag2_py + except ImportError: + print("rosbag2_py not available — skipping comparison") + return None + + results = [] + for i in range(runs): + t0 = time.perf_counter() + + reader = rosbag2_py.SequentialReader() + storage_options = rosbag2_py.StorageOptions(uri=path, storage_id="mcap") + converter_options = rosbag2_py.ConverterOptions("", "") + reader.open(storage_options, converter_options) + + t_open = time.perf_counter() + + count = 0 + total_bytes = 0 + t_first = None + + while reader.has_next(): + topic, data, log_time_ns = reader.read_next() + if t_first is None: + t_first = time.perf_counter() + count += 1 + total_bytes += len(data) + + t_done = time.perf_counter() + + results.append({ + "open_ms": (t_open - t0) * 1000, + "first_msg_ms": (t_first - t0) * 1000 if t_first else None, + "read_ms": (t_done - t_open) * 1000, + "total_ms": (t_done - t0) * 1000, + "count": count, + "total_bytes": total_bytes, + "mbps": total_bytes / max(t_done - t_open, 1e-9) / 1e6, + }) + + return _summarize(results, "rosbag2_py") + + +def _summarize(results: list, name: str) -> dict: + """Compute averages over runs.""" + n = len(results) + avg = {} + for key in results[0]: + vals = [r[key] for r in results if r[key] is not None] + avg[key] = sum(vals) / len(vals) if vals else None + + print(f"\n{'='*50}") + print(f" {name} ({n} runs)") + print(f"{'='*50}") + print(f" Open: {avg['open_ms']:.2f} ms") + print(f" First message: {avg['first_msg_ms']:.2f} ms" if avg['first_msg_ms'] else " First message: N/A") + print(f" Read: {avg['read_ms']:.2f} ms") + print(f" Total: {avg['total_ms']:.2f} ms") + print(f" Messages: {avg['count']:.0f}") + print(f" Throughput: {avg['mbps']:.2f} MB/s") + print(f" Data: {avg['total_bytes']/1e6:.2f} MB") + + return avg + + +def main(): + if len(sys.argv) < 2: + path = "tests/test_data.mcap" + else: + path = sys.argv[1] + + if not os.path.exists(path): + print(f"File not found: {path}") + sys.exit(1) + + file_size_mb = os.path.getsize(path) / 1e6 + print(f"Benchmarking: {path} ({file_size_mb:.2f} MB)") + + tachy = bench_tachy_raw(path) + rosbag = bench_rosbag2_py(path) + + if tachy and rosbag: + print(f"\n{'='*50}") + print(f" COMPARISON") + print(f"{'='*50}") + speedup = rosbag["total_ms"] / max(tachy["total_ms"], 0.001) + print(f" Speedup (total): {speedup:.2f}x") + open_speedup = rosbag["open_ms"] / max(tachy["open_ms"], 0.001) + print(f" Speedup (open): {open_speedup:.2f}x") + read_speedup = rosbag["read_ms"] / max(tachy["read_ms"], 0.001) + print(f" Speedup (read): {read_speedup:.2f}x") + + +if __name__ == "__main__": + main() diff --git a/tests/bench_test.py b/tests/bench_test.py new file mode 100644 index 0000000..6dfd360 --- /dev/null +++ b/tests/bench_test.py @@ -0,0 +1,91 @@ +"""pytest-benchmark based benchmarks: McapRawReader vs rosbag2_py.""" + +import resource +import pytest +from mcap_player._mcap_player_core import McapRawReader + + +def _rss_kb(): + return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + + +def test_bench_tachy_open(benchmark, mcap_file): + def _open(): + McapRawReader(mcap_file) + + benchmark(_open) + benchmark.extra_info["peak_rss_kb"] = _rss_kb() + + +def test_bench_tachy_read_all(benchmark, mcap_file): + reader = McapRawReader(mcap_file) + + def _read_all(): + reader.reset() + count = 0 + total_bytes = 0 + for _topic, _msg_type, data, _log_time_ns in reader: + count += 1 + total_bytes += len(data) + return count, total_bytes + + count, total_bytes = benchmark(_read_all) + benchmark.extra_info["message_count"] = count + benchmark.extra_info["total_bytes"] = total_bytes + benchmark.extra_info["mbps"] = total_bytes / max(benchmark.stats["mean"], 1e-9) / 1e6 + benchmark.extra_info["peak_rss_kb"] = _rss_kb() + + +def test_bench_tachy_topic_filter(benchmark, mcap_file): + reader = McapRawReader(mcap_file) + topics = reader.topics() + first_topic = next(iter(topics)) if topics else None + + def _read_filtered(): + r = McapRawReader(mcap_file, topics=[first_topic] if first_topic else None) + count = 0 + total_bytes = 0 + for _topic, _msg_type, data, _log_time_ns in r: + count += 1 + total_bytes += len(data) + return count, total_bytes + + benchmark(_read_filtered) + benchmark.extra_info["filtered_topic"] = first_topic + benchmark.extra_info["peak_rss_kb"] = _rss_kb() + + +def test_bench_rosbag2_open(benchmark, mcap_file): + rosbag2_py = pytest.importorskip("rosbag2_py") + + def _open(): + reader = rosbag2_py.SequentialReader() + storage_options = rosbag2_py.StorageOptions(uri=mcap_file, storage_id="mcap") + converter_options = rosbag2_py.ConverterOptions("", "") + reader.open(storage_options, converter_options) + + benchmark(_open) + benchmark.extra_info["peak_rss_kb"] = _rss_kb() + + +def test_bench_rosbag2_read_all(benchmark, mcap_file): + rosbag2_py = pytest.importorskip("rosbag2_py") + + def _read_all(): + reader = rosbag2_py.SequentialReader() + storage_options = rosbag2_py.StorageOptions(uri=mcap_file, storage_id="mcap") + converter_options = rosbag2_py.ConverterOptions("", "") + reader.open(storage_options, converter_options) + count = 0 + total_bytes = 0 + while reader.has_next(): + _topic, data, _log_time_ns = reader.read_next() + count += 1 + total_bytes += len(data) + return count, total_bytes + + count, total_bytes = benchmark(_read_all) + benchmark.extra_info["message_count"] = count + benchmark.extra_info["total_bytes"] = total_bytes + benchmark.extra_info["mbps"] = total_bytes / max(benchmark.stats["mean"], 1e-9) / 1e6 + benchmark.extra_info["peak_rss_kb"] = _rss_kb() diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..a82c8a2 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,14 @@ +import pytest + + +def pytest_addoption(parser): + parser.addoption( + "--mcap-file", + default="tests/test_data.mcap", + help="Path to MCAP file for benchmarks", + ) + + +@pytest.fixture +def mcap_file(request): + return request.config.getoption("--mcap-file") diff --git a/tests/create_test_mcap.py b/tests/create_test_mcap.py index 2e4e30f..1187dea 100644 --- a/tests/create_test_mcap.py +++ b/tests/create_test_mcap.py @@ -64,5 +64,76 @@ def create_test_mcap(path: str, num_messages: int = 50): return path +def create_multi_chunk_mcap(path: str, num_messages: int = 200, chunk_size: int = 256): + """Create a test MCAP with many small chunks to exercise lazy chunk decoding.""" + with open(path, "wb") as f: + writer = Writer(f, chunk_size=chunk_size) + writer.start() + + schema_id = writer.register_schema( + name="std_msgs/msg/String", + encoding="ros2msg", + data=b"string data", + ) + channel_id = writer.register_channel( + topic="/test/chunked", + message_encoding="cdr", + schema_id=schema_id, + ) + + base_time = 1_000_000_000 + for i in range(num_messages): + timestamp = base_time + i * 10_000_000 # 10ms apart + data = struct.pack(" 0 + + print(f"PASS: iterated {len(messages)} messages") + + +def test_topics(): + """Topics metadata matches expected test file contents.""" + reader = McapRawReader(TEST_MCAP) + topics = reader.topics() + + assert "/test/string" in topics, f"Missing /test/string, got {topics.keys()}" + assert "/test/int" in topics, f"Missing /test/int, got {topics.keys()}" + + msg_type_str, encoding_str = topics["/test/string"] + assert msg_type_str == "std_msgs/msg/String" + + msg_type_int, encoding_int = topics["/test/int"] + assert msg_type_int == "std_msgs/msg/Int32" + + print(f"PASS: topics = {list(topics.keys())}") + + +def test_topic_filter(): + """Topic filtering works correctly.""" + reader = McapRawReader(TEST_MCAP, topics=["/test/string"]) + + messages = list(reader) + assert all(t == "/test/string" for t, _, _, _ in messages) + assert len(messages) == 25, f"Expected 25 string messages, got {len(messages)}" + + print(f"PASS: filtered to {len(messages)} messages") + + +def test_exclude_topics(): + """Exclude topics filtering works correctly.""" + reader = McapRawReader(TEST_MCAP, exclude_topics=["/test/int"]) + + messages = list(reader) + assert all(t == "/test/string" for t, _, _, _ in messages) + assert len(messages) == 25 + + print(f"PASS: excluded /test/int, got {len(messages)} messages") + + +def test_message_order(): + """Messages are sorted by log_time_ns.""" + reader = McapRawReader(TEST_MCAP) + + timestamps = [ts for _, _, _, ts in reader] + assert timestamps == sorted(timestamps), "Messages not in timestamp order" + + print(f"PASS: {len(timestamps)} messages in correct order") + + +def test_duration_and_count(): + """Duration and message count are correct.""" + reader = McapRawReader(TEST_MCAP) + + assert reader.message_count() == 50 + assert reader.duration_ns() > 0 + + print(f"PASS: count={reader.message_count()}, duration={reader.duration_ns()}ns") + + +def test_reset(): + """Reset allows re-reading all messages.""" + reader = McapRawReader(TEST_MCAP) + + first_pass = list(reader) + assert len(first_pass) == 50 + + # After exhaustion, should return empty + assert reader.next_raw_message() is None + + # Reset and read again + reader.reset() + second_pass = list(reader) + assert len(second_pass) == 50 + + # Verify same data + for (t1, mt1, d1, ts1), (t2, mt2, d2, ts2) in zip(first_pass, second_pass): + assert t1 == t2 + assert mt1 == mt2 + assert d1 == d2 + assert ts1 == ts2 + + print("PASS: reset and re-read produced identical results") + + +def test_seek(): + """Seek skips messages before the target time.""" + reader = McapRawReader(TEST_MCAP) + + # Read all to get timestamps + all_msgs = list(reader) + reader.reset() + + # Seek to ~halfway (messages are at 100ms intervals, 50 msgs = 5 sec) + reader.seek(2.5) + + remaining = list(reader) + assert len(remaining) < len(all_msgs), "Seek should skip some messages" + assert len(remaining) > 0, "Seek should not skip all messages" + + print(f"PASS: seek(2.5s) skipped to {len(remaining)} remaining messages") + + +def test_no_timing_delay(): + """Raw reader should be much faster than real-time (no timing delays).""" + reader = McapRawReader(TEST_MCAP) + + start = time.perf_counter() + count = 0 + for _ in reader: + count += 1 + elapsed = time.perf_counter() - start + + # Test file has 5 seconds of data at 10Hz + # Raw reading should complete in well under 1 second + assert elapsed < 1.0, f"Raw read took {elapsed:.3f}s — should be near-instant" + assert count == 50 + + print(f"PASS: read {count} messages in {elapsed*1000:.1f}ms (no timing delay)") + + +def test_consistency_with_mcap_core(): + """McapRawReader produces same data as McapCore (minus timing).""" + raw_reader = McapRawReader(TEST_MCAP) + raw_messages = list(raw_reader) + + core = McapCore(TEST_MCAP) + core.play(rate=1000.0) # Very fast playback + core_messages = [] + while True: + msg = core.next_message() + if msg is None: + break + core_messages.append(msg) + + assert len(raw_messages) == len(core_messages), ( + f"Count mismatch: raw={len(raw_messages)}, core={len(core_messages)}" + ) + + for (rt, rmt, rd, rts), (ct, cmt, cd, cts) in zip(raw_messages, core_messages): + assert rt == ct, f"Topic mismatch: {rt} vs {ct}" + assert rmt == cmt, f"MsgType mismatch: {rmt} vs {cmt}" + # McapCore returns list[int], McapRawReader returns bytes — normalize + assert bytes(rd) == bytes(cd), f"Data mismatch at {rt}" + assert rts == cts, f"Timestamp mismatch: {rts} vs {cts}" + + print(f"PASS: {len(raw_messages)} messages match between RawReader and Core") + + +MULTI_CHUNK_MCAP = "tests/test_multi_chunk.mcap" +NO_SUMMARY_MCAP = "tests/test_no_summary.mcap" + + +def _ensure_test_files(): + """Generate multi-chunk and no-summary test files if they don't exist.""" + import os + import sys + sys.path.insert(0, "tests") + from create_test_mcap import create_multi_chunk_mcap, create_no_summary_mcap + if not os.path.exists(MULTI_CHUNK_MCAP): + create_multi_chunk_mcap(MULTI_CHUNK_MCAP) + if not os.path.exists(NO_SUMMARY_MCAP): + create_no_summary_mcap(NO_SUMMARY_MCAP) + + +def test_multi_chunk_iteration(): + """Lazy chunk decoding: all messages are iterated correctly across chunk boundaries.""" + _ensure_test_files() + reader = McapRawReader(MULTI_CHUNK_MCAP) + + messages = list(reader) + assert len(messages) == 200, f"Expected 200 messages, got {len(messages)}" + + # All must be from the expected topic + for topic, msg_type, data, log_time_ns in messages: + assert topic == "/test/chunked", f"Unexpected topic: {topic}" + assert isinstance(data, bytes) + assert log_time_ns > 0 + + # Timestamps must be sorted + timestamps = [ts for _, _, _, ts in messages] + assert timestamps == sorted(timestamps), "Messages not in timestamp order across chunks" + + print(f"PASS: iterated {len(messages)} messages across multiple chunks") + + +def test_lazy_seek_across_chunks(): + """Seek lands in the correct chunk and returns the right message.""" + _ensure_test_files() + reader = McapRawReader(MULTI_CHUNK_MCAP) + + all_msgs = list(reader) + assert len(all_msgs) == 200 + + # Seek to ~halfway through the file (messages at 10ms intervals over 2 seconds) + reader.reset() + reader.seek(1.0) # 1 second from start + + remaining = list(reader) + assert len(remaining) < len(all_msgs), "Seek should skip some messages" + assert len(remaining) > 0, "Seek should leave some messages" + + # First remaining message should have timestamp >= seek target + # start_time is 1_000_000_000 ns, seek(1.0) adds 1_000_000_000 ns + target_ns = all_msgs[0][3] + 1_000_000_000 + first_ts = remaining[0][3] + assert first_ts >= target_ns, ( + f"First message after seek ({first_ts}) should be >= target ({target_ns})" + ) + + print(f"PASS: seek(1.0s) returned {len(remaining)} messages, first ts={remaining[0][3]}") + + +def test_lazy_chunk_fallback(): + """Files without a summary section fall back to eager mode and still read correctly.""" + _ensure_test_files() + reader = McapRawReader(NO_SUMMARY_MCAP) + + messages = list(reader) + assert len(messages) == 20, f"Expected 20 messages, got {len(messages)}" + + for topic, msg_type, data, log_time_ns in messages: + assert topic == "/test/nosummary" + assert isinstance(data, bytes) + + timestamps = [ts for _, _, _, ts in messages] + assert timestamps == sorted(timestamps), "Messages not in order in no-summary file" + + print(f"PASS: no-summary file read {len(messages)} messages via eager fallback") + + +if __name__ == "__main__": + tests = [ + test_basic_iteration, + test_topics, + test_topic_filter, + test_exclude_topics, + test_message_order, + test_duration_and_count, + test_reset, + test_seek, + test_no_timing_delay, + test_consistency_with_mcap_core, + ] + + passed = 0 + failed = 0 + for test in tests: + try: + test() + passed += 1 + except Exception as e: + print(f"FAIL: {test.__name__}: {e}") + failed += 1 + + print(f"\n{'='*50}") + print(f"Results: {passed} passed, {failed} failed, {len(tests)} total")