Reading device state

A one-off snapshot

For a quick look at everything at once, collect_device_snapshot gathers the individual endpoints concurrently and tolerates partial failures:

from busylib.features import collect_device_snapshot

snapshot = await collect_device_snapshot(client)
print(f"Name: {snapshot.name or 'unknown'}")
print(f"Battery: {snapshot.power.battery_charge if snapshot.power else 'unknown'}")
print(f"Wi-Fi: {snapshot.wifi.state if snapshot.wifi else 'unknown'}")
print(f"Field errors: {snapshot.field_errors}")

Example output:

Name: Front desk
Battery: 88
Wi-Fi: connected
Field errors: {}

{} means every snapshot endpoint responded. A non-empty field_errors dictionary identifies only the unavailable fields, while usable values remain available in the same snapshot.

A field that fails lands in field_errors instead of aborting the whole snapshot, so one unavailable endpoint doesn't cost you the rest.

The underlying calls are available individually too: version(), status(), display_brightness(), audio_volume(), wifi_status(), ble_status(), storage_status(), time(), name().

Streaming updates

/api/status/ws pushes protobuf state messages as things change, which is far better than polling:

async for message in client.stream_status_ws():
    if isinstance(message, dict):
        print(message)

Example output after a power update:

{'updates': [{'power': {'known': {'battery_charge_percent': 88}}}]}

The exact payload changes with the device event. Each member of updates is a delta, not a full device snapshot, so retain earlier state for fields absent from that message.

Messages are decoded from the BSB_State.State schema into dictionaries with the original proto field names. Each one carries a list of updates, where every entry is one changed thing: device_name, power, brightness, audio_volume, wifi, timezone, ble, update_state, input, timer, or frame.

Keeping a snapshot in sync

DeviceStateStore applies those deltas onto a snapshot for you and notifies subscribers:

from busylib.features import DeviceStateStore, collect_device_snapshot

store = DeviceStateStore(await collect_device_snapshot(client))
store.on_state(lambda snapshot: print("now:", snapshot.name))
store.on_diff(lambda changed, snapshot: print("changed:", changed))

async for message in client.stream_status_ws():
    if isinstance(message, dict):
        store.apply_stream_message(message)

Example output after a battery update:

now: Front desk
changed: {'power'}

The first callback receives the complete merged snapshot; the second names the top-level fields changed by this particular stream message.

on_state receives the full updated snapshot; on_diff also receives the set of top-level fields that actually changed, which is handy for redrawing only what moved.

Screen frames arrive here too

Current firmware has no separate screen WebSocket. Instead, front-display frames come through this same stream as frame updates, carrying their own width, height, encoding, and pixel_format. The store decodes them into Frame objects on DeviceSnapshot.screen_front and screen_back:

def on_diff(changed, snapshot):
    if "screen_front" in changed and snapshot.screen_front is not None:
        render(snapshot.screen_front)


store.on_diff(on_diff)

Expected result: render receives a decoded Frame whenever the front screen changes, carrying RGB bytes with their geometry. This fragment intentionally has no terminal output; the observable result is the updated image in your renderer.

Because the frame describes its own encoding, no guessing by payload size is involved — see busylib.display.decode_frame_data. A decoded frame whose size doesn't match the target display is logged and dropped rather than stored, so a malformed update can't reach a renderer.

What an update leaves out

proto3 omits any field holding its type's default, so an update the bar did send can be missing the very thing it is about. Two of those bite:

• a number that is absent means zero - a muted volume, an unplugged USB, a panel at its dimmest;
• battery_status absent means DISCHARGING, because that is the first value of the firmware's enum. A bar simply running on its battery reports the most common state of all as nothing at all.

apply_state_stream_update reads both the way the device means them. What it cannot do is tell you whether a cable is plugged in: the firmware knows (vbus_present internally) but does not put it on the wire, and usb_voltage reads non-zero on a bar that is discharging with nothing attached.

Buttons, the selector and the wheel

Physical input arrives on the same stream, as input updates. input_events() turns one state message into the events it carried, so a consumer reacts to a press rather than to a nested dictionary of the firmware's enum names:

from busylib.features import input_events
from busylib.features.input_events import ButtonEvent, EncoderEvent, SelectorEvent

async for message in bb.stream_status_ws():
    for event in input_events(message):
        match event:
            case ButtonEvent(button="ok") if event.is_press:
                confirm()
            case SelectorEvent(position=position):
                print(f"selector moved to {position}")
            case EncoderEvent(delta=delta):
                scroll(delta)

Both halves of a press are reported, so a long press is distinguishable from a short one; is_press is there because most callers only want one of them.

Two things worth knowing. The bar reports the selector only when it moves - nothing answers "where is it now" - so its position is unknown until the first move after you start listening. And the same events come back when input is sent with input(), because the firmware makes no distinction between a button pressed by a finger and one pressed over HTTP:

await bb.input("ok")  # a key by name, or types.InputKey.OK