Skip to main content

MuJoCo Visualisation

Queries robot joint state data from a stored Mosaico sequence and replays the trajectory in a live MuJoCo simulation window, synchronized to the nanosecond timestamps stored in the archive.

Learn

Doc: The Query Workflow

Doc: The Reading Workflow

API Reference: Query Builders

API Reference: Query Response

API Reference: Data Retrieval

Run
mosaicolabs.examples mujoco_vis

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

The daemon must be running with data from the ROS Ingestion example. The full source is on GitHub.

MuJoCo is not included in the default SDK dependencies — install it before running:

Install MuJoCo
pip install mujoco
note

On first run, if the model files are not found locally at MUJOCO_XML_SCENE_PATH, the example automatically downloads them from the MuJoCo Menagerie GitHub repository. This requires outbound network access to GitHub.

Locating the Joint State Topic​

QuerySequence and QueryTopic are combined to find the right channel in a single server-side query. Filtering by RobotJoint.ontology_tag() ensures the result contains only joint state topics, regardless of how the channel is named in the bag.

Query for robot joint state
from mosaicolabs import MosaicoClient, QuerySequence, QueryTopic
from mosaicolabs.models.sensors.robot import RobotJoint

with MosaicoClient.connect(host=MOSAICO_HOST, port=MOSAICO_PORT) as client:
    result = client.query(
        QuerySequence().with_name(ROBOT_SEQUENCE_NAME),
        QueryTopic().with_ontology_tag(RobotJoint.ontology_tag()),
    )
    if result is None:
        print(f"Sequence '{ROBOT_SEQUENCE_NAME}' not found.")

Validating the Topic​

A TopicHandler confirms the matched topic carries the expected type before any data is streamed. The query already filters by ontology tag, so this check acts as a defensive guard for pipelines where the data source is not fully controlled.

Validate topic ontology
for items in result:
    for topic in items.topics:
        top_handler = client.topic_handler(items.sequence.name, topic.name)
        if top_handler is None:
            continue
        if top_handler.ontology_tag != RobotJoint.ontology_tag():
            print(f"Unexpected type: {top_handler.ontology_tag}")
            continue

Streaming into MuJoCo​

Each RobotJoint message is timestamped in nanoseconds relative to the recording start. That offset drives the simulation forward so the replay matches the original motion at the correct cadence. Joint positions are applied directly to the MuJoCo model state and the viewer is updated each frame.

note

The snippet below is a simplified illustration and omits setup boilerplate (loading the MJCF scene, syncing relative_ts to wall-clock time, joint-name-to-qpos-index lookups, error handling). See the full source on GitHub for a runnable version.

Set up the MuJoCo model, data, and viewer
import mujoco
import mujoco.viewer

mj_model = mujoco.MjModel.from_xml_path(MUJOCO_XML_SCENE_PATH)
mj_data = mujoco.MjData(mj_model)
viewer = mujoco.viewer.launch_passive(mj_model, mj_data)
Stream joints and drive the simulation
for joint_msg in top_handler.get_data_streamer():
    relative_ts = (
        joint_msg.timestamp_ns - top_handler.timestamp_ns_min
    ) / 1.0e9

    joints = joint_msg.get_data(RobotJoint)
    mj_data.qpos[:] = joints.positions
    mujoco.mj_forward(mj_model, mj_data)
    viewer.sync()

When running correctly, the MuJoCo viewer should open and replay the recorded trajectory: