> ## Documentation Index
> Fetch the complete documentation index at: https://docs.wuji.tech/llms.txt
> Use this file to discover all available pages before exploring further.

# Data Recording and Offline Processing

Real-time sensor data (tactile, IMU, hand tracking, and more) is ephemeral—recording persists it to files for offline analysis, algorithm debugging, or dataset building. Wuji SDK includes a built-in recording engine that synchronously writes multi-channel sensor data to MCAP files. The workflow takes three steps:

1. **Create a recorder** — Create a `TopicRecorder` and choose a [compression algorithm](#compression-options)
2. **Register channels** — Call `recorder.record(sub)` to register subscription channels, supporting [device and global resources](#recording-global-resources) in the same session
3. **Start recording** — Call `await recorder.start(path)` to begin recording, which returns a `RecordingHandle` for control

During recording, [pause and resume](#pause-and-resume) at any time, or use [episode switching](#episode-switching) to split a continuous capture into separate files. The engine includes built-in [recording monitoring](#recording-monitoring) that tracks frame drop rate, inter-frame jitter, and sync offset in real time. When recording stops, a [recording summary](#recording-summary) is returned with aggregate statistics.

<Info>
  For runnable code, see the [Examples](#examples) at the bottom of this page.
</Info>

## Compression Options

Choose a compression algorithm with the `compression` parameter when creating `TopicRecorder`:

| Option | Description |
| - | - |
| `"lz4"` | Low-latency compression for real-time use (default) |
| `"zstd"` | High compression ratio for storage and archiving |
| `"none"` | No compression, fastest write speed |

## Message Timestamps

Each message in the MCAP file carries two time fields. `log_time` is when the host received the frame. `publish_time` comes from the frame's `FrameHeader` `timestamp_us`, converted to nanoseconds, so it preserves the event time on the device. Keeping the two apart lets downstream tools align streams on `publish_time` without transport and queuing latency for messages that carry a valid device-side timestamp. Messages that fall back to the host receive time are not covered by this alignment.

Resources without a `FrameHeader` (such as plain scalar parameters) and frames whose `timestamp_us` is 0 fall back to the host receive time for `publish_time`. The meaning of `timestamp_us` depends on the time sync state. See [Time Synchronization](/en/docs/wuji-sdk/time-sync/).

## Recording Global Resources

Cross-device merged global resources (such as coordinate transforms) can also be recorded, mixed with device resources:

```python theme={null}
recorder.record(manager.tf_static().subscribe())  # Global resource
recorder.record(glove.tactile().subscribe())       # Device resource
```

## Pause and Resume

Pause and resume recording at any time. Device data continues to stream while paused, but nothing is written to the file.

```python theme={null}
await handle.pause()     # Pause — data is not written
await handle.resume()    # Resume recording
```

## Episode Switching

Call `start()` multiple times on the same `TopicRecorder` to switch output files without re-registering channels. Use this to split a continuous capture session into separate recording segments:

```python theme={null}
# Episode 1
handle1 = await recorder.start("./data/episode_001.mcap")
await asyncio.sleep(10)
await handle1.stop()

# Episode 2 — reuses the same channel configuration
handle2 = await recorder.start("./data/episode_002.mcap")
await asyncio.sleep(10)
await handle2.stop()
```

## Recording Monitoring

Monitor data quality metrics in real time during recording:

```python theme={null}
async for metrics in handle.subscribe_metrics():
    print(f"Drop rate: {metrics.frame_drop_rate:.4f}")
    print(f"Jitter: {metrics.frame_jitter_us:.1f} us")
    print(f"Sync offset: {metrics.sync_offset_ms:.2f} ms")
```

For the full field list, including per-channel health details in `channels`, see [Common Data Types — QualityMetrics](/en/docs/wuji-sdk/data-reference/#qualitymetrics).

### Quality Alerts

The recording engine includes a built-in SPC alert mechanism that triggers when quality metrics exceed thresholds for a sustained period:

```python theme={null}
async for alert in handle.subscribe_alerts():
    print(f"Alert: {alert.message}")
    print(f"  Metric: {alert.metric}, Current: {alert.current_value}, Threshold: {alert.threshold}")
```

### Recording Status

Monitor the runtime state of a recording:

```python theme={null}
async for status in handle.subscribe_status():
    print(f"State: {status.state}, Frames: {status.frame_count}, Duration: {status.duration_s:.1f}s")
```

## Recording Summary

`handle.stop()` returns a `RecordingSummary` with recording statistics:

```python theme={null}
summary = await handle.stop()
print(f"Total frames: {summary.total_frames}")
print(f"File size: {summary.file_size / 1024 / 1024:.2f} MB")
print(f"Duration: {summary.duration_s:.1f}s")
print(f"Drop rate: {summary.quality.frame_drop_rate:.4f}")
print(f"Sync offset: {summary.quality.avg_sync_offset_ms:.2f} ms")
```

## Examples

Demonstrates pause/resume, recording monitoring, episode switching, and recording summary:

```python theme={null}
import asyncio
import os
from datetime import datetime
from wuji_sdk import SdkManager, TopicRecorder

async def monitor_quality(handle):
    """Monitor recording quality metrics"""
    try:
        async for metrics in handle.subscribe_metrics():
            if metrics.frame_drop_rate > 0.01:
                print(f"  [Quality] Drop rate: {metrics.frame_drop_rate:.4f}, "
                      f"Jitter: {metrics.frame_jitter_us:.1f}us, "
                      f"Sync offset: {metrics.sync_offset_ms:.2f}ms")
    except asyncio.CancelledError:
        pass

async def main():
    manager = SdkManager.instance()
    os.makedirs("./data", exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

    # Episode 1: record 10 seconds with pause/resume demo
    glove = manager.auto_connect(device_name="glove")
    try:
        recorder = TopicRecorder(compression="lz4")
        recorder.record(glove.tactile().subscribe())
        recorder.record(glove.emf_poses().subscribe())
        recorder.record(glove.hand_skeleton().subscribe())

        path1 = f"./data/{timestamp}_episode_001.mcap"
        handle = await recorder.start(path1)
        metrics_task = asyncio.create_task(monitor_quality(handle))

        print(f"Recording: {path1}")
        await asyncio.sleep(5)

        print("Pausing...")
        await handle.pause()
        await asyncio.sleep(2)

        print("Resuming...")
        await handle.resume()
        await asyncio.sleep(5)

        metrics_task.cancel()
        try:
            await metrics_task
        except asyncio.CancelledError:
            pass
        summary = await handle.stop()
        print(f"Episode 1: {summary.total_frames} frames, "
              f"{summary.file_size / 1024 / 1024:.2f} MB, "
              f"{summary.duration_s:.1f}s")

        # Episode 2: switch output file, reuse the same recorder
        path2 = f"./data/{timestamp}_episode_002.mcap"
        handle2 = await recorder.start(path2)
        print(f"\nRecording: {path2}")
        await asyncio.sleep(5)

        summary2 = await handle2.stop()
        print(f"Episode 2: {summary2.total_frames} frames, "
              f"{summary2.file_size / 1024 / 1024:.2f} MB, "
              f"{summary2.duration_s:.1f}s")
    finally:
        glove.disconnect()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nStopped by user")
```

Expected output:

```
Recording: ./data/20260317_153943_episode_001.mcap
Pausing...
Resuming...
Episode 1: 3606 frames, 3.91 MB, 12.0s

Recording: ./data/20260317_153943_episode_002.mcap
Episode 2: 1812 frames, 1.97 MB, 5.0s
```

## Interface List

| Operation | Python | C |
| - | - | - |
| Create a recorder | `TopicRecorder(compression=..., chunk_size=...)` | No equivalent |
| Register a channel | `recorder.record(sub)` | No equivalent |
| Start recording | `await recorder.start(output_path)` | No equivalent |
| Pause and resume | `await handle.pause()`, `await handle.resume()` | No equivalent |
| Stop and read the summary | `await handle.stop()` | No equivalent |
| Subscribe to quality metrics | `handle.subscribe_metrics()` | No equivalent |
| Subscribe to recording status | `handle.subscribe_status()` | No equivalent |
| Subscribe to quality alerts | `handle.subscribe_alerts()` | No equivalent |

The recording engine ships in the Python API only, so the code samples on this page are Python only. A C program can reach the same frames through `wuji_<device>_subscribe_<stream>()`, but the on-disk format, quality statistics, and episode switching are up to the application. See the [C Interface Reference](/en/docs/wuji-sdk/data-reference/) for the subscription APIs.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.