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

# 数据录制

## 概述

本篇实践介绍如何使用 `TopicRecorder` 将订阅的数据流保存为 MCAP 文件。应用场景包括记录实验数据、回看采集过程和准备离线分析数据。

对应的 Python 示例为 [2.recording.py](https://github.com/wuji-technology/wuji-sdk/blob/main/examples/python/wuji_glove/2.recording.py)。 录制接口见 [TopicRecorder 接口清单](/docs/wuji-sdk/data-recording#接口清单)。

<Info>
  C SDK 没有对应的录制接口，本页介绍 Python 用法。
</Info>

## 前置条件

### 硬件

| 项目 | 要求 |
| - | - |
| Wuji Glove | 1 只 |
| 连接方式 | 以太网 |
| 磁盘 | 按下文的体积实测预留，六条流开 zstd 约每分钟 70 MB |

本示例只读取，不写任何设备参数。

### 软件

| 项目 | 要求 |
| - | - |
| 操作系统 | Ubuntu 22.04 LTS 及以上版本 |
| Wuji SDK | v2026.8.31 或更高版本 |
| Python | 3.10 及以上版本 |

```bash theme={null}
pip install wuji-sdk
# 想在本地解析 mcap 再装
pip install mcap
```

### 前序步骤

先看懂[数据订阅：回调](/docs/wuji-glove/sdk-practice/subscribe-callback)。录制的输入就是 `subscribe()` 返回的订阅句柄，队列语义完全一致。

## 数据流

```mermaid theme={null}
sequenceDiagram
    participant U as 应用
    participant G as Wuji Glove
    participant S as 订阅流
    participant R as TopicRecorder
    participant F as MCAP 文件

    U->>G: connect(sn=...)
    U->>S: stream.subscribe()
    S-->>R: record(subscription)
    U->>R: start(path)
    R->>F: 持续写入并压缩消息
    U->>R: handle.stop()
    R->>F: 写完索引并关闭文件
    F-->>U: summary 与 quality
```

## 文件结构实测

录 6 条设备流加 2 条全局流、5 秒，产出文件的通道如下：

| 通道 | 消息数 | 说明 |
| - | - | - |
| `/WG1KA00XXXXXXXXX/tactile` | 659 | 设备流按 `/{序列号}/{话题名}` 命名 |
| `/WG1KA00XXXXXXXXX/tactile_zones` | 659 | |
| `/WG1KA00XXXXXXXXX/emf_poses` | 659 | |
| `/WG1KA00XXXXXXXXX/hand_joint_angles` | 655 | |
| `/WG1KA00XXXXXXXXX/hand_skeleton` | 655 | |
| `/WG1KA00XXXXXXXXX/tactile_point_cloud` | 654 | |
| `/tf` | 4361 | **全局流不带序列号前缀**，是跨设备聚合数据 |
| `/tf_static` | 6 | |

消息编码是 `json`，每条通道自带一份 schema。用 Foxglove Studio 打开能直接看到字段名，不需要额外的消息定义文件。

文件头的 `library` 字段实测为 `mcap-rs-0.24.0`，写入器是 SDK 内置的 Rust 实现。

## 压缩模式实测

同样录 3 条流、5 秒、约 1970 帧：

| `compression` | 文件体积 | 相对无压缩 |
| - | - | - |
| `"none"` | 5.63 MB | 100% |
| `"lz4"` | 1.71 MB | 30% |
| `"zstd"` | 0.92 MB | 16% |

JSON 编码的重复字段名压缩率很高。长时间录制默认选 `zstd`，对写入延迟敏感的场合选 `lz4`，需要边录边被别的进程流式读取时才用 `none`。

单独看 lz4 的一个 chunk，实测 3101 字节原始数据压到 938 字节。

## 质量指标实测

`handle.stop()` 返回的 summary 带一个 `quality` 对象：

```text theme={null}
QualitySummary(total_frames=8308, dropped=0, drop_rate=0.0000, sync_rate=0.8571)
```

完整字段与实测取值：

| 字段 | 含义 | 3 条流实测 | 6 条流实测 | 8 条流（含 tf）实测 |
| - | - | - | - | - |
| `total_frames` | 落盘总帧数 | 1973 | 3937 | 8308 |
| `dropped_frames` | 丢帧数 | 0 | 0 | 0 |
| `frame_drop_rate` | 丢帧率，5 秒滑动窗口 | 0.0000 | 0.0000 | 0.0000 |
| `sync_rate` | 跨通道同步成功率 | 0.998 | 1.000 | 0.857 |
| `avg_sync_offset_ms` | 平均同步偏移 | 7.89 | 5.23 | — |
| `max_sync_offset_ms` | 最大同步偏移 | 2966.23 | 13.24 | — |
| `spc_alert_count` | 统计过程控制告警数 | 0 | 0 | — |

三处值得注意：

**`sync_rate` 在混入 `tf` 后掉到 0.857。** 六条设备流都在 130 Hz 附近，彼此对得齐。`tf` 实测 872 Hz、`tf_static` 只有 1 Hz，与其他通道的节奏差一个量级，同步判定自然失败。录制里同时放不同频段的流，这个数字偏低是预期行为，不代表数据有问题。

**`max_sync_offset_ms` 可能出现一个很大的孤立值。** 3 条流那轮实测到 2966 ms，同一轮的 `avg` 只有 7.89 ms。这是录制刚启动、各通道首帧尚未对齐造成的。判断录制质量看 `avg` 和 `frame_drop_rate`，不要被 `max` 的启动毛刺带偏。

**`duration_s` 是挂钟时长，不是有效录制时长。** 用了暂停就会对不上，见下一节。

## 暂停与恢复

`handle` 除了 `stop()` 还有 `pause()` 和 `resume()`，示例没有展示。实测录 2 秒、暂停 2 秒、再录 2 秒，3 条流：

| 指标 | 实测值 |
| - | - |
| `duration_s` | 6.0（挂钟时长，含暂停） |
| `total_frames` | 1574 |
| 按 4 秒有效时长推算 | 3 流 × 131 Hz × 4 s ≈ 1572 |

帧数对应的是 4 秒有效录制，暂停期间一帧不写。`duration_s` 仍然按挂钟走，所以拿 `total_frames / duration_s` 反推输出率会偏低。

暂停恢复这轮 `spc_alert_count` 变成 1。暂停期间通道静默，统计过程控制把它识别成一次异常。这是预期的，不是故障。

`handle` 还提供 `subscribe_alerts()`、`subscribe_metrics()`、`subscribe_status()` 三个订阅接口，可以在录制过程中实时拿到告警、质量指标和状态，不用等到 `stop()`。

## 集成步骤

完整示例见 [Python 示例](https://github.com/wuji-technology/wuji-sdk/blob/main/examples/python/wuji_glove/2.recording.py)。以下片段按接入顺序展示关键操作，参数解析、信号处理和完整错误分支见示例。

### 1. 连接目标手套

将 `serial_number` 替换为目标手套的序列号。连接成功后，后续步骤复用同一个设备句柄。C 片段需包含 `wuji_sdk.h`。

```python theme={null}
from wuji_sdk import SdkManager
manager = SdkManager.instance()
glove = manager.connect(sn=serial_number, device_name="glove_0")
```

### 2. 登记录制通道

创建录制器，再把需要保存的订阅登记进去。以下操作放在异步函数中，设备连接在外层 `finally` 中断开。

```python theme={null}
from wuji_sdk import TopicRecorder
recorder = TopicRecorder(compression="lz4")
recorder.record(glove.tactile().subscribe())
recorder.record(glove.emf_poses().subscribe())
recorder.record(glove.hand_skeleton().subscribe())
```

### 3. 启动录制

先创建输出目录，再启动录制。保留返回的句柄，用它停止写入并取得录制摘要。

```python theme={null}
from pathlib import Path
Path("./data").mkdir(exist_ok=True)
handle = await recorder.start("./data/session.mcap")
```

### 4. 等待文件写完再断开设备

停止录制会完成文件写入。沿用 [Python 示例](https://github.com/wuji-technology/wuji-sdk/blob/main/examples/python/wuji_glove/2.recording.py) 的取消保护，确保任务取消时仍等待 `stop()` 完成，然后在外层清理设备。

```python theme={null}
import asyncio
try:
    await asyncio.sleep(10)
finally:
    stop_task = asyncio.ensure_future(handle.stop())
    try:
        summary = await asyncio.shield(stop_task)
    except asyncio.CancelledError:
        summary = await stop_task
    print(summary.total_frames, summary.file_size, summary.duration_s)
# 外层 finally 中执行 manager.disconnect_all()
```

## 运行与输出

```bash theme={null}
python examples/python/wuji_glove/2.recording.py
```

```text theme={null}
  SN=WG1KA00XXXXXXXXX, Type=DeviceType.WujiGlove, Address=192.168.1.100:50001
Connected: WG1KA00XXXXXXXXX
Recording to ./data/<date>_<time>.mcap ...
Done — 3939 frames, 3.43 MB, 10.0s
```

以上为实机运行结果，设备是右手 Wuji Glove，手套静置无人佩戴，录 3 条流 10 秒，lz4 压缩。3939 帧分摊到 3 条通道各约 1313 帧，等效每条 131 Hz。

## 注意事项

| 分支点 | 表现 | 处理 |
| - | - | - |
| 未做手部模型标定 | 日志打 `offline hand_model: using built-in default URDF for Right` | 录制照常进行，用的是内置默认手模。需要贴合本人手型见[手部模型标定](/docs/wuji-glove/sdk-practice/hand-model-calibration) |
| 停止时报 `Unknown interest` | 一条 ERROR 级日志 | 来自 zenoh 传输层关闭时的竞态，不影响已落盘的文件，可以忽略 |
| 目标目录不存在 | `start()` 失败 | 示例用 `os.makedirs(..., exist_ok=True)` 先建好 |
| 磁盘写入跟不上 | `dropped_frames` 增长 | 换 zstd 压缩，或减少登记的通道 |
| 设备中途断连 | 对应通道停止产出 | `quality` 里该通道的 `last_downtime_ms` 会记录 |
| 录制期间被 `Ctrl+C` | 示例用 `asyncio.shield` 保住 `stop()` | 保证文件正常收尾，不留半截文件 |

示例里这段是有意为之：

```python theme={null}
stop_task = asyncio.ensure_future(handle.stop())
try:
    summary = await asyncio.shield(stop_task)
except asyncio.CancelledError:
    summary = await stop_task
```

`asyncio.shield` 让 `stop()` 不被外层取消打断。少了这层保护，`Ctrl+C` 可能在 MCAP 写完索引之前就把任务砍掉，文件读不出摘要。

## 读取录好的文件

装了 `mcap` 命令行工具直接看：

```bash theme={null}
mcap info ./data/<date>_<time>.mcap
```

用 Python 逐帧解析：

```python theme={null}
from mcap.reader import make_reader

with open("./data/<date>_<time>.mcap", "rb") as fp:
    reader = make_reader(fp)
    summary = reader.get_summary()
    for cid, channel in summary.channels.items():
        n = summary.statistics.channel_message_counts.get(cid, 0)
        print(f"{channel.topic}: {n} 条")

    for schema, channel, message in reader.iter_messages(topics=["/WG1KA00XXXXXXXXX/tactile"]):
        print(message.log_time, len(message.data))
```

消息体是 JSON，直接 `json.loads(message.data)` 就能拿到与在线订阅一致的字段结构。

## 扩展用法

### 按需登记通道，别全开

八条流全录、zstd 压缩，实测 5 秒 6.01 MB，折算每分钟约 72 MB。`tf` 一条就占了 4361 条消息，比六条设备流加起来还多。录长时间数据前先想清楚哪些流真的要。

### 降低输出率再录

订阅句柄支持降频。把 130 Hz 降到 30 Hz，文件体积直接掉到四分之一以下，见[输出率调节](/docs/wuji-glove/sdk-practice/stream-rate)。

### 录制期间监控质量

不想等 `stop()` 才知道是否丢帧，用 `handle.subscribe_metrics()` 在线拿指标，`handle.subscribe_alerts()` 拿告警。长时间无人值守的采集应该接上这两个。

## 相关内容

* [数据订阅：回调](/docs/wuji-glove/sdk-practice/subscribe-callback)：六条流的帧结构与无效通道处理
* [数据订阅：异步](/docs/wuji-glove/sdk-practice/subscribe-async)：订阅队列的缓冲边界
* [离线数据处理](/docs/wuji-glove/sdk-practice/offline-replay)：把录好的文件重新跑一遍算法
* [输出率调节](/docs/wuji-glove/sdk-practice/stream-rate)：录制前降频以控制体积
* [数据订阅机制](/docs/wuji-sdk/data-subscription/)：订阅模型与生命周期


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