Best Practices

from wuji_sdk import SdkManager, WujiException

try:
    manager = SdkManager.instance()
    glove = manager.auto_connect(device_name="glove")
    sub = glove.tactile().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}")

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="glove")
        except WujiException:
            wait = 2 ** attempt  # Exponential backoff
            print(f"Connection failed, retrying in {wait}s...")
            time.sleep(wait)
    raise RuntimeError("Max retries exceeded")

import asyncio
from wuji_sdk import SdkManager

async def collect_from_device(glove, name):
    sub = glove.tactile().subscribe()
    while True:
        frame = await sub.recv_async()
        print(f"[{name}] max={max(frame.data):.1f}")

async def main():
    manager = SdkManager.instance()
    devices = manager.scan()

    gloves = []
    for i, dev in enumerate(devices):
        glove = manager.connect(sn=dev.sn, device_name=f"glove_{i}")
        gloves.append((glove, f"glove_{i}"))

    tasks = [collect_from_device(g, n) for g, n in gloves]
    await asyncio.gather(*tasks)

asyncio.run(main())

manager = SdkManager.instance()
left = manager.connect(sn="WG1JA00XXXXXXXXX", device_name="left")
right = manager.connect(sn="WG1KA00XXXXXXXXX", device_name="right")

# Subscribe to global coordinate transforms — automatically merges left and right hand data
sub = manager.tf_static().subscribe()
transforms = await sub.recv_async()
for tf in transforms.transforms:
    print(f"{tf.parent_frame_id} → {tf.child_frame_id}")

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 logging

Privacy Notice
Wuji SDK collects anonymous usage data to improve the product.
No personal or sensitive information is collected.
Privacy Policy: https://wuji.tech/en/terms/privacy
If your environment has information security requirements, you can disable telemetry upload for the current process by setting:
WUJI_TELEMETRY_DISABLE=1 <your_command>

# Executable
WUJI_TELEMETRY_DISABLE=1 ./<your_program>

# Python script
WUJI_TELEMETRY_DISABLE=1 python <your_script.py>

import asyncio
from wuji_sdk import TopicRecorder

recorder = TopicRecorder(compression="lz4")
recorder.record(glove.tactile().subscribe())

for i in range(5):
    handle = await recorder.start(f"./data/episode_{i:03d}.mcap")
    await asyncio.sleep(60)  # 60 seconds per episode
    await handle.stop()

import asyncio

handle = await recorder.start("./data/session.mcap")

async def watch_alerts():
    async for alert in handle.subscribe_alerts():
        print(f"Warning: {alert.message}")

asyncio.create_task(watch_alerts())

# ✓ Subscribe only to needed data
sub_tactile = glove.tactile().subscribe()

# ✗ Don't subscribe to unused data
# sub_imu = glove.imu_palm().subscribe()

sub = glove.tactile().subscribe_with_callback(callback=on_data)
# ... done collecting data
sub.close()  # Release resources