Features¶
Higher-level helpers built on top of the raw client: notification layout for the front panel, timer state and control, a device snapshot for dashboards, a store that applies streamed protobuf state updates, and asset syncing.
For notifications, start with the Sending notifications guide - it covers the built-in templates and how to write your own. For timers, see Working with timers, which explains why a snapshot has to be advanced before it means anything.
module-attribute
¶
BUILT_IN_TEMPLATES: tuple[Template, ...] = (TWO_LINES, ONE_LINE)
Bases: BaseModel
class-attribute
instance-attribute
¶
update_available_version: str | None = None
class-attribute
instance-attribute
¶
field_errors: dict[str, str] = Field(default_factory=dict)
class-attribute
instance-attribute
¶
raw_time: object | None = Field(default=None, exclude=True)
class-attribute
instance-attribute
¶
screen_front: Frame | None = Field(default=None, exclude=True)
class-attribute
instance-attribute
¶
screen_back: Frame | None = Field(default=None, exclude=True)
DeviceStateStore(initial: DeviceSnapshot)
Stateful store for snapshot updates driven by status websocket stream.
The store applies protobuf state deltas, updates current snapshot, and
notifies subscribers through two callback channels:
- on_state: receives full updated snapshot;
- on_diff: receives changed top-level fields and full snapshot.
Create a state store with initial snapshot value.
on_state(callback: StateCallback) -> Callable[[], None]
Subscribe to full-state updates and return unsubscriber.
on_diff(callback: StateDiffCallback) -> Callable[[], None]
Subscribe to changed-fields updates and return unsubscriber.
apply_stream_message(state_message: dict[str, object]) -> DeviceSnapshot
Apply one stream message and emit callbacks when state changes.
dataclass
¶
ButtonEvent(button: Button, action: ButtonAction)
dataclass
¶
EncoderEvent(delta: int)
The wheel was turned, by delta steps: positive one way, negative the
other.
dataclass
¶
SelectorEvent(position: SelectorPosition)
The selector was moved to a position.
off is the bar's do-not-disturb rather than a power state.
Bases: Protocol
What these helpers need from a client.
Deliberately three methods rather than the whole client, so a caller can pass anything that speaks them - including a test double.
async
¶
busy_profile_set(slot: BusyProfileSlot, profile: BusyProfile) -> types.SuccessResponse
Bases: BusyBarError
Raised for a change that only makes sense while a session is running.
Pausing, resuming, moving to the next phase and setting the session's theme all rewrite the running session. With nothing running there is nothing to rewrite, and writing a session to make the change possible would start one nobody asked for.
PhaseTooShortError(field: str, given_ms: int, *, minimum_ms: int = MINIMUM_PHASE_MS, maximum_ms: int = MAXIMUM_PHASE_MS)
Bases: BusyBarError
Raised for a duration the device would refuse, at either end.
It never says so usefully. A card written with a two-minute work phase comes back with the phase it had before and every layer reports success; a session with one is refused as an unparseable snapshot. So a caller that is not told here finds out by watching a bar run the wrong timer.
The name is the common case - too short - but the same class carries the ceiling, since to a caller they are one mistake: a length the bar will not run.
dataclass
¶
TimerState(mode: TimerMode, is_paused: bool, phase: TimerPhase | None = None, interval: int | None = None, time_left_ms: int | None = None, is_finished: bool | None = None, elapsed_ms: int = 0)
What the timer is doing at one moment.
phase is None only once a session has finished, where no phase
applies any more.
UnknownThemeError(theme: str, available: Sequence[str])
Bases: BusyBarError
Raised for a theme this bar does not have.
The device does not raise it: a profile naming a theme that does not exist is stored and read back happily, and only the bar's own screen shows that something is wrong. So the check has to happen here, or a typo leaves a card pointing at nothing with every layer reporting success.
dataclass
¶
Asset(name: str, reference: str, kind: AssetKind, application: str | None = None)
One file on a bar, and how to name it in a call.
reference is what you pass, and which field it goes in depends on
where the file came from. A shipped asset is named by stock_path
and carries its folder: shared/images/clock_5x5.image. An upload is
named by path and is just the file name, because the device
resolves it inside the uploading application's own folder - so the
same name in two applications is two different files, and
application says which one this is.
dataclass
¶
NotificationSpec(line_1: str, line_2: str | None = None, icon: StockIcon | None = None, line_1_font: DisplayFontName = DEFAULT_FONT, line_2_font: DisplayFontName | None = None, line_1_color: ColorInput | None = None, line_2_color: ColorInput | None = None, background_color: ColorInput | None = None, duration: int | None = None, display: DisplaySpec = FRONT_DISPLAY)
Everything a template needs, already resolved and validated.
A template receives one of these rather than raw keyword arguments, so a custom template gets the icon already looked up, the geometry of the panel it is drawing on, and helpers for the parts that are easy to get wrong - text placement past an icon, and scrolling long lines.
property
¶
text_x: int
Where text starts: past the icon, or at the plain left margin.
The shipped icons are 5, 8 and 11px wide, so this is computed from the icon rather than fixed - a constant offset would overlap the narrow ones and leave a gap after the wide ones.
text(element_id: str, text: str, *, y: int, align: str, color: ColorInput | None = None, x: int | None = None, width: int | None = None, font: DisplayFontName | None = None) -> types.TextElement
Build a text element at y, scrolled if it cannot fit.
x defaults to text_x, which is where a left-anchored line starts
once the icon has had its share. A template anchoring to the right or
the centre has to say where instead: with mid_right and the default
x the line's right edge lands at the left margin, and the text runs
off the panel.
width is how much room the line has, and it is what scrolling is
decided against. It defaults to everything right of x, which is
only the room available when the line is anchored left - for any
other anchor, pass it. Getting this wrong does not misplace the
text, it scrolls it through a window a few pixels wide.
icon_element(element_id: str = _ICON_ID) -> types.ImageElement | None
Build the icon element, or None when there is no icon.
Bases: Protocol
A way of arranging a notification on the panel.
The built-in templates are ordinary implementations of this, so writing
one is the same work: declare which fonts it can place, say when it
applies, and return the elements. Everything version-dependent stays
outside - the background fill is added by build_notification, which
knows what the firmware can draw - so a template is only about layout.
property
¶
fonts: tuple[DisplayFontName, ...]
The fonts this template can place without clipping.
matches(spec: NotificationSpec) -> bool
Whether this template suits the notification, for auto-selection.
render(spec: NotificationSpec) -> list[types.DisplayElement]
Lay the notification out as display elements.
async
¶
sync_app_assets(client: AssetsClient, application_name: str, assets_dir: str | Path) -> list[str]
Sync local assets with device storage for the given application name. Uploads files that are missing or have different sizes.
apply_state_stream_update(snapshot: DeviceSnapshot, state_message: dict[str, object]) -> DeviceSnapshot
Apply one protobuf /api/status/ws state message to an existing snapshot.
The function updates only fields present in state updates and preserves all previously known values for absent fields.
async
¶
collect_device_snapshot(client: AsyncBusyBar) -> DeviceSnapshot
Collect a best-effort device snapshot with per-field diagnostics.
Each field request runs independently. Failed fields are set to None and
reported in field_errors without aborting the whole snapshot.
kind_of(settings: BusyTimerSettings) -> TimerKind
Which kind of timer a card holds, in this package's words.
kind_of_snapshot(snapshot: BusySnapshotVariant) -> TimerKind | None
Which kind of timer is running, or None if none is.
A snapshot has a state the three kinds do not - nothing started - so
this answers with None rather than pretending the bar is idling in
one of them.
phase_of(interval: int) -> TimerPhase
Name the phase an interval index belongs to.
current_interval is the firmware's own interval index, sent raw: it
starts at 0 for the first work period and rises by one at every change
of phase, so even indices are work and odd ones rest. Comparing the
interval's length against the configured durations would usually work
too, but not when work and rest are set equal - the index has no such
gap.
timer_state(snapshot: BusySnapshot, *, now_ms: int | None = None) -> TimerState
Advance a snapshot to now_ms and report the timer's state.
now_ms defaults to the current wall clock. A paused snapshot is
returned as it stands, since paused time does not pass.
Everything is derived from the interval index rather than from interval lengths, so a session configured with equal work and rest is no harder to read than any other: the index says which phase it is, and where the session ends.
async
¶
copy_to_application(client: AssetCopyClient, asset: Asset, application_name: str) -> Asset
Put an upload where another application can name it.
The device resolves an asset by name inside the folder of whichever application is drawing, so a picture uploaded by the BUSY app or the Draw Tool is one nobody else can draw. Copying it across is the way to use it - byte for byte, since it was converted for the device when it was first uploaded and converting it again would be a second guess at the same picture.
Answers with the asset as the copy: same name, same kind, the new application.
async
¶
discover_assets(client: AssetCatalogueClient) -> list[Asset]
Every asset this bar holds: shipped, and uploaded by applications.
Ordered by kind and then by name, so a list offered to a person is stable between calls and between bars.
This is a listing, not a read: nothing here opens a file, so an
image's width - which a layout needs and a file name does not always
carry - is read later, by notification.icon_at, for the one asset
that ends up being used.
build_notification(line_1: str, *, line_2: str | None = None, icon: str | StockIcon | None = None, line_1_font: DisplayFontName = DEFAULT_FONT, line_2_font: DisplayFontName | None = None, line_1_color: ColorInput | None = None, line_2_color: ColorInput | None = None, background_color: ColorInput | None = None, duration: int | None = None, priority: int = PRIORITY_DEFAULT, application_name: str = 'busylib', device_api_version: str | None = None, template: Template | None = None, display: DisplaySpec = FRONT_DISPLAY) -> types.DisplayElements
Lay out a notification, with a built-in template or one of your own.
Without template, the built-in that suits the arguments is chosen -
two lines if a second was given, otherwise one. Pass template to use
your own arrangement; it is handed the same NotificationSpec the
built-ins get, so it does not have to redo icon lookup, text placement
or scrolling.
The background fill is added here rather than by the template, because whether the firmware can fill at all depends on its version, and that is exactly the part a caller should not have to know.
device_api_version is what the bar reports. None means "unknown",
and a current device is then assumed - guessing "old" would disable the
feature for every caller that has not called version() yet.
elements = build_notification("Laundry done", icon="check") len(elements.elements) 2 elements.elements[0].type 'image' elements.elements[1].type 'text'
async
¶
notify(client: NotifyClient, line_1: str, *, line_2: str | None = None, icon: str | StockIcon | None = None, line_1_font: DisplayFontName = DEFAULT_FONT, line_2_font: DisplayFontName | None = None, line_1_color: ColorInput | None = None, line_2_color: ColorInput | None = None, background_color: ColorInput | None = None, sound: str | None = None, duration: int | None = None, priority: int = PRIORITY_DEFAULT, application_name: str = 'busylib', template: Template | None = None) -> types.SuccessResponse
Lay out a notification, draw it, and play its sound if it has one.
A notification with a sound is one intention, so sound belongs here
even though playing it is a second request - a caller composing this by
hand has to remember both, and to pass the same application_name to
each. Names come from STOCK_SOUNDS.
The device version is taken from the client, so a caller cannot forget to pass it and silently lose the check that a feature the firmware lacks is refused rather than drawn wrong.
There is deliberately no clearing helper: a drawing is withdrawn with
display_clear(application_name=...), which is already a single call to
a single endpoint, and wrapping it would only add a synonym.
Callers that want to place elements themselves can use
build_notification and display_draw separately, or skip both.
select_template(spec: NotificationSpec) -> Template
Pick the built-in template that suits this notification.
Timers¶
The control helpers are reached through the module - timer.start(bar) - so
that names like start and stop stay unambiguous at the call site.
Working out what a timer is doing right now.
The bar does not run a clock you can read. GET /api/busy/snapshot returns
the last snapshot that was applied, verbatim, together with the moment it
was applied - verified on firmware 27.7.0, where a snapshot with three
seconds left still reported three seconds left, and the same timestamp,
eleven seconds later. This is deliberate: it is how the apps keep a timer in
step, the freshest snapshot wins, and each client works out the rest.
So reading a snapshot tells you nothing on its own. This module does the arithmetic every consumer would otherwise repeat: advance the snapshot by the time that has passed and say what phase the timer is in.
Changing a timer works the same way round: there is no "pause" endpoint, only a snapshot to write, and the device takes the freshest one as the truth. The helpers here write the snapshot each change needs - which includes reading the card first, since a snapshot names one and inherits whatever it does not say for itself.
A session may say a great deal for itself, though: its own kind, its own lengths, its own theme, none of which touch the card. That is how the BUSY app runs a card with a timer the card does not hold, and it is what lets an automation start a forty-five minute countdown without rewriting the two cards their owner arranged.
dataclass
¶
TimerState(mode: TimerMode, is_paused: bool, phase: TimerPhase | None = None, interval: int | None = None, time_left_ms: int | None = None, is_finished: bool | None = None, elapsed_ms: int = 0)
What the timer is doing at one moment.
phase is None only once a session has finished, where no phase
applies any more.
Bases: BusyBarError
Raised for a change that only makes sense while a session is running.
Pausing, resuming, moving to the next phase and setting the session's theme all rewrite the running session. With nothing running there is nothing to rewrite, and writing a session to make the change possible would start one nobody asked for.
UnknownThemeError(theme: str, available: Sequence[str])
Bases: BusyBarError
Raised for a theme this bar does not have.
The device does not raise it: a profile naming a theme that does not exist is stored and read back happily, and only the bar's own screen shows that something is wrong. So the check has to happen here, or a typo leaves a card pointing at nothing with every layer reporting success.
PhaseTooShortError(field: str, given_ms: int, *, minimum_ms: int = MINIMUM_PHASE_MS, maximum_ms: int = MAXIMUM_PHASE_MS)
Bases: BusyBarError
Raised for a duration the device would refuse, at either end.
It never says so usefully. A card written with a two-minute work phase comes back with the phase it had before and every layer reports success; a session with one is refused as an unparseable snapshot. So a caller that is not told here finds out by watching a bar run the wrong timer.
The name is the common case - too short - but the same class carries the ceiling, since to a caller they are one mistake: a length the bar will not run.
Bases: Protocol
What these helpers need from a client.
Deliberately three methods rather than the whole client, so a caller can pass anything that speaks them - including a test double.
async
¶
busy_profile_set(slot: BusyProfileSlot, profile: BusyProfile) -> types.SuccessResponse
phase_of(interval: int) -> TimerPhase
Name the phase an interval index belongs to.
current_interval is the firmware's own interval index, sent raw: it
starts at 0 for the first work period and rises by one at every change
of phase, so even indices are work and odd ones rest. Comparing the
interval's length against the configured durations would usually work
too, but not when work and rest are set equal - the index has no such
gap.
timer_state(snapshot: BusySnapshot, *, now_ms: int | None = None) -> TimerState
Advance a snapshot to now_ms and report the timer's state.
now_ms defaults to the current wall clock. A paused snapshot is
returned as it stands, since paused time does not pass.
Everything is derived from the interval index rather than from interval lengths, so a session configured with equal work and rest is no harder to read than any other: the index says which phase it is, and where the session ends.
kind_of(settings: BusyTimerSettings) -> TimerKind
Which kind of timer a card holds, in this package's words.
kind_of_snapshot(snapshot: BusySnapshotVariant) -> TimerKind | None
Which kind of timer is running, or None if none is.
A snapshot has a state the three kinds do not - nothing started - so
this answers with None rather than pretending the bar is idling in
one of them.
async
¶
start(client: TimerClient, slot: BusyProfileSlot = 'busy', *, card_id: str | None = None, kind: TimerKind | None = None, duration_ms: int | None = None, rest_ms: int | None = None, cycles: int | None = None, theme: str | None = None, now_ms: int | None = None) -> None
Start a session, optionally with settings of its own.
With nothing but a slot, this runs what that card describes - the thing the bar's own switch would start.
Give a kind or any length and the session runs those instead, and the card is not touched: the settings travel in the snapshot. That is how the BUSY app starts a card with a different timer, and it is what makes "run a countdown for forty-five minutes" leave the two cards exactly as their owner arranged them.
The card is still named, because a snapshot names one and the app shows that card's name and its own decoration for the running session.
card_id names a card directly instead of taking the one in a slot,
and then nothing on the bar is read at all. The point is a card that
is in neither position: the bar keeps two, the app keeps more, and a
session naming one of the others runs without either of the bar's own
two being involved even by name. Such a session has no card to fall
back on, so its kind and lengths are the ones given here, or the
defaults below.
What the device will not take, as a parse error about the whole
snapshot rather than anything helpful: an interval phase outside five
minutes to eight hours, fewer than two or more than thirty-five work
phases, or a countdown over a day. Those are the firmware's own
bounds, in busy_timer_common.h, confirmed on a bar.
async
¶
stop(client: TimerClient, *, now_ms: int | None = None) -> None
End the session, leaving the bar with nothing running.
This is not the selector's off position, which is the bar's
do-not-disturb: it is the session going back to not started.
async
¶
set_paused(client: TimerClient, paused: bool, *, now_ms: int | None = None) -> None
Pause or resume the running session.
Pausing writes back how much time is actually left, not the figure the stored snapshot carries: that one was true when it was written, and resuming from it would hand back the time the session already spent.
async
¶
next_phase(client: TimerClient, *, now_ms: int | None = None) -> None
Move an interval session on to its next phase.
Work becomes rest and rest becomes the next work, at that phase's full length. Past the last interval the session is over, so it is stopped rather than wrapped around.
async
¶
set_session_theme(client: TimerClient, theme: str, *, known: Sequence[str] | None = None, now_ms: int | None = None) -> None
Change the theme the running session is showing.
This lasts as long as the session: the card keeps its own theme, and
the bar shows that one again next time it starts. Confirmed on
hardware - a session switched to dnd came back as the card's busy
once stopped.
async
¶
set_card_theme(client: TimerClient, slot: BusyProfileSlot, theme: str, *, known: Sequence[str] | None = None, now_ms: int | None = None) -> None
Change the theme one of the bar's cards starts with.
Unlike set_session_theme this outlasts the session, and it does not
touch what is on screen now - a session already running keeps the
theme it started with.
The profile is stamped with the moment of writing, for the same reason
a snapshot is: the device keeps whichever copy is newer and silently
discards the rest. A card read back and written unchanged carries the
stored timestamp, which is not newer - and a bar fresh from the
factory reports profile_timestamp_ms: 0, so the write is dropped
while still answering {"result": "OK"}. Confirmed on firmware r971.
async
¶
themes(client: ThemeCatalogueClient) -> list[str]
Which themes this bar can show.
Not a fixed set and not an enum: themes are assets, so one bar has what the firmware shipped, another has one the owner uploaded, and a third is missing one the owner deleted. It is read from the bar rather than written down anywhere that would go stale.
Two sources, because neither alone is the answer. The assets are one
directory each under THEMES_PATH. And whichever themes the bar's own
cards are set to are real by definition, which is how the firmware's
built-in default gets in - it has no directory, so a listing alone
would report a bar cannot show the theme it is showing right now.
async
¶
configure(client: TimerClient, slot: BusyProfileSlot = 'busy', *, kind: TimerKind | None = None, duration_ms: int | None = None, work_ms: int | None = None, rest_ms: int | None = None, cycles: int | None = None, total_ms: int | None = None, theme: str | None = None, known_themes: Sequence[str] | None = None, now_ms: int | None = None) -> types.BusyProfile
Change one of the bar's two cards, leaving the rest of it alone.
For a change that should last. A session can carry settings of its own
- start() takes them - and that is what a one-off wants; this is for
when the bar's own switch should start something different from now
on. The bar and the phone app see it too; that is the same thing they
do to each other.
Only what is given is changed. work_ms, rest_ms and cycles apply
to an interval card, total_ms to a countdown; giving one the card
cannot use is an error rather than a silent no-op, since the device
would treat it as one.
Returns the profile as written, so a caller can see what the card now holds.