数据来源与读取方式
两类数据的区别、callback、最新值、异步三种读取方式,以及外部输入来源。
数据来源与读取方式
两类数据的区别
SDK 发布两类数据:设备直接产出的原始数据,和 SDK 计算出的数据。读取方式相同,区别只在数据怎么产生:
| 原始数据 | SDK 处理后的数据 | |
|---|---|---|
| 典型 API | pose()、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。