数据订阅

Wuji SDK 通过订阅机制接收设备数据。每个数据流对应一种传感器或计算结果，通过语义化 API 访问。

订阅模式

异步接收

sub = glove.tactile().subscribe()
frame = await sub.recv_async()  # 等待下一帧数据

同步非阻塞接收

sub = glove.tactile().subscribe()
frame = sub.recv()  # 无数据时返回 None

回调接收

def on_data(frame):
    print(f"Max: {max(frame.data):.1f}")

sub = glove.tactile().subscribe_with_callback(callback=on_data)
# 后台自动接收，不阻塞主线程

关闭订阅

回调订阅需要手动关闭：

sub = glove.tactile().subscribe_with_callback(callback=on_data)
# ... 使用一段时间后
sub.close()

断开设备连接时，该设备上的所有订阅会自动关闭。

调节订阅流输出率

sub.set_rate(frequency_hz) 调节订阅流的固件推送频率，返回设备实际生效的输出率：

sub = glove.emf_poses().subscribe()

actual = sub.set_rate(30)   # 设备量化后返回实际生效值
print(f"requested 30 Hz, applied {actual} Hz")

sub.set_rate(0)             # 传 0 恢复设备默认满速

• 需要在未关闭的订阅句柄上调用，订阅存续期间设置持续生效。
• 只能调低，无法超过原生输出率。设备把请求量化到支持的档位（原生输出率的整数分频），返回值以实际生效为准。
• 输出率属于流而非句柄：同一流的所有订阅共享该设置，后写生效。
• 设置不持久化：该流的最后一个订阅关闭或设备重连后，输出率恢复设备默认满速。
• 仅设备直接产出的流可调。派生流跟随源流自动降频，tf_static 等全局资源由 SDK 聚合，对二者调用 set_rate 均抛 WujiException。
• 需要支持输出率调节的固件。固件不支持时抛 WujiException。

各设备的可调流与跟随的派生流：

Wuji Hand 2 的 joint_states 与 joint_diagnostics 共享同一条设备流，调节任意一个，两者同时生效。

C SDK 等价：wuji_sub_set_rate(sub, frequency_hz, &actual_hz)，不支持的流返回 WUJI_STATUS_ERR_UNSUPPORTED，见 C SDK 参考 — 订阅流输出率调节。

完整示例见 examples/python/wuji_glove/6.set_stream_rate.py 与 examples/python/wuji_hand_2/4.set_stream_rate.py。

跨设备合并订阅

全局资源会自动聚合所有已连接设备的数据，通过 SdkManager 级别的 API 订阅。当前版本的全局资源：

运行时通过 manager.global_topics() 查询完整的全局资源列表：

manager = SdkManager.instance()

# 查看所有全局资源
for topic in manager.global_topics():
    print(f"{topic.path} (sub={topic.can_sub})")

# 订阅全局资源——数据来自所有已连接设备的合并
sub = manager.tf_static().subscribe()
transforms = await sub.recv_async()

C SDK 等价：wuji_subscribe_tf(...) 与 wuji_subscribe_tf_static(...) 订阅全局坐标变换（回调签名 WujiFrameTransforms，设备无关）。

全局资源与设备资源的区别：

可用数据流

按产品索引

Wuji Hand 数据订阅

Wuji Hand 产品文档即将归档，订阅接口说明随本页维护。设备 USB 直连，由 wuji_sdk.WujiHand 暴露。

订阅关节状态

hand.joint_states().subscribe() 返回 Subscription[HandJointStates]，异步消费 20 关节实时位置，可选含速度与力矩：

sub = hand.joint_states().subscribe()
for _ in range(100):
    state = await sub.recv_async()
    print(f"seq={state.header.seq}, joint0={state.position[0]:+.3f}")
sub.close()

发布关节指令（实时控制）

hand.realtime_controller(LowPass(cutoff_hz=...)) 以 with 上下文打开实时控制会话，低通滤波抑制高频抖动。在会话内用 hand.joint_command().publish() 获取 PUB 句柄，调用 publisher.send(joints) 推送恰好 20 个 JointCommand（每项 position / velocity / effort），指令结构与 Wuji Hand 2 相同，控制代码可在 Wuji Hand 2 与 Wuji Hand 之间直接迁移：

from wuji_sdk import JointCommand, LowPass

with hand.realtime_controller(LowPass(cutoff_hz=5.0)) as ctrl:
    publisher = hand.joint_command().publish()
    joints = [JointCommand(0.0, 0.0, 0.0)] * 20
    joints[0] = JointCommand(0.3, 0.0, 0.0)
    publisher.send(joints)
    print(ctrl.get_actual_position()[0], ctrl.get_actual_effort()[0])
    publisher.close()

持续控制时按固定频率（典型 100 Hz）在 with 块内循环调用 publisher.send()。会话句柄 ctrl 支持随时读回实际状态：get_actual_position() 返回 20 关节实际位置（弧度），get_actual_effort() 返回 20 关节实际力矩（安培），两者都从非阻塞缓存读取，不影响控制频率。

ctrl 可以跨线程共享，不需要加锁。例如一个线程按固定频率下发目标，另一个线程读回实际位置。

电机力矩上限

hand.set_all_effort_limit(amps) 把 20 个关节的力矩上限统一设为指定安培值，hand.get_all_effort_limit() 读回长度 20 的 list[float]。

• 默认值 1.5 A，适用于大多数场景
• 合法范围 0.0–3.5 A，超过 3.5 会被限制到 3.5

hand.set_all_effort_limit(1.5)
limits = hand.get_all_effort_limit()

修改 effort limit 会改变关节最大输出能力。固件还会根据关节温度自动降低实际生效的值。完整说明与风险提示见 Wuji Hand SDK 使用说明 — joint_effort_limit。

触觉感知手套同步数据

手与触觉感知手套同 USB 总线接入且手性匹配时，SDK 在 connect 中自动配对。检测配对状态并订阅压力帧：

if hand.is_tactile_attached():
    tactile_sub = hand.tactile.subscribe_pressure_frame()
    frame = await tactile_sub.recv_async()
    print(f"max pressure: {max(frame.pressure):.2f}")
    tactile_sub.close()