ReductRos

The ReductROS extension provides tools that can be used to extract and transform data stored in ReductStore into formats that are compatible with the Robot Operating System (ROS). It allows users to query and convert historical data, such as sensor readings and camera images, into text-based formats like CSV or JSON, which can then be easily used for analysis, visualisation or integration with other systems.

info

The extension is only available under a commercial licence. If you wish to use it for non-commercial purposes, please contact us.

Supported File and Message Formats

The ReductROS extension supports parsing and extracting data exclusively from MCAP files containing messages encoded in the CDR (Common Data Representation) format—used as the default serialization method in ROS 2.

These .mcap files typically originate from ROS 2 logging tools and encapsulate serialized messages across multiple topics, along with metadata such as timestamps and message types. ReductROS provides functionality to:

• Decode CDR-encoded ROS 2 messages embedded in MCAP files.
• Extract specific topics from the MCAP file based on user-defined parameters and convert them into JSON format.
• Encode binary data (e.g., images, point clouds) into base64 strings or JPEG format for easier handling and analysis.
• Expose metadata, including topic names, timestamps, and message types alongside message content.

Input

• .mcap files containing ROS 2 messages in binary CDR format

Output

• JSON representations of ROS 2 messages
• timestamp from the message header
• computed_labels containing metadata such as topic name, message type, and encoding

This output format is ideal for analysis, visualization, integration with external systems, or further transformation to other formats.

Query Format

A user can use the ext query parameter to activate the ros extension and define the parameters for extracting and transforming ROS data in the following format:

{
    "ext": {
        "ros": {
            "extract": {
               "topic": "string",  # The name of the ROS topic to extract messages from
               "encode": "object"  # Dictionary of binary files to encode, e.g., {"data": "jpeg"} for JPEG encoding
            }
        }
    }
}

Data Extraction

The extract property allows you to specify the ROS topic from which to extract message and convert it to JSON format. The extension will read the MCAP file, decode the messages from the specified topic, and return each message as a JSON record with the timestamp from the message header.

Parameter	Type	Mandatory	Description
topic	string	Yes	The name of the ROS topic to extract messages from. This should match the topic names used in the MCAP file.
encode	object	No	A dictionary specifying how to encode binary data in the messages. For example, {"data": "jpeg"} for JPEG encoding.
encode.<field>	string	No	The encoding format for binary fields in the message. Supported values are base64 and jpeg. If not specified, the field will be returned as a JSON list.

Encoding Binary Data

The encode property allows you to specify how to handle binary data in the extracted messages. You can choose to encode binary fields in the following formats:

Format	Description
base64	Encodes binary data as a base64 string, suitable for text-based formats like JSON.
jpeg	Encodes binary data as a JPEG image encoded to a base64 string, suitable for image data.

info

Currently, the extension supports encoding only sensor_msgs/msg/Image messages with encoding set to rgb8, bgr8, or mono8.

Examples

The following examples demonstrate how to use the ReductROS extension to extract and transform ROS messages. Although this example is written in Python, it can be run using any of the official SDKs.

Extracting Messages as JSON

This example demonstrates how to use the ROS extension to extract topic from an MCAP file stored in ReductStore and convert it to JSON format.

Python

from time import time_ns
from pathlib import Path

from reduct import Client

HERE = Path(__file__).parent

async def main():
    async with Client("http://localhost:8383", api_token="my-token") as client:
        bucket = await client.create_bucket(
            "my-bucket",
            exist_ok=True,
        )
        # Write a mcap file with timestamps
        now = time_ns() // 1000

        data = b""
        with open(f"{HERE}/../data/file.mcap", "rb") as f:
            data = f.read()

        await bucket.write("mcap", data, content_length=len(data), timestamp=now, content_type="application/mcap")

        # Prepare the query with the 'ros' extension
        ext = {
            "ros": {     # name of the extension to use
                "extract": {
                    "topic": "/test"   # Specify the topic to extract from the mcap file
                },
            },
        }

        # Query the data with the 'ros' extension
        async for record in bucket.query("mcap", start=now, ext=ext):
            print(f"Record timestamp: {record.timestamp}")
            print(f"Record labels: {record.labels}")

            json = await record.read_all()
            print(json.decode("utf-8").strip())


# 5. Run the main function
if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Expected Output

The expected output of the above code is as follows:

Record timestamp: 24
Record labels: {'@encoding': 'cdr', '@schema': 'std_msgs/String', '@topic': '/test'}
{"data":"hello"}
Record timestamp: 25
Record labels: {'@encoding': 'cdr', '@schema': 'std_msgs/String', '@topic': '/test'}
{"data":"world"}

Explanation

• The extension extracts ROS 2 messages from an .mcap file stored in ReductStore.
• Only messages from the topic /test are selected using the topic filter in the ros.extract configuration.
• The content of each message is CDR-encoded and decoded by the extension.
• The decoded message is returned as JSON with the field data, matching the std_msgs/String schema.
• Each record corresponds to one ROS 2 message and includes:
  • The decoded JSON payload, e.g., {"data":"hello"}
  • Message metadata as labels, including:
    • topic: /test
    • schema: std_msgs/String
    • encoding: cdr

Extracting Messages as JSON with JPEG Encoding

This example demonstrates how to use the ROS extension to extract topic from an MCAP file stored in ReductStore and convert it to JSON format, while also encoding binary image data into JPEG format.

Python

import base64
import json
from time import time_ns
from pathlib import Path

from reduct import Client

HERE = Path(__file__).parent

async def main():
    async with Client("http://localhost:8383", api_token="my-token") as client:
        bucket = await client.create_bucket(
            "my-bucket",
            exist_ok=True,
        )
        # Write a mcap file with timestamps
        now = time_ns() // 1000

        data = b""
        with open(f"{HERE}/../data/camera_bag_0.mcap", "rb") as f:
            data = f.read()

        await bucket.write("mcap", data, content_length=len(data), timestamp=now, content_type="application/mcap")

        # Prepare the query with the 'ros' extension
        ext = {
            "ros": {     # name of the extension to use
                "extract": {
                    "topic": "/image_raw",
                    # encode the data filed in http://docs.ros.org/en/noetic/api/sensor_msgs/html/msg/Image.html
                    "encode": {
                        "data": "jpeg",
                    },
                },
            },
            "when": {  # optional filter to apply
                "$limit": 1, # return only one record
            }
        }

        # Query the data with the 'ros' extension
        async for record in bucket.query("mcap", start=now, ext=ext):
            print(f"Record timestamp: {record.timestamp}")
            print(f"Record labels: {record.labels}")
            content = await record.read_all()
            # Record content is a JSON object with metadata and base64-encoded data
            obj = json.loads(content)
            # Decode the base64-encoded data and save it as a JPEG file
            with open("output.jpg", "wb") as f:
                # Decode the base64-encoded data
                encoded = base64.decodebytes(obj["data"].encode("ascii"))
                f.write(encoded)

            # Print the image parameters without the data field
            del obj["data"]
            print(f"Image parameters: {obj}")


# 5. Run the main function
if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Expected Output

The expected output of the above code is as follows:

Record timestamp: 1753341400522732
Record labels: {'@encoding': 'cdr', '@schema': 'sensor_msgs/Image', '@topic': '/image_raw'}
Image parameters: {'height': 720, 'width': 1280, 'is_bigendian': 0, 'encoding': 'rgb8', 'step': 3840, 'header': {'frame_id': 'camera', 'stamp': {'sec': 1753341400, 'nanosec': 522732248}}}

Explanation

• The extension extracts ROS 2 messages from an .mcap file stored in ReductStore.
• Only messages from the topic /image_raw are selected using the topic filter in the ros.extract configuration.
• The content of each message is CDR-encoded and decoded by the extension.
• The decoded message is returned as JSON with the field data, which contains the image data encoded in JPEG format.
• Each record corresponds to one ROS 2 message and includes:
  • The decoded image parameters, such as height, width, encoding, and step.
  • The timestamp from the message header.
  • Message metadata as labels, including:
    • topic: /image_raw
    • schema: sensor_msgs/Image
    • encoding: cdr