# How to Build Replay Pipelines from Recorded Sensor Data for Robots with DimOS

> Learn to build robot replay pipelines from recorded sensor data with DimOS. Replay sensor streams via timestamped pickle files and Rx-observable pipelines without code changes.

- Repository: [Dimensional/dimos](https://github.com/dimensionalOS/dimos)
- Tags: how-to-guide
- Published: 2026-03-15

---

**DimOS provides a file-based replay system that stores sensor streams as timestamped pickle files and replays them through Rx-observable pipelines, allowing you to swap recorded data for live hardware without modifying downstream code.**

Robotic development requires testing perception, navigation, and control algorithms against consistent, reproducible datasets. The DimOS framework (dimensionalOS/dimos) implements a lightweight replay infrastructure that preserves original timestamps and integrates seamlessly with existing reactive sensor pipelines. This system enables developers to record live robot sessions to disk and later replay them through the exact same processing graphs used during real-world operation.

## Core Architecture of the Replay Pipeline

The replay system consists of three interoperating building blocks that handle storage, retrieval, and transparent playback.

### TimedSensorStorage for Recording

**TimedSensorStorage** (defined in [`dimos/utils/testing/replay.py`](https://github.com/dimensionalOS/dimos/blob/main/dimos/utils/testing/replay.py)) provides the entry point for persisting live sensor data. This class writes incoming messages to a designated subdirectory under the repository's `data/` folder, serializing each frame as a numbered pickle file (e.g., `000.pickle`, `001.pickle`) containing a `(timestamp, payload)` tuple.

Internally, `TimedSensorStorage` is a thin alias for `LegacyPickleStore`, exposing a simple interface for modules to consume observables and persist them durably.

### LegacyPickleStore as the Time-Series Backend

The **LegacyPickleStore** class (implemented in [`dimos/memory/timeseries/legacy.py`](https://github.com/dimensionalOS/dimos/blob/main/dimos/memory/timeseries/legacy.py)) serves as the generic time-series backend. It manages lazy iteration over stored frames, provides timestamp-indexed seeking, and exposes an Rx-observable interface that respects the original message timing during playback.

This store handles the low-level file I/O and chronological ordering, ensuring that replayed streams maintain the temporal relationships present in the original recording.

### ReplayConnection for Drop-In Playback

**ReplayConnection** (located in [`dimos/robot/unitree/go2/connection.py`](https://github.com/dimensionalOS/dimos/blob/main/dimos/robot/unitree/go2/connection.py)) acts as a drop-in replacement for live hardware connections. It instantiates `TimedSensorReplay` streams for each sensor type (lidar, odometry, video) and feeds them to downstream modules through the standard observable API.

Because `ReplayConnection` implements the same interface as `GO2Connection`, perception pipelines and control logic require no modifications to process recorded data.

## Recording Sensor Sessions with TimedSensorStorage

To capture a live robot run, subscribe to the inbound sensor observables and forward them to a `TimedSensorStorage` instance. The storage automatically manages file numbering and directory creation.

```python
from dimos.utils.testing.replay import TimedSensorStorage
from dimos.robot.unitree.go2.connection import GO2Connection

# Initialize live connection

conn = GO2Connection(ip="192.168.1.100")

# Create storage targets for each sensor type

lidar_store = TimedSensorStorage("my_recording/lidar")
odom_store = TimedSensorStorage("my_recording/odom")

# Begin consuming streams

lidar_store.consume_stream(conn.lidar_stream())
odom_store.consume_stream(conn.odom_stream())

```

The storage writes each frame sequentially to `data/my_recording/lidar/` as individual pickle files, preserving both the sensor payload and its acquisition timestamp.

## Replaying Data with TimedSensorReplay

The **TimedSensorReplay** class reads stored pickle directories and reconstructs the original observable streams. It iterates frames chronologically using `LegacyPickleStore._iter_files` and manages wall-clock timing to replicate the original sensor cadence.

```python
from dimos.utils.testing.replay import TimedSensorReplay

# Load recorded lidar data

replay = TimedSensorReplay("my_recording/lidar")

# Stream with original timing (speed=1.0)

replay.stream().subscribe(process_lidar_frame)

```

The resulting observable emits frames at the same relative intervals as the live recording, ensuring that time-dependent algorithms (such as SLAM or visual odometry) behave identically during replay.

## Swapping Live and Replay Connections Transparently

DimOS determines whether to instantiate live or replay connections through the `make_connection` factory logic (lines 86-92 in [`dimos/robot/unitree/go2/connection.py`](https://github.com/dimensionalOS/dimos/blob/main/dimos/robot/unitree/go2/connection.py)). The system checks `GlobalConfig` values (`replay_dir`, `unitree_connection_type`) and special IP aliases (`"replay"`, `"fake"`) to select the appropriate backend.

```python
from dimos.core.global_config import global_config
from dimos.robot.unitree.go2.connection import GO2Connection

# Live hardware connection

live_conn = GO2Connection(ip="192.168.1.100", cfg=global_config)

# Replay connection (no hardware required)

replay_conn = GO2Connection(ip="replay", cfg=global_config)

```

Both connection types expose identical methods (`lidar_stream()`, `odom_stream()`, `video_stream()`), allowing navigation stacks and perception modules to operate unchanged whether processing live or recorded data.

## Advanced Replay Controls

The replay system supports precise control over playback behavior through parameters passed to `TimedSensorReplay.stream()` or configured via CLI flags.

- **Speed control**: Adjust playback rate with `speed=2.0` for 2× fast-forward.
- **Seeking**: Start at a specific offset using `seek=10.0` to jump 10 seconds into the recording.
- **Duration limiting**: Process only a specific window with `duration=5.0`.
- **Looping**: Enable `loop=True` to repeat the dataset indefinitely for stress testing.

These parameters populate `ReplayConnection.replay_config` and can be controlled from the command line:

```bash
dimos --replay run unitree-go2-agentic \
  --replay-dir go2_sf_office \
  --seek 5 \
  --duration 30 \
  --loop

```

## Summary

- **TimedSensorStorage** captures live sensor streams to timestamped pickle files in the `data/` directory.
- **LegacyPickleStore** provides the underlying time-series storage engine with lazy iteration and timestamp-aware retrieval.
- **ReplayConnection** substitutes for live hardware connections, feeding recorded data through the same Rx-observable pipelines.
- The system supports variable playback speed, seeking, duration limits, and looping for comprehensive debugging and CI testing.
- Zero code changes are required in downstream modules when switching between live and replay data sources.

## Frequently Asked Questions

### What file format does DimOS use for sensor recordings?

DimOS stores sensor data as sequential pickle files (`.pickle`) in numbered sequences (`000.pickle`, `001.pickle`, etc.). Each file contains a tuple of `(timestamp, payload)`, where the timestamp preserves the original acquisition time and the payload holds the serialized sensor message. This format is managed by `LegacyPickleStore` in [`dimos/memory/timeseries/legacy.py`](https://github.com/dimensionalOS/dimos/blob/main/dimos/memory/timeseries/legacy.py).

### Can I replay multiple sensor streams simultaneously?

Yes. `ReplayConnection` creates independent `TimedSensorReplay` instances for each sensor type (lidar, odometry, video) from the same recording directory. Each stream maintains its own timing and can be subscribed to separately, allowing SLAM modules or multi-sensor fusion pipelines to process synchronized data exactly as they would during live operation.

### How does the replay system handle timing accuracy?

The `TimedSensorReplay.stream()` method measures wall-clock time and scales it by the specified `speed` factor to determine when to emit the next frame. It compares the elapsed time against the stored timestamps to maintain the original temporal relationships between messages, ensuring that time-dependent algorithms experience consistent inter-frame intervals.

### Is it possible to convert replay pipelines to live hardware without code changes?

Yes. Because `ReplayConnection` and `GO2Connection` implement identical interfaces and both return Rx-observables from methods like `lidar_stream()`, downstream modules perceive no difference between recorded and live data. Changing from replay to live operation requires only modifying the connection initialization (e.g., switching `ip="replay"` to `ip="192.168.1.100"`) without touching perception, navigation, or control code.