数据订阅

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

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

跨设备合并订阅

全局资源会自动聚合所有已连接设备的数据，通过 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 Glove SDK 数据参考 — 触觉、EMF 位姿、手部追踪、IMU、坐标变换

灵巧手数据订阅

SdkManager.connect() 按设备型号返回 WujiHand2 或 WujiHand 句柄。关节状态订阅、指令发布、触觉数据通过各自句柄上的 API 访问，下面分别说明。

Wuji Hand 2

Wuji Hand 2 由 wuji_sdk.WujiHand2 暴露，所有能力通过统一的资源式 API 访问：

• 资源访问：hand.<resource>() 操作整手，hand.joint(k).<resource>() 操作单关节，方法名一致
• 多态写入：传单个标量广播到全部关节，传 20 元素数组按关节生效
• 动作方法：enable() / disable() / clear_fault() / set_origin() / clear_origin() 各接受可选 joints=[0/1]*20 掩码触发子集，emergency_stop() 仅作用于整手
• 反馈订阅：joint_states() 与 joint_diagnostics() 返回仅含在线关节的变长帧，按 nid 识别每条
• 控制模式：设备默认运行 MIT 控制，Python API 无需也无法切换

订阅关节状态

hand.joint_states().subscribe() 返回 Subscription[JointStateFrame]，按帧推送在线关节的位置 / 速度 / 力矩。帧含 header（seq + timestamp_us + frame_id，frame_id 为 l_wrist 或 r_wrist）与变长 joints 列表：

sub = hand.joint_states().subscribe()
for _ in range(100):
    frame = await sub.recv_async()
    h = frame.header
    print(f"seq={h.seq} t={h.timestamp_us}us frame={h.frame_id} n={frame.num_joints}")
    for j in frame.joints:
        print(f"  nid={j.nid} pos={j.position:+.3f} vel={j.velocity:+.3f} eff={j.effort:+.3f}")
sub.close()

每条 joints 项的 nid 用于跨帧定位关节，离线关节不在帧内。字段定义见 Wuji Hand 2 类型 — JointStateFrame。

订阅关节诊断

hand.joint_diagnostics().subscribe() 返回 Subscription[JointDiagnosticsFrame]，与关节状态流同源，每条携带状态字、相电流、母线电压、温度与错误码：

sub = hand.joint_diagnostics().subscribe()
frame = await sub.recv_async()
for j in frame.joints:
    sw = j.status_word
    print(f"nid={j.nid} ext_state={sw.ext_state_name} "
          f"current={j.current:+.2f}A bus={j.vbus_v_fb:.1f}V "
          f"temp={j.mcu_temp_c_fb:.1f}C err=0x{j.error_code_current:04X}")
sub.close()

error_code_current 含 stop 与 warning 两类位，调 WujiHand2.describe_error(code) 解码出错误名与描述。每帧还携带通信健康信息：每关节的总线响应率/超时计数，以及帧级 comm 摘要（端到端以太网丢帧、RPC 重传/超时计数、指尖触觉在线位图）——由 SDK 自动维护，无需轮询。字段定义见 Wuji Hand 2 类型 — JointDiagnosticsFrame。

发布关节指令

hand.joint_command().publish() 返回 PUB 句柄，publisher.send(joints) 一次推送一帧 20 关节指令，每项为 JointCommand(position, velocity, effort)：

from wuji_sdk import JointCommand

publisher = hand.joint_command().publish()
zeros = [JointCommand(0.0, 0.0, 0.0) for _ in range(20)]
publisher.send(zeros)
publisher.close()

按固定频率循环 publisher.send() 维持持续控制（典型 200 Hz–1 kHz）。publisher.close() 显式释放，可省略（句柄 drop 时自动释放）。JointCommand 字段见 Wuji Hand 2 类型 — JointCommand。

动作方法：使能 / 失能 / 清错 / 用户零点 / 紧急停止

整手与单关节共用一套动作方法。整手默认对全部关节生效，传 joints=[0/1]*20 掩码可指定子集：

# 整手
hand.enable()
hand.clear_fault()

# 仅食指 4 关节（flat 索引 4..7）
mask = [0] * 20
for i in range(4, 8):
    mask[i] = 1
hand.enable(joints=mask)
hand.clear_fault(joints=mask)

# 单关节
hand.joint(0).enable()
hand.joint(0).clear_fault()

emergency_stop() 仅作用于整手且不接受掩码，触发后所有关节立即停止输出。set_origin() 与 clear_origin() 设置或清除用户零点（运行时偏移，不写 Flash），同样接受 joints 掩码。

MIT 参数与力矩上限

mit_params 与 effort_limit 是资源，写入支持多态：

# 广播到全部关节
hand.mit_params().set((3.0, 0.05))      # (kp, kd) 二元组
hand.effort_limit().set(1.5)            # 安培

# 按关节配置（20 元素数组）
hand.mit_params().set([(3.0, 0.05)] * 20)
hand.effort_limit().set([1.5] * 20)

# 读回长度 20 的列表
limits = hand.effort_limit().get()       # 离线关节为 None
params = hand.mit_params().get()         # 离线关节为 None

写入拒绝 NaN / Inf 与负值并抛 ValueError。mit_params 与 effort_limit 的 set 都会立即下发到设备并实时生效。SDK 目前不会将其持久化到设备（重启后丢失）。单关节同样可写：hand.joint(k).effort_limit().set(1.5)。

effort_limit 的取值还要通过设备侧校验。写入超出设备当前允许上限的值时，设备拒绝该次写入，SDK 抛 ValueError。上限由固件按型号设定，SDK 未提供查询接口，写入后读回确认实际生效值：

j = hand.joint(0)
try:
    j.effort_limit().set(1e6)
except ValueError as e:
    print(f"设备拒绝: {e}")
print(j.effort_limit().get())   # 读回确认当前生效值

两类参数错误都抛 ValueError，用一个 except ValueError 即可覆盖，与内部故障（RuntimeError）区分开。整手写入被拒时，逐关节生效情况同样以读回为准。

单关节与单手指访问

hand.joints() 返回 20 个 JointHandle，hand.fingers() 返回 5 个 FingerHandle（每个含 4 关节）。hand.joint(k) 按 flat 索引取单个关节：

# 遍历手指
for f in hand.fingers():
    print(f"finger joints: {[j.label for j in f.joints()]}")

# 单关节资源
j = hand.joint(0)        # 拇指 J1
j.effort_limit().set(2.0)
print(j.status_word().get().ext_state_name)
print(f"error=0x{j.error_code().get():04X}")

hand.online_joints_count().get() 返回当前在线关节数，配合订阅帧的 num_joints 字段做断连检测。

错误码描述

WujiHand2.describe_error(code) 是静态方法，把错误码（来自 joint_diagnostics 的 error_code_current 或单关节 error_code().get()）解码成对象：

from wuji_sdk import WujiHand2

desc = WujiHand2.describe_error(0x0101)
if desc is not None:
    print(desc)        # 含错误名、人类可读描述、stop / warning 分类

未知错误码返回 None。

订阅 IMU 数据

Wuji Hand 2 内置 IMU，100 Hz 上报加速度与角速度。hand.imu().subscribe() 返回 Subscription[ImuData]：

sub = hand.imu().subscribe()
for _ in range(100):
    s = await sub.recv_async()
    a = s.linear_acceleration   # m/s²
    g = s.angular_velocity      # rad/s
    print(f"seq={s.header.seq} accel=[{a.x:.2f} {a.y:.2f} {a.z:.2f}] gyro=[{g.x:.3f} {g.y:.3f} {g.z:.3f}]")
sub.close()

设备不做板载姿态融合，orientation 字段不可用。按 ROS sensor_msgs/Imu 约定，orientation_covariance[0] = -1 标记朝向无效。完整字段定义见 Wuji Hand 类型 — ImuData。

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 相同，控制代码可在两代手之间直接迁移：

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