数据录制

Wuji SDK 将多个可订阅资源写入同一个 MCAP 文件，供离线分析、算法调试和数据集构建使用。录制期间可暂停、恢复、监控数据质量或切换输出文件，停止后可查看统计摘要。

录制器配置

压缩选项

创建录制器时可分别设置压缩算法和 MCAP 分块大小。分块大小是每块未压缩数据的目标值，默认 4 MiB。默认使用 LZ4 压缩。

压缩算法	用途
LZ4	低延迟压缩，默认选项
Zstd	更高压缩比，适合存储归档
不压缩	不对数据进行压缩

Python
C

compression 接受 "lz4"、"zstd" 或 "none"。省略 chunk_size 使用默认大小。

from wuji_sdk import TopicRecorder

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

compression 接受 WUJI_RECORDING_COMPRESSION_LZ4、WUJI_RECORDING_COMPRESSION_ZSTD 或 WUJI_RECORDING_COMPRESSION_NONE。chunk_size=0 使用默认大小，options 传入 NULL 使用全部默认值。以下 C 片段中的 check、wait_seconds 和 cleanup 定义见示例。

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;

注册录制通道

每个通道对应一个可订阅的公开资源。先按设备发现与连接连接设备。设备通道在 MCAP 中以设备序列号为前缀，重复注册同一通道会报错。按设备选择以下资源：

设备	Python 订阅	C 资源路径
Wuji Hand 2	`hand2.joint_states().subscribe()`、`hand2.joint_diagnostics().subscribe()`	`joint_states`、`joint_diagnostics`
Wuji Glove	`glove.emf_poses().subscribe()`、`glove.imu_palm().subscribe()`	`emf_poses`、`imu_raw/palm`

下面以 Wuji Hand 2 为例，演示如何注册录制通道。录制 Wuji Glove 时，使用上表对应的资源，完整程序的修改步骤见示例。

Python
C

传入订阅对象，录制器读取其中的资源与设备信息。

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

传入设备句柄和公开资源路径，无需先创建 C 订阅。

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

录制全局资源

全局资源可与设备通道注册到同一录制器。例如，tf_static 保存全局静态坐标变换，其 MCAP 通道名不加设备前缀。

Python
C

使用连接示例中的 manager = SdkManager.instance()。

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

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

录制会话

开始录制

至少注册 1 个通道后，指定输出路径开始录制。每次启动都返回独立的录制会话，用于后续控制和监控。

启动会覆盖已有同名文件，且不会创建父目录。先创建输出目录，并为每次录制选择独立文件名。启动失败可能留下不完整文件。

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;

暂停与恢复

暂停、恢复操作均等待状态变更生效后完成。暂停期间仍接收设备数据，但不写入文件，也不在恢复后补录。

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 切换

同一录制器可重复启动，无需重新注册通道。每段停止后使用新路径开始下一段，生成独立的 MCAP 文件。以下每段录制 10 秒。

Python
C

在异步函数中运行，使用 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")

先将 handle 初始化为 NULL、summary 初始化为 {0}。每段结束后释放摘要与会话句柄，保留录制器。

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;
}

录制监控

录制期间监控丢帧率、帧间抖动和同步偏差。状态与质量指标只保留最新值，不能用通知次数计算录制帧数。完整指标与各通道健康明细见QualityMetrics。

Python
C

在异步函数中创建后台任务，同时继续执行录制控制。

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))

在文件作用域定义以下函数。开始录制后，在应用线程中调用 watch_metrics(handle, &metrics_sub) 并检查返回值，metrics_sub 初始为 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));
}

质量告警

质量指标持续超过阈值时，统计过程控制（SPC）监控发出告警，包含指标、当前值、阈值和说明。

Python
C

消费速度不足时，SDK 跳过未及时消费的告警并记录丢失数量，迭代继续。

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))

在应用线程中调用 watch_alerts(handle, &alerts_sub) 并检查返回值，alerts_sub 初始为 NULL。消费不及时导致告警丢失时，WUJI_FRAME_KIND_LAG 提供丢失数量。

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));
}

录制状态

状态通知包含运行状态、已写入消息数和经过时间。

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))

在应用线程中调用 watch_status(handle, &status_sub) 并检查返回值，status_sub 初始为 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));
}

监控生命周期

Python
C

结束监控时取消并等待任务。取消任务不会停止录制，监控流结束也不代表文件写入成功。调用 await handle.stop() 完成文件写入并检查错误。

每次 watch_* 调用返回独立的监控句柄，并使用独立的 SDK 线程。回调数据与字符串仅在本次调用中有效，需要保留时立即复制。

初始回调可能在 watch_* 返回前执行。注册前初始化 user_data，并保持其有效，直到关闭监控或停止、释放会话的调用返回
成功收尾发送 END，失败发送 ERROR。终止事件不释放监控句柄，仍需调用 wuji_recording_subscription_close
录制回调可以暂停或恢复其他录制会话，不能暂停或恢复当前会话
停止、释放任何录制会话，以及关闭任何录制监控，都需交由应用线程执行。在录制回调中，stop 和监控 close 返回 WUJI_STATUS_ERR_INVALID_ARG，free 设置 wuji_last_error() 且不释放句柄
若普通 Topic 订阅的回调可能停止或释放当前录制会话，不要从该录制回调中关闭这个 Topic 订阅，以免双方互相等待
较慢的回调会延迟 stop、free 和 close 返回。并发使用句柄时，由调用方同步释放操作

录制结果与资源释放

录制摘要

停止录制后获取总消息数、文件大小、持续时间、质量统计和逐通道消息数。通道按 ID 排序，包含未收到消息的通道。total_frames 是各 MCAP 通道的消息数之和。录制 Wuji Hand 2 的 joint_states 和 joint_diagnostics 时，两路消息分别计数，不能将总数当作设备原始采样数。duration_s 包含暂停和缓冲数据处理时间，不包含文件收尾、刷新及等待回调结束的时间。

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")

摘要先零初始化。使用后调用 wuji_recording_summary_free 释放内部数组和字符串，再复用该结构。

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);

消息时间戳

MCAP 中每条消息包含 log_time 和 publish_time。log_time 是主机收到消息的时刻。publish_time 取 FrameHeader.timestamp_us 并换算为纳秒，保留设备侧事件时间。对带有效设备侧时间戳的消息，下游可按 publish_time 对齐多路数据流，避免传输与排队延迟影响对齐。不带 FrameHeader 的资源（如普通标量参数），以及 timestamp_us=0 的帧，publish_time 回退为主机接收时刻，不适用上述对齐方式。设备时间戳的含义与同步状态有关，见时间同步与坐标变换。

控制与资源释放

显式停止录制并检查结果，然后结束监控、释放资源，最后断开设备。重复停止同一会话会报错。

Python
C

录制操作通过异常报告错误。以下示例结束前文创建的 3 个监控任务，应用应在 finally 中执行清理。

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

即使录制控制中途失败，也应在 finally 中停止录制。文件收尾错误由 await handle.stop() 抛出。

在录制回调之外关闭前文创建的 3 个监控句柄，再释放会话与录制器。stop 不释放这些句柄。

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 错误映射为 WUJI_STATUS_ERR_IO（-13），通过 wuji_last_error() 获取原因。录制期间及停止录制时写入剩余缓冲数据所产生的逐帧写入错误只记入 SDK 日志，stop 不会汇总返回这些错误。释放仍在运行的会话会同步完成文件收尾，但 wuji_recording_handle_free 没有状态码返回值，隐式收尾失败只记入日志。需要检查文件收尾结果时，先显式调用 stop，并查看 SDK 日志中的写入错误。开始录制后，释放录制器不会停止已启动的会话。

示例

以下示例以 Wuji Hand 2 为例。程序通过序列号连接设备，将 joint_states 和 joint_diagnostics 写入 MCAP 文件。录制 5 秒后暂停 2 秒，再录制 5 秒，最后输出总消息数和各通道消息数。运行时传入 DEVICE_SN 和输出文件路径，输出目录需预先存在。

Python
C

保存为 record_topics.py，运行 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):
    # 连接设备
    manager = SdkManager.instance()
    hand2 = manager.connect(sn=sn, device_name="recording_hand2")
    try:
        # 配置录制器并注册通道
        recorder = TopicRecorder(compression="lz4")
        recorder.record(hand2.joint_states().subscribe())
        recorder.record(hand2.joint_diagnostics().subscribe())

        # 开始录制，暂停 2 秒后继续
        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:
            # 停止录制并完成文件写入
            summary = await handle.stop()

        # 查看总消息数和各通道消息数
        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]))

保存为 record_topics.c。使用同一 SDK 发行包中的 wuji_sdk.h 和动态库，编译方式见 C 接口参考。示例使用 POSIX nanosleep，适用于 Linux 和 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;

    // 连接设备
    if (!check(wuji_connect(&target, "recording_hand2", NULL, &device))) goto cleanup;

    // 配置录制器并注册通道
    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;

    // 开始录制，暂停 2 秒后继续
    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_attempted = 1;
    if (!check(wuji_recording_handle_stop(handle, &summary))) goto cleanup;

    // 查看总消息数和各通道消息数
    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:
    // 失败时尝试停止录制，最后释放资源
    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;
}

保存 record_topics.c，将 SDK 设为解压后的 C SDK 包根目录的绝对路径。在文件所在目录中运行以下命令，将 DEVICE_SN 替换为设备序列号：

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

运行完成后，终端应输出总消息数和各通道消息数，指定路径应生成非空 MCAP 文件。若 Total frames 大于 0 且各通道有消息，说明所选资源的数据已写入文件。 录制 Wuji Glove 将示例中的设备连接和通道注册改为 Wuji Glove 对应的代码，录制控制步骤保持不变。

Python
C

在 record_topics.py 中修改：

将设备连接语句改为 glove = manager.connect(sn=sn, device_name="recording_glove")。
将两条通道注册语句改为 recorder.record(glove.emf_poses().subscribe()) 和 recorder.record(glove.imu_palm().subscribe())。
将 hand2.disconnect() 改为 glove.disconnect()。

在 record_topics.c 的 main 函数中，找到以下代码：

    // 连接设备
    if (!check(wuji_connect(&target, "recording_hand2", NULL, &device))) goto cleanup;

    // 配置录制器并注册通道
    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;

将这段代码替换为：

    // 连接设备
    if (!check(wuji_connect(&target, "recording_glove", NULL, &device))) goto cleanup;

    // 配置录制器并注册通道
    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;

重新运行对应示例，将 DEVICE_SN 换成 Wuji Glove 的序列号，并使用不同的输出路径，避免覆盖前一次录制文件。

接口清单

操作	Python	C
创建录制器	`TopicRecorder(...)`	`wuji_topic_recorder_create`
注册录制通道	`recorder.record(sub)`	`wuji_topic_recorder_record`
开始录制	`await recorder.start(path)`	`wuji_topic_recorder_start`
暂停与恢复	`await handle.pause()` / `await handle.resume()`	`wuji_recording_handle_pause` / `wuji_recording_handle_resume`
停止并取摘要	`await handle.stop()`	`wuji_recording_handle_stop`
订阅质量指标	`handle.subscribe_metrics()`	`wuji_recording_handle_watch_metrics`
订阅录制状态	`handle.subscribe_status()`	`wuji_recording_handle_watch_status`
订阅质量告警	`handle.subscribe_alerts()`	`wuji_recording_handle_watch_alerts`
关闭监控句柄	—	`wuji_recording_subscription_close`
释放录制摘要	—	`wuji_recording_summary_free`
释放会话与录制器	—	`wuji_recording_handle_free` / `wuji_topic_recorder_free`

“—”表示 Python 没有对应的手动释放接口。

概览

硬件

软件

更多

录制器配置

压缩选项

注册录制通道

录制全局资源

录制会话

开始录制

暂停与恢复

Episode 切换

录制监控

质量告警

录制状态

监控生命周期

录制结果与资源释放

录制摘要

消息时间戳

控制与资源释放

示例

接口清单

​录制器配置

​压缩选项

​注册录制通道

​录制全局资源

​录制会话

​开始录制

​暂停与恢复

​Episode 切换

​录制监控

​质量告警

​录制状态

​监控生命周期

​录制结果与资源释放

​录制摘要

​消息时间戳

​控制与资源释放

​示例

​接口清单

录制器配置

压缩选项

注册录制通道

录制全局资源

录制会话

开始录制

暂停与恢复

Episode 切换

录制监控

质量告警

录制状态

监控生命周期

录制结果与资源释放

录制摘要

消息时间戳

控制与资源释放

示例

接口清单