[Python] Extend API to obtain the msg_count of a given topic in a .mcap file

[nachovizzo]

So far, I couldn't get the number of messages from a given topic in a .mcap ROS2 bagfile. Using mcap info and ros2 bag info -s mcap works with no problems, but I need to do this inside a python script, and thus, would love to have it as part of the API

For those looking for a hacky solution, I'm now doing this ugly thing:

@staticmethod
def get_n_scans(mcap_file: str, topic: str) -> int:
    """TODO: Replace this horrible function by some pragmatically way of getting the msg_count
    from the mcap file. As far as I checked, the mcap python API from ROS2 or foxglove does not
    provide any function on their APIs to do this task.

    The following code was generated by CHAT-GPT, I don't even want to waste time on this."""
    import re
    import subprocess

    # Run a command and capture its output
    cmd = f"ros2 bag info -s mcap {mcap_file}"
    output = subprocess.check_output(cmd, shell=True)

    # Decode the output to a string
    output_str = output.decode("utf-8")

    # Define the string to search for in the lines

    # Find all the lines that contain the search string
    lines = [line for line in output_str.split("\n") if topic in line]

    # Find all the integer numbers in the lines
    int_regex = r"\b\d+\b"
    msg_count = []
    for line in lines:
        msg_count.extend(re.findall(int_regex, line))

    if len(msg_count) != 1:
        print(f"An error ocurred, please report it on the Github page")
        sys.exit(1)
    return int(msg_count[0])

This issue was first seen in PRBonn/kiss-icp#104

[jtbandes]

The rosbag2 python API get_metadata() was added in ros2/rosbag2#1082 and should be available in Rolling and the next major ROS release.

Using the mcap python API directly, you can do this:

>>> from mcap.reader import make_reader
>>> reader = make_reader(open("/path/to/file.mcap", "rb"))
>>> summary = reader.get_summary()

>>> summary.statistics
Statistics(attachment_count=0, channel_count=7, channel_message_counts={0: 52, 1: 234, 2: 774, 3: 156, 4: 156, 5: 156, 6: 78}, chunk_count=14, message_count=78, message_end_time=1490149587884601617, message_start_time=1490149580103843113, metadata_count=0, schema_count=7)

# NOTE: this code does not handle the case where 2 different channel IDs refer to the same topic name
>>> {summary.channels[id].topic: count for (id, count) in summary.statistics.channel_message_counts.items()}
{'/diagnostics': 52, '/image_color/compressed': 234, '/tf': 774, '/radar/points': 156, '/radar/range': 156, '/radar/tracks': 156, '/velodyne_points': 78}

[nachovizzo]

@jtbandes wow, thanks for the fast reply!

Indeed, I did see the get_metadata while I was navigating through the pybind of the repo, but I thought it was not ideal to break backwards compatibility with ros humble, etc, just because of this easy thing.

I will check the propossed solution and come back to you ;)

[nachovizzo]

Tested and solved in PRBonn/kiss-icp#109

Thanks!