Data Subscription

Wuji SDK uses a subscription mechanism to receive device data. Each data stream corresponds to a sensor or computed result, accessed through semantic APIs.

Subscription Modes

Async Receive

sub = glove.tactile().subscribe()
frame = await sub.recv_async()  # Wait for next data frame

Sync Non-blocking Receive

sub = glove.tactile().subscribe()
frame = sub.recv()  # Returns None when no data available

Callback Receive

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

sub = glove.tactile().subscribe_with_callback(callback=on_data)
# Receives automatically in background, non-blocking

Closing Subscriptions

Callback subscriptions must be manually closed:

sub = glove.tactile().subscribe_with_callback(callback=on_data)
# ... after some time
sub.close()

All subscriptions on a device are automatically closed when the device is disconnected.

Cross-device Merged Subscription

Global resources automatically aggregate data from all connected devices. Subscribe to them through the SdkManager-level API. Global resources in the current version:

Query the full list of global resources at runtime with manager.global_topics():

manager = SdkManager.instance()

# List all global resources
for topic in manager.global_topics():
    print(f"{topic.path} (sub={topic.can_sub})")

# Subscribe to a global resource — data is merged from all connected devices
sub = manager.tf_static().subscribe()
transforms = await sub.recv_async()

C SDK equivalents: wuji_subscribe_tf(...) and wuji_subscribe_tf_static(...) subscribe to the global coordinate transforms (callback signature WujiFrameTransforms, device-independent).

Differences between device resources and global resources:

Available Data Streams

For device-specific data streams and data structures, see the corresponding device documentation:

• Wuji Glove SDK data reference — Tactile, EMF poses, hand tracking, IMU, coordinate transforms

Dexterous Hand Data Subscription

SdkManager.connect() returns a WujiHand2 or WujiHand handle based on the device model. Joint-state subscription, command publishing, and tactile data are accessed through APIs on each handle — covered separately below.

Wuji Hand 2

Wuji Hand 2 is exposed as wuji_sdk.WujiHand2 and reaches every capability through a unified resource-style API:

• Resource access: hand.<resource>() operates on the whole hand and hand.joint(k).<resource>() operates on a single joint, using the same method names
• Polymorphic writes: pass a scalar to broadcast to all joints, pass a 20-element array to set per joint
• Action methods: enable(), disable(), clear_fault(), set_origin(), and clear_origin() each accept an optional joints=[0/1]*20 mask to target a subset, while emergency_stop() always acts on the whole hand
• Feedback streams: joint_states() and joint_diagnostics() return variable-length frames that include only online joints, identified by nid
• Control mode: the device runs MIT control by default — the Python API neither requires nor allows switching

Subscribing to joint state

hand.joint_states().subscribe() returns Subscription[JointStateFrame] and streams position / velocity / effort for every online joint. Each frame carries a header (seq, timestamp_us, frame_id — either l_wrist or r_wrist) and a variable-length joints list:

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

Use each entry's nid to identify joints across frames — offline joints do not appear in the frame. For field definitions, see Wuji Hand 2 Types — JointStateFrame.

Subscribing to joint diagnostics

hand.joint_diagnostics().subscribe() returns Subscription[JointDiagnosticsFrame], derived from the same realtime stream as joint_states. Each entry carries the status word, phase current, bus voltage, temperature, and error code:

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 carries both stop and warning bits — pass it to WujiHand2.describe_error(code) to get a name and description. Each frame also carries communication health: per-joint bus response rate / timeout counters, plus a frame-level comm summary with end-to-end (Ethernet) stream loss, RPC retry/timeout counters, and the fingertip tactile online bitmap — maintained automatically by the SDK, no polling required. For field definitions, see Wuji Hand 2 Types — JointDiagnosticsFrame.

Publishing joint commands

hand.joint_command().publish() returns a PUB handle. publisher.send(joints) sends one 20-joint frame, each entry a 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()

Call publisher.send() at a fixed rate (typically 200 Hz–1 kHz) to maintain continuous control. publisher.close() is optional — the PUB stream releases on drop. For JointCommand fields, see Wuji Hand 2 Types — JointCommand.

Action methods: enable / disable / clear faults / user origin / emergency stop

Whole-hand and single-joint share the same action methods. Whole-hand acts on every joint by default — pass joints=[0/1]*20 to target a subset:

# Whole hand
hand.enable()
hand.clear_fault()

# Index finger only (flat indices 4..7)
mask = [0] * 20
for i in range(4, 8):
    mask[i] = 1
hand.enable(joints=mask)
hand.clear_fault(joints=mask)

# Single joint
hand.joint(0).enable()
hand.joint(0).clear_fault()

emergency_stop() acts on the whole hand only and accepts no mask — every joint stops immediately when triggered. set_origin() and clear_origin() set or clear the user origin (a runtime offset, not written to Flash) and accept the same joints mask.

MIT params and effort limit

mit_params and effort_limit are resources with polymorphic writes:

# Broadcast to all joints
hand.mit_params().set((3.0, 0.05))      # (kp, kd) tuple
hand.effort_limit().set(1.5)            # amperes

# Per-joint (20-element array)
hand.mit_params().set([(3.0, 0.05)] * 20)
hand.effort_limit().set([1.5] * 20)

# Read back length-20 lists
limits = hand.effort_limit().get()       # offline joints are None
params = hand.mit_params().get()         # offline joints are None

Writes reject NaN, infinity, and negative values with ValueError. Both mit_params.set(...) and effort_limit.set(...) update the device immediately and take effect live. The SDK does not currently persist them across a device reboot. Single-joint writes work the same way: hand.joint(k).effort_limit().set(1.5).

effort_limit values must also pass device-side validation. Writing a value above the ceiling the device currently allows makes the device reject that write, and the SDK raises ValueError. Firmware sets the ceiling per model and the SDK doesn't expose a query for it, so read the value back to confirm what's in effect:

j = hand.joint(0)
try:
    j.effort_limit().set(1e6)
except ValueError as e:
    print(f"Device rejected: {e}")
print(j.effort_limit().get())   # read back what's in effect

Both kinds of parameter error raise ValueError, so a single except ValueError covers them, separate from internal faults (RuntimeError). When a whole-hand write is rejected, read back to see what each joint ended up with.

Per-joint and per-finger access

hand.joints() returns 20 JointHandles, hand.fingers() returns 5 FingerHandles (each with 4 joints), and hand.joint(k) returns a single joint by flat index:

# Walk the fingers
for f in hand.fingers():
    print(f"finger joints: {[j.label for j in f.joints()]}")

# Single-joint resources
j = hand.joint(0)        # thumb 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() returns the current online joint count, pair it with each frame's num_joints field to detect disconnects.

Error code description

WujiHand2.describe_error(code) is a static method that decodes an error code (from joint_diagnostics's error_code_current or single-joint error_code().get()) into an object:

from wuji_sdk import WujiHand2

desc = WujiHand2.describe_error(0x0101)
if desc is not None:
    print(desc)        # name, human-readable description, stop / warning category

Unknown codes return None.

Subscribing to IMU data

Wuji Hand 2 has an onboard IMU that streams acceleration and angular velocity at 100 Hz. hand.imu().subscribe() returns 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()

The device does not run onboard orientation fusion, so the orientation field is not available. Following the ROS sensor_msgs/Imu convention, orientation_covariance[0] = -1 marks the orientation as invalid. For the full field definition, see Wuji Hand Types — ImuData.

Wuji Hand (Gen 1)

Wuji Hand connects over USB and is exposed as wuji_sdk.WujiHand.

Subscribing to joint state

hand.joint_states().subscribe() returns Subscription[HandJointStates] and delivers real-time 20-joint position, with optional velocity and effort:

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

Publishing joint commands (realtime control)

hand.realtime_controller(LowPass(cutoff_hz=...)) opens a realtime control session via a with context, with the lowpass filter suppressing high-frequency jitter. Inside the session, get a PUB handle from hand.joint_command().publish() and call publisher.send(joints) with exactly 20 JointCommand entries (position / velocity / effort each). The command shape matches Wuji Hand 2, so control code is portable between the two hands:

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

For continuous control, call publisher.send() at a fixed rate (typically 100 Hz) inside the with block. The session handle ctrl reads back actual state at any time: get_actual_position() returns the 20-joint actual position (radians) and get_actual_effort() the 20-joint actual effort (amps) — both read from a non-blocking cache, so control frequency is unaffected.

You can share ctrl across threads without locking — for example, one thread streaming targets at a fixed rate while another polls actual positions.

Effort limits

hand.set_all_effort_limit(amps) sets the torque limit on all 20 joints to the same amperage, and hand.get_all_effort_limit() reads back a length-20 list[float].

• Default is 1.5 A, suitable for most applications
• Valid range is 0.0 to 3.5 A — values above 3.5 are clamped to 3.5

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

Modifying the effort limit changes the maximum output capability of each joint. The firmware also reduces the effective limit based on joint temperature. For the full reference and risk notes, see Wuji Hand SDK Guide — joint_effort_limit.

Tactile glove companion data

When a tactile glove shares the USB bus and matches the hand's handedness, the SDK auto-pairs it during connect. Check the pairing status and subscribe to pressure frames:

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