最佳实践

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("未找到设备，请检查连接")
    elif "Disconnected" in error_msg:
        print("设备已断开")
    elif "Connection timeout" in error_msg:
        print("连接超时，请检查网络和 IP 设置")
    elif "Operation timeout" in error_msg:
        print("操作超时，设备无响应，正在重试...")
    else:
        print(f"错误: {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  # 指数退避
            print(f"连接失败，{wait}s 后重试...")
            time.sleep(wait)
    raise RuntimeError("达到最大重试次数")

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

# 订阅全局坐标变换——自动合并左右手数据
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")   # 显示调试信息
wuji_sdk.set_log_level("trace")   # 显示所有日志
wuji_sdk.set_log_level("info")    # 默认级别
wuji_sdk.set_log_level("warn")    # 仅警告
wuji_sdk.set_log_level("error")   # 仅错误
wuji_sdk.set_log_level("off")     # 关闭日志

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>

# 可执行程序
WUJI_TELEMETRY_DISABLE=1 ./<your_program>

# Python 脚本
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)  # 每个 episode 60 秒
    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"告警: {alert.message}")

asyncio.create_task(watch_alerts())

# ✓ 只订阅需要的数据
sub_tactile = glove.tactile().subscribe()

# ✗ 不要订阅用不到的数据
# sub_imu = glove.imu_palm().subscribe()

sub = glove.tactile().subscribe_with_callback(callback=on_data)
# ... 完成数据采集
sub.close()  # 释放资源