src.bluetooth_sig.device.connected

Connection-related functionality for Device.

Manages GATT connection operations for a BLE device using the composition pattern. This class is accessed via device.connected.

Based on patterns from bleak (BLEDevice + BleakClient) and real-world implementations.

Attributes

Name

Description

logger

Classes

Name

Description

DeviceConnected

Manages GATT connection operations for a device.

DeviceEncryption

Encryption state for connected device.

DeviceService

Wrapper for a discovered GATT service.

Module Contents

class src.bluetooth_sig.device.connected.DeviceConnected(mac_address: str, connection_manager: bluetooth_sig.device.connection.ConnectionManagerProtocol | None = None)

Manages GATT connection operations for a device.

Accessed via device.connected.

services

Discovered GATT services by UUID string.

encryption

Current encryption state.

Example

device = Device(mac_address=”AA:BB:CC:DD:EE:FF”, translator=translator)

# Connect and discover services await device.connected.connect() services = await device.connected.discover_services()

# Read a characteristic battery = await device.connected.read(

BluetoothUUID(“00002a19-0000-1000-8000-00805f9b34fb”)

)

# Subscribe to notifications async def on_heart_rate(value):

print(f”Heart rate: {value}”)

await device.connected.subscribe(

BluetoothUUID(“00002a37-0000-1000-8000-00805f9b34fb”), on_heart_rate,

)

await device.connected.disconnect()

cache_characteristic(char_uuid: bluetooth_sig.types.uuid.BluetoothUUID, char_instance: bluetooth_sig.gatt.characteristics.base.BaseCharacteristic[Any]) None

Store characteristic instance in services cache.

Parameters:
  • char_uuid – UUID of the characteristic.

  • char_instance – BaseCharacteristic instance to cache.

async connect(*, timeout: float = 10.0) None

Establish GATT connection.

Parameters:

timeout – Connection timeout in seconds.

Raises:

RuntimeError – If no connection manager is set.

async disconnect() None

Disconnect from device.

Raises:

RuntimeError – If no connection manager is set.

async discover_services() list[DeviceService]

Discover and cache GATT services.

Returns:

List of discovered services.

Raises:

RuntimeError – If no connection manager is set.

get_cached_characteristic(char_uuid: bluetooth_sig.types.uuid.BluetoothUUID) bluetooth_sig.gatt.characteristics.base.BaseCharacteristic[Any] | None

Get cached characteristic instance from services.

Parameters:

char_uuid – UUID of the characteristic to find.

Returns:

BaseCharacteristic instance if found, None otherwise.

async pair() None

Pair with the device.

Raises:

RuntimeError – If no connection manager is set.

async read(characteristic_uuid: bluetooth_sig.types.uuid.BluetoothUUID | str) Any

Read a characteristic value.

Parameters:

characteristic_uuid – UUID of the characteristic to read.

Returns:

Parsed characteristic value.

Raises:
async read_descriptor(descriptor_uuid: bluetooth_sig.types.uuid.BluetoothUUID | str) bytes

Read a descriptor value.

Parameters:

descriptor_uuid – UUID of the descriptor to read.

Returns:

Raw descriptor bytes.

Raises:

RuntimeError – If no connection manager is set.

async read_rssi() int

Read the RSSI (signal strength) of the connection.

Returns:

RSSI value in dBm.

Raises:

RuntimeError – If no connection manager is set.

set_disconnected_callback(callback: collections.abc.Callable[[], None]) None

Set a callback to be invoked when the device disconnects.

Parameters:

callback – Function to call when disconnection occurs.

Raises:

RuntimeError – If no connection manager is set.

async subscribe(characteristic_uuid: bluetooth_sig.types.uuid.BluetoothUUID | str, callback: collections.abc.Callable[[Any], None]) None

Subscribe to characteristic notifications.

Parameters:
  • characteristic_uuid – UUID of the characteristic to subscribe to.

  • callback – Function called with parsed value on each notification.

Raises:

RuntimeError – If no connection manager is set.

async unpair() None

Unpair from the device.

Raises:

RuntimeError – If no connection manager is set.

async unsubscribe(characteristic_uuid: bluetooth_sig.types.uuid.BluetoothUUID | str) None

Unsubscribe from characteristic notifications.

Parameters:

characteristic_uuid – UUID of the characteristic to unsubscribe from.

Raises:

RuntimeError – If no connection manager is set.

async write(characteristic_uuid: bluetooth_sig.types.uuid.BluetoothUUID | str, value: Any, *, response: bool = True) None

Write a value to a characteristic.

Parameters:
  • characteristic_uuid – UUID of the characteristic to write.

  • value – Value to write (will be encoded if characteristic is known).

  • response – Whether to wait for write response.

Raises:

RuntimeError – If no connection manager is set.

async write_descriptor(descriptor_uuid: bluetooth_sig.types.uuid.BluetoothUUID | str, data: bytes) None

Write data to a descriptor.

Parameters:
  • descriptor_uuid – UUID of the descriptor to write.

  • data – Raw bytes to write.

Raises:

RuntimeError – If no connection manager is set.

property connection_manager: bluetooth_sig.device.connection.ConnectionManagerProtocol | None

Current connection manager.

encryption
property is_connected: bool

Whether currently connected.

property mac_address: str

Device MAC address.

property mtu_size: int

Get the MTU size of the connection.

Returns:

MTU size in bytes.

Raises:

RuntimeError – If no connection manager is set.

services: dict[str, DeviceService]
class src.bluetooth_sig.device.connected.DeviceEncryption

Bases: msgspec.Struct

Encryption state for connected device.

paired

Whether the device is paired.

bonded

Whether the device is bonded (persistent pairing).

encrypted

Whether the current connection is encrypted.

bonded: bool = False
encrypted: bool = False
paired: bool = False
class src.bluetooth_sig.device.connected.DeviceService

Bases: msgspec.Struct

Wrapper for a discovered GATT service.

uuid

Service UUID.

service_class

The GATT service class, or None if unknown.

characteristics

Discovered characteristics by UUID string.

characteristics: dict[str, bluetooth_sig.gatt.characteristics.base.BaseCharacteristic[Any]]
service_class: type[bluetooth_sig.gatt.services.base.BaseGattService] | None = None
uuid: bluetooth_sig.types.uuid.BluetoothUUID
src.bluetooth_sig.device.connected.logger