SDK 接口

from wuji_sdk import SdkManager

manager = SdkManager.instance()
devices = manager.scan()

for dev in devices:
    print(f"SN: {dev.sn}, 类型: {dev.device_type}, 地址: {dev.address}")

from wuji_sdk import SdkManager, Handedness

manager = SdkManager.instance()

# 自动发现并连接局域网内首个 Wuji Hand 2
hand = manager.auto_connect(device_name="wuji_hand_2")

# 已知序列号或地址时显式连接
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")

# 或按左右手直接连接，无需序列号
hand = manager.connect(handedness=Handedness.Right, device_name="wuji_hand_2")

from wuji_sdk import SdkManager, ConnectOptions

opts = ConnectOptions(
    timeout_ms=1000,
    retry_count=3,
    enable_bridge=True,    # 默认 True：允许多个客户端（Wuji Studio + 录制脚本 + 业务程序）同时连接
)
hand = manager.connect(sn=<device_sn>, device_name="wuji_hand_2", options=opts)

left = manager.connect(handedness=Handedness.Left, device_name="left_hand")
right = manager.connect(handedness=Handedness.Right, device_name="right_hand")

# 获取已连接设备列表
all_devices = manager.get_connected_devices()
for name, device in all_devices:
    print(f"{name}: {device.serial_number}")

# 按名称获取设备
left = manager.get_device(device_name="left_hand")

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

# 断开指定设备
manager.disconnect(device_name="wuji_hand_2")

# 或通过设备对象断开
hand.disconnect()

sub = hand.joint_states().subscribe()
frame = await sub.recv_async()  # 等待下一帧数据

sub = hand.joint_states().subscribe()
frame = sub.recv()  # 无数据时返回 None

def on_data(frame):
    print(frame.header.seq)

sub = hand.joint_states().subscribe_with_callback(callback=on_data)
# 后台自动接收，不阻塞主线程

sub = hand.joint_states().subscribe_with_callback(callback=on_data)
# ... 使用一段时间后
sub.close()

sub = hand.joint_states().subscribe()

actual = sub.set_rate(200)   # 设备量化后返回实际生效值
print(f"requested 200 Hz, applied {actual} Hz")

sub.set_rate(0)              # 传 0 恢复设备默认满速

side = hand.handedness().get()   # → "left" 或 "right"

n = hand.online_joints_count().get()   # → int，0–20

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

diag = hand.comm_diag().get()           # → HandCommunicationDiagnostics
for finger in diag.fingers:             # 5 根手指各自的吞吐 / 错误率
    print(finger.tx_kbps, finger.rx_kbps, finger.error_per_sec)

hand.effort_limit().set(1.5)            # 全部关节设为 1.5 A
limits = hand.effort_limit().get()      # → list[Optional[float]]

hand.mit_params().set((1.0, 0.05))           # 全部关节统一 kp / kd
hand.mit_params().set([(1.0, 0.05)] * 20)    # 长度 20 的列表逐关节设置
mp = hand.mit_params().get()                 # → list[Optional[MitParam]]，离线关节为 None

hand.clear_fault()                  # 清除全部关节故障
hand.clear_fault(joints=mask)       # 可选：长度 20 的 0/1 掩码，只清除掩码位为 1 的关节

hand.set_origin()                   # 整手：所有关节刷新用户零点
hand.clear_origin()                 # 整手：清除所有关节的用户零点
hand.set_origin(joints=mask)        # 可选：长度 20 的 0/1 掩码，只作用于掩码位为 1 的关节

hand.enable()               # 使能全部关节
hand.disable()              # 失能全部关节
hand.enable(joints=mask)    # 可选：长度 20 的 0/1 掩码，作用于掩码位为 1 的关节
hand.emergency_stop()       # 急停：整手动作，不接受掩码

sub = hand.joint_states().subscribe()   # → Subscription[JointStateFrame]
frame = sub.recv()                      # 同步非阻塞，None = 暂无数据
frame = await sub.recv_async()          # 异步等待
print(frame.header.seq, frame.header.timestamp_us)
for j in frame.joints:
    print(j.nid, j.position, j.velocity, j.effort)
sub.close()

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

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

# 导出当前固件的运行日志（绝大多数排查场景用这个）
result = await hand.dump_hand_logs(bank="current", out_dir="./logs")

result = await hand.dump_hand_logs(bank="current")
print(result["session_dir"])    # str: 本次会话目录绝对路径
print(result["files"])          # list[str]: 已写入的 .log 文件路径

hand.ip().get()                    # 读当前 IP，如 "192.168.3.110"
hand.ip().set("192.168.2.111")     # 写入 flash，重启后生效

hand.ip().set("192.168.2.111")
hand.reboot()
manager.disconnect_all()
# 等设备重启且以太网就绪，约 8 秒
hand = manager.connect(address="192.168.2.111:50001", device_name="wuji_hand_2")
hand.ip().get()                    # → "192.168.2.111"

hand.reboot()

sub = hand.imu().subscribe()                 # → Subscription[ImuData]
s = await sub.recv_async()
a, g = s.linear_acceleration, s.angular_velocity   # m/s² 与 rad/s
print(s.header.seq, a.x, a.y, a.z, g.x, g.y, g.z)
sub.close()

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 个关节，按 S1..S4 顺序
        print(joint.label)                 # "thumb_S1", "thumb_S2", ...

# 所有资源
for res in hand.resources():
    print(res.path)

# 可读写参数
for param in hand.params():
    print(param.path)

# 可订阅 Topic
for topic in hand.topics():
    print(topic.path)

from wuji_sdk import SdkManager

manager = SdkManager.instance()

# 创建用户
alice = manager.create_user("Alice", description="右手操作员")

# 切换到指定用户
manager.switch_user(alice["user_id"])

# 查看当前用户
print(manager.current_user())

# 列出所有用户
for user in manager.list_users():
    marker = "*" if user["is_default"] else " "
    print(f"{marker} {user['display_name']} ({user['user_id']})")

# 切回默认用户
manager.switch_to_default_user()

class JointDiagnosticsFrame:
    header: FrameHeader                    # seq / timestamp_us / frame_id
    num_joints: int
    joints: list[JointDiagnosticsEntry]
    comm: Hand2CommSummary                 # 帧级通信健康摘要，1 Hz 刷新（见 Hand2CommSummary）

class JointDiagnosticsEntry:
    nid: int                               # 节点编号，跨帧定位关节
    status_word: StatusWord                # 状态字
    current: float                         # 电流 (A)
    vbus_v_fb: float                       # 母线电压 (V)
    mcu_temp_c_fb: float                   # MCU 温度 (°C)
    error_code_current: int                # 当前错误码（整数值，打印条目对象时按 16 进制显示），用 describe_error() 解码
    comm_response_rate_pct: int            # 该关节最近 1 秒的总线响应率，0–100
    comm_timeout_total: int                # 该关节累计总线超时次数

class MitParam:
    kp: float
    kd: float

class JointCommand:
    position: float    # 位置 (rad)
    velocity: float    # 速度 (rad/s)
    effort: float      # 力矩前馈 (A)

class HandCommunicationDiagnostics:
    fingers: list[FingerCommunicationDiagnostics]   # 长度 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_pct

设备时间 = uptime_us + offset_us

from wuji_sdk import ConnectOptions

# 默认 30 秒
options = ConnectOptions()

# 自定义为 10 秒
options = ConnectOptions(auto_time_sync_interval_ms=10_000)

# 关闭后台自动同步
options = ConnectOptions(auto_time_sync_interval_ms=None)

result = hand.sync_time()
print(f"偏移量: {result.offset_us} us")
print(f"往返时延: {result.round_trip_us} us")
print(f"同步完成时刻: {result.synced_at_us} us")

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

await handle.pause()     # 暂停——数据不会写入
await handle.resume()    # 恢复录制

# Episode 1
handle1 = await recorder.start("./data/episode_001.mcap")
await asyncio.sleep(10)
await handle1.stop()

# Episode 2——复用已注册的通道配置
handle2 = await recorder.start("./data/episode_002.mcap")
await asyncio.sleep(10)
await handle2.stop()

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

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

from wuji_sdk import WujiException

try:
    hand.joint_states().subscribe()
except WujiException as e:
    print(f"SDK error: {e}")

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("未找到设备，请检查连接")
    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="wuji_hand_2")
        except WujiException:
            wait = 2 ** attempt  # 指数退避
            print(f"连接失败，{wait}s 后重试...")
            time.sleep(wait)
    raise RuntimeError("达到最大重试次数")

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")     # 关闭日志

#include "wuji_sdk.h"

if (wuji_init(NULL) != WUJI_STATUS_OK) {
    fprintf(stderr, "init failed: %s\n", wuji_last_error());
    return 1;
}

// ... 业务逻辑

wuji_shutdown();