SDKPython SDK

采集

采集的作用、引导式交互、采集进度与控制和手动发起采集。

当前版本 · 1.4.0

采集

先读：设备发现与连接（设备连接成功后才能开始采集） · 读完你能：要么让默认流程跑完采集，要么在程序里接管采集交互

采集配置

自动流程

默认自动流程在创建 Sdk 时从 AlgorithmsConfig.capture 读取采集配置。配置前先按目的选择：

目的配置
在交互式终端由 SDK 显示动作提示interaction=CaptureInteractionMode.AUTO_TERMINAL
由 GUI、ROS2 或远程程序显示提示和控件interaction=CaptureInteractionMode.EXTERNAL
忽略可复用的旧采集，完整重新采样force=True
按佩戴者分别查找和保存结果operator_id="operator-a"

例如，使用默认终端引导，并强制 operator-a 完整重新采集：

from qnbot_sdk import (
    AlgorithmsConfig,
    CaptureConfig,
    CaptureInteractionMode,
    Sdk,
)

sdk = Sdk(
    devices=(device_config,),
    targets=(target_config,),
    algorithms=AlgorithmsConfig(
        capture=CaptureConfig(
            interaction=CaptureInteractionMode.AUTO_TERMINAL,
            force=True,
            operator_id="operator-a",
        ),
    ),
)

字段类型、默认值与约束见 CaptureConfig。

手动流程

手动流程在 start_capture() 调用中直接提供本次采集所需的参数：

目的参数
选择本次使用的算法包和采集计划package_ids=("<package-id>",)
按佩戴者查找和保存结果operator_id="operator-a"
忽略可复用的旧采集，完整重新采样force=True

例如，强制 operator-a 使用一个算法包完整重新采集：

session = device.start_capture(
    package_ids=("<package-id>",),
    operator_id="operator-a",
    force=True,
)

如果你只想使用已有采集重新计算标定，先看概览中两个 force 的区别。

采集状态与控制

默认自动流程会自行检查并推进采集，不需要调用下面的接口。只有两种接管场景需要它们：

  • 应用决定何时启动：capture_status() → 判断状态 → start_capture()。
  • 外部 UI 接管提示：capture_progress() → 展示当前动作 → capture_control()。

判断是否启动手动采集

先查询当前算法包和佩戴者是否仍需采集：

readiness = device.capture_status(
    package_ids=("<package-id-1>", "<package-id-2>"),
    operator_id="operator-a",
)
render_capture_readiness(readiness.state, readiness.plan, readiness.stages)

根据 readiness.state 决定下一步：

状态下一步
NEEDS_CAPTURE调用 start_capture() 创建采集会话
READY复用已有采集，继续标定
NOT_REQUIRED本次选择无需采集，继续后续流程

plan 和 stages 用于向佩戴者展示本次计划。完整字段见 CaptureReadiness，启动代码见手动发起采集。

在外部 UI 中推进采集

订阅进度，展示当前动作，并用同一个进度中的标识响应：

from qnbot_sdk.glove import CaptureStageState

progress = device.capture_progress()
control = device.capture_control()

def on_progress(sample) -> None:
    stage = sample.value.stage
    render_prompt(stage.prompt, stage.collected_sample_count,
                  stage.required_sample_count)
    if (stage.state is not CaptureStageState.AWAITING_CONFIRMATION
            or stage.request_id is None):
        return
    if stage.optional and should_skip():
        control.skip(stage.request_id)
    else:
        control.confirm(stage.request_id)

progress.subscribe(on_progress)

render_prompt() 和 should_skip() 由应用实现。重试阶段时调用 control.retry(stage_id)，取消当前请求时调用 control.cancel(request_id)；标识必须来自最新进度。完整流程见外部接管引导交互，字段定义见 CaptureProgress。

采集方式

引导式采集交互

默认 auto_terminal 会在交互控制台中引导采集：前台 run_forever() 可以启用终端交互；run_background() 启动时若仍需采集或标定，也会先完成终端引导。真实 TTY 会自动启用，其他控制台的判断见 终端输出与交互。

终端输出示例：

[Capture] glove.primary.right
open_hand: Open your hand
Press Enter to begin...
[####################] 2/2

[Calibration] glove.primary.right -> hand.primary.right completed

glove.primary.right 是右手手套，hand.primary.right 是输出目标。按 Enter 后，进度条在同一行刷新；动作提示和采样数由算法包决定。非交互环境或手动更新流程需要引导采集时，应使用 external 模式，由应用接管确认等操作。

外部接管引导交互

将 interaction 设为 external 后，GUI、ROS2 等外部程序负责展示提示，并通过手套控制接口调用 confirm()、skip()、retry() 或 cancel()。SDK 仍负责采集与标定；各 target 的状态见 标定。

下例沿用前文的 device_config、target_config；UI 函数和事件循环由应用实现。完整示例见 external_input_run_background.py：

from qnbot_sdk import (
    AlgorithmsConfig,
    CalibrationReadinessState,
    CaptureConfig,
    CaptureInteractionMode,
    Sample,
    Sdk,
)
from qnbot_sdk.glove import (
    CalibrationProgress,
    CaptureProgress,
    CaptureStageState,
)

sdk = Sdk(
    devices=(device_config,),
    targets=(target_config,),
    algorithms=AlgorithmsConfig(
        capture=CaptureConfig(
            interaction=CaptureInteractionMode.EXTERNAL,
        ),
    ),
)
glove = sdk.glove()
device = glove.device()
readiness = device.calibration_status(name="dexhand")
if readiness.state is CalibrationReadinessState.NEEDS_CALIBRATION:
    open_calibration_ui(readiness.operator_id)
capture_progress = device.capture_progress()
calibration_progress = device.calibration_progress(name="dexhand")
control = device.capture_control()

def on_confirm(request_id: str) -> None:
    control.confirm(request_id)

def on_skip(request_id: str) -> None:
    control.skip(request_id)

def on_cancel(request_id: str) -> None:
    control.cancel(request_id)

def render_capture(sample: Sample[CaptureProgress]) -> None:
    value = sample.value
    stage = value.stage
    render_prompt(stage.prompt, stage.collected_sample_count,
                  stage.required_sample_count)
    if stage.request_id is None:
        return
    if stage.state is CaptureStageState.AWAITING_CONFIRMATION:
        bind_waiting_actions(
            request_id=stage.request_id,
            optional=stage.optional,
            on_confirm=on_confirm,
            on_skip=on_skip,
            on_cancel=on_cancel,
        )
    else:
        bind_cancel_action(stage.request_id, on_cancel)

def render_calibration(sample: Sample[CalibrationProgress]) -> None:
    render_calibration_job(sample.value.job)

capture_progress.subscribe(render_capture)
calibration_progress.subscribe(render_calibration)

glove.start()
glove.run_background()
try:
    run_application_event_loop()
finally:
    glove.request_stop()
    glove.join()
    glove.close()

手动发起采集

需要自行决定何时查询和启动采集时，用本节的两阶段流程：先查询准备情况，再在选定的 GloveDevice 上创建采集会话并推进它。启动时机由应用控制，不同于只接管引导交互的 external 模式。

启动参数和强制重采的最小调用见采集配置中的手动流程。

回环表示继续读取同一个会话的状态，不会再次创建 Session；标定接收仍打开的已完成采集会话：

流程图渲染中

下面是示例文件里 capture_side() 的主流程摘录：查询准备情况、创建会话、推进会话、取采集结果。

# 摘自 examples/glove/python/src/host_capture_calibration.py 的 capture_side()。
# 代码逐字保留（`# ...` 标记省略），中文注释为文档补充；导入与常量见示例文件。

def capture_side(
    device: GloveDevice,
    glove: Glove,
    package_ids: tuple[str, ...],
    force: bool,
    start_runtime: bool,
) -> CaptureSession:
    # 查询准备情况：capture_status() 只报告计划，不创建 Session。
    readiness = device.capture_status(
        package_ids=package_ids,
        operator_id=OPERATOR_ID,
    )
    # ... 示例在此打印 readiness.state ...
    for stage in readiness.stages:
        print(f"  {stage.stage_id}: {stage.state.value}")

    # 创建会话：一个 Session 执行一只手，采集与标定必须使用同一个 device。
    session = device.start_capture(
        package_ids=package_ids,
        operator_id=OPERATOR_ID,
        force=force,
    )
    if start_runtime:
        glove.start()
    deadline = time.monotonic() + WAIT_SECONDS
    while time.monotonic() < deadline:
        snapshot = session.snapshot()
        # 完成：取采集结果，并把仍打开的 Session 交给标定（见 calibrate()）。
        if snapshot.state is CaptureSessionState.COMPLETED:
            if session.result() is None:
                raise RuntimeError("completed capture session has no result")
            print(f"{device.source_id}: capture completed")
            return session
        # 失败或取消：报错退出，由调用方关闭已创建的会话。
        if snapshot.state in (
            CaptureSessionState.FAILED,
            CaptureSessionState.CANCELLED,
        ):
            raise RuntimeError(
                snapshot.failure.message
                if snapshot.failure is not None
                else f"{device.source_id} capture stopped"
            )

        current = next(
            (
                stage
                for stage in snapshot.stage_progress
                if stage.stage_id == snapshot.current_stage
            ),
            None,
        )
        # 阶段等待确认：用当前快照的 request_id 确认；可选阶段用 skip()，重采用 retry(stage_id)。
        if (
            current is not None
            and current.state is CaptureStageState.AWAITING_CONFIRMATION
        ):
            if snapshot.request_id is None:
                session.cancel()
                raise RuntimeError("capture confirmation is missing its request token")
            prompt = next(
                (
                    stage.prompt
                    for stage in snapshot.plan.stages
                    if stage.stage_id == snapshot.current_stage
                ),
                snapshot.current_stage or "Continue capture",
            )
            answer = input(f"{device.source_id} {prompt} [Y/n]: ").strip().lower()
            if answer in ("", "confirm", "y", "yes"):
                session.confirm(snapshot.request_id)
            else:
                session.cancel()
                raise RuntimeError("capture cancelled by operator")
            continue

        # 仍在运行：让 SDK 推进任务，然后回到读取快照。
        update = glove.update()
        update.sleep()

    session.cancel()
    raise TimeoutError(f"{device.source_id} capture timed out")

关键点

  • capture_status() 只查询、不创建 Session；状态对应的下一步见判断是否启动手动采集。
  • CaptureSession 一次执行一只手：左右手顺序创建两个 Session，不交替采集，采集与标定共用同一个 GloveDevice。
  • 采集完成后把仍打开的 Session 交给 start_calibration()，无需接触原始帧文件或内部标识，见 手动发起标定。
  • 左右手顺序采集只启动一次运行：左手 start_runtime=True，右手 start_runtime=False；两手都结束后再关闭 Session 与 glove。

参考