Creating a Robot Sensor Data Recording and Replay System

Every robot learning project eventually needs the same infrastructure: a way to record everything the robot sees and does, and a way to play it back later — for debugging, dataset inspection, offline evaluation, or regenerating labels without touching the physical robot again. This tutorial builds a general-purpose recording and replay system you can drop into any robotics pipeline.

Why You Need This

Recording raw sensor streams alongside your teleoperation or autonomous runs pays off in ways that aren't obvious until you need it:

• Debugging policy failures without needing the robot present
• Re-deriving labels (e.g., re-running object detection or segmentation on old camera frames)
• Regression testing — replaying the exact same episode against a new model version
• Sharing datasets with collaborators who don't have access to the hardware

Architecture

[Sensors/Robot State] --> [Recorder] --> [Storage: rosbag / hdf5 / mcap]
                                                    |
                                                    v
                                              [Replay Engine]
                                                    |
                              -----------------------------------------
                              |                     |                 |
                        [Visualization]      [Re-labeling]     [Offline Eval]

Step 1: Define What to Record

Decide on your sensor set up front — changing it mid-project fragments your dataset. A typical manipulation setup records:

• RGB (and optionally depth) from one or more cameras
• Robot joint positions, velocities, and torques
• End-effector pose
• Gripper state / force
• Any external sensors (F/T sensor, tactile sensors, IMU)
• Timestamps for everything, from a single shared clock

SENSOR_SCHEMA = {
    "camera_front_rgb": {"shape": (480, 640, 3), "dtype": "uint8", "hz": 30},
    "camera_wrist_rgb": {"shape": (480, 640, 3), "dtype": "uint8", "hz": 30},
    "joint_positions": {"shape": (7,), "dtype": "float32", "hz": 100},
    "joint_torques": {"shape": (7,), "dtype": "float32", "hz": 100},
    "ee_pose": {"shape": (7,), "dtype": "float32", "hz": 100},
    "ft_sensor": {"shape": (6,), "dtype": "float32", "hz": 500},
}

Step 2: The Recorder

Sensors often run at different rates. Rather than forcing everything into a single fixed-rate loop, record each stream at its native rate with its own timestamp, and resolve synchronization at replay/read time.

import threading
import queue
import time

class SensorRecorder:
    def __init__(self, storage_backend):
        self.storage = storage_backend
        self.queues = {}
        self.threads = []
        self.running = False

    def register_sensor(self, name, read_fn, hz):
        self.queues[name] = queue.Queue()

        def _loop():
            dt = 1.0 / hz
            while self.running:
                start = time.time()
                data = read_fn()
                self.queues[name].put((time.time(), data))
                time.sleep(max(0, dt - (time.time() - start)))

        self.threads.append(threading.Thread(target=_loop, daemon=True))

    def start(self):
        self.running = True
        for t in self.threads:
            t.start()

    def stop_and_flush(self, episode_id):
        self.running = False
        for name, q in self.queues.items():
            records = []
            while not q.empty():
                ts, data = q.get()
                records.append((ts, data))
            self.storage.write(episode_id, name, records)

Step 3: Storage Backend

For robotics, three formats dominate: rosbag/rosbag2 (if you're in the ROS ecosystem), MCAP (a modern, ROS-agnostic container format gaining popularity), and plain HDF5 (simplest, good for non-ROS pipelines). Here's a minimal HDF5 backend:

import h5py
import numpy as np

class HDF5Storage:
    def __init__(self, base_path):
        self.base_path = base_path

    def write(self, episode_id, sensor_name, records):
        path = f"{self.base_path}/episode_{episode_id}.hdf5"
        with h5py.File(path, "a") as f:
            grp = f.require_group(sensor_name)
            timestamps = np.array([r[0] for r in records])
            data = np.stack([r[1] for r in records])
            grp.create_dataset("timestamps", data=timestamps, compression="gzip")
            grp.create_dataset("data", data=data, compression="gzip")

If you expect to scale past a few thousand episodes or need cross-language tooling, MCAP is worth the extra setup — it has strong support for indexed random access and streaming playback.

Step 4: The Replay Engine

Replay needs to reconstruct a synchronized view across sensors that were recorded at different rates. The standard approach is nearest-timestamp alignment to a chosen reference clock (often the slowest sensor, e.g., the camera).

class ReplayEngine:
    def __init__(self, hdf5_path):
        self.file = h5py.File(hdf5_path, "r")

    def get_synced_frame(self, reference_sensor, index):
        ref_ts = self.file[reference_sensor]["timestamps"][index]
        frame = {reference_sensor: self.file[reference_sensor]["data"][index]}

        for sensor_name in self.file.keys():
            if sensor_name == reference_sensor:
                continue
            timestamps = self.file[sensor_name]["timestamps"][:]
            nearest_idx = np.argmin(np.abs(timestamps - ref_ts))
            frame[sensor_name] = self.file[sensor_name]["data"][nearest_idx]

        return frame

    def __len__(self):
        return len(self.file["camera_front_rgb"]["timestamps"])

Step 5: Playback and Visualization

A simple playback loop lets you scrub through an episode visually — invaluable for spotting labeling errors or teleoperation glitches:

import cv2

def visualize_episode(replay_engine, reference_sensor="camera_front_rgb", fps=30):
    for i in range(len(replay_engine)):
        frame = replay_engine.get_synced_frame(reference_sensor, i)
        img = frame[reference_sensor]

        cv2.putText(img, f"t={i}", (10, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 1)
        cv2.imshow("Replay", img)

        if cv2.waitKey(int(1000 / fps)) & 0xFF == ord('q'):
            break

    cv2.destroyAllWindows()

Step 6: Replaying Actions Against a Simulator or Policy

Beyond visualization, you can replay recorded actions into a simulator to validate physical plausibility, or feed recorded observations into a trained policy for offline comparison against the original human action:

def offline_policy_comparison(replay_engine, policy, stats):
    errors = []
    for i in range(len(replay_engine)):
        frame = replay_engine.get_synced_frame("camera_front_rgb", i)
        pred_action = policy.predict(frame)
        true_action = frame.get("action")
        if true_action is not None:
            errors.append(np.linalg.norm(pred_action - true_action))
    return np.mean(errors)

This gives you a fast, hardware-free sanity check before committing to a real robot evaluation.

Practical Tips

• Record a single monotonic clock source across all sensors if possible (e.g., a shared NTP-synced or hardware trigger clock) — this eliminates most synchronization headaches.
• Store raw, unprocessed sensor data. Apply cropping, resizing, or filtering at load time, not at record time — you can't undo lossy preprocessing later.
• Include a manifest file per episode (task name, success flag, sensor list, schema version) so tooling can validate compatibility before loading.
• Compress images inline (e.g., JPEG) if storage is tight, but keep at least one high-fidelity/lossless recording session for cases where compression artifacts matter (e.g., fine texture-based manipulation).

Wrapping Up

Across this series, we've covered the full loop: building a teleoperation system, capturing clean human demonstrations, turning them into a trained imitation learning policy, using VR controllers to make teleoperation more natural, and finally recording/replaying sensor data for debugging and offline evaluation. Together, these form the core data infrastructure behind most modern robot learning pipelines.

Useful Links

Website: www.v-modal.com
SDK Flutter: https://github.com/v-modal/vmodal_sdk_flutter
SDK Android: https://github.com/v-modal/vmodal_sdk_android
Discord: https://discord.gg/K72z28KUx