Creating a Robot Sensor Data Recording and Replay System

A robust recording-and-replay system is the backbone of any serious robot learning workflow. Recording lets you build datasets and debug incidents after the fact; replay lets you re-run recorded sensor streams through your perception or control stack without needing the physical robot, which massively speeds up development and debugging. This tutorial covers building both halves as a cohesive system.

Why replay matters as much as recording

It's easy to treat recording as "just logging," but a good replay system pays for itself quickly:

• Debugging without hardware: reproduce a bug from last week's session without booking robot time.
• Regression testing: replay a fixed set of recorded sessions through a new version of your perception pipeline and diff the outputs.
• Dataset iteration: re-extract features, re-label, or re-sample previously recorded sessions as your pipeline evolves, without recollecting data.
• Simulation grounding: compare simulated sensor output against real recorded sensor output for the same nominal trajectory.

Core design: a session as a set of synchronized streams

Model a recording session as multiple independent streams (camera, joint states, IMU, force-torque, etc.), each with its own timestamped samples, plus a session-level index that lets you query "what did every stream look like at time T."

Session/
  metadata.json          # session info: robot config, sensors present, start time
  streams/
    camera_wrist/
      timestamps.npy
      frames/            # or a video file + frame index
    joint_states/
      timestamps.npy
      data.npy
    gripper/
      timestamps.npy
      data.npy

The recorder

The recorder's job is simple in principle: subscribe to every sensor stream, timestamp each sample, and write it to disk without blocking the control loop.

import threading
import queue
import time

class StreamRecorder:
    def __init__(self, name, writer):
        self.name = name
        self.writer = writer
        self.queue = queue.Queue()
        self.thread = threading.Thread(target=self._worker, daemon=True)
        self.running = False

    def start(self):
        self.running = True
        self.thread.start()

    def record(self, data):
        # Called from the sensor callback thread — must be fast and non-blocking
        self.queue.put((time.time(), data))

    def _worker(self):
        while self.running or not self.queue.empty():
            try:
                timestamp, data = self.queue.get(timeout=0.1)
                self.writer.write(self.name, timestamp, data)
            except queue.Empty:
                continue

    def stop(self):
        self.running = False
        self.thread.join()

The key design point: record() is called from whatever thread the sensor callback lives on, and it must return immediately. All the actual disk I/O happens on a dedicated writer thread per stream. This decoupling is what prevents a slow disk write from stalling your control loop — a very common source of "why does my robot judder every few seconds" bugs.

Handling different sensor rates cleanly

Cameras, joint encoders, and force-torque sensors rarely run at the same rate. Rather than forcing everything onto one clock during recording, record each stream at its native rate with accurate timestamps, and defer alignment to query time:

import numpy as np

def get_nearest_sample(timestamps: np.ndarray, data: np.ndarray, query_time: float):
    idx = np.searchsorted(timestamps, query_time)
    idx = np.clip(idx, 1, len(timestamps) - 1)
    before, after = idx - 1, idx
    if abs(timestamps[before] - query_time) < abs(timestamps[after] - query_time):
        return data[before]
    return data[after]

This "query by nearest timestamp, per stream" pattern is more flexible than forcing a single global sample rate at recording time — it lets you resample to whatever rate a downstream consumer (a training pipeline, a replay tool) actually needs, without having thrown away information at recording time.

Building the replay engine

A replay engine reconstructs a session and plays it back through the same interfaces your live system uses, so your downstream code doesn't need separate "live" and "replay" code paths.

class SessionReplayer:
    def __init__(self, session_path):
        self.streams = self._load_streams(session_path)
        self.start_time = min(s["timestamps"][0] for s in self.streams.values())
        self.end_time = max(s["timestamps"][-1] for s in self.streams.values())

    def _load_streams(self, session_path):
        # Load timestamps + data arrays per stream from disk
        raise NotImplementedError

    def get_state_at(self, t):
        return {
            name: get_nearest_sample(s["timestamps"], s["data"], t)
            for name, s in self.streams.items()
        }

    def play(self, callback, speed=1.0, dt=0.033):
        t = self.start_time
        wall_start = time.time()
        while t <= self.end_time:
            state = self.get_state_at(t)
            callback(state, t)

            t += dt * speed
            target_wall_time = wall_start + (t - self.start_time) / speed
            sleep_time = target_wall_time - time.time()
            if sleep_time > 0:
                time.sleep(sleep_time)

By designing the callback(state, t) interface to match what your live perception/control code expects as input, you can point the exact same downstream code at either live sensors or a replayed session — this is the single biggest productivity win of building a proper replay system rather than ad hoc debug scripts.

Practical tips for a system that scales

• Compress images as you go, not after the fact — writing raw uncompressed frames during a live session will exhaust disk bandwidth quickly on any camera above VGA resolution. JPEG or a lightweight video codec (H.264) with a keyframe interval short enough for random access works well.
• Write an index file per session (start/end time, sensors present, robot config, any metadata like task label or operator id) so you can filter and search sessions without opening every stream file.
• Version your schema. Sensor configurations change over a project's lifetime — added cameras, changed resolutions, new sensors. Store a schema version in each session's metadata so your loader can handle old and new sessions correctly.
• Validate on write, not just on read. Catch dropped frames, clock jumps, and sensor disconnects at recording time when you can still restart the session, rather than discovering the corruption weeks later during training.

Where this fits in the bigger picture

This recording-and-replay system is the shared infrastructure underneath every other tutorial in this series — teleoperation sessions, human demonstrations, and imitation learning training data are all, at the storage layer, just sessions recorded and replayed through this same system. Getting this layer right early pays dividends across the entire robot learning pipeline.

Useful Links

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