Subscribing Asynchronously

SDK Practice

Overview

Callback subscriptions push frames to you. Async subscriptions let you pull them. subscribe() returns a subscription handle, await sub.recv_async() takes one frame at a time, and six streams each run their own coroutine driven concurrently by asyncio.gather.

Compared with callbacks, async gives you control over pacing: the code reads sequentially and can await other I/O directly. The cost is that the subscription queue fills up while you’re not reading, and silently drops frames once it’s full. This page measures exactly where that boundary sits.

Matching example: examples/python/wuji_glove/1.subscribe_async.py.

For API and field definitions, see Hand Tracking, Tactile, and Coordinate Transforms.

Prerequisites

Hardware

This example only reads. It writes no parameters.

Software

pip install wuji-sdk

Before You Start

Read the callback page first. Frame structures for all six streams, the -1.0 invalid-channel marker, and how the numbers 744, 572, and 526 relate all live there, and this page doesn’t repeat them.

Data Flow

The queues live inside the SDK, filled by firmware pushes and decoupled from your consuming coroutines. That decoupling is where the frame-loss risk comes from.

Queue Measurements

The Queue Holds 1024 Frames

The queue fills from the moment a subscription opens, even if you never read. Measured on the tactile stream at 132 Hz:

In the first two rows, the lag equals the production count, so nothing was lost. The third row stops at 1023, meaning everything beyond that was already discarded. The queue holds 1024 frames, matching the SDK’s internal broadcast channel setting.

Drops Take the Oldest Frames, Never a Sample Here and There

Across all three runs, reading 400 consecutive frames after the first one gives a header.seq step set of exactly [1]. On overflow the queue discards the oldest batch at its head. What remains stays contiguous, with no holes punched in the middle.

Drops Raise No Exception, Only a WARN Log

After a 12-second backlog, reading resumes and the SDK prints:

[<timestamp> WARN  wuji_sdk] Subscription for 'tactile' lagged 549 messages

recv_async() returns the next frame normally, raises nothing, and carries no drop marker in its return value. Your application code has no visibility into those 549 lost frames. The log is the only signal.

For work that depends on data completeness, set the SDK log level to WARN or higher and route it into your own logging system rather than letting it scroll past in a terminal.

Fast Consumers Lose Nothing

Keep up and you lose nothing. Reading 200 frames with consecutive await recv_async() calls takes 1.52 seconds, an effective 131 Hz that matches the stream’s own rate, with every seq step equal to 1.

Integration Steps

See Python example for the complete example. These excerpts show the integration order. See the example for argument parsing, signal handling, and full error handling.

1. Connect to the Target Glove

Replace serial_number with the target glove’s serial number. Reuse the connected device handle in later steps.

from wuji_sdk import SdkManager
manager = SdkManager.instance()
glove = manager.connect(sn=serial_number, device_name="glove_0")

2. Create a Subscription and Read Frames in a Coroutine

Create the subscription before the consumption loop. Each recv_async() waits for the next frame. Close the subscription in finally when the task ends or is cancelled.

async def consume_tactile(glove):
    sub = glove.tactile().subscribe()
    try:
        while True:
            frame = await sub.recv_async()
            valid = [v for v in frame.data if v >= 0.0]
            print(frame.header.seq, max(valid, default=0.0))
    finally:
        sub.close()

3. Start the Consumer and Clean Up the Connection

For one stream, await its consumer coroutine directly. For concurrent streams, create a coroutine for each stream and pass them to asyncio.gather(), as shown in Python example.

import asyncio
async def main():
    try:
        await consume_tactile(glove)
    finally:
        manager.disconnect_all()

asyncio.run(main())

Running It

python examples/python/wuji_glove/1.subscribe_async.py --sn WG1KA00XXXXXXXXX

Found 1 device(s)
  SN=WG1KA00XXXXXXXXX, Type=DeviceType.WujiGlove, Address=192.168.1.101:50001
Connected: WG1KA00XXXXXXXXX (FW=0.11.2, right)
Subscribed to 6 streams. Ctrl+C to stop.

[glove_0][EmfPoses] thumb=[+0.067, +0.000, -0.059]
[glove_0][Tactile] max=0.3
[glove_0][JointAngles] conf=['0.47', '0.41', '0.74', '0.71', '0.47']
[glove_0][Skeleton] wrist=[+0.000, +0.000, +0.000] joints=21
[glove_0][Zones] palm=-0.1 thumb=-0.1
[glove_0][PointCloud] points=526 frame=r_wrist

This sample output shows a right-hand Wuji Glove resting on a desk, unworn, using its default IP address, 192.168.1.101. Use the address reported by the scan.

Five seconds produced 3031 lines, with per-stream frame counts of 506, 506, 505, 502, 502, and 501. No two streams differ by more than 5 frames, so the event loop favors none of them and gather schedules fairly.

The palm=-0.1 on the Zones line comes from the example’s avg() including -1.0 invalid channels in its mean, not from the glove sensing a reversed force. The callback page explains the cause and the fix.

Notes

The example’s while True loop has no exit condition and stops on KeyboardInterrupt. Its finally block only calls manager.disconnect_all(), leaving subscription handles to be reclaimed at process exit. A long-running service needs its own sub.close().

Adapting It

Callback or Async

Queue semantics are identical, and the 1024-frame boundary applies to callbacks too. Callbacks keep your event loop free. Async composes naturally with coroutines such as HTTP requests and database writes.

Don’t Block Inside a Coroutine

The event loop is single-threaded. Calling time.sleep(), writing files synchronously, or making blocking HTTP requests inside a coroutine backs up all six streams at once. Hand blocking work to a thread pool:

loop = asyncio.get_running_loop()
await loop.run_in_executor(None, write_frame_to_disk, frame)

Subscribe Only to What You Need

All six streams open means about 640 recv_async() wakeups per second at rest. For tactile analysis, keep tactile or tactile_point_cloud alone and event-loop pressure drops immediately.

Lower the Stream Rate

130 Hz is higher than most applications need. Subscription handles support rate control, described in Adjusting Stream Rate. At a lower rate, those 1024 queued frames cover a longer window, which also buys more tolerance for jitter.

Related

• Subscribing With Callbacks: frame structures for all six streams and invalid-channel handling
• Recording Data: write these streams to a replayable file
• Adjusting Stream Rate: bring 130 Hz down to what your application needs
• Data Subscription: subscription model and lifecycle
• Common Data Types: FrameHeader, TactileFrame, HandSkeleton, and related structures