Developer Guide

SDK version: 2.1.0

This guide covers the Revo3 SDK 2.x asynchronous programming model, continuous data subscriptions, servo control, operation waiting and cancellation, production error handling, and offline diagnostics. Examples use Python. See the Revo3 API Reference for the corresponding C++ interfaces and complete signatures.

1. Asynchronous Programming with asyncio

All Revo3 Python 2.x device I/O APIs are asynchronous and must run in an asyncio event loop:

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

• Wrap synchronous entry points with asyncio.run(). Do not create an event loop directly inside callbacks or at library-module import time.
• SDK waits do not block the event loop. Other coroutines, subscriptions, and controls for other hands continue to run. Use asyncio.gather() for concurrent operations.
• The manager.discover() on_found callback runs synchronously on the SDK scan thread. It is not a coroutine and cannot use await. Return False to stop scanning early.
• Properties and non-I/O methods, such as hand.device_info and session.state, are not coroutines. Only device I/O and wait operations require await.

2. State, Touch, and Health Subscriptions

The Manager and Hand object APIs provide pull-based asynchronous subscriptions for State, Touch, and Health.

2.1 State Subscription

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 contains operating_states, positions_deg, velocities_rpm, currents_ma, and timestamp. Read low-rate raw motor status codes through hand.health.snapshot().motor_fault_codes. This compatibility field includes both fault bits and non-fault state bits.

2.2 Touch Subscription

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 contains the modules available for the detected layout. Depending on module capabilities, each entry can contain taxel values, calibrated regional force in regional_forces_mn, three-axis force, two-axis torque, resultant_force_mn, and status fields. Three-axis and resultant force values use mN.

2.3 Health Subscription

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 Subscription Semantics

• subscribe() returns a pull-based subscription. Call next() to wait for the next read.
• Each next() reads the current State, Touch, or Health snapshot after the configured interval. Subscriptions do not retain historical samples.
• close() is idempotent. A closed subscription performs no further reads.
• State, Touch, and Health periods are independent and must be finite positive values in seconds.
• Timestamp.clock identifies the clock domain. Compare or subtract timestamps only when their clock values match.
• Subscriptions are for monitoring. Start servo control through hand.motion.open_servo().

3. Servo Control

Use open_servo() for vision tracking, teleoperation, or algorithmic control that sends targets at a fixed period.

The following example demonstrates session lifetime and periodic position commands. Before running it, secure the hand, remove mechanical interference, configure an appropriate current limit, and keep an independent power disconnect within reach. The application remains responsible for target generation, joint-limit checks, and recovery after failures.

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("Servo command result is indeterminate; current positions:", state.positions_deg)
    raise
finally:
    session.close()

In addition to send_position, a servo session provides send_velocity, send_current, send_impedance, and send_mit. See Hand Domain APIs for parameters and units.

• command_timeout_ms monitors the interval between consecutive servo commands. It is not a background heartbeat. When omitted, the session uses RuntimeOptions.servo_command_timeout_ms.
• After a timeout, ServoSession.state becomes Expired. The SDK releases software ownership and rejects additional commands from that session.
• Session expiry does not send a stop command and does not prove that the motors stopped or released torque. Apply firmware watchdog requirements and hardware-validated recovery behavior separately.

4. Waiting for and Cancelling Operations

Motion calls such as move_to, move_finger, move_thumb, and move_joint return an OperationHandle that tracks the final outcome:

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("Motion completed")
    elif result == sdk.OperationState.Indeterminate:
        # The device may be moving. Read the actual position before deciding what to do.
        state = await hand.state.snapshot()
        print("Result is indeterminate; current positions:", state.positions_deg)
except sdk.SdkError as error:
    print(f"Motion failed [{error.code}]: {error.message}")

• wait(timeout) waits for a terminal state. A timeout only stops the application from waiting; physical motion already accepted by the device can continue.
• handle.state is always readable. After a terminal state, handle.error contains the associated SdkError, or None when no error occurred.
• handle.cancel() requests cooperative cancellation. The SDK lets the active register request finish, stops sending trajectory points at the next control-cycle boundary, and releases software ownership. It does not discard an in-flight serial request. If the final physical position cannot be confirmed, the operation ends as Indeterminate.
• See Waiting, Cancellation, and Motion Conflicts for the full state machine.

5. Production Error Handling and Retry

Object API failures raise SdkError. In addition to code and message, the following machine-readable fields support recovery decisions:

Field	Type	Purpose
operation_effect	OperationEffect	Whether a write command may have taken effect
retryable	bool	Whether the SDK considers an automatic retry safe; limited to side-effect-free reads with connection or timeout errors
recovery_requirement	RecoveryRequirement	Required action: None_, Retry, Reconnect, or OperatorAction; Python uses None_ because None is a keyword, while C++ uses None
low_level_cause	str | None	Original driver-level cause for diagnostic records

For failed write commands, inspect operation_effect first:

• NotApplied: the device did not execute the command, for example because validation failed before transmission.
• PartiallyApplied: part of the operation took effect, such as a manager close that could not release every transport.
• Indeterminate: the command was sent but its response was lost. Read State to determine the current physical condition. Do not automatically repeat the write.

The following pattern retries only when the SDK marks a read as retryable:

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  # Let the caller decide whether to reconnect.
            if not error.retryable or attempt + 1 >= max_retries:
                raise
            await asyncio.sleep(0.1 * (attempt + 1))

Warning: Automate retries only for read operations with retryable=True. Write operations, including motion, configuration, and firmware updates, always report retryable=False. Recover according to operation_effect and recovery_requirement.

6. Diagnostics, Recording, and Offline Replay

After installing the public examples as described in Quickstart, run the diagnostics and capture CLI from the example repository root:

# 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

Offline replay only reads and displays the recording. It does not send historical motion to a device.

6.1 Teach and Motion Replay

The manager_cli record/replay commands above record state for offline inspection. The separate teach/replay motion APIs record manual movement as a trajectory and execute that trajectory on a device:

try:
    trajectory = await asyncio.wait_for(
        hand.motion.teach_hand(duration=10.0, dt=0.02),
        timeout=15.0,
    )

    # Replay produces physical motion. Complete the mechanical and power checks first.
    await asyncio.wait_for(
        hand.motion.replay_hand(trajectory, dt=0.02),
        timeout=15.0,
    )
except (asyncio.TimeoutError, sdk.SdkError) as error:
    # A timeout or communication failure does not prove that the device did not move.
    state = await hand.state.snapshot()
    print("Teach or replay did not complete with confirmation; current positions:", state.positions_deg)
    raise

Use teach_joint and replay_joint for a single joint. See Hand Domain APIs for control semantics and safety boundaries.