SDK Reference
The Python semantic API for Wuji Hand 2. Unified pattern: hand.{resource}().{action}(). All operations target the whole hand (20 joints).
This page is an interface reference for lookup. For control methods—unit conventions, the MIT control law, the joint table, and parameter safety limits—see the Control Guide.
This page is self-contained: it covers the Wuji Hand 2–specific semantic API plus the shared SDK capabilities you need to operate a Wuji Hand 2—installation, connection, subscription, time synchronization, recording, and troubleshooting—so there's no need to jump to another product's docs. The SDK builds on the shared Wuji SDK (wuji_sdk, pip install wuji-sdk). For source, see the wuji-sdk repository. The complete C API reference and hand retargeting stay in the Wuji SDK docs.
Installation and Environment
System Requirements
| Item | Requirement |
|---|---|
| OS | Ubuntu 22+ |
| Network | Ethernet (same subnet as the Wuji Hand 2) |
| Python | 3.10+ |
| C compiler | gcc / clang, only for the C API |
Installation
| Language | Installation |
|---|---|
| Python | PyPI package wuji-sdk: pip install wuji-sdk |
| C | Download the platform tarball wuji-sdk-c-<version>-<target>.tar.gz from the Releases page. The extracted archive contains lib/libwuji_sdk_c.so and include/wuji_sdk.h |
For C initialization and calling conventions, see C API Overview.
Connection
Network prerequisite: Wuji Hand 2 uses a static IP, not DHCP. Each device ships with a fixed address by handedness — left hand 192.168.1.110, right hand 192.168.1.111, gateway 192.168.1.1, subnet mask 255.255.255.0. Before the first connection, set your host NIC to the same subnet (for example 192.168.1.100), then connect by address. The factory IP is changeable later — see ip (SET/GET).
Device Discovery
Use SdkManager.scan() to discover Wuji devices on the local network:
from wuji_sdk import SdkManager
manager = SdkManager.instance()
devices = manager.scan()
for dev in devices:
print(f"SN: {dev.sn}, Type: {dev.device_type}, Address: {dev.address}")The returned DiscoveredDevice contains the serial number, address, and device type. device_type is a DeviceType enum and reads WujiHand2 for a Wuji Hand 2, so the device type is known before connecting. Unknown means the type was not available at scan time or isn't recognized by this SDK build.
Connect via SdkManager (Recommended)
from wuji_sdk import SdkManager, Handedness
manager = SdkManager.instance()
# Auto-discover and connect to the first Wuji Hand 2 on the LAN
hand = manager.auto_connect(device_name="wuji_hand_2")
# Connect explicitly by serial number or address
hand = manager.connect(sn=<device_sn>, device_name="wuji_hand_2")
hand = manager.connect(address="192.168.3.110:50001", device_name="wuji_hand_2")
# Or connect by handedness directly, without a serial number
hand = manager.connect(handedness=Handedness.Right, device_name="wuji_hand_2")The device_name="wuji_hand_2" overloads of auto_connect and connect return a WujiHand2 instance directly (with type hints). handedness is mutually exclusive with sn and address. If multiple Hand 2 units of the same handedness are on the network, AmbiguousHandedness is raised—specify sn instead.
ConnectOptions
from wuji_sdk import SdkManager, ConnectOptions
opts = ConnectOptions(
timeout_ms=1000,
retry_count=3,
enable_bridge=True, # Default True: lets multiple clients (Wuji Studio + recording scripts + your app) connect at once
)
hand = manager.connect(sn=<device_sn>, device_name="wuji_hand_2", options=opts)Set enable_bridge=False for exclusive single-client mode.
Instance Attributes
| Attribute | Type | Description |
|---|---|---|
serial_number | str | Device serial number |
device_name | str | Name given at connect (default "wuji_hand_2") |
info | Optional[DeviceInfo] | Device info: serial_number, firmware_version |
is_connected | bool | Connection state |
hand.hw_version().get() returns the factory hardware version HwVersion(major, minor, patch). 0.0.0 means it wasn't written at the factory.
Multi-device Management
One process can hold both hands at once, each identified by a unique device_name:
left = manager.connect(handedness=Handedness.Left, device_name="left_hand")
right = manager.connect(handedness=Handedness.Right, device_name="right_hand")
# Get all connected devices
all_devices = manager.get_connected_devices()
for name, device in all_devices:
print(f"{name}: {device.serial_number}")
# Get a device by name
left = manager.get_device(device_name="left_hand")Use different device_name values to keep the left and right hand in the same process. Reconnecting to an already-active device raises SessionAlreadyExists without dropping the existing connection — keep using the existing handle.
To capture from both hands at once, consume each subscription concurrently with asyncio:
import asyncio
from wuji_sdk import SdkManager, Handedness
async def collect(hand, name):
sub = hand.joint_states().subscribe()
while True:
frame = await sub.recv_async()
print(f"[{name}] seq={frame.header.seq}")
async def main():
manager = SdkManager.instance()
left = manager.connect(handedness=Handedness.Left, device_name="left_hand")
right = manager.connect(handedness=Handedness.Right, device_name="right_hand")
await asyncio.gather(collect(left, "left"), collect(right, "right"))
asyncio.run(main())Disconnecting
# Disconnect a specific device
manager.disconnect(device_name="wuji_hand_2")
# Or disconnect via the device object
hand.disconnect()When a device is disconnected, all subscriptions on that device are automatically closed. Subscriptions have to be re-established after reconnecting.
Data Subscription
Real-time data from a Wuji Hand 2 arrives through subscriptions. Each stream corresponds to a sensor or computed result, accessed through semantic APIs. For what each stream carries, see Hand-Level Resources.
Subscription Modes
Async receive:
sub = hand.joint_states().subscribe()
frame = await sub.recv_async() # Wait for next data frameSync non-blocking receive:
sub = hand.joint_states().subscribe()
frame = sub.recv() # Returns None when no data availableCallback receive:
def on_data(frame):
print(frame.header.seq)
sub = hand.joint_states().subscribe_with_callback(callback=on_data)
# Receives automatically in background, non-blockingClosing Subscriptions
Callback subscriptions must be manually closed:
sub = hand.joint_states().subscribe_with_callback(callback=on_data)
# ... after some time
sub.close()Subscribe only to the streams you actually consume, and call close() as soon as a subscription is no longer needed, to avoid wasting bandwidth and CPU.
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 = hand.joint_states().subscribe()
actual = sub.set_rate(200) # the device may quantize the request
print(f"requested 200 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.
- Requires firmware with stream rate control support. Firmware without it raises
WujiException.
joint_states and joint_diagnostics share one device stream, so adjusting either changes both.
C API equivalent: wuji_sub_set_rate(sub, frequency_hz, &actual_hz). Streams without rate control return WUJI_STATUS_ERR_UNSUPPORTED.
Available Data Streams
| Stream | Native output rate | Frame type |
|---|---|---|
| joint_states | 1000 Hz | JointStateFrame |
| joint_diagnostics | Shares the joint_states device stream | JointDiagnosticsFrame |
| imu | 100 Hz | ImuData |
Joint commands go the other direction — see joint_command (PUB).
Hand-Level Resources
This section covers the whole-hand API (20 joints). Single-joint reads and actions go through the JointHandle returned by hand.joint(k) / hand.joints() (see Joint Traversal). The device runs in MIT control mode by default — the control mode isn't set from the Python API.
handedness (GET)
Get the handedness result (left / right).
side = hand.handedness().get() # → "left" or "right"online_joints_count (GET)
Get the online joint count (0–20).
n = hand.online_joints_count().get() # → int, 0–20joint_diagnostics (SUB)
Joint diagnostics stream: subscribe to per-joint status word, current, bus voltage, temperature, and error code. Frames are variable-length and contain only online joints — identify each entry by its nid. This stream and joint_states ride the same device stream, so lowering the push rate of one lowers both — see joint_states (SUB).
sub = hand.joint_diagnostics().subscribe() # → Subscription[JointDiagnosticsFrame]
frame = await sub.recv_async()
for j in frame.joints:
print(j.nid, j.vbus_v_fb, j.mcu_temp_c_fb, f"0x{j.error_code_current:04X}")
sub.close()Decode error_code_current with the static method WujiHand2.describe_error(code). The returned object carries code, the error name, desc, cause, resolution, severity (Warning / DeferredStop / ImmediateStop / Fatal), and clear_policy (AutoClear / ManualClear / NonClearable). Use cause and resolution to identify and address the fault. Unknown codes return None. Printing a diagnostics entry object shows error codes in hex.
For joint angles, use position from the joint_states stream — the diagnostics stream doesn't carry joint angles.
comm_diag (GET, 1 Hz)
Communication diagnostics: returns the throughput / error rate of the 5 fingers, once per second.
diag = hand.comm_diag().get() # → HandCommunicationDiagnostics
for finger in diag.fingers: # throughput / error rate for each of the 5 fingers
print(finger.tx_kbps, finger.rx_kbps, finger.error_per_sec)Use it to troubleshoot communication between the joints and the control board.
effort_limit (SET/GET)
Effort limit: read / write the per-joint torque cap (A).
hand.effort_limit().set(1.5) # set all joints to 1.5 A
limits = hand.effort_limit().get() # → list[Optional[float]]Writes reject NaN, Inf, negative values, and values above the device ceiling — whole-hand and per-joint writes validate the same way. A rejected write fails with an error and leaves the current value unchanged. The firmware sets the ceiling. For common errors and remedies, see Script Throws an Exception.
mit_params (SET/GET)
MIT impedance parameters: read / write per-joint kp / kd.
hand.mit_params().set((1.0, 0.05)) # same kp / kd for all joints
hand.mit_params().set([(1.0, 0.05)] * 20) # list of 20, per joint
mp = hand.mit_params().get() # → list[Optional[MitParam]]; offline joints are NoneWrites reject NaN, Inf, and negative values.
clear_fault (EXEC)
Clear faults: a direct action. Without arguments it acts on the whole hand.
hand.clear_fault() # clear faults on all joints
hand.clear_fault(joints=mask) # optional: 0/1 mask of length 20 — only joints with a 1 are clearedUser Origin
Calibrate the current physical position as the joint-side zero. If a joint is moving, it takes effect at the next IDLE.
hand.set_origin() # whole hand: refresh the user origin of all joints
hand.clear_origin() # whole hand: clear the user origin of all joints
hand.set_origin(joints=mask) # optional: 0/1 mask of length 20 — only joints with a 1 are affectedEnable / Disable / Emergency Stop
hand.enable() # enable all joints
hand.disable() # disable all joints
hand.enable(joints=mask) # optional: 0/1 mask of length 20 — acts on joints with a 1
hand.emergency_stop() # emergency stop: whole-hand action, no maskjoint_states (SUB)
Joint state stream: subscribe to the real-time state frames of the 20 joints, at a native output rate of 1000 Hz. Lower the push rate through the subscription handle: sub.set_rate(hz) returns the rate in effect, and 0 restores the native rate — see Adjusting the Stream Output Rate for full usage. This stream and joint_diagnostics ride the same device stream, so lowering the rate of one lowers both.
Push-rate tuning applies to the joint_states, joint_diagnostics, and imu streams. Upgrade the firmware and the Wuji SDK together to versions that support it — see the release notes for the version requirements.
sub = hand.joint_states().subscribe() # → Subscription[JointStateFrame]
frame = sub.recv() # synchronous, non-blocking; None = no data yet
frame = await sub.recv_async() # asynchronous wait
print(frame.header.seq, frame.header.timestamp_us)
for j in frame.joints:
print(j.nid, j.position, j.velocity, j.effort)
sub.close()Callback mode:
def on_state(frame):
print(frame.header.seq, [j.position for j in frame.joints[:4]])
cb = hand.joint_states().subscribe_with_callback(on_state)
# ...
cb.close()position is the joint-side angle (rad) and velocity the joint-side angular velocity (rad/s)—this is the only recommended path for joint angle and speed. Frames are variable-length and contain only online joints — identify each entry by its nid. The joint_diagnostics stream doesn't carry joint angles.
joint_command (PUB)
Joint command: publish positions / velocities / torque feedforward for the 20 joints. Each send takes exactly 20 JointCommand entries, one position / velocity / effort per joint.
from wuji_sdk import JointCommand
pub = hand.joint_command().publish() # → JointCommandPublisher
pub.send([JointCommand(position=p, velocity=0.0, effort=0.0) for p in positions])
pub.close()To turn human hand keypoints into these 20-joint commands, use Wuji SDK Hand Retargeting.
Flash Log Export (Diagnostics)
# Export the running logs of the current firmware (use this for most troubleshooting)
result = await hand.dump_hand_logs(bank="current", out_dir="./logs")bank | Meaning | When to use |
|---|---|---|
"current" | Logs written by the firmware currently running | Default — troubleshoot live issues |
"other" | Logs left by the firmware version before an upgrade or rollback | Only when you need to trace behavior before an upgrade / rollback |
Each call creates a separate session directory <sn>-<unix_ts>/ under out_dir, containing joint{0..19}.log for each joint plus one sboard.log (JSONL: one {"timestamp_ms", "level", "target", "message"} per line). out_dir is optional and defaults to ~/.wuji/hand_logs. Each call returns a dict:
result = await hand.dump_hand_logs(bank="current")
print(result["session_dir"]) # str: absolute path of this session directory
print(result["files"]) # list[str]: paths of the written .log filesip (SET/GET)
Read and write the device's static IP address. set writes the new IP to the device's flash. The firmware doesn't hot-swap the running Ethernet stack, so the new IP takes effect only after the next reboot. Between set and reboot, get still returns the current IP.
hand.ip().get() # read the current IP, e.g. "192.168.3.110"
hand.ip().set("192.168.2.111") # write to flash, takes effect after rebootFull round-trip to change the IP: set → reboot → reconnect at the new IP → get to confirm.
hand.ip().set("192.168.2.111")
hand.reboot()
manager.disconnect_all()
# wait for the device to reboot and Ethernet to come up, about 8 seconds
hand = manager.connect(address="192.168.2.111:50001", device_name="wuji_hand_2")
hand.ip().get() # → "192.168.2.111"The new IP takes effect only after a reboot. set writes to flash without changing the live connection. Before the reboot, get still returns the current IP.
reboot (EXEC)
Reboot the device. The device disconnects and needs a fresh connection. Pair it with ip().set() to apply a new IP written to flash.
hand.reboot()imu (SUB)
Onboard IMU stream: pushes acceleration and angular velocity at a native output rate of 100 Hz, adjustable down through the subscription handle. The device does no onboard orientation fusion, so the orientation field is unavailable and orientation_covariance[0] = -1 marks it invalid per the ROS sensor_msgs/Imu convention. Data is reported in the IMU sensor's own frame, and the mapping between its axes and the hand structure will be defined in a future hand model update.
sub = hand.imu().subscribe() # → Subscription[ImuData]
s = await sub.recv_async()
a, g = s.linear_acceleration, s.angular_velocity # m/s² and rad/s
print(s.header.seq, a.x, a.y, a.z, g.x, g.y, g.z)
sub.close()ImuData is a cross-device shared type. See ImuData for field definitions.
Joint Traversal
JointHandle provides label / index (for indexing into 20-element return arrays such as effort_limit() / mit_params()), plus single-joint resources and actions. FingerHandle traverses joints by finger.
| Method | Return type | Description |
|---|---|---|
hand.joints() | list[JointHandle] | All 20 joints |
hand.fingers() | list[FingerHandle] | All 5 fingers |
for joint in hand.joints():
print(joint.label, joint.index) # e.g. "thumb_S1" 0
for finger in hand.fingers():
for joint in finger.joints(): # 4 joints per finger, in S1..S4 order
print(joint.label) # "thumb_S1", "thumb_S2", ...JointHandle
Joint handle: provides label / index, single-joint resources, and single-joint actions.
| Member | Type | Description |
|---|---|---|
label | str | Property. Joint label, format {finger}_S{1..4}, e.g. "thumb_S1", "pinky_S4" |
index | int | Property. Global index (0–19) |
effort_limit() | Resource (SET/GET) | Single-joint torque cap (A) |
error_code() | Resource (GET) | The joint's current error code — decode with describe_error(). Returns an integer, so format it as f"0x{code:04X}" if you print it directly |
status_word() | Resource (GET) | The joint's status word |
enable() / disable() | Action | Enable / disable this joint |
clear_fault() | Action | Clear this joint's faults |
set_origin() / clear_origin() | Action | Set / clear this joint's user origin |
FingerHandle
Finger handle: traverse the 4 joints of this finger.
| Method | Return type | Description |
|---|---|---|
joints() | list[JointHandle] | The 4 joints of this finger, in S1..S4 order |
Listing All Resources
List all available resources, parameters, and topics on a device:
# All resources
for res in hand.resources():
print(res.path)
# Read/write parameters
for param in hand.params():
print(param.path)
# Subscribable topics
for topic in hand.topics():
print(topic.path)SDK User Management
When several operators share the same device, local parameters from one operator overwrite another's. The SDK keeps a local user profile per operator at ~/.wuji/sdk/users/<user_id>/, isolating each user's parameter files. Without a named user, parameter files land in ~/.wuji/sdk/params/ directly. Create a named user to isolate by operator:
from wuji_sdk import SdkManager
manager = SdkManager.instance()
# Create a user
alice = manager.create_user("Alice", description="right-hand operator")
# Switch to a specific user
manager.switch_user(alice["user_id"])
# Inspect the current user
print(manager.current_user())
# List all users
for user in manager.list_users():
marker = "*" if user["is_default"] else " "
print(f"{marker} {user['display_name']} ({user['user_id']})")
# Switch back to the default user
manager.switch_to_default_user()Switching reloads parameters for connected devices immediately, no reconnect needed. Deleting the current user automatically switches back to the default user. The C API offers the same user management calls.
Data Types
This section lists the cross-device common types first, then the Wuji Hand 2–specific ones.
FrameHeader
Header information for each data frame.
| Field | Type | Description |
|---|---|---|
seq | int | Incrementing sequence number |
timestamp_us | int | Device timestamp (microseconds) — see Data Frame Timestamps |
frame_id | str | Coordinate frame ID (e.g., "l_wrist"), max 32 characters |
Vector3 / Vector3F64
Three-dimensional vector. Vector3 uses f32 precision, Vector3F64 uses f64 precision (used for IMU data, ROS compatible).
| Field | Type | Description |
|---|---|---|
x | float | X component |
y | float | Y component |
z | float | Z component |
Quaternion
f64 precision quaternion representing 3D rotation.
| Field | Type | Description |
|---|---|---|
x | float | X component |
y | float | Y component |
z | float | Z component |
w | float | W component |
Handedness
Device handedness enum, used for connecting by handedness.
| Value | Description |
|---|---|
Handedness.Left | Left hand (fourth character of serial number is J) |
Handedness.Right | Right hand (fourth character of serial number is K) |
ImuData
IMU sensor data, following the ROS sensor_msgs/Imu convention. It is the frame type of the imu (SUB) stream.
| Field | Type | Description |
|---|---|---|
header | FrameHeader | Frame header |
orientation | Quaternion | Orientation quaternion |
orientation_covariance | list[float] | Orientation covariance (length 9). A first element of -1 marks the orientation as unavailable |
angular_velocity | Vector3F64 | Angular velocity (rad/s) |
angular_velocity_covariance | list[float] | Angular velocity covariance (length 9) |
linear_acceleration | Vector3F64 | Linear acceleration (m/s²) |
linear_acceleration_covariance | list[float] | Linear acceleration covariance (length 9) |
Wuji Hand 2 does not run onboard orientation fusion, so orientation_covariance[0] is always -1.
JointDiagnosticsFrame / JointDiagnosticsEntry
Joint diagnostics frame: frame type of the hand.joint_diagnostics().subscribe() stream. Variable-length, online joints only.
class JointDiagnosticsFrame:
header: FrameHeader # seq / timestamp_us / frame_id
num_joints: int
joints: list[JointDiagnosticsEntry]
comm: Hand2CommSummary # frame-level comm health summary, 1 Hz refresh (see Hand2CommSummary)
class JointDiagnosticsEntry:
nid: int # node ID — identifies the joint across frames
status_word: StatusWord # status word
current: float # current (A)
vbus_v_fb: float # bus voltage (V)
mcu_temp_c_fb: float # MCU temperature (°C)
error_code_current: int # current error code (integer value, shown in hex when the entry is printed) — decode with describe_error()
comm_response_rate_pct: int # bus response rate over the last second, 0–100
comm_timeout_total: int # cumulative bus timeouts for this jointMitParam
MIT parameters: element of mit_params().get().
class MitParam:
kp: float
kd: floatset accepts a single (kp, kd) (applied to all joints) or a list of 20 (per joint). get returns 20 entries — offline joints are None.
JointStateFrame
Joint state frame: return value of hand.joint_states().subscribe().recv(). Variable-length, online joints only.
| Attribute | Type | Description |
|---|---|---|
header | FrameHeader | seq / timestamp_us / frame_id |
num_joints | int | Joint count in this frame |
joints | list[JointStateEntry] | Joint state array |
JointStateEntry
Single-joint state.
| Attribute | Type | Description |
|---|---|---|
nid | int | Node ID — identifies the joint across frames |
position | float | Position (rad, joint-side) |
velocity | float | Velocity (rad/s) |
effort | float | Torque (A) |
JointCommand
Joint command: element of the joint_command().publish().send() argument.
class JointCommand:
position: float # position (rad)
velocity: float # velocity (rad/s)
effort: float # torque feedforward (A)HandCommunicationDiagnostics
Communication diagnostics data.
class HandCommunicationDiagnostics:
fingers: list[FingerCommunicationDiagnostics] # length 5
class FingerCommunicationDiagnostics:
tx_frame_total: int
rx_frame_total: int
tx_kbps: int
rx_kbps: int
error_per_sec: int
crc_error_total: int
frame_format_error_total: int
uart_hw_error_total: int
transfer_stats: list[TransferStats]
nodes: list[NodeDiagnostics] # per-node: online / ms_since_last_response / response_rate_pctStatusWord
Status word: the type of the status_word field on JointDiagnosticsEntry, the public status view exposed by the SDK.
| Field | Type | Description |
|---|---|---|
ext_state | int | Extended state value (such as Init / Ready / Enabled / Stopped) |
ext_state_name | str | Semantic name of ext_state |
position_limit_active | bool | Position limit triggered |
velocity_limit_active | bool | Velocity limit triggered |
current_limit_active | bool | Current limit triggered |
Hand2CommSummary
Frame-level communication health summary: the type of the comm field on JointDiagnosticsFrame, combining the device-internal bus view with the SDK-local end-to-end (Ethernet) view. The device-internal part comes from a snapshot the SDK refreshes automatically at 1 Hz — no polling required.
| Field | Type | Description |
|---|---|---|
age_ms | int | Age of the device-internal snapshot in milliseconds. 65535 = never sampled successfully (the device-internal fields are zeros). 65534 = saturated, the snapshot is 65.5 s or older |
e2e_received | int | Frames received across all subscribed streams (cumulative) |
e2e_lost | int | Frames lost on the Ethernet segment, detected via sequence gaps (cumulative) |
e2e_reordered | int | Late/reordered frames (cumulative) |
e2e_duplicates | int | Duplicate frames (cumulative) |
e2e_window_loss_x100 | int | Loss rate over the last second, in 0.01% units |
rpc_total | int | Requests sent (cumulative). 0 means this transport does not report RPC statistics (only the wuji-proto transport does, and a live wuji-proto connection has always sent at least one request), so a zero here is not the same as "no retries happened" |
rpc_retries | int | Request retransmissions (cumulative) |
rpc_timeouts | int | Requests that ultimately timed out (cumulative) |
comm_get_failures | int | Failed internal snapshot refreshes (cumulative) |
sdk_dropped | int | Frames dropped inside the SDK because one of its in-process hops fell behind (cumulative) — either your subscription consumer or the internal stream handler. Distinct from e2e_lost, which is network loss |
Reading the device-internal fields. age_ms and the per-joint comm_* fields
all come from one snapshot the SDK refreshes at 1 Hz. The refresh is deliberately
skipped while a firmware upgrade is running, so a briefly stale snapshot is expected
rather than a fault. Tell the two cases apart with
comm_get_failures: a growing age_ms while comm_get_failures stays flat means the
refresh is being skipped on purpose, while comm_get_failures increasing means the refresh
itself is failing.
Counters saturate, they do not wrap. Every counter in this table stops at its
maximum (65535 for the 16-bit ones — e2e_reordered, e2e_duplicates,
rpc_retries, rpc_timeouts, comm_get_failures — and 4294967295 for the 32-bit ones)
and stays there. On a 1 kHz stream a long-running session can reach those ceilings, so
read these as deltas between two samples rather than as absolute lifetime totals.
Joint Numbering
| Finger | S1 | S2 | S3 | S4 |
|---|---|---|---|---|
| thumb | 0 | 1 | 2 | 3 |
| index | 4 | 5 | 6 | 7 |
| middle | 8 | 9 | 10 | 11 |
| ring | 12 | 13 | 14 | 15 |
| pinky | 16 | 17 | 18 | 19 |
Label format: {finger}_S{1..4} (e.g. thumb_S1, pinky_S4). For the full table with motion ranges and model naming, see the Control Guide.
Time Synchronization
Every data frame the device outputs carries a timestamp. After the SDK connects, it synchronizes time automatically so device timestamps align with the host's UTC clock.
Data Frame Timestamps
Data frames share the same FrameHeader, in which timestamp_us is a 64-bit microsecond timestamp. Its meaning depends on whether time synchronization has completed:
- After sync:
timestamp_usis a UTC timestamp in microseconds (a Unix timestamp), directly aligned with the host clock - Before sync:
timestamp_usis device uptime, the number of microseconds since power-on
The Wuji Hand 2 has no real-time clock (RTC) and counts from 0 at power-on. Time synchronization aligns this uptime counter to real UTC time. The firmware stamps timestamp_us on feedback frames at send time.
How Synchronization Works
The SDK uses client-side time setting: the device does not fetch time on its own. Instead, the SDK pushes the host clock to the device.
The sync flow follows the NTP four-timestamp model — the SDK and device exchange a request/response pair and record four moments (request sent, device received, device replied, response received). From these, the SDK computes offset_us, the device clock's offset relative to host UTC, and pushes it to the device. The device then derives:
device_time = uptime_us + offset_usThis is a PTP-like lightweight synchronization that does not rely on standard PTP hardware. Sync accuracy depends on the network round-trip time between the SDK and the device.
Periodic auto-sync requires recent firmware. Earlier firmware applies the offset only on the first sync after connecting. Subsequent periodic syncs do not refresh the device clock. Upgrade to the latest firmware for full time synchronization.
Automatic Synchronization
After the SDK connects to a device, it maintains time synchronization automatically, and applications usually do not need to intervene:
- First sync: a full sync runs immediately when
connect()completes - Periodic sync: after connecting, a background sync runs every 30 seconds by default to counter device crystal drift
The background sync interval is configured through ConnectOptions.auto_time_sync_interval_ms:
from wuji_sdk import ConnectOptions
# default 30 seconds
options = ConnectOptions()
# custom 10 seconds
options = ConnectOptions(auto_time_sync_interval_ms=10_000)
# disable background auto-sync
options = ConnectOptions(auto_time_sync_interval_ms=None)| Value | Behavior |
|---|---|
| Not set | Default 30000 (30 seconds) |
None | Disables background auto-sync |
| Integer (milliseconds) | Custom interval, must be ≥ 100 |
When a session closes, the sync state clears automatically and re-syncs on the next connection.
Manual Synchronization
To force a sync at a moment of your choice (for example, before a long recording or a time-sensitive operation), call the device's sync_time():
result = hand.sync_time()
print(f"Offset: {result.offset_us} us")
print(f"Round trip: {result.round_trip_us} us")
print(f"Synced at: {result.synced_at_us} us")sync_time() is a blocking call. It triggers a full sync, pushes the offset to the device, and returns a TimeSyncResult:
| Field | Type | Description |
|---|---|---|
offset_us | int | Device clock offset. device_time = uptime_us + offset_us |
round_trip_us | int | Network round-trip time of this sync (microseconds), measured with a monotonic clock. Smaller is more accurate |
synced_at_us | int | Host UTC timestamp (microseconds) when this sync completed |
The SDK already runs a built-in 30-second background auto-sync, so most use cases do not need to call sync_time() manually. Manual calls are useful in two cases: forcing an immediate sync before a time-sensitive operation, or inspecting the quality of the latest sync through the returned round_trip_us.
Timestamp Monotonicity and Sync Accuracy
Device-stamped timestamp_us is strictly monotonically increasing across all sync scenarios. Even when the host clock steps backward or a periodic sync computes a reverse offset, the device smooths the correction so the timestamp never jumps backward. Each frame's timestamp_us is greater than the previous one, so the field can be used directly for ordering frames and computing sample intervals.
Sync accuracy is determined mainly by the network round-trip time between the SDK and the device. TimeSyncResult.round_trip_us reflects this metric. The smaller the round-trip time, the more accurate the offset estimate. Over a direct wired Ethernet connection, round-trip time is typically on the order of a few hundred microseconds. The periodic background sync continuously corrects device crystal drift, keeping the device clock aligned with host UTC even over long runs.
Data Recording
Real-time joint state and diagnostics data is ephemeral—recording persists it to files for offline analysis, algorithm debugging, or dataset building. The SDK includes a built-in recording engine that synchronously writes multi-channel data to MCAP files. The workflow takes three steps:
- Create a recorder — Create a
TopicRecorderand choose a compression algorithm - Register channels — Call
recorder.record(sub)to register subscription channels - Start recording — Call
await recorder.start(path)to begin recording, which returns aRecordingHandlefor control
import asyncio
from wuji_sdk import TopicRecorder
recorder = TopicRecorder(compression="lz4")
recorder.record(hand.joint_states().subscribe())
recorder.record(hand.joint_diagnostics().subscribe())
handle = await recorder.start("./data/session.mcap")
await asyncio.sleep(60)
summary = await handle.stop()Compression Options
Choose a compression algorithm with the compression parameter when creating TopicRecorder:
| Option | Description |
|---|---|
"lz4" | Low-latency compression for real-time use (default) |
"zstd" | High compression ratio for storage and archiving |
"none" | No compression, fastest write speed |
Prefer "lz4" for live capture and "zstd" for long-term archiving. joint_states runs at a native 1000 Hz, so it counts as a high-rate channel — keep it on "lz4" to stop compression becoming the bottleneck.
Pause and Resume
Pause and resume recording at any time. Device data continues to stream while paused, but nothing is written to the file.
await handle.pause() # Pause — data is not written
await handle.resume() # Resume recordingEpisode Switching
Call start() multiple times on the same TopicRecorder to switch output files without re-registering channels. Use this to split a continuous capture session into separate recording segments:
# Episode 1
handle1 = await recorder.start("./data/episode_001.mcap")
await asyncio.sleep(10)
await handle1.stop()
# Episode 2 — reuses the same channel configuration
handle2 = await recorder.start("./data/episode_002.mcap")
await asyncio.sleep(10)
await handle2.stop()Recording Monitoring
Monitor data quality metrics in real time during recording, and subscribe to quality alerts and run state as needed:
async for metrics in handle.subscribe_metrics():
print(f"Drop rate: {metrics.frame_drop_rate:.4f}")
print(f"Jitter: {metrics.frame_jitter_us:.1f} us")
print(f"Sync offset: {metrics.sync_offset_ms:.2f} ms")The recording engine includes a built-in SPC alert mechanism: when quality metrics exceed thresholds for a sustained period, handle.subscribe_alerts() delivers an alert. handle.subscribe_status() reports the current state, frames recorded, and elapsed duration. For the full field definitions of all three monitoring streams, see the Wuji SDK data reference.
Recording Summary
handle.stop() returns a RecordingSummary with recording statistics:
summary = await handle.stop()
print(f"Total frames: {summary.total_frames}")
print(f"File size: {summary.file_size / 1024 / 1024:.2f} MB")
print(f"Duration: {summary.duration_s:.1f}s")
print(f"Drop rate: {summary.quality.frame_drop_rate:.4f}")Recording Types
TopicRecorder: MCAP recording session configurator. Register channels, then call start() to begin recording.
| Method | Parameters | Description |
|---|---|---|
__init__() | compression: str = "lz4", chunk_size: int = None | Create a recorder. Supports "lz4", "zstd", "none" |
record() | sub: Subscription | Register a subscription channel |
start() | output_path: str | Start recording, returns RecordingHandle |
RecordingHandle: recording control handle returned by TopicRecorder.start().
| Method | Returns | Description |
|---|---|---|
pause() | — | Pause recording |
resume() | — | Resume recording |
stop() | RecordingSummary | Stop recording and return summary |
subscribe_metrics() | MetricsStream | Subscribe to real-time quality metrics |
subscribe_status() | StatusStream | Subscribe to recording status updates |
subscribe_alerts() | AlertStream | Subscribe to quality alerts |
RecordingSummary: recording statistics returned by handle.stop().
| Field | Type | Description |
|---|---|---|
total_frames | int | Total frames recorded |
file_size | int | MCAP file size in bytes |
duration_s | float | Recording duration in seconds |
quality | QualitySummary | Quality statistics summary |
QualitySummary: aggregate recording quality statistics, the type of RecordingSummary.quality.
| Field | Type | Description |
|---|---|---|
total_frames | int | Total frames received |
dropped_frames | int | Total frames dropped |
frame_drop_rate | float | Drop rate (0.0–1.0) |
avg_sync_offset_ms | float | Average sync offset in milliseconds |
max_sync_offset_ms | float | Maximum sync offset in milliseconds |
sync_rate | float | Sync success rate (0.0–1.0) |
spc_alert_count | int | Total SPC alerts triggered |
duration_s | float | Recording duration in seconds |
Exception Handling
All SDK operation errors raise a unified WujiException whose message carries an error-type prefix (Disconnected, Timeout, PathNotFound, SchemaMismatch, SerializeError, …). Use try / except as needed.
from wuji_sdk import WujiException
try:
hand.joint_states().subscribe()
except WujiException as e:
print(f"SDK error: {e}")Common Error Types
| Error | Description |
|---|---|
DeviceNotFound | Specified device not found |
DeviceMismatch | Device type mismatch |
Disconnected | Device has been disconnected |
ConnectionTimeout | Connection timed out |
OperationTimeout | Operation timed out |
PathNotFound | Resource path does not exist |
SchemaMismatch | Data type mismatch |
NoData | No data available |
StreamClosed | Subscription stream has been closed |
Branch on the error type:
from wuji_sdk import SdkManager, WujiException
try:
manager = SdkManager.instance()
hand = manager.auto_connect(device_name="wuji_hand_2")
sub = hand.joint_states().subscribe()
frame = sub.recv()
except WujiException as e:
error_msg = str(e)
if "DeviceNotFound" in error_msg:
print("Device not found, check connection")
elif "Disconnected" in error_msg:
print("Device disconnected")
elif "Connection timeout" in error_msg:
print("Connection timed out, check network/IP settings")
elif "Operation timeout" in error_msg:
print("Operation timed out, device not responding, retrying...")
else:
print(f"Error: {error_msg}")Device Fault Codes
error_code_current in the joint diagnostics and error_code in the error history are both u16 device fault codes. Decode them with WujiHand2.describe_error().
Device fault codes lay their hex digits out as 0xSCNN, but don't decode the digits yourself — the severity and clear_policy fields returned by describe_error() carry the same information as plain strings and stay correct if the layout ever changes. Treat the code as an opaque identifier: log it, show it to the user, quote it in a bug report. When decoding succeeds, use the returned severity and clear_policy to decide how to react. When describe_error() returns None (an unknown code), don't access those fields — keep the original value for logging and bug reports.
Reconnection Strategy
After disconnection, reconnect and resubscribe:
import time
from wuji_sdk import SdkManager, WujiException
manager = SdkManager.instance()
def connect_with_retry(sn, max_retries=5):
for attempt in range(max_retries):
try:
return manager.connect(sn=sn, device_name="wuji_hand_2")
except WujiException:
wait = 2 ** attempt # Exponential backoff
print(f"Connection failed, retrying in {wait}s...")
time.sleep(wait)
raise RuntimeError("Max retries exceeded")Logging and Troubleshooting
This section covers SDK-side troubleshooting. For device-side symptoms — device not discovered, IP conflicts, offline joints, enable failures, over-temperature and over-current faults — see Troubleshooting.
SDK Log Level
Adjust the SDK log level for debugging:
import wuji_sdk
wuji_sdk.set_log_level("debug") # Show debug information
wuji_sdk.set_log_level("trace") # Show all logs
wuji_sdk.set_log_level("info") # Default level
wuji_sdk.set_log_level("warn") # Warnings only
wuji_sdk.set_log_level("error") # Errors only
wuji_sdk.set_log_level("off") # Disable loggingRetrieve the device firmware's own runtime logs with Flash Log Export.
SDK Installation Fails
- Confirm Python version ≥ 3.10
- Update pip with
pip install --upgrade pip - If you encounter compilation errors, try installing the pre-built wheel package
Script Throws an Exception
| Exception Message | Cause | Resolution |
|---|---|---|
DeviceNotFound | Device not discovered | Check physical connection and network |
Disconnected | Connection interrupted | Check cables, reconnect |
ConnectionTimeout | Connection timed out | Check network connection or increase timeout_ms parameter |
OperationTimeout | Operation timed out | Increase timeout_ms parameter or check device status |
StreamClosed | Subscription stream closed | Device may have disconnected, reconnect |
SessionAlreadyExists | A session for the same device already exists | Keep using the existing handle, or disconnect() and reconnect |
ValueError | A write to hand.mit_params() (kp / kd both required non-negative), hand.effort_limit() (non-negative), or the joint_command realtime publisher receives NaN / infinity / negative values | Filter invalid values before writing |
ValueError | A write to hand.effort_limit() exceeds the ceiling the device currently allows, so the device rejects the write | Lower the value. Firmware sets the ceiling — read effort_limit back to confirm what's in effect |
Data Subscription Has Latency or Frame Drops
- Check network bandwidth and latency (wired connection preferred over WiFi)
- Reduce the number of simultaneously subscribed data streams
- Ensure callback functions don't contain blocking operations
- Subscribe to
hand.joint_diagnostics()and read the frame-levelcommsummary to locate where frames go missing:e2e_lostis network loss, whilesdk_droppedmeans the consumer is falling behind. For field definitions, see Hand2CommSummary
SDK Version Is Incompatible with the Firmware
- Check the SDK release notes for version compatibility
- Keep SDK and firmware on the same minor version (e.g., both 0.6.x)
C API Overview
The SDK also ships a C API (libwuji_sdk_c.so + wuji_sdk.h) with the same semantics as the Python interface. This section covers initialization, calling conventions, and how the two interfaces map onto each other. For the complete C structs, typed callbacks, and function signatures, see the Wuji SDK C API reference.
Initialization and Conventions
Call once at process start and release at shutdown:
#include "wuji_sdk.h"
if (wuji_init(NULL) != WUJI_STATUS_OK) {
fprintf(stderr, "init failed: %s\n", wuji_last_error());
return 1;
}
// ... your code
wuji_shutdown();Return value taxonomy:
WujiStatus— most operations (wuji_init/wuji_scan/wuji_connect/ opening subscriptions, etc.).WUJI_STATUS_OKmeans success. On failure, retrieve the per-thread error string viawuji_last_error().int32_t— metadata readers (wuji_dev_serial_number/wuji_dev_device_name) return the number of bytes written.uint8_t— status querywuji_dev_is_connected.void—wuji_shutdownand resource cleanup functions.
Resource cleanup functions:
wuji_discovered_free— release the device list returned bywuji_scanwuji_dev_release— paired withwuji_connect, release the device handle after disconnectingwuji_sub_close— close the subscription handle
Usage constraints (ignoring them leads to use-after-free, deadlocks, or stale error strings):
- A typed callback's
framepointer is valid only for the duration of the callback. Heap fields are released after the callback returns. Copy whatever you need to retain inside the callback. Do not store the pointer. - Never call
wuji_sub_closefrom inside that subscription's own callback.closejoins the worker thread and will deadlock if called from the callback itself. Close from a different thread. - The return value of
wuji_last_error()is only valid until the nextwuji_*call on the same thread. Copy the message immediately if you need to retain it. ENDandERRORare terminal frames — no further data will arrive after them, but you still need to callwuji_sub_closeto release the subscription handle.- String getters use a two-call query: first call with
buf=NULL, buf_len=0to read*needed(required bytes including NUL), then allocate and call again to fill. Passing an undersized buffer returnsWUJI_STATUS_ERR_BUFFER_TOO_SMALLand never truncates. - Whole-hand batch reads (
get_all_effort_limit/get_all_mit_params) return a flat-20 array plus an online bitmap. Use theWUJI_JOINT_ONLINE(mask, i)macro to check slot validity, otherwise you will read offline-joint placeholders (0 or NaN).
Python Equivalents
| Feature | C | Python |
|---|---|---|
| Initialization | wuji_init(NULL) | SdkManager.instance() |
| Scan | wuji_scan(&devs, &count) | manager.scan() |
| Connect | wuji_connect(&target, alias, &opts, &dev) | manager.connect(...) / manager.auto_connect(...) |
| Connect option defaults | wuji_connect_options_default() | ConnectOptions() |
| Stream output rate | wuji_sub_set_rate(sub, hz, &actual) | sub.set_rate(hz) |
| Control actions (with 20-joint mask) | wuji_hand_2_enable(dev, mask) | hand.enable(joints=mask) |
| Realtime commands | wuji_hand_2_joint_command_publish + wuji_joint_command_publisher_send | hand.joint_command().publish().send([...]) |
| Joint state subscription | wuji_hand_2_subscribe_joint_states | hand.joint_states().subscribe() |
| Joint diagnostics subscription | wuji_hand_2_subscribe_joint_diagnostics | hand.joint_diagnostics().subscribe() |
| IMU subscription | wuji_hand_2_subscribe_imu | hand.imu().subscribe() |
| Error code description | wuji_hand_2_describe_error(code, &info) | WujiHand2.describe_error(code) |
| SDK user management | wuji_create_user / wuji_switch_user / wuji_list_users | manager.create_user() / switch_user() / list_users() |
| Disconnect | wuji_dev_disconnect + wuji_dev_release | device.disconnect() |
For a CMake project example and the minimal link command, see C API Reference · CMake project example.
Interface Availability on Beta 1 Devices
The interfaces on this page work on both Beta 1 and Beta 2 devices. The capabilities below depend on hardware, so they stay unavailable or behave differently on Beta 1 devices even after you upgrade the firmware and the SDK:
| Interface or capability | On Beta 1 devices | Notes |
|---|---|---|
| Status light behavior | Requires the matching hardware batch | Beta 1 devices without status light hardware produce no light output, and the calls still succeed |
Device-internal fields in Hand2CommSummary | Depends on the firmware version | The exact version threshold isn't available yet and will be added later |
For the full hardware and software mapping, see Version Identification and Compatibility.