开发指南

SDK 版本：2.1.0

本文为 Revo3 SDK 2.0 开发者实战指南，涵盖异步编程模型、持续数据订阅、高频实时流式控制、运动命令的等待与取消、生产环境错误处理与重试策略，以及数据诊断与离线回放。示例以 Python 为主；C++ 对应接口请参阅 Revo3 API 参考手册。

完整对象架构、数据结构字段和方法签名请参阅 Revo3 API 参考手册。

1. 异步编程模型 (asyncio)

Revo3 Python 2.0 的所有设备 I/O API 都是异步的，必须在 asyncio 事件循环中调用：

import asyncio
from bc_revo3_sdk import main_mod as sdk


async def main() -> None:
    manager = sdk.Manager()
    try:
        hand = await manager.connect_auto()
        state = await hand.state.snapshot()
        print(state.positions_deg)
    finally:
        await manager.close()


asyncio.run(main())

• 同步代码入口用 asyncio.run() 包装；不要在回调或库模块顶层直接创建事件循环。
• SDK 的等待不会阻塞事件循环：等待期间其他协程（订阅、其他手的控制）照常调度，可用 asyncio.gather() 并发操作多台设备。
• manager.discover() 的 on_found 回调在 SDK 扫描线程上同步执行，不是协程：回调内不能 await，返回 False 可提前终止扫描。
• 对象方法与属性本身不是协程（如 hand.device_info、session.state），可直接使用；只有涉及设备 I/O 或等待的方法需要 await。

2. 状态与触觉持续订阅 (Subscription)

Revo3 Python 2.0 通过 Manager/Hand 对象 API 提供 State、Touch 和 Health 的异步拉取订阅。

2.1 电机状态订阅

from bc_revo3_sdk import main_mod as sdk


async def monitor_state(hand: sdk.Hand) -> None:
    subscription = hand.state.subscribe(period=0.02)
    try:
        while True:
            state = await subscription.next()
            print(state.timestamp.clock, state.positions_deg)
    finally:
        subscription.close()

HandState 包含 operating_states、positions_deg、velocities_rpm、currents_ma 和 timestamp。逐电机原始状态码属于低频诊断数据，通过 hand.health.snapshot().motor_fault_codes 读取；该兼容字段同时包含故障位和非故障状态位。

2.2 Touch 触觉订阅

async def monitor_touch(hand: sdk.Hand) -> None:
    subscription = hand.touch.subscribe(period=0.05)
    try:
        while True:
            frame = await subscription.next()
            print(len(frame.modules), frame.modules[0].points)
    finally:
        subscription.close()

TouchFrame 统一提供 modules 数组；每个 module 按能力选择性包含点阵、模块级合力 regional_forces_mn、三轴力、二轴力矩、合力 resultant_force_mn 和状态字段。三轴力和合力统一使用 mN。

2.3 Health 健康订阅

async def monitor_health(hand: sdk.Hand) -> None:
    subscription = hand.health.subscribe(period=0.2)
    try:
        while True:
            health = await subscription.next()
            print(health.safety_state, health.error_code)
    finally:
        subscription.close()

2.4 订阅语义与注意事项

• subscribe() 返回拉取式订阅；调用 next() 等待下一次读取结果。
• 每次 next() 在配置间隔后直接读取当前 State、Touch 或 Health 快照；订阅不保留历史样本。
• close() 可以重复调用；关闭后该订阅不再读取数据。
• State、Touch 和 Health 的 period 相互独立，并且必须是有限正数（秒）。
• Timestamp.clock 标识时钟域；只有 clock 相同时，timestamp 才能比较或相减。
• 订阅只用于监控；Servo 命令仍从 hand.motion.open_servo() 进入。

3. 高频实时流式控制 (Servo)

对于机械臂视觉跟随、遥操作或实时算法控制场景，推荐使用 open_servo() 建立高频流式会话。

以下片段只展示会话生命周期和固定周期发送方式。执行前必须固定灵巧手、排除机构干涉、设置合理电源限流，并让独立断电开关保持可触达；目标生成、限位检查和异常后的安全恢复按应用要求实现。

import asyncio


period = 0.02
state = await hand.state.snapshot()
target_positions = list(state.positions_deg)
session = hand.motion.open_servo(command_timeout_ms=100)
try:
    for _ in range(100):
        await session.send_position(target_positions)
        await asyncio.sleep(period)
except sdk.SdkError as error:
    if error.operation_effect == sdk.OperationEffect.Indeterminate:
        state = await hand.state.snapshot()
        print("流式命令结果不确定，当前位置:", state.positions_deg)
    raise
finally:
    session.close()

除 send_position 外，Servo 会话还提供 send_velocity、send_current、send_impedance 与 send_mit，参数与单位见 API 参考手册 - Servo。

• command_timeout_ms 监视相邻两次流式控制命令的时间间隔，不是后台心跳；未显式传入时使用 RuntimeOptions.servo_command_timeout_ms；
• 超时后 ServoSession.state 变为 Expired，SDK 释放软件控制权，并拒绝该会话继续发送命令；
• 该机制不发送停止命令，也不能证明电机已经停止、卸力或进入安全状态。固件通信 Watchdog 的触发条件和机械行为必须按固件规格及真机结果处理。

4. 运动命令的等待与取消 (OperationHandle)

move_to、move_finger、move_thumb、move_joint 等运动命令返回 OperationHandle，用于跟踪一次运动的最终结果：

motion = await hand.motion.move_to(target_positions, duration=1.5)
try:
    result = await motion.wait(timeout=5.0)
    if result == sdk.OperationState.Succeeded:
        print("运动完成")
    elif result == sdk.OperationState.Indeterminate:
        # 设备可能已在运动：先读实际位置，再决策，禁止直接重发
        state = await hand.state.snapshot()
        print("结果不确定，当前位置:", state.positions_deg)
except sdk.SdkError as error:
    print(f"运动失败 [{error.code}]: {error.message}")

• wait(timeout) 等待终态；超时仅表示停止等待，硬件已接收的物理运动仍会继续执行。
• handle.state 随时可读；handle.error 在终态后携带绑定的 SdkError（无错误时为 None）。
• handle.cancel() 请求协作式取消：SDK 会让当前寄存器请求完整结束，在下一个控制周期边界停止发送轨迹点并释放软件控制权，不会中途丢弃串口请求。取消请求发出后，若设备最终位置无法确认，终态为 Indeterminate。
• 完整状态机（Pending/Running/Succeeded/Preempted/Failed/Indeterminate 等）见 API 参考手册 - 等待、取消和运动冲突。

5. 生产环境错误处理与重试策略

对象调用失败时会抛出 SdkError。除 code 和 message 外，SdkError 还提供四个机器可读字段用于生产环境的自动化决策：

字段	类型	用途
operation_effect	OperationEffect	写命令是否可能已生效（见下）
retryable	bool	SDK 确认可安全自动重试（仅限无副作用的只读操作遇连接/超时类错误）
recovery_requirement	RecoveryRequirement	建议恢复动作：None_/Retry/Reconnect/OperatorAction（Python 中 None 为关键字，枚举成员名为 None_；C++ 为 None）
low_level_cause	str | None	底层驱动原始原因，用于排障记录

写命令失败时必须先检查 operation_effect：

• NotApplied：命令确定未被设备执行（只读失败或发包前校验失败），无副作用。
• PartiallyApplied：部分生效（如 Manager 关闭未能释放全部传输）。
• Indeterminate：命令已发出但响应丢失，无法确认是否执行。应先读取 hand.state.snapshot() 确认当前机械位置，禁止直接无脑重发写命令。

含重连决策的重试示例：

async def read_with_retry(hand: sdk.Hand, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return await hand.state.snapshot()
        except sdk.SdkError as error:
            print(f"[{error.code}] {error.message}")
            if error.recovery_requirement == sdk.RecoveryRequirement.Reconnect:
                # 传输层已失步或断开：先重建连接再重试
                await hand.close()
                raise  # 由上层决定是否重新 connect
            if not error.retryable or attempt + 1 >= max_retries:
                raise
            await asyncio.sleep(0.1 * (attempt + 1))  # 退避后重试

警告： 重试自动化仅限 retryable=True 的只读操作。写命令（运动、配置、固件）的 retryable 恒为 False：写失败时按 operation_effect 与 recovery_requirement 走人工确认或状态核实流程。

6. 数据诊断、录制与离线回放

按照快速入门获取并安装公开示例后，在示例仓库根目录运行诊断与状态采集 CLI：

# Inspect device information and current state.
python \
  python/revo3/manager_cli.py --port /dev/ttyUSB0 inspect

# Record state data as JSONL.
python \
  python/revo3/manager_cli.py --port /dev/ttyUSB0 \
  record state.jsonl --duration 10

# Replay the recording offline.
python \
  python/revo3/manager_cli.py replay state.jsonl

回放默认只读取并展示记录，不向设备发送运动命令，避免把历史轨迹误当成已验证指令。

6.1 示教与轨迹回放 (Teach/Replay)

注意区分：上面的 manager_cli record/replay 是状态数据的离线查看；SDK 另提供示教运动 API，把人工拖动记录为轨迹并回放到设备执行：

try:
    # 示教：拖动手部，SDK 按 dt 采样记录整手轨迹
    trajectory = await asyncio.wait_for(
        hand.motion.teach_hand(duration=10.0, dt=0.02),
        timeout=15.0,
    )

    # 回放会产生真实运动；执行前必须完成固定、限流和独立断电检查
    await asyncio.wait_for(
        hand.motion.replay_hand(trajectory, dt=0.02),
        timeout=15.0,
    )
except (asyncio.TimeoutError, sdk.SdkError) as error:
    # 超时或通信失败不能证明设备未执行，先读取实际状态再决定恢复动作
    state = await hand.state.snapshot()
    print("示教或回放未确认完成，当前位置:", state.positions_deg)
    raise

单关节版本为 teach_joint / replay_joint。示教与回放的控制语义和安全边界见 API 参考手册。