MCAPLogger

SUMMARY

MCAPLogger captures every message published on a Zenoh (BabyROS) network into a single .mcap file. Use it to record robot runs, replay them for debugging, and convert them into training datasets.

MCAP is an open container format for robotics data — timestamped, multi-channel, seekable, and self-describing. MCAPLogger records every topic on the network with no per-topic configuration. Reading back is a single generator call.

Install

pip install telekinesis-dataengine

Import

from telekinesis.dataengine import MCAPLogger

---

Writing

MCAPLogger is a context manager. Open it before publishers start; close it when the run ends. It subscribes to "**" by default, capturing every topic on the Zenoh session.

MCAPLogger(
    path,           # output .mcap file path
    key="**",       # Zenoh key expression to subscribe to
    depth=1024,     # internal subscriber queue depth
)

Parameters

Parameter	Type	Description
path	str | Path	Output file path, e.g. "results/run.mcap".
key	str	Zenoh key expression. "**" captures all topics (default). Use a prefix like "camera/**" to filter.
depth	int	Subscriber queue depth. Increase for high-frequency topics.

Methods

Method	Returns	Description
num_messages()	int	Total messages written so far.
topics()	list[str]	List of topic strings seen during the recording.

Example

import babyros
from telekinesis.dataengine import MCAPLogger

# ------------------------------------------------
# 1. Start publishers on the Zenoh network
# ------------------------------------------------
camera_pub  = babyros.node.Publisher(topic="camera/rgb")
joint_pub   = babyros.node.Publisher(topic="robot/joint_states")

# ------------------------------------------------
# 2. Record everything to an MCAP file
# ------------------------------------------------
with MCAPLogger("results/run.mcap") as logger:
    for step in range(100):
        camera_pub.publish(frame)
        joint_pub.publish(joint_states)

# ------------------------------------------------
# 3. Inspect the recording
# ------------------------------------------------
print(f"Messages: {logger.num_messages()}")
print(f"Topics:   {logger.topics()}")

---

Reading

MCAPLogger.read() is a generator that yields (topic, object) pairs in recording order.

for topic, obj in MCAPLogger.read(path):
    print(topic, obj)

Parameters

Parameter	Type	Description
path	str | Path	Path to an existing .mcap file.

Returns

Each iteration yields a (topic: str, obj: Any) tuple. obj is the decoded Telekinesis datatype (e.g. Image, Points3D) or a plain dict if the original type is not available.

Example

from collections import defaultdict
from telekinesis.dataengine import MCAPLogger

# ------------------------------------------------
# 1. Read back a recording and count per topic
# ------------------------------------------------
counts = defaultdict(int)

for topic, obj in MCAPLogger.read("results/run.mcap"):
    counts[topic] += 1

for topic, count in counts.items():
    print(f"{topic}: {count} messages")

---

Full Pipeline Example

Record a multi-rate robot run, then read it back and inspect the contents.

import time
import threading
import babyros
from telekinesis.dataengine import MCAPLogger

# ------------------------------------------------
# 1. Define publishers
# ------------------------------------------------
imu_pub    = babyros.node.Publisher(topic="sensors/imu")
camera_pub = babyros.node.Publisher(topic="camera/rgb")
joints_pub = babyros.node.Publisher(topic="robot/joint_states")

def publish_loop(pub, make_msg, hz, stop_event):
    while not stop_event.is_set():
        pub.publish(make_msg())
        time.sleep(1.0 / hz)

stop = threading.Event()

# ------------------------------------------------
# 2. Record for 5 seconds
# ------------------------------------------------
with MCAPLogger("results/robot_run.mcap") as logger:
    threads = [
        threading.Thread(target=publish_loop, args=(imu_pub,    make_imu,    100, stop), daemon=True),
        threading.Thread(target=publish_loop, args=(camera_pub, make_frame,   30, stop), daemon=True),
        threading.Thread(target=publish_loop, args=(joints_pub, make_joints,   5, stop), daemon=True),
    ]
    for t in threads:
        t.start()

    time.sleep(5)
    stop.set()

print(f"Recorded {logger.num_messages()} messages across {len(logger.topics())} topics")

# ------------------------------------------------
# 3. Read back and count per topic
# ------------------------------------------------
from collections import defaultdict

counts = defaultdict(int)
for topic, obj in MCAPLogger.read("results/robot_run.mcap"):
    counts[topic] += 1

for topic, count in sorted(counts.items()):
    print(f"  {topic}: {count}")