small-mcap

Lightweight Python library for reading and writing MCAP files.

Installation

uv add small-mcap

# With compression support
uv add small-mcap[compression]  # ZSTD + LZ4
uv add small-mcap[zstd]         # ZSTD only
uv add small-mcap[lz4]          # LZ4 only

Reader

Basic read

from small_mcap import read_message

with open("input.mcap", "rb") as f:
    for schema, channel, message in read_message(f):
        print(f"{channel.topic}: {message.data}")

Read multiple inputs

from small_mcap import read_message

with open("recording1.mcap", "rb") as f1, \
     open("recording2.mcap", "rb") as f2, \
     open("recording3.mcap", "rb") as f3:
    for schema, channel, message in read_message([f1, f2, f3]):
        print(f"{channel.topic}: {message.log_time}")

Read with topic filtering

from small_mcap import read_message, include_topics

with open("input.mcap", "rb") as f:
    topics = ["/camera/image", "/lidar/points"]
    for schema, channel, message in read_message(f, should_include=include_topics(topics)):
        print(f"{channel.topic}: {len(message.data)} bytes")

Read with time range

from small_mcap import read_message

with open("input.mcap", "rb") as f:
    start = 1000000000  # nanoseconds
    end = 2000000000
    for schema, channel, message in read_message(f, start_time_ns=start, end_time_ns=end):
        print(f"{channel.topic} at {message.log_time}")

Reuse a file across seeks

McapFile keeps the file summary and recently decompressed chunks available for repeated reads. Each iterator has independent position and ordering state.

from small_mcap import McapFile

with McapFile.open("input.mcap") as recording:
    forward = recording.read_message(start_time_ns=1_000_000_000)
    snapshot = recording.read_message(end_time_ns=2_000_000_000, reverse=True)

Recovery of an incomplete file is opt-in. It reconstructs chunk indexes in memory, preserves complete on-disk message indexes, ignores only an incomplete tail, and never modifies the source:

with McapFile.open("incomplete.mcap", recover=True) as recording:
    print(recording.is_recovered, recording.supports_reverse)
    for schema, channel, message in recording.read_message():
        ...

Follow an append-only file

McapFollower polls a local growing file without blocking. Every poll has finite message and byte budgets; partial record headers and bodies remain uncommitted until a later poll completes them.

from small_mcap import McapFollower

with McapFollower.open("recording.mcap", validate_crc=True) as follower:
    while True:
        batch = follower.poll_messages(max_messages=1000, max_bytes=16 * 1024 * 1024)
        for schema, channel, message in batch.messages:
            consume(channel.topic, message)
        if batch.is_final:
            break

The follower emits a complete chunk without waiting for its message indexes and preserves schemas and channels across polls. Footer plus trailing magic marks the file final. Truncation and inode replacement raise McapFileTruncatedError and McapFileReplacedError; the follower never silently reopens a different file.

Read decoded messages

from small_mcap import read_message_decoded
import json

class JsonDecoderFactory:
    def decoder_for(self, message_encoding, schema):
        if message_encoding == "json":
            return lambda data: json.loads(bytes(data))
        return None

with open("input.mcap", "rb") as f:
    for msg in read_message_decoded(f, decoder_factories=[JsonDecoderFactory()]):
        print(f"{msg.channel.topic}: {msg.decoded_message}")

Read summary/metadata

from small_mcap import get_summary, get_header

with open("input.mcap", "rb") as f:
    summary = get_summary(f)
    print(f"Messages: {summary.statistics.message_count}")
    print(f"Duration: {summary.statistics.message_start_time} - {summary.statistics.message_end_time}")

    for channel in summary.channels.values():
        print(f"  {channel.topic}: {channel.message_encoding}")

Writer

Basic write

from small_mcap import McapWriter

with open("output.mcap", "wb") as f:
    writer = McapWriter(f)
    writer.start(profile="", library="my-app")

    # Add schema
    schema_id = 1
    writer.add_schema(schema_id, "MySchema", "json", b'{"type": "object"}')

    # Add channel
    channel_id = 1
    writer.add_channel(channel_id, "/my/topic", "json", schema_id)

    # Add messages
    for i in range(100):
        writer.add_message(
            channel_id,
            log_time=i * 1000000,  # nanoseconds
            data=b'{"value": 42}',
            publish_time=i * 1000000
        )

    writer.finish()

Write with compression

from small_mcap import McapWriter, CompressionType

with open("output.mcap", "wb") as f:
    writer = McapWriter(
        f,
        compression=CompressionType.ZSTD,
        chunk_size=1024 * 1024  # 1MB chunks
    )
    writer.start(profile="", library="my-app")

    schema_id = 1
    writer.add_schema(schema_id, "MySchema", "json", b"{}")
    channel_id = 1
    writer.add_channel(channel_id, "/topic", "json", schema_id)

    for i in range(1000):
        writer.add_message(channel_id, log_time=i*1000, data=b"data", publish_time=i*1000)

    writer.finish()

Write with encoder factory

from small_mcap import McapWriter
import json

class JsonEncoderFactory:
    """Implements EncoderFactoryProtocol for JSON messages."""
    profile = ""
    encoding = "jsonschema"
    message_encoding = "json"

    def encoder_for(self, schema):
        return lambda msg: json.dumps(msg).encode()

with open("output.mcap", "wb") as f:
    writer = McapWriter(f, encoder_factory=JsonEncoderFactory())
    writer.start(profile="", library="my-app")

    schema_id = 1
    writer.add_schema(schema_id, "SensorData", "jsonschema", b'{"type": "object"}')

    channel_id = 1
    writer.add_channel(channel_id, "/sensor/data", "json", schema_id)

    for i in range(100):
        msg = {"timestamp": i, "value": i * 2}
        writer.add_message_encode(channel_id, i * 1000, msg, publish_time=i * 1000)

    writer.finish()

Features

• Zero dependencies for core functionality
• Optional compression support (ZSTD, LZ4)
• Lazy chunk loading for efficient memory usage
• Topic and time-range filtering
• Automatic schema/channel registration
• CRC validation
• Fast summary/metadata access

Performance

small-mcap is optimized for high-performance MCAP file reading with zero-copy operations and lazy chunk loading:

Key Optimizations:

• Zero-copy memory access: Uses memoryview to avoid unnecessary data copies
• Lazy chunk loading: Only decompresses chunks when needed
• Parallel chunk decompression: num_workers threads decompress chunks ahead of the reader (zstd/lz4 release the GIL)
• Binary search: Efficient time-range filtering using chunk indexes
• Heap-based merging: Optimal multi-file reading with automatic ID remapping

Parallel Prefetch (num_workers)

Pass num_workers to read_message to decompress chunks in parallel using a thread pool. The main thread reads raw bytes sequentially while worker threads decompress ahead.

with open("large.mcap", "rb") as f:
    for schema, channel, message in read_message(f, num_workers=4):
        ...

Benchmarked on the included nuScenes MCAP file (431 MB, 560 zstd chunks, 30,900 messages; median of 5 runs):

Comparison with other libraries:

Benchmarks

Median runtime from pytest-benchmark on the included nuScenes dataset (data/data/nuScenes-v1.0-mini-scene-0061-ros2.mcap, 30,900 messages, 19.15s duration, 560 zstd chunks):

Note: rosbags and pybag require seekable streams and are skipped for the non-seekable cases.

Summary:

• small-mcap was fastest in all six scenarios
• 1.17-1.24x faster than mcap (official) across all scenarios
• 1.06-4.02x faster than rosbags where rosbags supports the scenario
• 1.20-1.31x faster than pybag on seekable streams

Links

• MCAP Specification