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:

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

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

For runnable code, see the Examples at the bottom of this page.

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.

Recording Global Resources

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

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.

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:

# 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:

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.

Quality Alerts

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

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:

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:

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:

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 for the subscription APIs.

Overview

Hardware

Software

More

Data Recording and Offline Processing

Compression Options

Message Timestamps

Recording Global Resources

Pause and Resume

Episode Switching

Recording Monitoring

Quality Alerts

Recording Status

Recording Summary

Examples

Interface List

​Compression Options

​Message Timestamps

​Recording Global Resources

​Pause and Resume

​Episode Switching

​Recording Monitoring

​Quality Alerts

​Recording Status

​Recording Summary

​Examples

​Interface List

Compression Options

Message Timestamps

Recording Global Resources

Pause and Resume

Episode Switching

Recording Monitoring

Quality Alerts

Recording Status

Recording Summary

Examples

Interface List