MCAP protobuf: How to index .mcap files?

[aeon0]

I am currently using the from mcap_protobuf.writer import Writer to create an .mcap foxglove visu file from protobuf messages.

I can write a bunch of messages and finish() it an that works. But I need to index the messages because we have some big .mcap files which can not be opened otherwise (<1gb). These .mcap files will show:

It seems that finish() does add an chunk index. But I am not really sure how to debug why this is not properly indexing the file.

[achim-k]

How are you writing the mcap file? Could you share some of your code?

Maybe also take a look at the following protobuf examples which outputs an indexed mcap file:
https://github.com/foxglove/mcap/blob/main/python/examples/protobuf/point_cloud_example.py

[aeon0]

That is pretty much how I create the mcap files. In my use-case I write to the mcap file from a callback function, thus I save the mcap writer in a member variable and reuse it.

import time
from typing import Dict
from pathlib import Path
from mcap_protobuf.writer import Writer

class SomeClass:
    # ....
    def __init__(self):
        self.rec_files = []

    def write_messages(self, data: Dict[str, any], rec_path: Path):
        """Write a message to a mcap foxglove rec file

        Args:
            data (Dict[str, any]): data that should be sent in a dict with keys being topic_name and
                value being the portobuf data
            rec_path (Path): Path of the .mcap rec file
        """
        if rec_path.suffix != ".mcap":
            rec_path = rec_path.with_suffix(".mcap")
        if rec_path in self.rec_files:
            mcap_writer = self.rec_files[rec_path]
        else:
            f = open(rec_path, "wb")
            mcap_writer = Writer(f)
            self.rec_files[rec_path] = mcap_writer
        for topic_name, proto_msg in data.items():
            mcap_writer.write_message(
                topic=topic_name,
                message=proto_msg,
                log_time=time.time_ns(),
                publish_time=time.time_ns(),
            )

    def write_finish(self):
        for _, writer in self.rec_files.items():
            writer.finish()

def some_callback_batch_end(results):
  # ....
  some_class.write_messages(data, rec_path)

def some_callback_run_end():
  some_class.write_finish()

[aeon0]

It works if I dont write the data each callback but instead just keep the data in memory and write everything at the end:

    def write_messages(self, data: Dict[str, any], rec_path: Path):
        """Write a message to a mcap foxglove rec file

        Args:
            data (Dict[str, any]): data that should be sent in a dict with keys being topic_name and
                value being the portobuf data
            rec_path (Path): Path of the .mcap rec file
        """
        if rec_path.suffix != ".mcap":
            rec_path = rec_path.with_suffix(".mcap")
        if rec_path not in self.data_arr:
            self.data_arr[rec_path] = []
        self.data_arr[rec_path].append(data)

    def write_finish(self):
        for rec_path, data_arr in self.data_arr.items():
            with open(rec_path, "wb") as f, Writer(f) as mcap_writer:
                for data in data_arr:
                    for topic_name, proto_msg in data.items():
                        mcap_writer.write_message(
                            topic=topic_name,
                            message=proto_msg,
                            log_time=time.time_ns(),
                            publish_time=time.time_ns(),
                        )

But to me the two implementations should have the same result. Since the __exit__() on the Writer just calls finish(). The data can get quite large and I don't necessarily want to keep it in memory.

[achim-k]

But to me the two implementations should have the same result. Since the exit() on the Writer just calls finish()

Yes, both implementations should work. But the first implementation is more prone to race conditions e.g. when write_messages is called in parallel or when write_finish is called before the very last write_messages, ... . Can you confirm that this is not the problem?

[aeon0]

Its ensured that this doesnt happen in the code that calls the methods. But maybe should also put a check in the class itself. Good point.