juno.bluetooth¶

Talk to Bluetooth Low Energy peripherals (central role).

Wraps Apple’s Core Bluetooth central API. Open scans yield Advertisement events, Peripheral.connect() returns a context-manager Connection, and characteristic reads / writes / notifications are exposed as plain Python calls and iterators.

Typical usage — Nordic UART pattern:

from juno import bluetooth as bt

NUS_SERVICE = "6E400001-B5A3-F393-E0A9-E50E24DCCA9E"
NUS_TX = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"   # write to device
NUS_RX = "6E400003-B5A3-F393-E0A9-E50E24DCCA9E"   # notify from device

bt.request_authorization()                        # blocks for prompt

# Find the device by name, scanning for advertisements that
# include the UART service UUID.
target = None
for adv in bt.scan(services=[NUS_SERVICE], timeout=10):
    if adv.name == "M5-BB-Cardputer":
        target = adv.peripheral
        break
if target is None:
    raise RuntimeError("device not in range")

with target.connect() as conn:
    service = conn.service(NUS_SERVICE)
    rx = service.characteristic(NUS_RX)
    tx = service.characteristic(NUS_TX)

    # Send a payload (no ack required).
    tx.write(b"hello\n", response=False)

    # Receive incoming notifications.
    for chunk in rx.notifications():
        print("got:", chunk)
        if chunk == b"bye\n":
            break

Authorization:

CBManager.authorization is read-only — there is no explicit Apple-provided request method, so the system prompt fires on the first real radio operation (scan / connect). request_authorization() triggers a brief scan to provoke the prompt up-front and waits until the status resolves.

Streaming queue & overflow:

scan() and Characteristic.notifications() are backed by bounded queues with drop-oldest overflow (advertisements: 128, notifications: 1024). The radio is never stalled — when the consumer can’t keep up, the oldest events are discarded. Iterators expose a dropped counter so a slow loop can detect this.

Lifecycle:

Peripheral, Connection, Service, Characteristic, and Descriptor represent native iOS resources. Use with blocks (or call close / disconnect explicitly) to release them deterministically; __del__ is a best-effort safety net only.

Stop is responsive throughout: long-running scans, notification iterators, and authorization waits all surface as KeyboardInterrupt when Juno’s Stop button is pressed.

class juno.bluetooth.AuthorizationStatus¶

Bases: object

Bluetooth authorization status string constants.

class juno.bluetooth.State¶

Bases: object

Adapter state string constants returned by state().

exception juno.bluetooth.BluetoothError¶

Bases: Exception

Base class for module errors.

exception juno.bluetooth.DisconnectedError¶

Bases: BluetoothError

Raised from a notifications iterator when the peer disconnects mid-stream. Lets reconnect-on-disconnect loops distinguish a user-initiated unsubscribe (StopIteration) from an unexpected drop.

exception juno.bluetooth.NotConnectedError¶

Bases: BluetoothError

Operation requires an active connection but the connection is closed or the peripheral has disconnected.

exception juno.bluetooth.BluetoothTimeoutError¶

Bases: BluetoothError

A blocking operation exceeded its timeout. Named explicitly (rather than reusing the builtin TimeoutError) so users can handle Bluetooth timeouts separately from filesystem / network ones.

class juno.bluetooth.Advertisement(peripheral, name=None, rssi=0, is_connectable=True, service_uuids=(), manufacturer_data=None, service_data=None, tx_power_level=None)¶

Bases: object

A single advertisement received during a scan.

Parameters:

peripheral (Peripheral)
name (str | None)
rssi (int)
is_connectable (bool)
service_uuids (tuple[str, ...])
manufacturer_data (bytes | None)
service_data (dict[str, bytes] | None)
tx_power_level (int | None)

peripheral¶

The advertising peripheral. Pass to Peripheral.connect() to open a session.

Type:: juno.bluetooth.Peripheral

name¶

Local name from the advertisement, or the peripheral’s GAP name if available; None if neither is present.

Type:: str | None

rssi¶

Received signal strength in dBm (typical range -100..-30).

Type:: int

is_connectable¶

True if the peripheral advertised itself as connectable. Some beacon-style devices advertise read-only.

Type:: bool

service_uuids¶

Service UUIDs the peripheral advertises. UUIDs are returned in the canonical 36-char form (or 4-char short form for SIG-defined services depending on what the peripheral advertised).

Type:: tuple[str, …]

manufacturer_data¶

Raw vendor-specific bytes (the first two bytes are the company identifier per the Bluetooth SIG list); None if not present.

Type:: bytes | None

service_data¶

Per-service-UUID payload bytes (e.g. iBeacon data, sensor frames). Empty dict when absent.

Type:: dict[str, bytes] | None

tx_power_level¶

Advertised TX power in dBm; None when the peripheral didn’t advertise it.

Type:: int | None

class juno.bluetooth.Peripheral¶

Bases: object

A Bluetooth peripheral.

Peripheral instances are deduped by their identifier UUID: multiple advertisements for the same device produce the same Peripheral. Hold a reference across scans / disconnects to skip rescanning when reconnecting.

Construct via scan() (advertised) or Peripheral.from_identifier() (cached identifier from a previous run).

classmethod from_identifier(identifier)¶

Resolve a peripheral by its identifier UUID (the value of Peripheral.identifier from a previous run).

Backed by CBCentralManager.retrievePeripherals(withIdentifiers:), so no scan is required — useful for reconnecting to a device without burning power on a discovery cycle.

Parameters:

identifier (str) – Full 36-character UUID string.

Raises:

TypeError – If identifier is not a string.
ValueError – If identifier is not a valid UUID.
BluetoothError – If the system has no record of that identifier (typical after device-side bonding wipe, factory reset, or first-time pairing).

Return type:

Peripheral

property identifier: str¶: Identifier UUID. Stable across scans and reboots.

property name: str | None¶: Last-known advertised / GAP name. None when unknown.

property state: str¶: disconnected / connecting / connected / disconnecting.

connect(*, timeout=15.0)¶

Open a connection to this peripheral.

Parameters:

timeout (float) – Seconds to wait for didConnect / didFailToConnect. 0 waits indefinitely.

Returns:

A Connection context manager. Use with blocks for deterministic teardown.

Raises:

BluetoothTimeoutError – If the connection didn’t complete in time.
NotConnectedError – If the system rejected the connect.

Return type:

Connection

class juno.bluetooth.Connection¶

Bases: object

An active Core Bluetooth connection (context manager).

Construct via Peripheral.connect(). Use with blocks so the connection is torn down deterministically:

with peripheral.connect() as conn:
    for service in conn.services():
        ...

property peripheral: Peripheral¶: The peripheral this connection was opened against.

property is_connected: bool¶: Whether the connection is currently active.

services(*uuids)¶

Discover services on the peripheral.

Parameters:

*uuids (str) – Optional UUID filter. Forwarded to CBPeripheral.discoverServices(_:) — passing UUIDs is faster and avoids discovering hidden services some peripherals expose only when the filter is nil. Pass no arguments to discover all.

Returns:

List of Service records.

Raises:

NotConnectedError – If the connection is closed.
BluetoothTimeoutError – If discovery doesn’t complete in 30 seconds.

Return type:

list[Service]

service(uuid)¶

Convenience: services(uuid)[0].

Raises:

KeyError – If no service with that UUID is exposed.
ValueError – If multiple services advertise the same UUID (rare; call services() to disambiguate).

Parameters:

uuid (str)

Return type:

Service

read_rssi(*, timeout=5.0)¶

Read the live RSSI in dBm.

Returns:

Signal strength. Negative values close to 0 mean strong signal; -100 means barely connected.

Raises:

NotConnectedError – If the connection is closed.
BluetoothTimeoutError – If the read doesn’t complete in time.

Parameters:

timeout (float)

Return type:

int

wait_ready_to_write(*, timeout=5.0)¶

Block until peripheralIsReadyToSendWriteWithoutResponse fires (or until timeout elapses).

Pair with Characteristic.write() calls that use response=False to avoid dropped packets when the controller’s outgoing buffer fills.

Returns:: True on signal, False on timeout.
Raises:: NotConnectedError – If the connection is closed.
Parameters:: timeout (float)
Return type:: bool

wait_disconnect(*, timeout=None)¶

Block until the peripheral disconnects.

Parameters:: timeout (float | None) – Maximum wait in seconds. None waits indefinitely.
Returns:: Disconnect reason string.
Return type:: str

disconnect(*, timeout=5.0)¶

Disconnect and wait for the system to confirm.

Idempotent — calling on an already-disconnected connection returns immediately with the prior reason.

Returns:: Disconnect reason string.
Parameters:: timeout (float)
Return type:: str

class juno.bluetooth.Service¶

Bases: object

A GATT service exposed by a connected peripheral.

property uuid: str¶: The service UUID (canonical 36-char form).

property is_primary: bool¶: True for primary services, False for secondary.

property connection: Connection¶: The owning connection.

characteristics(*uuids)¶

Discover characteristics for this service.

Parameters:: *uuids (str) – Optional UUID filter (forwarded to CBPeripheral.discoverCharacteristics(_:for:)).
Return type:: list[Characteristic]

characteristic(uuid)¶

Convenience: characteristics(uuid)[0].

Raises:

KeyError – If no characteristic with that UUID exists.
ValueError – If multiple characteristics share the UUID (rare; use characteristics()).

Parameters:

uuid (str)

Return type:

Characteristic

class juno.bluetooth.Characteristic¶

Bases: object

A GATT characteristic on a service.

property properties: frozenset[str]¶: Subset of {"read", "write", "writeWithoutResponse", "notify", "indicate", "broadcast", "authenticatedSignedWrites", "extended"}.

read(*, timeout=30.0)¶

Read the current value (blocking).

Raises:

NotConnectedError – If the connection is closed.
BluetoothTimeoutError – If no value arrives in time.

Parameters:

timeout (float)

Return type:

bytes

write(data, *, response=True, timeout=30.0)¶

Write a value to the characteristic.

Parameters:

data (bytes | bytearray | memoryview) – Bytes payload.
response (bool) – When True, wait for the peripheral’s ack via didWriteValueFor:. When False, fire-and-forget (faster; pair with Connection.wait_ready_to_write() for backpressure on tight streams).
timeout (float) – Seconds to wait for the ack. Ignored when response is False.

Raises:

TypeError – If data is not bytes-like.
NotConnectedError – If the connection is closed.
BluetoothTimeoutError – If response=True and no ack arrives in time.

Return type:

None

notifications(*, subscribe_timeout=30.0)¶

Subscribe to value-change notifications.

Returns an iterator that yields raw bytes payloads. The underlying setNotifyValue:true: is sent on iteration start; setNotifyValue:false: runs on iterator close.

Pre-flight rejects characteristics without notify or indicate properties (would otherwise park the iterator indefinitely on a silent Core Bluetooth error). The wrapper also blocks for the didUpdateNotificationStateFor: ack — peer-side rejection (peripheral refused subscription) surfaces here as a BluetoothError instead of a hung iterator.

Parameters:: subscribe_timeout (float) – Seconds to wait for the system / peer to confirm the subscription. Default 30s comfortably covers slow stacks; tune down for tighter UX requirements.
Return type:: _NotificationIterator

Closing semantics:

Iterating with a for loop and break: the iterator’s close() runs (Python’s generator-close protocol), which unsubscribes.
Peer-initiated disconnect: the iterator raises DisconnectedError from the next __next__. Catch this to drive a reconnect loop.

Raises:

NotConnectedError – If the connection is closed.
BluetoothError – If the characteristic doesn’t support notifications, the peer rejects the subscription, or the confirmation times out.

Parameters:

subscribe_timeout (float)

Return type:

_NotificationIterator

descriptors(*, timeout=30.0)¶

Discover descriptors for this characteristic.

Parameters:: timeout (float)
Return type:: list[Descriptor]

class juno.bluetooth.Descriptor¶

Bases: object

A GATT descriptor on a characteristic.

Mostly relevant for advanced cases — Client Characteristic Configuration (CCCD) management is handled implicitly when you use Characteristic.notifications(). Read / write are exposed for vendor-defined descriptors and Characteristic User Description.

juno.bluetooth.authorization_status()¶

Return the current Bluetooth authorization status.

One of the AuthorizationStatus string constants.

Return type:: str

juno.bluetooth.is_authorized()¶

True iff the app has allowedAlways Bluetooth permission.

Return type:: bool

juno.bluetooth.request_authorization()¶

Provoke the Bluetooth permission prompt and block until the status resolves.

Core Bluetooth has no explicit “request” API — the prompt fires when a real radio operation runs against a freshly- instantiated central. This call instantiates the central and runs a brief throwaway scan to trigger the prompt, then polls authorization_status() until it stops being notDetermined (timeout-free; the user may dismiss the prompt minutes later).

Returns:: Resolved status, one of the AuthorizationStatus string constants.
Return type:: str

Note

Calling this method when status is already determined is a no-op — it returns the existing status without instantiating anything new.

juno.bluetooth.state()¶

Return the adapter state string.

Useful before a scan or connect to verify the radio is up:

if bt.state() != bt.State.POWERED_ON:
    raise RuntimeError("Bluetooth is off")

The very first call from a fresh process instantiates the CBCentralManager lazily, and iOS delivers the initial centralManagerDidUpdateState: callback asynchronously a few hundred microseconds later. So the first call can return “unknown” even when Bluetooth is on and authorization is granted. This wrapper polls briefly (capped at _STATE_SETTLE_TIMEOUT) so callers see the settled state instead of the transient placeholder. Subsequent calls return immediately because the central is already instantiated.

Return type:: str

juno.bluetooth.reset()¶

Cancel every in-flight scan and connection and release every open handle.

Mirrors Pythonista’s cb.reset(). Safe to call from a notebook teardown cell to return the module to a clean state for the next run.

Return type:: None

juno.bluetooth.scan(*, services=None, allow_duplicates=False, timeout=None)¶

Iterate advertisements until the iterator is closed.

Parameters:

services (list[str] | None) – If given, only peripherals advertising one of these service UUIDs are reported. Equivalent to CoreBluetooth’s scanForPeripherals(withServices:).
allow_duplicates (bool) – When False (the default), the system reports each peripheral once per scan. When True, the same peripheral can be reported multiple times with updated RSSI — useful for proximity loops at the cost of higher battery use.
timeout (float | None) – Stop scanning after this many seconds (in addition to closing the iterator on break). None means scan indefinitely.

Yields:

Advertisement records as they arrive.

Raises:

TypeError – If services is not a list of strings, or timeout is not a number.
ValueError – If a UUID is malformed or timeout is negative.
BluetoothError – If the adapter is not powered on at scan-start time.

Return type:

Iterator[Advertisement]

Note

The iterator is backed by a bounded queue with drop-oldest overflow — the underlying central queue must not block. The _ScanIterator exposes a dropped count so callers can detect a slow consumer.