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 |
|---|---|
Classes¶
Name | Description |
|---|---|
Manages GATT connection operations for a device. |
|
Encryption state for connected device. |
|
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:
RuntimeError – If no connection manager is set.
ValueError – If characteristic is unknown.
- 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 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.StructEncryption 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.
- class src.bluetooth_sig.device.connected.DeviceService¶
Bases:
msgspec.StructWrapper 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¶
- src.bluetooth_sig.device.connected.logger¶