Skip to main content

ROS Ingestion

Downloads the NVIDIA R2B Dataset 2024 and ingests it from .mcap ROS 2 bags into a structured, queryable Mosaico archive. Asset download, type translation, batching, and catalog verification are all handled automatically.

Learn

Doc: ROS Bridge

API Reference: ROS Bridge

Run
mosaicolabs.examples ros_injection

Run the command with --help to see all available options.

The daemon must be running before you start. See the setup guide if you have not started it yet.

The full source is on GitHub.

Custom Ontology Definition​

The NVIDIA Isaac Nova encoder produces isaac_ros_nova_interfaces/msg/EncoderTicks, a message type with no equivalent in the built-in ontology. Before the injector can run, we define a custom model by inheriting from Serializable. This registers the type automatically at import time. From then on, it is available as an ontology_type for topic creation, with the .Q query proxy enabled on its fields.

Custom EncoderTicks model
from mosaicolabs import MosaicoField, MosaicoType, Serializable

class EncoderTicks(Serializable):
    left_ticks: MosaicoType.uint32 = MosaicoField(
        description="Cumulative counts from the left wheel encoder."
    )
    right_ticks: MosaicoType.uint32 = MosaicoField(
        description="Cumulative counts from the right wheel encoder."
    )
    encoder_timestamp: MosaicoType.uint64 = MosaicoField(
        description="Timestamp of the encoder ticks."
    )

ROS Adapter​

Because EncoderTicks is a custom type, no default adapter exists. We subclass ROSAdapterBase and implement from_dict to map ROS dictionary fields onto the model. The @register_default_adapter decorator ensures the RosbagInjector selects this adapter automatically whenever it encounters the matching schema name in the bag.

This adapter only implements from_dict, so it only supports the ROS → Mosaico direction. To also support Mosaico → ROS — reconstructing a native ROS message from EncoderTicks — implement to_ros() as well.

EncoderTicks ROS adapter
from mosaicolabs import Message
from mosaicolabs.ros_bridge import ROSMessage, ROSAdapterBase, register_default_adapter
from mosaicolabs.ros_bridge.adapters.helpers import _validate_msgdata
from .isaac import EncoderTicks

@register_default_adapter(is_default=True)
class EncoderTicksAdapter(ROSAdapterBase[EncoderTicks]):
    ros_msgtype = ("isaac_ros_nova_interfaces/msg/EncoderTicks",)
    __mosaico_ontology_type__ = EncoderTicks
    _REQUIRED_KEYS = ("left_ticks", "right_ticks", "encoder_timestamp")

    @classmethod
    def from_dict(cls, ros_data: dict) -> EncoderTicks:
        _validate_msgdata(cls, ros_data)
        return EncoderTicks(
            left_ticks=ros_data["left_ticks"],
            right_ticks=ros_data["right_ticks"],
            encoder_timestamp=ros_data["encoder_timestamp"],
        )

    @classmethod
    def translate(cls, ros_msg: ROSMessage, **kwargs) -> Message:
        return super().translate(ros_msg, **kwargs)

Writing an Adapter Is Optional​

EncoderTicks is the only message type in the R2B dataset without built-in ontology coverage, and above we gave it a hand-written adapter to get a fully-typed, queryable EncoderTicks model. That step is a choice, not a requirement.

Even if EncoderTicksAdapter didn't exist, RosbagInjector would still ingest every isaac_ros_nova_interfaces/msg/EncoderTicks message without error: the bridge synthesizes an Unmodeled Adapter for it at runtime, converting the message's schema, embedded directly in the bag, into an equivalent Mosaico Unmodeled ontology class on the fly. The data is still ingested and fully queryable, just without the dedicated EncoderTicks Python class. See Advanced: Ingesting Unmodeled Ontologies for the full mechanics.

Ingestion Pipeline​

The example loops over each bag file in the dataset and runs three phases: download, ingest, verify.

Download. download_asset (defined in the example's helpers.py) fetches the raw .mcap from NVIDIA with a progress bar and caches it locally so re-runs skip the download.

Ingest. RosbagInjector handles the full write pipeline: it opens the MCAP, routes each message through its registered adapter, batches the resulting Message objects, and transmits them to the daemon over gRPC. The sequence name is derived from the filename so each bag becomes a separate, independently queryable recording session.

Verify. A list_sequences() call after ingestion confirms the sequence is visible in the catalog before the script exits.

Download, ingest, and verify
for bag_path in BAG_FILES_PATH:
    out_bag_file = download_asset(BASE_BAGFILE_URL + bag_path, ASSET_DIR)

    config = ROSInjectionConfig(
        host=MOSAICO_HOST,
        port=MOSAICO_PORT,
        file_path=out_bag_file,
        sequence_name=out_bag_file.stem,
        metadata={
            "source_url": BAGFILE_URL,
            "ingested_via": "mosaico_example_ros_injection",
            "download_time_utc": str(downloaded_time),
        },
    )
    RosbagInjector(config).run()

    with MosaicoClient.connect(host=MOSAICO_HOST, port=MOSAICO_PORT) as client:
        if config.sequence_name in client.list_sequences():
            print(f"'{config.sequence_name}' is now queryable.")

Once complete, the ingested sequences are available for the Data Inspection, Querying Catalogs, and MuJoCo Visualisation examples.