Data Subscription Mechanism

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.

Adjusting the Stream Output Rate

sub.set_rate(frequency_hz) adjusts the firmware push rate of a subscribed stream and returns the rate the device applies:

sub = glove.emf_poses().subscribe()

actual = sub.set_rate(30)   # the device may quantize the request
print(f"requested 30 Hz, applied {actual} Hz")

sub.set_rate(0)             # pass 0 to restore the device default

• Call it on an open subscription handle. The setting stays in effect for the life of the subscription.
• Rates can only go down, not above the native rate. The device quantizes the request to a supported rate (an integer division of the native rate). The return value is the rate that took effect.
• The rate belongs to the stream, not to the handle: all subscribers of the stream share the setting, and the last write wins.
• The setting doesn't persist: the rate resets to the device default full rate when the last subscription to the stream is closed or the device reconnects.
• Only streams produced directly by the device support rate control. Derived streams follow their source rate automatically, and global resources such as tf_static are aggregated by the SDK. Calling set_rate on either raises WujiException.
• Requires firmware with stream rate control support. Firmware without it raises WujiException.

Rate-adjustable streams and the derived streams that follow them, per device:

joint_states and joint_diagnostics on the Wuji Hand 2 share one device stream, so adjusting either changes both.

C SDK equivalent: wuji_sub_set_rate(sub, frequency_hz, &actual_hz). Streams without rate control return WUJI_STATUS_ERR_UNSUPPORTED — see C SDK Reference — Stream output rate control.

For complete examples, see examples/python/wuji_glove/6.set_stream_rate.py and examples/python/wuji_hand_2/4.set_stream_rate.py.

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

Index by Product

Wuji Hand Data Subscription

Wuji Hand product docs will be archived soon. The subscription APIs are maintained on this page. The device 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 Wuji Hand 2 and 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()

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 sensing glove companion data

When a tactile sensing 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()