# MCAP

MCAP is a modular container format and logging library for pub/sub messages with arbitrary message serialization. It is primarily designed for robotics applications and works well under various workloads, resource constraints, and durability requirements. The format supports efficient random access, indexing, and compression while remaining serialization-agnostic, meaning it can store messages encoded in any format including Protobuf, JSON, ROS messages, FlatBuffers, and more.

The MCAP library provides implementations in multiple languages including Python, Go, TypeScript/JavaScript, C++, Rust, and Swift. Each implementation offers APIs for reading and writing MCAP files with support for chunking, compression (zstd, lz4), and automatic indexing. The format's self-describing nature with embedded schemas makes files fully interpretable without external dependencies, while the optional summary section enables fast random access to messages by timestamp or topic without scanning the entire file.

## Python Writer API

The Python Writer class provides methods for creating MCAP files with schemas, channels, messages, metadata, and attachments. It supports chunking with configurable compression (zstd, lz4) and automatic index generation for efficient random access reading.

```python
import json
from time import time_ns
from mcap.writer import Writer, CompressionType

# Create an MCAP file with JSON-encoded messages
with open("output.mcap", "wb") as stream:
    writer = Writer(
        stream,
        chunk_size=1024 * 1024,  # 1MB chunks
        compression=CompressionType.ZSTD
    )

    writer.start(profile="", library="my-app v1.0")

    # Register a schema (returns schema_id)
    schema_id = writer.register_schema(
        name="sensor_data",
        encoding="jsonschema",
        data=json.dumps({
            "type": "object",
            "properties": {
                "temperature": {"type": "number"},
                "humidity": {"type": "number"},
                "timestamp": {"type": "integer"}
            }
        }).encode()
    )

    # Register a channel (returns channel_id)
    channel_id = writer.register_channel(
        topic="/sensors/environmental",
        message_encoding="json",
        schema_id=schema_id,
        metadata={"sensor_type": "DHT22", "location": "room1"}
    )

    # Write messages to the channel
    for i in range(100):
        message_data = json.dumps({
            "temperature": 22.5 + i * 0.1,
            "humidity": 45.0 + i * 0.5,
            "timestamp": time_ns()
        }).encode()

        writer.add_message(
            channel_id=channel_id,
            log_time=time_ns(),
            publish_time=time_ns(),
            data=message_data,
            sequence=i
        )

    # Add metadata
    writer.add_metadata(
        name="recording_info",
        data={"operator": "test_user", "environment": "lab"}
    )

    # Add attachment
    writer.add_attachment(
        log_time=time_ns(),
        create_time=time_ns(),
        name="calibration.json",
        media_type="application/json",
        data=b'{"offset": 0.5}'
    )

    writer.finish()
```

## Python Reader API

The McapReader provides efficient message iteration with support for topic filtering, time range queries, and optional decoded message iteration when decoder factories are provided.

```python
from mcap.reader import make_reader

# Basic reading - iterate all messages
with open("input.mcap", "rb") as f:
    reader = make_reader(f)

    # Get file summary (statistics, channels, schemas)
    summary = reader.get_summary()
    if summary and summary.statistics:
        print(f"Total messages: {summary.statistics.message_count}")
        print(f"Duration: {summary.statistics.message_end_time - summary.statistics.message_start_time}ns")

    # Iterate messages with filtering
    for schema, channel, message in reader.iter_messages(
        topics=["/sensors/environmental", "/camera/image"],
        start_time=1700000000000000000,  # nanoseconds
        end_time=1700001000000000000,
        log_time_order=True,  # Sort by timestamp
        reverse=False
    ):
        print(f"Topic: {channel.topic}")
        print(f"Schema: {schema.name if schema else 'None'}")
        print(f"Log time: {message.log_time}")
        print(f"Data size: {len(message.data)} bytes")

# Reading attachments and metadata
with open("input.mcap", "rb") as f:
    reader = make_reader(f)

    for attachment in reader.iter_attachments():
        print(f"Attachment: {attachment.name} ({attachment.media_type})")
        print(f"Size: {len(attachment.data)} bytes")

    for metadata in reader.iter_metadata():
        print(f"Metadata '{metadata.name}': {metadata.metadata}")
```

## Go Writer API

The Go Writer provides a high-performance API for creating MCAP files with configurable chunking, compression, and indexing options.

```go
package main

import (
    "bytes"
    "os"
    "time"

    "github.com/foxglove/mcap/go/mcap"
)

func main() {
    file, _ := os.Create("output.mcap")
    defer file.Close()

    writer, _ := mcap.NewWriter(file, &mcap.WriterOptions{
        Chunked:          true,
        ChunkSize:        1024 * 1024, // 1MB
        Compression:      mcap.CompressionZSTD,
        CompressionLevel: mcap.CompressionLevelDefault,
        IncludeCRC:       true,
    })

    // Write header
    writer.WriteHeader(&mcap.Header{
        Profile: "ros1",
        Library: "my-go-app v1.0",
    })

    // Write schema
    schema := &mcap.Schema{
        ID:       1,
        Name:     "sensor_msgs/Temperature",
        Encoding: "ros1msg",
        Data:     []byte("float64 temperature\nstring frame_id"),
    }
    writer.WriteSchema(schema)

    // Write channel
    channel := &mcap.Channel{
        ID:              0,
        SchemaID:        1,
        Topic:           "/temperature",
        MessageEncoding: "ros1",
        Metadata:        map[string]string{"sensor": "thermocouple"},
    }
    writer.WriteChannel(channel)

    // Write messages
    for i := 0; i < 1000; i++ {
        timestamp := uint64(time.Now().UnixNano())
        data := []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x59, 0x40} // 100.0 as float64

        writer.WriteMessage(&mcap.Message{
            ChannelID:   0,
            Sequence:    uint32(i),
            LogTime:     timestamp,
            PublishTime: timestamp,
            Data:        data,
        })
    }

    // Write metadata
    writer.WriteMetadata(&mcap.Metadata{
        Name:     "recording_metadata",
        Metadata: map[string]string{"robot": "robot1", "trial": "001"},
    })

    // Write attachment
    attachmentData := bytes.NewReader([]byte(`{"calibration": "data"}`))
    writer.WriteAttachment(&mcap.Attachment{
        LogTime:    uint64(time.Now().UnixNano()),
        CreateTime: uint64(time.Now().UnixNano()),
        Name:       "calibration.json",
        MediaType:  "application/json",
        DataSize:   uint64(attachmentData.Len()),
        Data:       attachmentData,
    })

    writer.Close()
}
```

## Go Reader API

The Go Reader supports both indexed and unindexed reading modes, with options for topic filtering, time range queries, and message ordering.

```go
package main

import (
    "fmt"
    "io"
    "os"

    "github.com/foxglove/mcap/go/mcap"
)

func main() {
    file, _ := os.Open("input.mcap")
    defer file.Close()

    reader, _ := mcap.NewReader(file)
    defer reader.Close()

    // Get file info (requires seekable reader)
    info, _ := reader.Info()
    fmt.Printf("Messages: %d\n", info.Statistics.MessageCount)
    fmt.Printf("Channels: %d\n", info.Statistics.ChannelCount)
    fmt.Printf("Duration: %d ns\n",
        info.Statistics.MessageEndTime - info.Statistics.MessageStartTime)

    // List channels and schemas
    for id, channel := range info.Channels {
        fmt.Printf("Channel %d: %s\n", id, channel.Topic)
        if schema, ok := info.Schemas[channel.SchemaID]; ok {
            fmt.Printf("  Schema: %s (%s)\n", schema.Name, schema.Encoding)
        }
    }

    // Iterate messages with options
    iterator, _ := reader.Messages(
        mcap.WithTopics([]string{"/camera/image", "/lidar/points"}),
        mcap.WithStart(1700000000000000000),
        mcap.WithEnd(1700001000000000000),
        mcap.InOrder(mcap.LogTimeOrder),
    )

    for {
        schema, channel, message, err := iterator.Next(nil)
        if err == io.EOF {
            break
        }
        if err != nil {
            panic(err)
        }

        fmt.Printf("[%d] %s: %d bytes\n",
            message.LogTime, channel.Topic, len(message.Data))
        _ = schema // Use schema to decode message.Data
    }

    // Read attachments
    for _, attachmentIndex := range info.AttachmentIndexes {
        attachmentReader, _ := reader.GetAttachmentReader(attachmentIndex.Offset)
        data, _ := io.ReadAll(attachmentReader.Data())
        fmt.Printf("Attachment: %s (%s) - %d bytes\n",
            attachmentReader.Name, attachmentReader.MediaType, len(data))
    }
}
```

## TypeScript/JavaScript Writer API

The McapWriter class provides an async API for creating MCAP files in Node.js and browser environments, with support for chunking, compression, and automatic indexing.

```typescript
import { McapWriter } from "@mcap/core";
import { FileHandleWritable } from "@mcap/nodejs";
import * as fs from "fs";

async function writeExample() {
  const fileHandle = await fs.promises.open("output.mcap", "w");
  const writable = new FileHandleWritable(fileHandle);

  const writer = new McapWriter({
    writable,
    useStatistics: true,
    useSummaryOffsets: true,
    useChunks: true,
    chunkSize: 1024 * 1024,
    compressChunk: (data: Uint8Array) => ({
      compression: "zstd",
      compressedData: zstdCompress(data), // Provide your compressor
    }),
  });

  await writer.start({
    profile: "",
    library: "my-ts-app v1.0",
  });

  // Register schema
  const schemaId = await writer.registerSchema({
    name: "geometry_msgs/Point",
    encoding: "ros2msg",
    data: new TextEncoder().encode("float64 x\nfloat64 y\nfloat64 z"),
  });

  // Register channel
  const channelId = await writer.registerChannel({
    schemaId,
    topic: "/robot/position",
    messageEncoding: "cdr",
    metadata: new Map([["frame_id", "world"]]),
  });

  // Write messages
  const encoder = new TextEncoder();
  for (let i = 0; i < 100; i++) {
    const timestamp = BigInt(Date.now()) * 1_000_000n; // Convert to nanoseconds
    await writer.addMessage({
      channelId,
      sequence: i,
      logTime: timestamp,
      publishTime: timestamp,
      data: encoder.encode(JSON.stringify({ x: i, y: i * 2, z: 0 })),
    });
  }

  // Add metadata
  await writer.addMetadata({
    name: "session_info",
    metadata: new Map([
      ["session_id", "abc123"],
      ["user", "operator1"],
    ]),
  });

  // Add attachment
  await writer.addAttachment({
    logTime: BigInt(Date.now()) * 1_000_000n,
    createTime: BigInt(Date.now()) * 1_000_000n,
    name: "config.yaml",
    mediaType: "application/yaml",
    data: encoder.encode("robot_name: my_robot\nmax_speed: 1.5"),
  });

  await writer.end();
  await fileHandle.close();
}
```

## TypeScript/JavaScript Indexed Reader API

The McapIndexedReader provides efficient random-access reading using the file's index, with support for topic filtering, time ranges, and async iteration.

```typescript
import { McapIndexedReader } from "@mcap/core";
import { FileHandleReadable } from "@mcap/nodejs";
import * as fs from "fs";
import { decompress as zstdDecompress } from "@foxglove/wasm-zstd";

async function readExample() {
  const fileHandle = await fs.promises.open("input.mcap", "r");
  const readable = new FileHandleReadable(fileHandle);

  const reader = await McapIndexedReader.Initialize({
    readable,
    decompressHandlers: {
      zstd: (buffer, decompressedSize) =>
        zstdDecompress(buffer, Number(decompressedSize)),
      lz4: (buffer, decompressedSize) =>
        lz4Decompress(buffer, Number(decompressedSize)),
    },
  });

  // Access file metadata
  console.log(`Profile: ${reader.header.profile}`);
  console.log(`Library: ${reader.header.library}`);

  if (reader.statistics) {
    console.log(`Total messages: ${reader.statistics.messageCount}`);
    console.log(`Channels: ${reader.statistics.channelCount}`);
    console.log(`Schemas: ${reader.statistics.schemaCount}`);
  }

  // List channels and schemas
  for (const [id, channel] of reader.channelsById) {
    console.log(`Channel ${id}: ${channel.topic}`);
    const schema = reader.schemasById.get(channel.schemaId);
    if (schema) {
      console.log(`  Schema: ${schema.name} (${schema.encoding})`);
    }
  }

  // Read messages with filtering
  const topics = ["/camera/image", "/lidar/points"];
  for await (const message of reader.readMessages({
    topics,
    startTime: 1700000000000000000n,
    endTime: 1700001000000000000n,
    reverse: false,
    validateCrcs: true,
  })) {
    const channel = reader.channelsById.get(message.channelId)!;
    console.log(`[${message.logTime}] ${channel.topic}: ${message.data.length} bytes`);
  }

  // Read metadata
  for await (const metadata of reader.readMetadata({ name: "session_info" })) {
    console.log(`Metadata '${metadata.name}':`, Object.fromEntries(metadata.metadata));
  }

  // Read attachments
  for await (const attachment of reader.readAttachments({
    mediaType: "application/json",
    validateCrcs: true,
  })) {
    console.log(`Attachment: ${attachment.name} (${attachment.mediaType})`);
    console.log(`  Data: ${new TextDecoder().decode(attachment.data)}`);
  }

  await fileHandle.close();
}
```

## Rust Writer API

The Rust Writer provides a zero-copy API for creating MCAP files with strong type safety and support for async I/O via the tokio feature.

```rust
use std::{collections::BTreeMap, fs::File, io::BufWriter};
use mcap::{Channel, Writer, WriteOptions, Compression, records::MessageHeader};

fn write_example() -> mcap::McapResult<()> {
    let file = BufWriter::new(File::create("output.mcap")?);

    let mut writer = WriteOptions::new()
        .profile("ros2")
        .compression(Some(Compression::Zstd))
        .chunk_size(Some(1024 * 1024))
        .create(file)?;

    // Add schema (returns schema_id)
    let schema_id = writer.add_schema(
        "sensor_msgs/msg/PointCloud2",
        "ros2msg",
        b"# Point cloud message definition...",
    )?;

    // Add channel (returns channel_id)
    let mut metadata = BTreeMap::new();
    metadata.insert("frame_id".to_string(), "lidar_link".to_string());

    let channel_id = writer.add_channel(
        schema_id,
        "/lidar/points",
        "cdr",
        &metadata,
    )?;

    // Write messages
    for i in 0..1000u32 {
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos() as u64;

        let message_data = vec![0u8; 1024]; // Your serialized message

        writer.write_to_known_channel(
            &MessageHeader {
                channel_id,
                sequence: i,
                log_time: timestamp,
                publish_time: timestamp,
            },
            &message_data,
        )?;
    }

    // Write attachment
    writer.attach(
        "calibration.yaml",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos() as u64,
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos() as u64,
        "application/yaml",
        b"camera_matrix: [1, 0, 0, 0, 1, 0, 0, 0, 1]",
    )?;

    writer.finish()?;
    Ok(())
}
```

## Rust Reader API

The Rust MessageStream provides iterator-based access to messages with automatic schema and channel resolution, supporting both memory-mapped and streaming reads.

```rust
use std::fs;
use memmap2::Mmap;
use mcap::{MessageStream, Summary};

fn read_example() -> mcap::McapResult<()> {
    // Memory-map the file for efficient reading
    let file = fs::File::open("input.mcap")?;
    let mapped = unsafe { Mmap::map(&file)? };

    // Get file summary
    let summary = Summary::read(&mapped)?;
    if let Some(stats) = &summary.statistics {
        println!("Messages: {}", stats.message_count);
        println!("Channels: {}", stats.channel_count);
        println!("Schemas: {}", stats.schema_count);
        println!("Duration: {} ns", stats.message_end_time - stats.message_start_time);
    }

    // List channels and schemas
    for (id, channel) in &summary.channels {
        println!("Channel {}: {} ({})", id, channel.topic, channel.message_encoding);
        if let Some(schema) = &channel.schema {
            println!("  Schema: {} ({})", schema.name, schema.encoding);
        }
    }

    // Iterate messages
    for message_result in MessageStream::new(&mapped)? {
        let message = message_result?;

        println!(
            "[{}] {}: {} bytes (seq {})",
            message.log_time,
            message.channel.topic,
            message.data.len(),
            message.sequence
        );

        // Access schema for decoding
        if let Some(schema) = &message.channel.schema {
            println!("  Schema: {} ({})", schema.name, schema.encoding);
        }
    }

    // Read attachments
    for attachment in summary.attachments.iter() {
        println!(
            "Attachment: {} ({}) - {} bytes",
            attachment.name,
            attachment.media_type,
            attachment.data.len()
        );
    }

    Ok(())
}
```

## MCAP CLI Tool

The MCAP command-line tool provides utilities for inspecting, converting, merging, and manipulating MCAP files. It supports local files and remote files in S3/GCS.

```bash
# Installation
brew install mcap
# Or download from https://github.com/foxglove/mcap/releases

# Get file information
mcap info recording.mcap
# Output:
# library: mcap go v1.0.0
# profile: ros2
# messages: 15234
# duration: 120.5s
# channels:
#   (0) /camera/image     1000 msgs (8.30 Hz) : sensor_msgs/msg/Image [cdr]
#   (1) /lidar/points     2400 msgs (19.92 Hz): sensor_msgs/msg/PointCloud2 [cdr]

# Convert ROS bag to MCAP
mcap convert recording.bag output.mcap
mcap convert recording.db3 output.mcap --ament-prefix-path /opt/ros/humble

# Echo messages as JSON
mcap cat recording.mcap --topics /camera/info --json
# Output:
# {"topic":"/camera/info","sequence":1,"log_time":1700000000000000000,...}

# Filter messages to new file
mcap filter recording.mcap filtered.mcap \
  --include-topics /camera/image,/lidar/points \
  --start 1700000000000000000 \
  --end 1700000100000000000

# Merge multiple files
mcap merge input1.mcap input2.mcap -o merged.mcap

# List chunks in file
mcap list chunks recording.mcap
# Output:
# offset    length   start              end                compression  ratio
# 43        4529455  1700000000000000   1700000001000000   zstd        0.48

# Compress/decompress
mcap compress input.mcap -o compressed.mcap --compression zstd
mcap decompress compressed.mcap -o uncompressed.mcap

# Check file integrity
mcap doctor recording.mcap

# Recover data from corrupt file
mcap recover corrupt.mcap -o recovered.mcap

# Remote file support (S3/GCS)
mcap info gs://bucket/recording.mcap
AWS_REGION=us-west-2 mcap info s3://bucket/recording.mcap

# Add attachment to existing file
mcap add attachment recording.mcap \
  --name calibration.json \
  --media-type application/json \
  --file calibration.json

# Add metadata to existing file
mcap add metadata recording.mcap \
  --name session_info \
  --key operator --value "user1" \
  --key trial --value "001"
```

## C++ Reader/Writer API

The C++ library provides a header-only implementation for reading and writing MCAP files with minimal dependencies.

```cpp
#include <mcap/mcap.hpp>
#include <fstream>
#include <chrono>

// Writing MCAP files
void writeExample() {
    mcap::McapWriter writer;

    mcap::McapWriterOptions options("my-cpp-app");
    options.compression = mcap::Compression::Zstd;
    options.chunkSize = 1024 * 1024;

    std::ofstream file("output.mcap", std::ios::binary);
    writer.open(file, options);

    // Register schema
    mcap::Schema schema;
    schema.name = "sensor_msgs/msg/Temperature";
    schema.encoding = "ros2msg";
    schema.data = reinterpret_cast<const std::byte*>("float64 temperature");
    writer.addSchema(schema);

    // Register channel
    mcap::Channel channel;
    channel.topic = "/temperature";
    channel.messageEncoding = "cdr";
    channel.schemaId = schema.id;
    channel.metadata["sensor_type"] = "thermocouple";
    writer.addChannel(channel);

    // Write messages
    for (int i = 0; i < 1000; ++i) {
        auto now = std::chrono::system_clock::now();
        auto timestamp = std::chrono::duration_cast<std::chrono::nanoseconds>(
            now.time_since_epoch()).count();

        std::vector<std::byte> data(8);
        double temp = 25.0 + i * 0.1;
        std::memcpy(data.data(), &temp, sizeof(temp));

        mcap::Message msg;
        msg.channelId = channel.id;
        msg.sequence = i;
        msg.logTime = timestamp;
        msg.publishTime = timestamp;
        msg.data = data.data();
        msg.dataSize = data.size();

        writer.write(msg);
    }

    // Add attachment
    mcap::Attachment attachment;
    attachment.name = "config.json";
    attachment.mediaType = "application/json";
    attachment.logTime = std::chrono::duration_cast<std::chrono::nanoseconds>(
        std::chrono::system_clock::now().time_since_epoch()).count();
    std::string configData = R"({"setting": "value"})";
    attachment.data = reinterpret_cast<const std::byte*>(configData.data());
    attachment.dataSize = configData.size();
    writer.write(attachment);

    writer.close();
}

// Reading MCAP files
void readExample() {
    std::ifstream file("input.mcap", std::ios::binary);
    mcap::McapReader reader;

    auto status = reader.open(file);
    if (!status.ok()) {
        std::cerr << "Failed to open: " << status.message << std::endl;
        return;
    }

    // Get summary statistics
    auto summary = reader.readSummary(mcap::ReadSummaryMethod::AllowFallbackScan);
    if (summary.statistics) {
        std::cout << "Messages: " << summary.statistics->messageCount << std::endl;
        std::cout << "Channels: " << summary.statistics->channelCount << std::endl;
    }

    // List channels
    for (const auto& [id, channel] : reader.channels()) {
        std::cout << "Channel " << id << ": " << channel->topic << std::endl;
        if (channel->schemaId != 0) {
            auto schema = reader.schema(channel->schemaId);
            if (schema) {
                std::cout << "  Schema: " << schema->name << std::endl;
            }
        }
    }

    // Read messages
    mcap::ReadMessageOptions options;
    options.topics = {"/camera/image", "/lidar/points"};
    options.startTime = 1700000000000000000;
    options.endTime = 1700001000000000000;

    auto messageView = reader.readMessages([](const auto&) {}, options);
    for (auto it = messageView.begin(); it != messageView.end(); ++it) {
        const auto& msg = it->message;
        const auto& channel = it->channel;

        std::cout << "[" << msg.logTime << "] "
                  << channel->topic << ": "
                  << msg.dataSize << " bytes" << std::endl;
    }

    reader.close();
}
```

## Summary

MCAP serves as a universal container format for robotics data logging, providing a standardized way to store timestamped pub/sub messages regardless of the underlying serialization format. Its primary use cases include ROS bag replacement with better compression and random access, multi-sensor data recording in autonomous vehicles and robots, data interchange between different robotics frameworks, and long-term archival of experimental data. The format's self-describing nature with embedded schemas makes files portable across systems and tools.

Integration patterns typically involve using MCAP as the storage layer while keeping existing message serialization (Protobuf, ROS messages, JSON, etc.). Writers register schemas and channels once, then efficiently stream messages with automatic chunking and indexing. Readers can perform random access queries by topic and time range without scanning entire files. The CLI tool enables inspection and manipulation without writing code, while the multi-language library support allows integration into existing robotics stacks. The format is particularly well-suited for Foxglove visualization and analysis workflows, but remains open and interoperable with any tools that implement the specification.