快速开始 | Qnbot Docs Center

这一章只放能直接跑的代码。每个场景对应一个独立示例文件，都在 examples/glove/python/src/，互不依赖，可以只跑你需要的那个。全部示例（含每个文件对应的主要 API）见示例索引。

开始前先确认已经按安装与环境要求装好并验证过 SDK，并且手套已经连上。所有示例都用 Ctrl+C 结束。

建议从这里开始：首次接入先运行 quick_start.py，确认能持续收到姿态；需要 SDK 处理后的骨骼数据时运行 skeleton.py；已安装算法包并需要灵巧手输出时运行 foreground_runtime.py。每个示例的命令和预期输出都在下文。

读取原始姿态数据

姿态由设备直接产出，不经 SDK 计算，也不需要算法包：

文件：quick_start.py

零参数配置，自动发现唯一连接的受支持手套，用 callback 逐帧接收姿态。这是最短的一条路径：

from __future__ import annotations

from qnbot_sdk import Sample, Sdk
from qnbot_sdk.glove import GloveConfig, GlovePose


def print_pose(sample: Sample[GlovePose]) -> None:
    print(
        f"pose sequence={sample.sequence} "
        f"fingertips={sample.value.payload.fingertip_local}"
    )


def main() -> None:
    sdk = Sdk(devices=(GloveConfig(),))
    glove = sdk.glove()
    pose = glove.device().pose()
    pose.subscribe(print_pose)

    glove.start()
    print("running; press Ctrl+C to stop", flush=True)
    glove.run_forever()


if __name__ == "__main__":
    main()

运行：

python examples/glove/python/src/quick_start.py

预期输出（SDK 启动时会先打印 sdk_initialized 与 device_connected 两条连接日志，这里是示例自己的输出；姿态值已省略）：

running; press Ctrl+C to stop
pose sequence=1 fingertips={<GloveFinger.INDEX: 'index'>: GloveNodePose(position=(…), quaternion_xyzw=(…)), …}
pose sequence=2 fingertips={…}

进程不退出，持续打印。如果发现多只手套，Sdk(...) 会直接报错，改用 port 或 serial_number 选择，见设备发现与连接。

读取 Skeleton

文件：skeleton.py

在读取姿态的基础上增加 SDK 处理后的数据：订阅 OpenXR 骨骼，并分别读取关节角与骨骼节点姿态。 Skeleton 由 SDK 提供，不需要算法包；首次运行需要完成一次采集标定，之后直接复用。定义见读取 Retargeting 数据。

持续读取时按运行方式推进 SDK 调度。

from __future__ import annotations

import argparse

from qnbot_sdk import Sample, Sdk, SerialConnection, Side
from qnbot_sdk.glove import GloveConfig, HandJointCommand, HandSkeletonPose


def print_joint_angles(sample: Sample[HandJointCommand]) -> None:
    print(
        f"skeleton sequence={sample.sequence} "
        f"target={sample.value.target} joints={sample.value.joints}"
    )


def print_pose(sample: Sample[HandSkeletonPose]) -> None:
    wrist = sample.value.positions_local[0]
    index_tip = sample.value.positions_local[9]
    print(
        f"pose sequence={sample.sequence} frame={sample.value.coordinate_frame} "
        f"wrist={wrist} index_tip={index_tip}"
    )


def main() -> None:
    arguments = argparse.ArgumentParser(description="Read SDK-owned skeleton output")
    arguments.add_argument("--port", required=True, help="Serial port for the glove")
    arguments.add_argument(
        "--side",
        choices=(Side.LEFT.value, Side.RIGHT.value),
        required=True,
        help="Physical glove side",
    )
    options = arguments.parse_args()
    sdk = Sdk(
        devices=(
            GloveConfig(
                side=Side(options.side),
                connection=SerialConnection(port=options.port),
            ),
        )
    )
    glove = sdk.glove()
    skeleton = glove.device().skeleton()
    skeleton.joint_angles().subscribe(print_joint_angles)
    skeleton.pose().subscribe(print_pose)

    glove.start()
    print("running; press Ctrl+C to stop", flush=True)
    glove.run_forever()


if __name__ == "__main__":
    main()

运行：

python examples/glove/python/src/skeleton.py --port <串口> --side right

预期输出（关节值与坐标已省略）：

running; press Ctrl+C to stop
skeleton sequence=1 target=hand.skeleton.right joints={…}
pose sequence=1 frame=wrist_local wrist=(0.0, 0.0, 0.0) index_tip=(…)

两条线各印一行：skeleton 是关节角，pose 是骨骼节点姿态。

读取灵巧手输出

三种映射的区别只在于手套与灵巧手的对应关系：

场景	映射	示例文件
一对一	一只手套对应一个具名 target	`foreground_runtime.py`
一对多	一只手套驱动多个灵巧手	`multiple_outputs.py`
多对多	多只手套驱动多个灵巧手	`device_lifecycle.py`

三种映射都需要先安装算法包，见读取 Retargeting 数据——这些值由算法包按 target 做 retargeting 计算得出。首次运行还需要完成一次采集标定，SDK 会按算法包的提示引导，完成后会被后续运行复用。

一对一：一只手套对应一个具名 target

在当前线程持续运行，并用 callback 接收目标输出。

from __future__ import annotations

import argparse

from qnbot_sdk import (
    DeviceSelector,
    Health,
    Sample,
    Sdk,
    SerialConnection,
    Side,
    TargetAlgorithm,
    TargetConfig,
)
from qnbot_sdk.glove import GloveConfig, HandJointCommand

DEFAULT_TARGET_NAME = "openxr_hand"


def create_sdk(port: str, side: Side, target_name: str, package_id: str) -> Sdk:
    source = DeviceSelector(type="glove", name="primary")
    return Sdk(
        devices=(
            GloveConfig(
                name="primary",
                side=side,
                connection=SerialConnection(port=port),
            ),
        ),
        targets=(
            TargetConfig(
                name=target_name,
                side=side,
                source=source,
                algorithms=(TargetAlgorithm(id=package_id),),
            ),
        ),
    )


def print_output(sample: Sample[HandJointCommand]) -> None:
    print(
        f"retargeting sequence={sample.sequence} "
        f"target={sample.value.target} joints={sample.value.joints}"
    )


def print_health(label: str, health: Health) -> None:
    devices = ", ".join(
        f"{source_id}(connected={device.connected}, stale={device.stale})"
        for source_id, device in health.devices.items()
    )
    print(
        f"{label} health ok={health.ok} warnings={health.warning_count} "
        f"errors={health.error_count} devices=[{devices}]"
    )


def main() -> None:
    arguments = argparse.ArgumentParser(
        description="Run the Glove domain on the current thread"
    )
    arguments.add_argument("--port", required=True, help="Serial port for the glove")
    arguments.add_argument(
        "--side",
        choices=(Side.LEFT.value, Side.RIGHT.value),
        required=True,
        help="Physical glove side",
    )
    arguments.add_argument("--target-name", default=DEFAULT_TARGET_NAME)
    arguments.add_argument("--package-id", required=True)
    options = arguments.parse_args()

    sdk = create_sdk(
        options.port,
        Side(options.side),
        options.target_name,
        options.package_id,
    )
    glove = sdk.glove()
    output = glove.device().output(name=options.target_name)
    output.subscribe(print_output)
    glove.start()
    print_health("running", glove.health())

    print("running; press Ctrl+C to stop", flush=True)
    glove.run_forever()


if __name__ == "__main__":
    main()

运行：

python examples/glove/python/src/foreground_runtime.py \
  --port <串口> --side right --package-id <package-id>

预期输出（关节值已省略）：

running health ok=True warnings=0 errors=0 devices=[glove.<SN>.right(connected=True, stale=False)]
running; press Ctrl+C to stop
retargeting sequence=1 target=hand.openxr_hand.right joints={…}

一直收不到输出时，确认算法包 ID、target 配置和标定状态，见连接成功但收不到数据。

一对多：一只手套驱动多个灵巧手

为同一个 source 配置多个 target，再按 name 分别读取输出。手套连接和数据流仍然只有一份。

from __future__ import annotations

import argparse
from collections.abc import Callable

from qnbot_sdk import (
    DeviceSelector,
    Sample,
    Sdk,
    SerialConnection,
    Side,
    TargetAlgorithm,
    TargetConfig,
)
from qnbot_sdk.glove import GloveConfig, HandJointCommand


def create_sdk(
    port: str,
    side: Side,
    package_id: str,
) -> Sdk:
    source = DeviceSelector(type="glove", name="primary")
    return Sdk(
        devices=(
            GloveConfig(
                name="primary",
                side=side,
                connection=SerialConnection(port=port),
            ),
        ),
        targets=(
            TargetConfig(
                name="primary",
                side=side,
                source=source,
                algorithms=(TargetAlgorithm(id=package_id),),
            ),
            TargetConfig(
                name="backup",
                side=side,
                source=source,
                algorithms=(TargetAlgorithm(id=package_id),),
            ),
        ),
    )


def print_output(name: str) -> Callable[[Sample[HandJointCommand]], None]:
    def show(sample: Sample[HandJointCommand]) -> None:
        print(
            f"callback {name} sequence={sample.sequence} "
            f"target={sample.value.target} joints={sample.value.joints}"
        )

    return show


def main() -> None:
    arguments = argparse.ArgumentParser(
        description="Read multiple hand outputs driven by one glove"
    )
    arguments.add_argument("--port", required=True, help="Serial port for the glove")
    arguments.add_argument(
        "--side",
        choices=(Side.LEFT.value, Side.RIGHT.value),
        required=True,
        help="Physical glove side",
    )
    arguments.add_argument(
        "--package-id",
        required=True,
        help="Installed algorithm package ID used by both output instances",
    )
    options = arguments.parse_args()

    sdk = create_sdk(
        options.port,
        Side(options.side),
        options.package_id,
    )
    glove = sdk.glove()
    device = glove.device()
    primary = device.output(name="primary")
    backup = device.output(name="backup")
    primary.subscribe(print_output("primary"))
    backup.subscribe(print_output("backup"))

    glove.start()
    print("running; press Ctrl+C to stop", flush=True)
    glove.run_forever()


if __name__ == "__main__":
    main()

运行：

python examples/glove/python/src/multiple_outputs.py \
  --port <串口> --side right --package-id <package-id>

预期输出（每个 target 的输出带自己的名称，关节值已省略）：

running; press Ctrl+C to stop
callback primary sequence=1 target=hand.primary.right joints={…}
callback primary sequence=2 target=hand.primary.right joints={…}
…
callback backup sequence=1 target=hand.backup.right joints={…}

完整的 target 配置见读取 Retargeting 数据。

多对多：多只手套驱动多个灵巧手

两只手套各自配置 target，选择各自的设备后读取自己的输出。脚本会先等待左右手输出，然后停止右手设备确认左手继续推进，再重启右手。

from __future__ import annotations

import argparse
import time

from qnbot_sdk import (
    AlgorithmsConfig,
    CaptureConfig,
    CaptureControl,
    CaptureInteractionMode,
    DeviceSelector,
    ReadChannel,
    Sample,
    Sdk,
    SerialConnection,
    Side,
    TargetAlgorithm,
    TargetConfig,
)
from qnbot_sdk.glove import (
    CalibrationJobState,
    CalibrationProgress,
    CaptureProgress,
    CaptureSessionState,
    CaptureStageState,
    GloveConfig,
    HandJointCommand,
)

DEFAULT_TARGET_NAME = "openxr_hand"


def create_sdk(
    left_port: str,
    right_port: str,
    target_name: str,
    package_id: str,
) -> Sdk:
    return Sdk(
        devices=(
            GloveConfig(
                name="left",
                side=Side.LEFT,
                connection=SerialConnection(port=left_port),
            ),
            GloveConfig(
                name="right",
                side=Side.RIGHT,
                connection=SerialConnection(port=right_port),
            ),
        ),
        targets=(
            TargetConfig(
                name=target_name,
                side=Side.LEFT,
                source=DeviceSelector(type="glove", name="left"),
                algorithms=(TargetAlgorithm(id=package_id),),
            ),
            TargetConfig(
                name=target_name,
                side=Side.RIGHT,
                source=DeviceSelector(type="glove", name="right"),
                algorithms=(TargetAlgorithm(id=package_id),),
            ),
        ),
        algorithms=AlgorithmsConfig(
            capture=CaptureConfig(
                interaction=CaptureInteractionMode.EXTERNAL,
            ),
        ),
    )


def print_output(device_name: str, sample: Sample[HandJointCommand]) -> None:
    print(
        f"{device_name} retargeting sequence={sample.sequence} "
        f"target={sample.value.target} joints={sample.value.joints}"
    )


def handle_capture_prompt(
    capture_progress: ReadChannel[CaptureProgress],
    capture_control: CaptureControl,
    calibration_progress: ReadChannel[CalibrationProgress],
    handled_request_ids: set[str],
) -> None:
    capture = capture_progress.latest()
    if capture is not None:
        value = capture.value
        if value.session_state is CaptureSessionState.FAILED:
            message = value.failure.message if value.failure else "unknown error"
            raise RuntimeError(f"capture failed: {message}")
        stage = value.stage
        if (
            stage.state is CaptureStageState.AWAITING_CONFIRMATION
            and stage.request_id is not None
            and stage.request_id not in handled_request_ids
        ):
            answer = (
                input(f"{stage.prompt or 'Continue capture'} [Y/n]: ").strip().lower()
            )
            if answer in ("", "y", "yes"):
                capture_control.confirm(stage.request_id)
            else:
                capture_control.cancel(stage.request_id)
                raise RuntimeError("capture was cancelled")
            handled_request_ids.add(stage.request_id)
    calibration = calibration_progress.latest()
    if (
        calibration is not None
        and calibration.value.job.state is CalibrationJobState.FAILED
    ):
        failure = calibration.value.job.failure
        message = failure.message if failure else "unknown error"
        raise RuntimeError(f"calibration failed: {message}")


def main() -> None:
    arguments = argparse.ArgumentParser(
        description="Control one physical glove in a multi-glove domain"
    )
    arguments.add_argument("--left-port", required=True)
    arguments.add_argument("--right-port", required=True)
    arguments.add_argument("--updates", type=int, default=10)
    arguments.add_argument("--target-name", default=DEFAULT_TARGET_NAME)
    arguments.add_argument("--package-id", required=True)
    options = arguments.parse_args()
    if options.updates < 1:
        arguments.error("--updates must be at least 1")

    sdk = create_sdk(
        options.left_port,
        options.right_port,
        options.target_name,
        options.package_id,
    )
    glove = sdk.glove()

    left = glove.device(name="left")
    right = glove.device(side=Side.RIGHT)
    left_output = left.output(name=options.target_name)
    right_output = right.output(name=options.target_name)
    left_capture_progress = left.capture_progress()
    right_capture_progress = right.capture_progress()
    left_calibration_progress = left.calibration_progress(name=options.target_name)
    right_calibration_progress = right.calibration_progress(name=options.target_name)
    left_capture_control = left.capture_control()
    right_capture_control = right.capture_control()
    left.start()
    right.start()

    def wait_for_output(
        output: ReadChannel[HandJointCommand],
        capture_progress: ReadChannel[CaptureProgress],
        capture_control: CaptureControl,
        calibration_progress: ReadChannel[CalibrationProgress],
    ) -> None:
        handled_request_ids: set[str] = set()
        deadline = time.monotonic() + 300.0
        while output.latest() is None and time.monotonic() < deadline:
            update = glove.update()
            handle_capture_prompt(
                capture_progress,
                capture_control,
                calibration_progress,
                handled_request_ids,
            )
            update.sleep()
        if output.latest() is None:
            raise RuntimeError("retargeting output was not ready before the timeout")

    wait_for_output(
        left_output,
        left_capture_progress,
        left_capture_control,
        left_calibration_progress,
    )
    wait_for_output(
        right_output,
        right_capture_progress,
        right_capture_control,
        right_calibration_progress,
    )

    for _ in range(options.updates):
        update = glove.update()
        update.sleep()
        left_sample = left_output.latest()
        if left_sample is not None:
            print_output("left", left_sample)
        right_sample = right_output.latest()
        if right_sample is not None:
            print_output("right", right_sample)

    left_before_stop = left_output.latest()
    if left_before_stop is None:
        raise RuntimeError("both gloves must produce output before lifecycle control")

    right.stop()
    print("right device stopped; left remains active")
    right_stopped_baseline = right_output.latest()
    if right_stopped_baseline is None:
        raise RuntimeError("right output was unavailable after stop")
    left_sample = None
    for _ in range(options.updates):
        update = glove.update()
        update.sleep()
        right_candidate = right_output.latest()
        if (
            right_candidate is not None
            and right_candidate.sequence > right_stopped_baseline.sequence
        ):
            raise RuntimeError("right output advanced while the device was stopped")
        candidate = left_output.latest()
        if candidate is not None and candidate.sequence > left_before_stop.sequence:
            left_sample = candidate
            break
    if left_sample is None:
        raise RuntimeError("left output did not advance while right was stopped")
    print("left output while right stopped")
    print_output("left", left_sample)

    right_before_restart = right_output.latest()
    if right_before_restart is None:
        raise RuntimeError("right output was unavailable before restart")
    right.start()
    print("right device restarted")
    right_sample = None
    for _ in range(options.updates):
        update = glove.update()
        update.sleep()
        candidate = right_output.latest()
        if candidate is not None and candidate.sequence > right_before_restart.sequence:
            right_sample = candidate
            break
    if right_sample is None:
        raise RuntimeError("right output did not advance after restart")
    print("right output after restart")
    print_output("right", right_sample)

    health = glove.health()
    print(
        f"health ok={health.ok} warnings={health.warning_count} "
        f"errors={health.error_count}"
    )
    glove.close()


if __name__ == "__main__":
    main()

运行（需要同时接上左右手两只手套）：

python examples/glove/python/src/device_lifecycle.py \
  --left-port <左手串口> --right-port <右手串口> --package-id <package-id>

预期输出：终端分别打印左右手输出，并打印 right device stopped; left remains active、 left output while right stopped、right device restarted、right output after restart 等步骤提示，脚本完成检查后退出。完整映射配置见读取 Retargeting 数据。