Data channels & read methods

The two data families

The SDK publishes two families of data: raw values produced by the device, and values the SDK computes. They are read the same way; only how the data is produced differs:

Raw data is covered in Reading raw data; SDK-processed data (retargeting) in Reading retargeting data.

Data reading methods

All three read methods deliver the same Sample[T]. Callback subscription is the recommended default: each sample is handled as it is produced, with no polling; use latest() when you only need the current value, and asynchronous reading in an asyncio program.

latest() and next() obtain a Sample[T]; subscribe() returns a Subscription, and the registered callback receives Sample[T]; each item of async iteration is also a Sample[T]. Field definitions: Sample[T].

Raw pose, IMU, SDK-processed data and device state all support latest(), subscribe(), next() and async iteration. After creating the SDK, select the device and prepare to read; with callbacks, subscribe before start():

device = glove.device()
device.pose().subscribe(on_pose)
glove.start()

After startup you can still use next(), async for or add a new subscribe(); the scope of the related calls is:

To read SDK-processed data or use automatic reconnection, choose a runtime mode too.

Callback subscription

from qnbot_sdk import Sample
from qnbot_sdk.glove import GlovePose

def on_pose(sample: Sample[GlovePose]) -> None:
    print(sample.sequence, sample.value)

device.pose().subscribe(on_pose)

Call subscribe() before start() to register a callback. The SDK holds registered subscriptions until glove.close() or sdk.close() releases them; save the Subscription returned by subscribe() and call unsubscribe() only to stop one callback early while the SDK keeps running.

The SDK runs the callback when data arrives; keep it short. For long-running work, see Runtime modes; for exception handling, see Common questions.

Getting the latest value

pose = device.pose()
sample = pose.latest()

if sample is None:
    print("No pose received yet")
else:
    print(sample.sequence, sample.value)

latest() does not block. It returns None when there is no data yet, otherwise it returns Sample[T]. It is suitable for reading the current value from other logic during runtime, or for reading the last piece of data after run_forever() returns and before close().

Asynchronous reading

await next() waits for the next piece of data, and async for reads continuously. Raw pose/IMU can be read right after start() succeeds, without choosing a run mode; just keep the asyncio event loop running:

import asyncio

from qnbot_sdk import Sdk
from qnbot_sdk.glove import GloveConfig

async def main() -> None:
    sdk = Sdk(devices=(GloveConfig(),))
    glove = sdk.glove()
    try:
        pose = glove.device().pose()
        glove.start()
        try:
            first = await pose.next()
            print(first.value)

            async for sample in pose:
                print(sample.value)
        finally:
            glove.stop()
    finally:
        await asyncio.to_thread(glove.close)

if __name__ == "__main__":
    asyncio.run(main())

Cancelling the task that awaits next() or iterates asynchronously raises Python's native asyncio.CancelledError.

Asynchronous reading of IMU works the same way as pose.

External input

When there is no physical glove, or when the frames come from your own device or a simulated hand, ExternalConnection and push_frame() feed external hand frames as the input source. External frames join the same data path as a physical glove: raw poses are read directly, and the SDK computes the retargeting result.

Configuration and usage

External input has no device identity handshake, so side must be configured explicitly in GloveConfig:

from qnbot_sdk import Side
from qnbot_sdk.glove import ExternalConnection, GloveConfig

glove_config = GloveConfig(
    side=Side.RIGHT,
    connection=ExternalConnection(),
)

After configuring the target, first subscribe to the output and start the SDK, then keep submitting external frames:

device = glove.device()
output = device.output(name="hand")
output.subscribe(lambda sample: print(sample.value.joints))
device.start()
glove.run_background()

pose = device.push_frame(frame)

push_frame() synchronously returns the GlovePose corresponding to the current frame; the target joint result is read via output(), not the return value of push_frame(). The example uses background mode.

External input uses the background or manual runtime: keep calling push_frame(frame) in background mode; in manual mode, submit each frame before calling update(). If the first run requires calibration, complete Capture and Calibration before expecting target output. See the complete background and manual update examples.

Frame constraints

• The five fingertip poses (thumb, index, middle, ring, pinky) are required, and each is a GloveNodePose.
• position is three finite floating-point numbers, in meters.
• quaternion_xyzw is four finite XYZW components with a non-zero overall norm; the SDK normalizes them before use.
• timestamp_ms may be omitted; when provided it must be a finite number of milliseconds and strictly increasing for the same device; when omitted, the current shared time reference is used. See Timestamps & external time sync for the time unit, default reference, and synchronization flow.
• timestamp_sec is retained only for compatibility with existing code. The SDK multiplies it by 1000 to convert it to milliseconds; do not set it together with timestamp_ms.

The SDK defines the input units and quaternion component order, but the coordinate origin, X/Y/Z axis directions, and relative rotation reference frame are determined by the algorithm package, and must be consistent with the product documentation of the selected algorithm package. side is not passed with the frame; the device identifier and sequence are generated by the SDK from the current configuration. For fields, types and nullability, see ExternalGloveFrame and GloveNodePose.