The Data Platform for Robotics and Physical AI

from mosaicolabs import MosaicoClient, IMU, SessionLevelErrorPolicy
 
with MosaicoClient.connect("localhost", 6726) as client:
    with client.sequence_create(
        sequence_name="imu_session_01",
        metadata={"source": "sensor_rig_v3"},
        on_error=SessionLevelErrorPolicy.Delete,
    ) as swriter:
        imu_writer = swriter.topic_create("sensors/imu", ontology_type=IMU)
        for msg in stream_imu("imu.csv"):  # user-supplied generator
            imu_writer.push(message=msg)

from mosaicolabs import MosaicoClient, QuerySequence
 
with MosaicoClient.connect("localhost", 6726) as client:
    results = client.query(
        QuerySequence()
            .with_name_match("test_drive")
            .with_user_metadata("project.name", eq="Apollo")
            .with_user_metadata("environment.visibility", lt=50)
    )
    if results:
        for item in results:
            print(item.sequence.name, [t.name for t in item.topics])

from mosaicolabs import MosaicoClient
 
with MosaicoClient.connect("localhost", 6726) as client:
    handler = client.sequence_handler("mission_alpha")
    if handler:
        stream = handler.get_data_streamer(
            topics=["/sensors/gps", "/sensors/imu", "/sensors/pressure"],
            start_timestamp_ns=1738508778000000000,
            end_timestamp_ns  =1738509618000000000,
        )
        print(f"Replay starts at: {stream.next_timestamp()}")
        for topic, msg in stream:
            print(f"[{topic}] {msg.timestamp_ns} — {type(msg.data).__name__}")
        handler.close()