Data Recording

Developer Guide

Wuji SDK writes multiple subscribable resources to one MCAP file for offline analysis, algorithm debugging, and dataset creation. During recording, you can pause, resume, monitor data quality, or switch output files. After stopping, you can view a statistical summary.

Recorder Configuration

Compression Options

Set the compression algorithm and MCAP chunk size independently. The default chunk size targets 4 MiB of uncompressed data per chunk. LZ4 is the default compression algorithm.

• Python

• C

compression accepts "lz4", "zstd", or "none". Omit chunk_size to use the default.

from wuji_sdk import TopicRecorder

recorder = TopicRecorder(compression="lz4", chunk_size=4 * 1024 * 1024)

compression accepts WUJI_RECORDING_COMPRESSION_LZ4, WUJI_RECORDING_COMPRESSION_ZSTD, or WUJI_RECORDING_COMPRESSION_NONE. A zero chunk_size uses the default size. Pass NULL as options for all defaults.

The check, wait_seconds, and cleanup helpers used in the C snippets are defined in the Examples section.

WujiTopicRecorder *recorder = NULL;
WujiRecordingOptions options = {
    .compression = WUJI_RECORDING_COMPRESSION_LZ4,
    .chunk_size = 4 * 1024 * 1024,
};
if (!check(wuji_topic_recorder_create(&options, &recorder))) goto cleanup;

Register Channels

Each channel represents a public resource that supports subscriptions. Connect a device as described in Device Discovery and Connection. Device channel names in MCAP have a serial-number prefix. Registering the same channel twice returns an error. Choose resources for the connected device:

The following examples show how to register recording channels for Wuji Hand 2. To record Wuji Glove, use the corresponding resources in the table above and follow the substitutions in Examples.

• Python

• C

Pass a subscription object. The recorder reads its resource and device metadata.

recorder.record(hand2.joint_states().subscribe())
recorder.record(hand2.joint_diagnostics().subscribe())

Pass a device handle and a public resource path. You don’t need an existing C subscription.

if (!check(wuji_topic_recorder_record(recorder, device, "joint_states"))) goto cleanup;
if (!check(wuji_topic_recorder_record(recorder, device, "joint_diagnostics"))) goto cleanup;

Recording Global Resources

Register global resources alongside device channels. For example, tf_static records global static transforms without a device prefix in its MCAP channel name.

• Python

• C

Use manager = SdkManager.instance() from the connection example.

recorder.record(manager.tf_static().subscribe())

if (!check(wuji_topic_recorder_record(recorder, NULL, "tf_static"))) goto cleanup;

Recording Session

Start Recording

Register at least one channel, then start recording with an output path. Each start returns an independent session for control and monitoring.

• Python

• C

handle = await recorder.start("./data/session.mcap")

WujiRecordingHandle *handle = NULL;
if (!check(wuji_topic_recorder_start(recorder, "./data/session.mcap", &handle))) goto cleanup;

Pause and Resume

Pause and resume complete after the state change takes effect. Device data still arrives during a pause, but it isn’t written to the file or replayed when recording resumes.

• Python

• C

await handle.pause()
# ...
await handle.resume()

if (!check(wuji_recording_handle_pause(handle))) goto cleanup;
// ...
if (!check(wuji_recording_handle_resume(handle))) goto cleanup;

Episode Switching

Reuse a recorder without registering its channels again. Stop each episode, then start the next with a new path to create a separate MCAP file. Each episode below records for 10 seconds.

• Python

• C

Run this inside an async function with import asyncio.

for path in ("./data/episode_001.mcap", "./data/episode_002.mcap"):
    handle = await recorder.start(path)
    try:
        await asyncio.sleep(10)
    finally:
        summary = await handle.stop()
    print(f"{path}: {summary.total_frames} frames")

Initialize handle to NULL and summary to {0}. Free each episode’s summary and session handle, and retain the recorder.

const char *paths[] = {"./data/episode_001.mcap", "./data/episode_002.mcap"};
for (size_t i = 0; i < 2; ++i) {
    if (!check(wuji_topic_recorder_start(recorder, paths[i], &handle))) goto cleanup;
    if (!wait_seconds(10)) goto cleanup;
    if (!check(wuji_recording_handle_stop(handle, &summary))) goto cleanup;
    printf("%s: %" PRIu64 " frames\n", paths[i], summary.total_frames);
    wuji_recording_summary_free(&summary);
    wuji_recording_handle_free(handle);
    handle = NULL;
}

Recording Monitoring

Monitor frame drop rate, inter-frame jitter, and sync offset during recording. Status and quality updates coalesce to the latest value, so don’t count notifications to measure recorded frames. See QualityMetrics for all fields and per-channel health details.

• Python

• C

Create a background task inside an async function so recording control can continue.

async def monitor_metrics(handle):
    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")

metrics_task = asyncio.create_task(monitor_metrics(handle))

Define these functions at file scope. After starting recording, call watch_metrics(handle, &metrics_sub) from an application thread and check its return value. Initialize metrics_sub to NULL.

static void on_metrics(WujiFrameKind kind, const WujiRecordingMetrics *value,
                        uint64_t lag_count, const char *error, void *user_data) {
    (void)user_data;
    (void)lag_count;
    if (kind == WUJI_FRAME_KIND_OK) {
        printf("Drop rate: %.4f, jitter: %.1f us, sync offset: %.2f ms\n",
               value->frame_drop_rate, value->frame_jitter_us, value->sync_offset_ms);
    } else if (kind == WUJI_FRAME_KIND_ERROR) {
        fprintf(stderr, "%s\n", error);
    }
}

static int watch_metrics(WujiRecordingHandle *handle,
                        WujiRecordingSubscription **sub) {
    return check(wuji_recording_handle_watch_metrics(handle, on_metrics, NULL, sub));
}

Quality Alerts

Statistical process control (SPC) monitoring emits alerts when quality metrics exceed thresholds for a sustained period. Each alert includes the metric, current value, threshold, and message.

• Python

• C

If the consumer falls behind, the SDK skips missed alerts, logs the count, and continues the iterator.

async def monitor_alerts(handle):
    async for alert in handle.subscribe_alerts():
        print(f"{alert.message}: {alert.metric}={alert.current_value}, "
              f"threshold={alert.threshold}")

alerts_task = asyncio.create_task(monitor_alerts(handle))

Call watch_alerts(handle, &alerts_sub) from an application thread and check its return value. Initialize alerts_sub to NULL. If the consumer falls behind, WUJI_FRAME_KIND_LAG reports the number of dropped alerts.

static void on_alerts(WujiFrameKind kind, const WujiRecordingAlert *value,
                        uint64_t lag_count, const char *error, void *user_data) {
    (void)user_data;
    if (kind == WUJI_FRAME_KIND_OK) {
        printf("%s: %s=%g, threshold=%g\n",
               value->message, value->metric, value->current_value, value->threshold);
    } else if (kind == WUJI_FRAME_KIND_LAG) {
        fprintf(stderr, "Dropped alerts: %" PRIu64 "\n", lag_count);
    } else if (kind == WUJI_FRAME_KIND_ERROR) {
        fprintf(stderr, "%s\n", error);
    }
}

static int watch_alerts(WujiRecordingHandle *handle,
                        WujiRecordingSubscription **sub) {
    return check(wuji_recording_handle_watch_alerts(handle, on_alerts, NULL, sub));
}

Recording Status

Status updates include the recording state, written message count, and elapsed time.

• Python

• C

async def monitor_status(handle):
    async for status in handle.subscribe_status():
        print(f"State: {status.state}, frames: {status.frame_count}, "
              f"duration: {status.duration_s:.1f} s")

status_task = asyncio.create_task(monitor_status(handle))

Call watch_status(handle, &status_sub) from an application thread and check its return value. Initialize status_sub to NULL.

static void on_status(WujiFrameKind kind, const WujiRecordingStatus *value,
                        uint64_t lag_count, const char *error, void *user_data) {
    (void)user_data;
    (void)lag_count;
    if (kind == WUJI_FRAME_KIND_OK) {
        printf("State: %d, frames: %" PRIu64 ", duration: %.1f s\n",
               (int)value->state, value->frame_count, value->duration_s);
    } else if (kind == WUJI_FRAME_KIND_ERROR) {
        fprintf(stderr, "%s\n", error);
    }
}

static int watch_status(WujiRecordingHandle *handle,
                        WujiRecordingSubscription **sub) {
    return check(wuji_recording_handle_watch_status(handle, on_status, NULL, sub));
}

Monitor Lifecycle

• Python

• C

Cancel and await monitoring tasks when monitoring ends. Canceling a task doesn’t stop recording. A stream ending doesn’t confirm that the file was written successfully. Call await handle.stop() to finish the file and check for errors.

Each watch_* call returns a separate monitor handle and uses a dedicated SDK thread. Callback data and strings are valid only during that call. Copy any data you need to retain.

• An initial callback can run before watch_* returns. Initialize user_data before registration, and keep it valid until monitor close or session stop/free returns
• Successful finalization sends END, and failure sends ERROR. Terminal events don’t free monitor handles. Call wuji_recording_subscription_close to release them
• A recording callback can pause or resume other recording sessions, but not its own session
• Stop or free any recording session, and close any recording monitor, from an application thread. Inside a recording callback, stop and monitor close return WUJI_STATUS_ERR_INVALID_ARG. free sets wuji_last_error() without freeing the handle
• Don’t close a regular topic subscription from a recording callback if that subscription’s callback might stop or free the recording session. The callbacks could wait for each other
• Slow callbacks delay stop, free, and close. Synchronize handle cleanup with concurrent use

Recording Results and Cleanup

Recording Summary

Stop recording to retrieve the total message count, file size, duration, quality statistics, and per-channel message counts. Channels are sorted by ID and include channels with no messages. total_frames sums MCAP messages across channels. When recording Wuji Hand 2 joint_states and joint_diagnostics, count each stream separately. The total does not represent the number of raw device samples. duration_s includes pauses and buffer draining, but excludes file finalization, flushing, and waiting for callbacks to exit.

• Python

• C

summary = await handle.stop()
print(f"Total frames: {summary.total_frames}")
print(f"File size: {summary.file_size / 1024 / 1024:.2f} MiB")
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")
for channel in summary.channels:
    print(f"{channel.channel_name}: {channel.frame_count} frames")

Zero-initialize the summary. After use, call wuji_recording_summary_free to release its arrays and strings before reusing the structure.

WujiRecordingSummary summary = {0};
if (!check(wuji_recording_handle_stop(handle, &summary))) goto cleanup;
printf("Total frames: %" PRIu64 "\n", summary.total_frames);
printf("File size: %.2f MiB\n", (double)summary.file_size / 1024 / 1024);
printf("Duration: %.1f s\n", summary.duration_s);
printf("Drop rate: %.4f\n", summary.quality.frame_drop_rate);
printf("Sync offset: %.2f ms\n", summary.quality.avg_sync_offset_ms);
for (size_t i = 0; i < summary.channels_len; ++i) {
    printf("%s: %" PRIu64 " frames\n", summary.channels[i].channel_name,
           summary.channels[i].frame_count);
}
wuji_recording_summary_free(&summary);

Message Timestamps

Each MCAP message includes log_time and publish_time. log_time is when the host receives the message. publish_time comes from FrameHeader.timestamp_us, converted to nanoseconds to preserve the device-side event time. For messages with valid device timestamps, downstream tools can align streams on publish_time without transport and queuing latency.

For resources without a FrameHeader (such as scalar parameters) and frames with timestamp_us=0, publish_time falls back to the host receive time. The alignment described above doesn’t apply to those messages. Device timestamp meaning depends on synchronization state. See Time Synchronization and Coordinate Transforms.

Control and Cleanup

Stop recording explicitly and check the result, then finish monitoring, release resources, and disconnect the device. Stopping the same session again returns an error.

• Python

• C

Recording operations report errors through exceptions. This snippet finishes the three monitoring tasks created above. Run cleanup in a finally block.

tasks = (metrics_task, alerts_task, status_task)
for task in tasks:
    task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)

Stop recording in a finally block even if recording control fails. await handle.stop() raises file-finalization errors.

Outside recording callbacks, close the three monitor handles created above, then free the session and recorder. stop doesn’t free these handles.

if (check(wuji_recording_subscription_close(metrics_sub))) metrics_sub = NULL;
if (check(wuji_recording_subscription_close(alerts_sub))) alerts_sub = NULL;
if (check(wuji_recording_subscription_close(status_sub))) status_sub = NULL;
wuji_recording_handle_free(handle);
handle = NULL;
wuji_topic_recorder_free(recorder);
recorder = NULL;

I/O errors returned by API operations such as file creation and finalization map to WUJI_STATUS_ERR_IO (-13). Read the details with wuji_last_error(). Per-frame write errors during recording and when writing remaining buffered data at stop are only logged in the SDK logs. stop doesn’t aggregate and return these errors. Freeing an active session synchronously finalizes the file, but wuji_recording_handle_free returns no status code and only logs implicit finalization errors. Call stop explicitly to check the file finalization result, and check the SDK logs for write errors. Freeing the recorder after starting a session doesn’t stop that session.

Examples

The following examples use Wuji Hand 2. They connect to the device by serial number and write joint_states and joint_diagnostics to an MCAP file. They record for 5 seconds, pause for 2 seconds, record for another 5 seconds, and print total and per-channel message counts. Pass DEVICE_SN and an output file path as arguments. The output directory must already exist.

• Python

• C

Save as record_topics.py, then run python record_topics.py DEVICE_SN ./data/session.mcap.

import asyncio
import sys
from wuji_sdk import SdkManager, TopicRecorder

async def main(sn, output_path):
    # Connect to the device
    manager = SdkManager.instance()
    hand2 = manager.connect(sn=sn, device_name="recording_hand2")
    try:
        # Configure the recorder and register channels
        recorder = TopicRecorder(compression="lz4")
        recorder.record(hand2.joint_states().subscribe())
        recorder.record(hand2.joint_diagnostics().subscribe())

        # Start recording, pause for 2 seconds, then resume
        handle = await recorder.start(output_path)
        try:
            await asyncio.sleep(5)
            await handle.pause()
            await asyncio.sleep(2)
            await handle.resume()
            await asyncio.sleep(5)
        finally:
            # Stop recording and finish writing the file
            summary = await handle.stop()

        # Show total and per-channel message counts
        print(f"Total frames: {summary.total_frames}")
        for channel in summary.channels:
            print(f"{channel.channel_name}: {channel.frame_count} frames")
    finally:
        hand2.disconnect()

if __name__ == "__main__":
    if len(sys.argv) != 3:
        raise SystemExit(f"Usage: {sys.argv[0]} DEVICE_SN OUTPUT.mcap")
    asyncio.run(main(sys.argv[1], sys.argv[2]))

Save as record_topics.c. Use wuji_sdk.h and the shared library from the same SDK release. See C SDK Reference for build instructions. The example uses POSIX nanosleep on Linux and macOS.

#define _POSIX_C_SOURCE 200809L
#include "wuji_sdk.h"
#include <errno.h>
#include <inttypes.h>
#include <stdio.h>
#include <time.h>

static int check(WujiStatus status) {
    if (status == WUJI_STATUS_OK) return 1;
    fprintf(stderr, "SDK error: %s\n", wuji_last_error());
    return 0;
}

static int wait_seconds(time_t seconds) {
    struct timespec remaining = {seconds, 0};
    while (nanosleep(&remaining, &remaining) != 0) {
        if (errno != EINTR) { perror("nanosleep"); return 0; }
    }
    return 1;
}

int main(int argc, char **argv) {
    if (argc != 3) {
        fprintf(stderr, "Usage: %s DEVICE_SN OUTPUT.mcap\n", argv[0]);
        return 2;
    }
    int result = 1;
    int stop_attempted = 0;
    WujiDevice *device = NULL;
    WujiTopicRecorder *recorder = NULL;
    WujiRecordingHandle *handle = NULL;
    WujiRecordingSummary summary = {0};
    WujiConnectTarget target = {
        .kind = WUJI_CONNECT_TARGET_KIND_SN, .value = argv[1]
    };
    if (!check(wuji_init(NULL))) return 1;

    // Connect to the device
    if (!check(wuji_connect(&target, "recording_hand2", NULL, &device))) goto cleanup;

    // Configure the recorder and register channels
    if (!check(wuji_topic_recorder_create(NULL, &recorder))) goto cleanup;
    if (!check(wuji_topic_recorder_record(recorder, device, "joint_states"))) goto cleanup;
    if (!check(wuji_topic_recorder_record(recorder, device, "joint_diagnostics"))) goto cleanup;

    // Start recording, pause for 2 seconds, then resume
    if (!check(wuji_topic_recorder_start(recorder, argv[2], &handle))) goto cleanup;
    if (!wait_seconds(5)) goto cleanup;
    if (!check(wuji_recording_handle_pause(handle))) goto cleanup;
    if (!wait_seconds(2)) goto cleanup;
    if (!check(wuji_recording_handle_resume(handle))) goto cleanup;
    if (!wait_seconds(5)) goto cleanup;

    // Stop recording and finish writing the file
    stop_attempted = 1;
    if (!check(wuji_recording_handle_stop(handle, &summary))) goto cleanup;

    // Show total and per-channel message counts
    printf("Total frames: %" PRIu64 "\n", summary.total_frames);
    for (size_t i = 0; i < summary.channels_len; ++i) {
        printf("%s: %" PRIu64 " frames\n", summary.channels[i].channel_name,
               summary.channels[i].frame_count);
    }
    result = 0;
cleanup:
    // Try to stop after an earlier failure, then release resources
    if (handle && !stop_attempted) {
        (void)check(wuji_recording_handle_stop(handle, &summary));
    }
    wuji_recording_summary_free(&summary);
    wuji_recording_handle_free(handle);
    wuji_topic_recorder_free(recorder);
    if (device) { wuji_dev_disconnect(device); wuji_dev_release(device); }
    wuji_shutdown();
    return result;
}

Save record_topics.c and set SDK to the absolute path of the extracted C SDK package root. From the directory containing the file, run the following commands. Replace DEVICE_SN with the device’s serial number:

mkdir -p data
cc record_topics.c -I "$SDK/include" -L "$SDK/lib" -Wl,-rpath,"$SDK/lib" -lwuji_sdk_c -o record_topics &&
./record_topics DEVICE_SN ./data/session.mcap

After the program finishes, the terminal should show the total and per-channel message counts, and the output path should contain a nonempty MCAP file. If Total frames is greater than 0 and each channel has messages, the selected resources were written to the file.

Recording Wuji Glove

Replace the device connection and channel registration code with the Wuji Glove versions. Keep the recording controls unchanged.

• Python

• C

In record_topics.py:

• Change the device connection to glove = manager.connect(sn=sn, device_name="recording_glove").
• Register recorder.record(glove.emf_poses().subscribe()) and recorder.record(glove.imu_palm().subscribe()).
• Replace hand2.disconnect() with glove.disconnect().

In the main function of record_topics.c, find this code:

    // Connect to the device
    if (!check(wuji_connect(&target, "recording_hand2", NULL, &device))) goto cleanup;

    // Configure the recorder and register channels
    if (!check(wuji_topic_recorder_create(NULL, &recorder))) goto cleanup;
    if (!check(wuji_topic_recorder_record(recorder, device, "joint_states"))) goto cleanup;
    if (!check(wuji_topic_recorder_record(recorder, device, "joint_diagnostics"))) goto cleanup;

Replace that block with:

    // Connect to the device
    if (!check(wuji_connect(&target, "recording_glove", NULL, &device))) goto cleanup;

    // Configure the recorder and register channels
    if (!check(wuji_topic_recorder_create(NULL, &recorder))) goto cleanup;
    if (!check(wuji_topic_recorder_record(recorder, device, "emf_poses"))) goto cleanup;
    if (!check(wuji_topic_recorder_record(recorder, device, "imu_raw/palm"))) goto cleanup;

Run the corresponding example again with the Wuji Glove serial number as DEVICE_SN. Use a different output path to avoid overwriting the previous recording.

Interface List

“—” means Python has no corresponding manual release API.