Clients¶
BusyBar is the synchronous client and AsyncBusyBar is its async/await
counterpart. Both compose the same set of endpoint mixins, so the method
surface is identical apart from the coroutines.
from busylib import AsyncBusyBar, BusyBar
bb = BusyBar("10.0.4.20")
async_bb = AsyncBusyBar(addr="10.0.4.20", token="my-access-key")
Creating either client has no terminal output and does not contact the device. The first endpoint call performs the connection; see the quick start for its expected output.
BusyBar(addr: str | None = None, *, token: str | None = None, timeout: float | Timeout | None = None, max_retries: int = 2, backoff: float = DEFAULT_BACKOFF, transport: BaseTransport | None = None, api_version: str | None = None, compatibility_mode: CompatibilityMode = 'warn', is_cloud: bool | None = None, device_api_version: str | None = None)
Bases: AccessMixin, AccountMixin, BusyMixin, TimeMixin, UpdaterMixin, FirmwareMixin, StorageMixin, AssetsMixin, DisplayMixin, AudioMixin, WifiMixin, InputMixin, SmartHomeMixin, StateStreamMixin, BleMixin, SyncClientBase
HTTPX-based client for the BUSY Bar API.
Build a client for one bar.
addr is a device address; leaving it out with a token reaches the
bar through the cloud, at the host BUSYLIB_CLOUD_URL names. Pass
is_cloud=True only to name a cloud host per client, which is what
stops an address like api.dev.busy.app being taken for a device.
property
¶
is_usb_connected: bool
Returns True if a USB device was found and connected.
class-attribute
instance-attribute
¶
connection_type: Literal['local', 'cloud', 'network'] = 'network'
instance-attribute
¶
client = httpx2.Client(base_url=self.base_url, headers=headers or None, timeout=self._timeout, transport=transport)
property
¶
device_api_version: str | None
The device's API version, if it is known yet.
None until version() has been called or a value was supplied to the
constructor.
property
¶
is_cloud: bool
Check whether connection uses cloud mode.
Returns True for cloud connection_type.
property
¶
is_local: bool
Check whether connection uses local mode.
Returns True for local connection_type.
usb_reboot(*, raise_on_error: bool = False) -> bool
Attempt to reboot the device via USB.
Returns True on success and False on failure by default. If raise_on_error is True, re-raises BusyBarUsbError.
usb_reset(*, raise_on_error: bool = False) -> bool
Alias for usb_reboot().
Provided for callers that prefer "reset" naming.
device_at_least(minimum: str) -> bool | None
Whether the bar is known to run at least minimum.
None means the version is not known, which is not the same as "older": a caller deciding between two request contracts should send the current one rather than assume the device is behind.
method_compatibility(method_name: str) -> versioning.MethodCompatibility | None
Return declarative OpenAPI compatibility metadata for a client method.
is_local_available() -> bool
Check local API reachability on base_url.
Returns True when /api/version responds without network errors.
api_request(method: str, path: str, *, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, session_id: str | None = None, application_name: str | None = None, json_payload: JsonType | None = None, data: bytes | Iterable[bytes] | None = None, expect_bytes: bool = False, allow_text: bool = False, timeout: float | Timeout | None = None) -> JsonType | bytes | str
Execute a raw API request through the current client session.
Advanced callers can control path, params, headers, body, and request context without creating a separate HTTP client.
prepare_request(method: str, path: str, *, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, session_id: str | None = None, application_name: str | None = None, json_payload: JsonType | None = None, data: bytes | Iterable[bytes] | None = None, expect_bytes: bool = False, allow_text: bool = False, timeout: float | Timeout | None = None) -> PreparedRequest
Build a prepared request without executing network I/O.
External integrations can inspect the prepared payload, route it to
custom transports, or execute later via execute_prepared_request.
execute_prepared_request(prepared: PreparedRequest, *, client: Client | None = None) -> JsonType | bytes | str
Execute a previously prepared request.
By default the current httpx2.Client is used. Callers may inject a
custom client while preserving error mapping. Prepared streaming content
is single-use and should be regenerated for repeated executions.
stream_status_ws() -> None
Stream device status updates via WebSocket /api/status/ws.
Sync iteration is not supported because websocket lifecycle and frame handling are asynchronous in this client implementation.
smart_home_pairing() -> types.SmartHomePairingInfo
Fetch smart home pairing status via GET /api/smart_home/pairing.
smart_home_pairing_start() -> types.SmartHomePairingPayload
Start smart home pairing via POST /api/smart_home/pairing.
smart_home_pairing_stop() -> types.SuccessResponse
Stop smart home pairing via DELETE /api/smart_home/pairing.
smart_home_switch() -> types.SmartHomeSwitchState
Fetch smart home switch state via GET /api/smart_home/switch.
smart_home_switch_set(state: bool, *, startup: Literal['off', 'on', 'toggle', 'last'] | None = None) -> types.SuccessResponse
Set smart home switch state via POST /api/smart_home/switch.
wifi_enable() -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/wifi/enable, so calling this raises
BusyBarRemovedEndpointError. Use wifi_connect() / wifi_disconnect() instead.
wifi_disable() -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/wifi/disable, so calling this raises
BusyBarRemovedEndpointError. Use wifi_connect() / wifi_disconnect() instead.
wifi_networks() -> types.NetworkResponse
Scan for nearby networks via GET /api/wifi/networks.
The device cannot scan while it is associated: doing so returns
400 "Scan not possible when connected" as a BusyBarAPIError.
Disconnect first with wifi_disconnect(), or skip the scan and pass
the SSID to wifi_connect() directly.
audio_play(*, path: str | None = None, stock_path: str | None = None, payload: AudioPlayRequest | dict[str, Any] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Play audio through POST /api/audio/play.
The endpoint payload may reference either an uploaded asset path or a
stock path. Explicit path and stock_path keyword arguments override
the same keys provided in payload. Request context such as
application_name and session_id is accepted through request kwargs;
use api_request for fully custom bodies.
audio_stop() -> types.SuccessResponse
Stop audio through DELETE /api/audio/play.
Uses API-like naming for callers that mirror firmware endpoints.
display_draw(display_data: DisplayElements | dict[str, Any], *, clear_before_draw: bool = False, sanitize_text: bool = False, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
display(display_data: DisplayElements | dict[str, Any], *, clear_before_draw: bool = False, sanitize_text: bool = False, audio_payload: AudioPlayRequest | dict[str, Any] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Render display content and optionally play audio after draw.
Operations are sequential, not atomic: clear may succeed before draw fails, and audio failure may occur after display content is visible. Exceptions include the failed endpoint path for diagnostics.
display_clear(*, element_ids: Sequence[str] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Clear display content through DELETE /api/display/draw.
Without element_ids this removes everything the application drew.
With them, only those elements go - which is how a drawing built
from several pieces loses one of them without being redrawn whole.
Uses API-like naming for callers that mirror firmware endpoints.
screen(display_id: DisplaySpecLike) -> bytes
Fetch a single display frame via GET /api/screen.
Returns RGB bytes, three per pixel: 3456 for the front display (72x16) and 38400 for the back (160x80, sent L4-packed and expanded to grey triples here). The device sends colour as BGR and it is swapped on the way out.
The response body is base64-encoded, uncompressed framebuffer bytes
(the Content-Type: image/bmp header is misleading, there is no
real BMP header).
frame(display_id: DisplaySpecLike) -> Frame
Fetch a display frame as a Frame via GET /api/screen.
The same bytes as screen(), with the geometry and the display
attached, so callers can read pixels, rows or a PNG without tracking
the layout themselves.
assets_upload(application_name: str, filename: str, data: bytes, *, timeout: float | Timeout | None = ASSET_UPLOAD_TIMEOUT) -> types.SuccessResponse
Upload an asset file for the given application.
Uses a longer default timeout to tolerate large payload uploads.
storage_write(path: str, data: bytes, *, append: bool = False, timeout: float | None = 60.0, progress_callback: Callable[[int, int], None] | None = None, chunk_size: int = 64 * 1024) -> types.SuccessResponse
Write a file, converting it for the device first.
append adds to the file instead of replacing it, creating it if
it is not there - which is what makes a log or a growing asset
possible without reading the whole file back first.
storage_mkdir(path: str) -> types.SuccessResponse
Create a storage directory via POST /api/storage/mkdir.
storage_rename(old_path: str, new_path: str) -> types.SuccessResponse
Rename a storage entry via POST /api/storage/rename.
transport() -> types.NetworkInterfaceInfo
Fetch active network transport via GET /api/transport.
status_device() -> types.StatusDevice
Fetch device manufacturing status via GET /api/status/device.
status_firmware() -> types.StatusFirmware
Fetch firmware status via GET /api/status/firmware.
status_system() -> types.StatusSystem
Fetch runtime status via GET /api/status/system.
log_dump(filename: str | None = None, *, path: str | None = None) -> types.LogDumpResponse
Dump the in-memory device log buffer to a storage file.
The contract changed at OpenAPI 25.0.0 and the two are not translatable, so both are kept rather than one being dropped:
filenameis a bare name matching^[a-zA-Z0-9_-]+$; the device adds the extension and the storage path. 25.0.0 and later.pathis a full destination path such as/ext/dump.log. Before 25.0.0.
With neither, the device writes to its own default and the request is
identical on both, which is why plain log_dump() needs no version at
all.
When the device version is known - after version(), or from
device_api_version= - asking for the wrong one raises here instead
of being refused by the bar with a bare 400.
update(firmware_data: bytes) -> types.SuccessResponse
Upload firmware update TAR and initiate update.
update_status() -> types.UpdateStatus
Get firmware update status with progress information.
update_changelog(version: str) -> types.UpdateChangelogResponse
Fetch update changelog for a specific version.
update_install(version: str) -> types.SuccessResponse
Start firmware update installation by version.
update_abort_download() -> types.SuccessResponse
Abort an ongoing firmware download.
update_autoupdate() -> types.AutoupdateSettings
Fetch autoupdate settings via GET /api/update/autoupdate.
update_autoupdate_set(settings: AutoupdateSettings | dict[str, object]) -> types.SuccessResponse
Set autoupdate settings via POST /api/update/autoupdate.
time_timezone_info() -> types.TimezoneInfo
Fetch current device timezone via GET /api/time/timezone.
time_timezone_list() -> types.TimezoneListResponse
Fetch supported device timezones via GET /api/time/tzlist.
time_timestamp(timestamp: str) -> types.SuccessResponse
Set device time via POST /api/time/timestamp.
time_timezone(timezone: str) -> types.SuccessResponse
Set device timezone via POST /api/time/timezone.
busy_snapshot() -> types.BusySnapshot
Fetch busy snapshot via GET /api/busy/snapshot.
busy_snapshot_set(snapshot: BusySnapshot) -> types.SuccessResponse
Set busy snapshot via PUT /api/busy/snapshot.
busy_profile(slot: BusyProfileSlot) -> types.BusyProfile
Fetch a busy profile slot via GET /api/busy/profiles/{slot}.
busy_profile_set(slot: BusyProfileSlot, profile: BusyProfile | dict[str, object]) -> types.SuccessResponse
Set a busy profile slot via PUT /api/busy/profiles/{slot}.
account_unlink() -> types.SuccessResponse
Unlink the device from the account via DELETE /api/account.
account_link() -> types.AccountLink
Request an account link code via POST /api/account/link.
account_info() -> types.AccountInfo
Fetch linked account info via GET /api/account/info.
account_status() -> types.AccountState
Fetch MQTT status info via GET /api/account/status.
account_backend() -> types.AccountBackend
Fetch MQTT backend settings via GET /api/account/backend.
account_backend_set(backend: AccountBackend | dict[str, object]) -> types.SuccessResponse
Set MQTT backend settings via PUT /api/account/backend.
account_profile() -> types.AccountProfile
Removed from the device API.
No supported firmware serves GET /api/account/profile, so calling this raises
BusyBarRemovedEndpointError. Use account_backend() instead.
account_profile_set(profile: AccountProfileName, custom_url: str | None = None) -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/account/profile, so calling this raises
BusyBarRemovedEndpointError. Use account_backend_set() instead.
access_set(mode: HttpAccessMode, key: str) -> types.SuccessResponse
Set HTTP access mode via POST /api/access.
access_tokens_list() -> types.AccessTokensInfo
List issued access tokens via GET /api/access/tokens.
The secret itself is returned only when a token is minted, so every
token here has token set to None.
Added by busy-app/busybar-firmware#886 and served from OpenAPI 27.5.0, first released in firmware 1.2.3; earlier firmware 404s.
access_token_mint(name: str) -> types.AccessToken
Issue a new access token via POST /api/access/tokens.
This is the only time the device discloses the secret; it cannot be read back afterwards.
Added by busy-app/busybar-firmware#886 and served from OpenAPI 27.5.0, first released in firmware 1.2.3; earlier firmware 404s.
access_tokens_delete_all() -> types.SuccessResponse
Revoke every access token via DELETE /api/access/tokens.
Added by busy-app/busybar-firmware#886 and served from OpenAPI 27.5.0, first released in firmware 1.2.3; earlier firmware 404s.
access_tokens_revoke(short_id: str) -> types.SuccessResponse
Revoke one access token via DELETE /api/access/tokens/{short_id}.
Added by busy-app/busybar-firmware#886 and served from OpenAPI 27.5.0, first released in firmware 1.2.3; earlier firmware 404s.
AsyncBusyBar(addr: str | None = None, *, token: str | None = None, timeout: float | Timeout | None = None, max_retries: int = 2, backoff: float = DEFAULT_BACKOFF, transport: AsyncBaseTransport | None = None, api_version: str | None = None, compatibility_mode: CompatibilityMode = 'warn', is_cloud: bool | None = None, device_api_version: str | None = None)
Bases: AsyncAccessMixin, AsyncAccountMixin, AsyncBusyMixin, AsyncTimeMixin, AsyncUpdaterMixin, AsyncFirmwareMixin, AsyncStorageMixin, AsyncAssetsMixin, AsyncDisplayMixin, AsyncAudioMixin, AsyncWifiMixin, AsyncInputMixin, AsyncSmartHomeMixin, AsyncStateStreamMixin, AsyncBleMixin, AsyncClientBase
Async HTTPX-based client for the BUSY Bar API.
Build a client for one bar.
addr is a device address; leaving it out with a token reaches the
bar through the cloud, at the host BUSYLIB_CLOUD_URL names. Pass
is_cloud=True only to name a cloud host per client, which is what
stops an address like api.dev.busy.app being taken for a device.
property
¶
is_usb_connected: bool
Returns True if a USB device was found and connected.
class-attribute
instance-attribute
¶
connection_type: Literal['local', 'cloud', 'network'] = 'network'
instance-attribute
¶
client = httpx2.AsyncClient(base_url=self.base_url, headers=headers or None, timeout=self._timeout, transport=transport)
property
¶
device_api_version: str | None
The device's API version, if it is known yet.
None until version() has been called or a value was supplied to the
constructor.
property
¶
is_cloud: bool
Check whether connection uses cloud mode.
Returns True for cloud connection_type.
property
¶
is_local: bool
Check whether connection uses local mode.
Returns True for local connection_type.
async
¶
usb_reboot(*, raise_on_error: bool = False) -> bool
Attempt to reboot the device via USB.
async
¶
usb_reset(*, raise_on_error: bool = False) -> bool
Alias for usb_reboot().
Provided for callers that prefer "reset" naming.
device_at_least(minimum: str) -> bool | None
Whether the bar is known to run at least minimum.
None means the version is not known, which is not the same as "older": a caller deciding between two request contracts should send the current one rather than assume the device is behind.
method_compatibility(method_name: str) -> versioning.MethodCompatibility | None
Return declarative OpenAPI compatibility metadata for a client method.
async
¶
is_local_available() -> bool
Check local API reachability on base_url.
Returns True when /api/version responds without network errors.
async
¶
api_request(method: str, path: str, *, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, session_id: str | None = None, application_name: str | None = None, json_payload: JsonType | None = None, data: bytes | AsyncIterable[bytes] | None = None, expect_bytes: bool = False, allow_text: bool = False, timeout: float | Timeout | None = None) -> JsonType | bytes | str
Execute a raw async API request through the current client session.
Advanced callers can control path, params, headers, body, and request context without creating a separate HTTP client.
prepare_request(method: str, path: str, *, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, session_id: str | None = None, application_name: str | None = None, json_payload: JsonType | None = None, data: bytes | AsyncIterable[bytes] | None = None, expect_bytes: bool = False, allow_text: bool = False, timeout: float | Timeout | None = None) -> PreparedRequest
Build a prepared async request without executing network I/O.
External integrations can inspect the prepared payload, route it to
custom transports, or execute later via execute_prepared_request.
Prepared request should be executed with an async executor.
async
¶
execute_prepared_request(prepared: PreparedRequest, *, client: AsyncClient | None = None) -> JsonType | bytes | str
Execute a previously prepared async request.
By default the current httpx2.AsyncClient is used. Callers may inject
a custom client while preserving error mapping. Prepared streaming
content is single-use and should be regenerated for repeated executions.
async
¶
stream_status_ws(*, enable: bool = True, decode_protobuf: bool = True) -> AsyncIterator[dict[str, Any] | bytes | str]
Open /api/status/ws and yield status updates.
When decode_protobuf=True, binary frames are decoded using
BSB_State.State protobuf schema from bsb-protobuf and converted into
dictionaries with original proto field names.
async
¶
smart_home_pairing() -> types.SmartHomePairingInfo
Fetch smart home pairing status via GET /api/smart_home/pairing.
async
¶
smart_home_pairing_start() -> types.SmartHomePairingPayload
Start smart home pairing via POST /api/smart_home/pairing.
async
¶
smart_home_pairing_stop() -> types.SuccessResponse
Stop smart home pairing via DELETE /api/smart_home/pairing.
async
¶
smart_home_switch() -> types.SmartHomeSwitchState
Fetch smart home switch state via GET /api/smart_home/switch.
async
¶
smart_home_switch_set(state: bool, *, startup: Literal['off', 'on', 'toggle', 'last'] | None = None) -> types.SuccessResponse
Set smart home switch state via POST /api/smart_home/switch.
async
¶
wifi_enable() -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/wifi/enable, so calling this raises
BusyBarRemovedEndpointError. Use wifi_connect() / wifi_disconnect() instead.
async
¶
wifi_disable() -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/wifi/disable, so calling this raises
BusyBarRemovedEndpointError. Use wifi_connect() / wifi_disconnect() instead.
async
¶
wifi_connect(config: ConnectRequestConfig | dict[str, Any]) -> types.SuccessResponse
async
¶
wifi_networks() -> types.NetworkResponse
Scan for nearby networks via GET /api/wifi/networks.
The device cannot scan while it is associated: doing so returns
400 "Scan not possible when connected" as a BusyBarAPIError.
Disconnect first with wifi_disconnect(), or skip the scan and pass
the SSID to wifi_connect() directly.
async
¶
audio_play(*, path: str | None = None, stock_path: str | None = None, payload: AudioPlayRequest | dict[str, Any] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Play audio through async POST /api/audio/play.
The endpoint payload may reference either an uploaded asset path or a
stock path. Explicit path and stock_path keyword arguments override
the same keys provided in payload. Request context such as
application_name and session_id is accepted through request kwargs;
use api_request for fully custom bodies.
async
¶
audio_stop() -> types.SuccessResponse
Stop audio through async DELETE /api/audio/play.
Uses API-like naming for callers that mirror firmware endpoints.
async
¶
display_draw(display_data: DisplayElements | dict[str, Any], *, clear_before_draw: bool = False, sanitize_text: bool = False, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
async
¶
display(display_data: DisplayElements | dict[str, Any], *, clear_before_draw: bool = False, sanitize_text: bool = False, audio_payload: AudioPlayRequest | dict[str, Any] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Render display content and optionally play audio after draw.
Operations are sequential, not atomic: clear may succeed before draw fails, and audio failure may occur after display content is visible. Exceptions include the failed endpoint path for diagnostics.
async
¶
display_clear(*, element_ids: Sequence[str] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Clear display content through async DELETE /api/display/draw.
Without element_ids this removes everything the application drew.
With them, only those elements go - which is how a drawing built
from several pieces loses one of them without being redrawn whole.
Uses API-like naming for callers that mirror firmware endpoints.
async
¶
display_brightness_set(value: BrightnessValue) -> types.SuccessResponse
async
¶
screen(display_id: DisplaySpecLike) -> bytes
Fetch a single display frame via GET /api/screen.
Returns RGB bytes, three per pixel: 3456 for the front display (72x16) and 38400 for the back (160x80, sent L4-packed and expanded to grey triples here). The device sends colour as BGR and it is swapped on the way out.
The response body is base64-encoded, uncompressed framebuffer bytes
(the Content-Type: image/bmp header is misleading, there is no
real BMP header).
async
¶
frame(display_id: DisplaySpecLike) -> Frame
Fetch a display frame as a Frame via GET /api/screen.
The same bytes as screen(), with the geometry and the display
attached, so callers can read pixels, rows or a PNG without tracking
the layout themselves.
async
¶
assets_upload(application_name: str, filename: str, data: bytes, *, timeout: float | Timeout | None = ASSET_UPLOAD_TIMEOUT) -> types.SuccessResponse
Upload an asset file for the given application.
Uses a longer default timeout to tolerate large payload uploads.
async
¶
storage_write(path: str, data: bytes, *, append: bool = False, timeout: float | None = 60.0, progress_callback: Callable[[int, int], None] | None = None, chunk_size: int = 64 * 1024) -> types.SuccessResponse
async
¶
storage_mkdir(path: str) -> types.SuccessResponse
Create a storage directory via POST /api/storage/mkdir.
async
¶
storage_rename(old_path: str, new_path: str) -> types.SuccessResponse
Rename a storage entry via POST /api/storage/rename.
async
¶
transport() -> types.NetworkInterfaceInfo
Fetch active network transport via GET /api/transport.
async
¶
status_device() -> types.StatusDevice
Fetch device manufacturing status via GET /api/status/device.
async
¶
status_firmware() -> types.StatusFirmware
Fetch firmware status via GET /api/status/firmware.
async
¶
status_system() -> types.StatusSystem
Fetch runtime status via GET /api/status/system.
async
¶
status_power() -> types.StatusPower
Fetch power status via GET /api/status/power.
async
¶
log_dump(filename: str | None = None, *, path: str | None = None) -> types.LogDumpResponse
Dump the in-memory device log buffer to a storage file.
The contract changed at OpenAPI 25.0.0 and the two are not translatable, so both are kept rather than one being dropped:
filenameis a bare name matching^[a-zA-Z0-9_-]+$; the device adds the extension and the storage path. 25.0.0 and later.pathis a full destination path such as/ext/dump.log. Before 25.0.0.
With neither, the device writes to its own default and the request is
identical on both, which is why plain log_dump() needs no version at
all.
When the device version is known - after version(), or from
device_api_version= - asking for the wrong one raises here instead
of being refused by the bar with a bare 400.
async
¶
update(firmware_data: bytes) -> types.SuccessResponse
Upload firmware update TAR and initiate update.
async
¶
update_check() -> types.SuccessResponse
Start asynchronous firmware update check.
async
¶
update_status() -> types.UpdateStatus
Get firmware update status with progress information.
async
¶
update_changelog(version: str) -> types.UpdateChangelogResponse
Fetch update changelog for a specific version.
async
¶
update_install(version: str) -> types.SuccessResponse
Start firmware update installation by version.
async
¶
update_abort_download() -> types.SuccessResponse
Abort an ongoing firmware download.
async
¶
update_autoupdate() -> types.AutoupdateSettings
Fetch autoupdate settings via GET /api/update/autoupdate.
async
¶
update_autoupdate_set(settings: AutoupdateSettings | dict[str, object]) -> types.SuccessResponse
Set autoupdate settings via POST /api/update/autoupdate.
async
¶
time_timezone_info() -> types.TimezoneInfo
Fetch current device timezone via GET /api/time/timezone.
async
¶
time_timezone_list() -> types.TimezoneListResponse
Fetch supported device timezones via GET /api/time/tzlist.
async
¶
time_timestamp(timestamp: str) -> types.SuccessResponse
Set device time via POST /api/time/timestamp.
async
¶
time_timezone(timezone: str) -> types.SuccessResponse
Set device timezone via POST /api/time/timezone.
async
¶
busy_snapshot() -> types.BusySnapshot
Fetch busy snapshot via GET /api/busy/snapshot.
async
¶
busy_snapshot_set(snapshot: BusySnapshot) -> types.SuccessResponse
Set busy snapshot via PUT /api/busy/snapshot.
async
¶
busy_profile(slot: BusyProfileSlot) -> types.BusyProfile
Fetch a busy profile slot via GET /api/busy/profiles/{slot}.
async
¶
busy_profile_set(slot: BusyProfileSlot, profile: BusyProfile | dict[str, object]) -> types.SuccessResponse
Set a busy profile slot via PUT /api/busy/profiles/{slot}.
async
¶
account_unlink() -> types.SuccessResponse
Unlink the device from the account via DELETE /api/account.
async
¶
account_link() -> types.AccountLink
Request an account link code via POST /api/account/link.
async
¶
account_info() -> types.AccountInfo
Fetch linked account info via GET /api/account/info.
async
¶
account_status() -> types.AccountState
Fetch MQTT status info via GET /api/account/status.
async
¶
account_backend() -> types.AccountBackend
Fetch MQTT backend settings via GET /api/account/backend.
async
¶
account_backend_set(backend: AccountBackend | dict[str, object]) -> types.SuccessResponse
Set MQTT backend settings via PUT /api/account/backend.
async
¶
account_profile() -> types.AccountProfile
Removed from the device API.
No supported firmware serves GET /api/account/profile, so calling this raises
BusyBarRemovedEndpointError. Use account_backend() instead.
async
¶
account_profile_set(profile: AccountProfileName, custom_url: str | None = None) -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/account/profile, so calling this raises
BusyBarRemovedEndpointError. Use account_backend_set() instead.
async
¶
access_set(mode: HttpAccessMode, key: str) -> types.SuccessResponse
Set HTTP access mode via POST /api/access.
async
¶
access_tokens_list() -> types.AccessTokensInfo
List issued access tokens via GET /api/access/tokens.
The secret itself is returned only when a token is minted, so every
token here has token set to None.
Added by busy-app/busybar-firmware#886 and served from OpenAPI 27.5.0, first released in firmware 1.2.3; earlier firmware 404s.
async
¶
access_token_mint(name: str) -> types.AccessToken
Issue a new access token via POST /api/access/tokens.
This is the only time the device discloses the secret; it cannot be read back afterwards.
Added by busy-app/busybar-firmware#886 and served from OpenAPI 27.5.0, first released in firmware 1.2.3; earlier firmware 404s.
async
¶
access_tokens_delete_all() -> types.SuccessResponse
Revoke every access token via DELETE /api/access/tokens.
Added by busy-app/busybar-firmware#886 and served from OpenAPI 27.5.0, first released in firmware 1.2.3; earlier firmware 404s.
async
¶
access_tokens_revoke(short_id: str) -> types.SuccessResponse
Revoke one access token via DELETE /api/access/tokens/{short_id}.
Added by busy-app/busybar-firmware#886 and served from OpenAPI 27.5.0, first released in firmware 1.2.3; earlier firmware 404s.
Prepared requests¶
Bases: BaseModel
Prepared low-level request ready for execution by HTTP clients.
This object stores normalized request attributes after payload encoding. Callers can execute it with built-in client transport or pass fields to an external request executor. Serialization is not guaranteed because timeout and streaming content may contain runtime-only objects.
Streaming content based on iterables/generators is single-use and should not be reused across multiple executions.
class-attribute
instance-attribute
¶
headers: dict[str, str] | None = Field(default=None, repr=False)
class-attribute
instance-attribute
¶
content: bytes | Iterable[bytes] | AsyncIterable[bytes] | None = Field(default=None, repr=False)
class-attribute
instance-attribute
¶
model_config = ConfigDict(arbitrary_types_allowed=True)
Using your own HTTP session¶
A host application often already owns a connection pool and wants everything
to go through it. Home Assistant, for example, hands integrations a shared
aiohttp session; opening a second pool beside it wastes connections and
sidesteps the host's timeouts and tracing.
httpx2 transports are the seam for that, and busylib[aiohttp] ships one:
from busylib import AsyncBusyBar
from busylib.transports import AiohttpTransport
bb = AsyncBusyBar("10.0.4.20", transport=AiohttpTransport(session))
Every client method then rides the session you passed. The session is
borrowed, never closed — aclose() on the client leaves it usable, because
whoever handed it over still owns it.
Status streaming is unaffected either way: it speaks WebSocket through
websockets rather than the HTTP transport, so it keeps its own connection.
Run this client's requests over someone else's HTTP session.
A host application often already owns a connection pool and wants everything
it does to go through it - Home Assistant hands integrations a shared
aiohttp session, and opening a second pool beside it wastes connections and
sidesteps the host's own timeouts and tracing.
httpx2 already has the seam for this: a transport is one method. So nothing
in the client changes, and the default stays httpx2's own transport.
from busylib import AsyncBusyBar
from busylib.transports import AiohttpTransport
bb = AsyncBusyBar("10.0.4.20", transport=AiohttpTransport(session))
Status streaming is not affected: it speaks WebSocket through websockets,
not through the HTTP transport, so it keeps its own connection either way.
module-attribute
¶
AIOHTTP_MISSING = 'AiohttpTransport needs aiohttp, which is optional: install busylib[aiohttp].'
AiohttpTransport(session: ClientSession)
Bases: AsyncBaseTransport
An httpx2 transport backed by an aiohttp session you already have.
The session is borrowed, never closed: the caller owns its lifetime, and closing it here would break every other user of it.
async
¶
handle_async_request(request: Request) -> httpx2.Response
Perform one request on the borrowed session.
Errors are translated into httpx2 transport errors, because that is
the family the client catches to retry and to report; letting an
aiohttp exception through would escape that handling entirely.