SDKPython SDK

数据来源与读取方式

两类数据的区别、callback、最新值、异步三种读取方式,以及外部输入来源。

当前版本 · 1.4.0

数据来源与读取方式

两类数据的区别

SDK 发布两类数据:设备直接产出的原始数据,和 SDK 计算出的数据。读取方式相同,区别只在数据怎么产生:

原始数据SDK 处理后的数据
典型 APIpose()、imu()、status()skeleton()、output()
需要算法包不需要灵巧手输出需要,见 算法包管理;Skeleton 由 SDK 提供,不需要
需要运行方式不需要需要,见 运行方式

原始数据见 读取原始数据,SDK 处理后的数据(Retargeting)见 读取 Retargeting 数据。

数据读取方式

三种读取方式拿到的是同一批 Sample[T]。推荐 callback 订阅:数据在产生时立即处理,不需要轮询;只在需要当前值时用 latest(),asyncio 程序用异步读取。

latest() 和 next() 取得 Sample[T];subscribe() 返回 Subscription,注册的 callback 接收 Sample[T];异步迭代的每一项也是 Sample[T]。字段定义见 Sample[T]。

原始姿态、IMU、SDK 处理后的数据和设备状态都支持 latest()、subscribe()、next() 和异步迭代。创建 SDK 后先选择设备并准备读取数据;使用 callback 时,在 start() 前完成订阅:

device = glove.device()
device.pose().subscribe(on_pose)
glove.start()

启动后仍可使用 next()、async for 或新增 subscribe();相关调用的作用范围如下:

调用作用范围
subscribe()、latest()、next()、async for启动后仍可使用;对应设备数据尚未启用时,SDK 先完成设备侧启用,第一次等待因此比后续慢,启用失败按 SDK 错误返回
unsubscribe()只停止该 callback
stop()关闭数据流
close()回收订阅注册

需要 SDK 处理后的数据或自动重连时,还要选择一种 运行方式。

callback 订阅

from qnbot_sdk import Sample
from qnbot_sdk.glove import GlovePose

def on_pose(sample: Sample[GlovePose]) -> None:
    print(sample.sequence, sample.value)

device.pose().subscribe(on_pose)

在 start() 前调用 subscribe() 注册 callback。SDK 会持有已注册的订阅,直到 glove.close() 或 sdk.close() 回收;只有需要在 SDK 继续运行期间提前停止某个 callback 时,才保存 subscribe() 返回的 Subscription 并调用 unsubscribe()。

SDK 收到数据时会执行 callback,callback 应保持简短;耗时操作见 运行方式,异常处理见 常见问题。

获取最新值

pose = device.pose()
sample = pose.latest()

if sample is None:
    print("尚未收到姿态")
else:
    print(sample.sequence, sample.value)

latest() 不阻塞。还没有数据时返回 None,否则返回 Sample[T]。它适合在运行期间从其他逻辑读取当前值,或在 run_forever() 返回后、close() 之前读取最后一条数据。

异步读取

await next() 等待下一条数据,async for 持续读取。原始 pose/IMU 在 start() 成功后即可读取,不需要选择运行方式;保持 asyncio event loop 运行即可:

import asyncio

from qnbot_sdk import Sdk
from qnbot_sdk.glove import GloveConfig

async def main() -> None:
    sdk = Sdk(devices=(GloveConfig(),))
    glove = sdk.glove()
    try:
        pose = glove.device().pose()
        glove.start()
        try:
            first = await pose.next()
            print(first.value)

            async for sample in pose:
                print(sample.value)
        finally:
            glove.stop()
    finally:
        await asyncio.to_thread(glove.close)

if __name__ == "__main__":
    asyncio.run(main())

等待 next() 或异步迭代的任务被取消时,会抛出 Python 原生的 asyncio.CancelledError。

IMU 的异步读取方式与 pose 相同。

外部输入

没有实体手套,或要把自研设备、仿真手部的数据接进来时,用 ExternalConnection + push_frame() 把外部手部帧作为输入源。外部手部帧与实体手套进入同一条数据链路:原始姿态直接读取,Retargeting 结果由 SDK 计算。

配置与使用

外部输入没有设备身份握手,因此必须在 GloveConfig 中显式配置 side:

from qnbot_sdk import Side
from qnbot_sdk.glove import ExternalConnection, GloveConfig

glove_config = GloveConfig(
    side=Side.RIGHT,
    connection=ExternalConnection(),
)

配置 target 后,先订阅输出并启动 SDK,再由应用持续提交帧:

device = glove.device()
output = device.output(name="hand")
output.subscribe(lambda sample: print(sample.value.joints))
device.start()
glove.run_background()

pose = device.push_frame(frame)

push_frame() 同步返回当前帧的原始 GlovePose;target 关节结果通过 output() 读取,不是 push_frame() 的返回值。上例使用后台模式。

外部输入沿用运行方式中的后台或手动模式:后台模式持续提交 push_frame(frame);手动模式每轮先提交帧,再调用 update()。首次需要标定时,完成采集与标定后才会产生 target 输出。完整示例见后台运行和手动 update。

帧约束

  • 五个指尖姿态(thumb、index、middle、ring、pinky)必填,每个都是一个 GloveNodePose。
  • position 是 3 个有限浮点数,单位为米。
  • quaternion_xyzw 是 4 个有限的 XYZW 分量,整体范数非零;SDK 会在使用前归一化。
  • timestamp_ms 可省略;填写时必须是有限毫秒数,并对同一设备严格递增;省略时沿用当前共享时间基准。时间单位、默认基准和授时方式见时间戳与外部授时。
  • timestamp_sec 仅用于兼容旧代码,SDK 会将其乘以 1000 转为毫秒;不得与 timestamp_ms 同时设置。

公共合同统一单位和 quaternion 分量顺序,但坐标原点、X/Y/Z 轴方向和相对旋转参考系由算法包决定,必须与所选算法包的产品说明一致。side 不随帧传入;设备标识和 sequence 由 SDK 根据当前配置生成。字段、类型和可空行为见 ExternalGloveFrame 与 GloveNodePose。