tessl/pypi-bleak

Cross-platform Bluetooth Low Energy GATT client library for asynchronous BLE communication

Overall
score

97%

Overview

Eval results

Files

GATT Client Operations

Name: tessl/pypi-bleak
Author: tessl

Bleak's BleakClient provides comprehensive GATT client functionality for connecting to BLE devices and performing operations on services, characteristics, and descriptors. It supports async context management and provides a unified API across all platforms.

Capabilities

Client Initialization and Configuration

Initialize GATT clients with device specifications, connection parameters, and platform-specific options.

class BleakClient:
    def __init__(
        self,
        address_or_ble_device: Union[BLEDevice, str],
        disconnected_callback: Optional[Callable[[BleakClient], None]] = None,
        services: Optional[Iterable[str]] = None,
        *,
        timeout: float = 10.0,
        pair: bool = False,
        winrt: WinRTClientArgs = {},
        backend: Optional[type[BaseBleakClient]] = None,
        **kwargs: Any,
    ) -> None:
        """
        Initialize GATT client.
        
        Args:
            address_or_ble_device: BLEDevice from scanner or Bluetooth address string
            disconnected_callback: Function called when device disconnects
            services: Optional list of service UUIDs to resolve
            timeout: Connection timeout in seconds
            pair: Attempt pairing before connecting (not supported on macOS)
            winrt: Windows-specific client arguments
            backend: Custom backend implementation
        """

Connection Management

Establish and manage GATT connections with async context manager support for automatic lifecycle management.

async def connect(self, **kwargs: Any) -> None:
    """
    Connect to the GATT server.
    
    Args:
        **kwargs: Backward compatibility arguments (should not be used)
    """

async def disconnect(self) -> None:
    """Disconnect from the GATT server."""

async def __aenter__(self) -> Self:
    """Async context manager entry - connects to device."""

async def __aexit__(
    self,
    exc_type: Optional[type[BaseException]],
    exc_val: Optional[BaseException],
    exc_tb: Optional[TracebackType],
) -> None:
    """Async context manager exit - disconnects from device."""

Device Information Access

Access basic device information and connection status.

@property
def name(self) -> str:
    """Human-readable name for the peripheral device."""

@property
def address(self) -> str:
    """Bluetooth address of this device (UUID on macOS)."""

@property
def mtu_size(self) -> int:
    """Negotiated MTU size in bytes for the active connection."""

@property
def is_connected(self) -> bool:
    """Check connection status between client and GATT server."""

Pairing and Security

Manage device pairing and security operations.

async def pair(self, *args: Any, **kwargs: Any) -> None:
    """
    Pair with the GATT server.
    
    Note: Not available on macOS. User will be prompted automatically
    when accessing characteristics requiring authentication.
    
    Args:
        *args, **kwargs: Backend-specific pairing arguments
    """

async def unpair(self) -> None:
    """
    Unpair from the GATT server.
    
    Note: Only available on Windows and Linux. Unpairing also disconnects.
    """

Service Discovery

Access the device's GATT service hierarchy after connection.

@property
def services(self) -> BleakGATTServiceCollection:
    """
    Collection of GATT services available on the device.
    
    Returns:
        BleakGATTServiceCollection containing all discovered services
        
    Raises:
        BleakError: If service discovery has not been performed
    """

Characteristic Operations

Read from and write to GATT characteristics with support for different write modes.

async def read_gatt_char(
    self,
    char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
    **kwargs: Any,
) -> bytearray:
    """
    Read data from a GATT characteristic.
    
    Args:
        char_specifier: Characteristic specified by handle, UUID, or object
        **kwargs: Backend-specific read arguments
        
    Returns:
        Bytearray containing the read data
        
    Raises:
        BleakCharacteristicNotFoundError: If characteristic not found
    """

async def write_gatt_char(
    self,
    char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
    data: Buffer,
    response: Optional[bool] = None,
) -> None:
    """
    Write data to a GATT characteristic.
    
    Args:
        char_specifier: Characteristic specified by handle, UUID, or object
        data: Data to write (supports buffer protocol)
        response: True for write-with-response, False for write-without-response,
                 None for automatic selection based on characteristic properties
                 
    Raises:
        BleakCharacteristicNotFoundError: If characteristic not found
    """

Notification and Indication Handling

Subscribe to and manage notifications and indications from GATT characteristics.

async def start_notify(
    self,
    char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
    callback: Callable[
        [BleakGATTCharacteristic, bytearray], Union[None, Awaitable[None]]
    ],
    *,
    cb: CBStartNotifyArgs = {},
    **kwargs: Any,
) -> None:
    """
    Activate notifications/indications on a characteristic.
    
    Args:
        char_specifier: Characteristic specified by handle, UUID, or object
        callback: Function called when notifications are received
        cb: CoreBluetooth-specific notification arguments
        **kwargs: Backend-specific notification arguments
        
    Raises:
        BleakCharacteristicNotFoundError: If characteristic not found
        BleakError: If not connected
    """

async def stop_notify(
    self, char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID]
) -> None:
    """
    Deactivate notifications/indications on a characteristic.
    
    Args:
        char_specifier: Characteristic specified by handle, UUID, or object
        
    Raises:
        BleakCharacteristicNotFoundError: If characteristic not found
    """

Descriptor Operations

Read from and write to GATT descriptors for advanced characteristic configuration.

async def read_gatt_descriptor(
    self,
    desc_specifier: Union[BleakGATTDescriptor, int],
    **kwargs: Any,
) -> bytearray:
    """
    Read data from a GATT descriptor.
    
    Args:
        desc_specifier: Descriptor specified by handle or object
        **kwargs: Backend-specific read arguments
        
    Returns:
        Bytearray containing the read data
        
    Raises:
        BleakError: If descriptor not found
    """

async def write_gatt_descriptor(
    self,
    desc_specifier: Union[BleakGATTDescriptor, int],
    data: Buffer,
) -> None:
    """
    Write data to a GATT descriptor.
    
    Args:
        desc_specifier: Descriptor specified by handle or object
        data: Data to write (supports buffer protocol)
        
    Raises:
        BleakError: If descriptor not found
    """

Usage Examples

Basic Connection and Reading

import asyncio
from bleak import BleakClient

async def connect_and_read():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    
    async with BleakClient(address) as client:
        print(f"Connected to {client.name}")
        
        # Read device name characteristic
        device_name_uuid = "00002a00-0000-1000-8000-00805f9b34fb"
        name_bytes = await client.read_gatt_char(device_name_uuid)
        print(f"Device name: {name_bytes.decode()}")
        
        # Read battery level if available
        battery_uuid = "00002a19-0000-1000-8000-00805f9b34fb"
        try:
            battery_bytes = await client.read_gatt_char(battery_uuid)
            battery_level = int.from_bytes(battery_bytes, byteorder='little')
            print(f"Battery level: {battery_level}%")
        except Exception as e:
            print(f"Battery level not available: {e}")

asyncio.run(connect_and_read())

Service and Characteristic Exploration

import asyncio
from bleak import BleakClient

async def explore_services():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    
    async with BleakClient(address) as client:
        print(f"Connected to {client.name}")
        print(f"MTU Size: {client.mtu_size}")
        
        # Explore all services and characteristics
        for service in client.services:
            print(f"\nService: {service.uuid} - {service.description}")
            
            for char in service.characteristics:
                print(f"  Char: {char.uuid} - {char.description}")
                print(f"    Properties: {char.properties}")
                print(f"    Handle: {char.handle}")
                
                # List descriptors
                for desc in char.descriptors:
                    print(f"    Desc: {desc.uuid} - {desc.description}")

asyncio.run(explore_services())

Writing Data

import asyncio
from bleak import BleakClient

async def write_data():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    char_uuid = "12345678-1234-5678-9012-123456789abc"  # Replace with actual UUID
    
    async with BleakClient(address) as client:
        print(f"Connected to {client.name}")
        
        # Write with response (reliable)
        data = b"Hello, BLE!"
        await client.write_gatt_char(char_uuid, data, response=True)
        print("Data written with response")
        
        # Write without response (faster)
        await client.write_gatt_char(char_uuid, data, response=False)
        print("Data written without response")

asyncio.run(write_data())

Notifications

import asyncio
from bleak import BleakClient, BleakGATTCharacteristic

def notification_handler(characteristic: BleakGATTCharacteristic, data: bytearray):
    """Handle incoming notifications."""
    print(f"Notification from {characteristic.uuid}: {data.hex()}")

async def handle_notifications():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    notify_uuid = "12345678-1234-5678-9012-123456789abc"  # Replace with actual UUID
    
    async with BleakClient(address) as client:
        print(f"Connected to {client.name}")
        
        # Start notifications
        await client.start_notify(notify_uuid, notification_handler)
        print("Notifications started")
        
        # Keep connection alive to receive notifications
        await asyncio.sleep(30)
        
        # Stop notifications (automatic on disconnect)
        await client.stop_notify(notify_uuid)
        print("Notifications stopped")

asyncio.run(handle_notifications())

Async Notification Callbacks

import asyncio
from bleak import BleakClient, BleakGATTCharacteristic

async def async_notification_handler(characteristic: BleakGATTCharacteristic, data: bytearray):
    """Async notification handler for complex processing."""
    print(f"Processing notification from {characteristic.uuid}")
    
    # Simulate async processing
    await asyncio.sleep(0.1)
    
    value = int.from_bytes(data, byteorder='little', signed=False)
    print(f"Processed value: {value}")

async def async_notifications():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    notify_uuid = "12345678-1234-5678-9012-123456789abc"  # Replace with actual UUID
    
    async with BleakClient(address) as client:
        await client.start_notify(notify_uuid, async_notification_handler)
        await asyncio.sleep(30)  # Receive notifications for 30 seconds

asyncio.run(async_notifications())

Connection with Pairing

import asyncio
from bleak import BleakClient

async def connect_with_pairing():
    address = "00:11:22:33:44:55"  # Replace with actual device address
    
    # Enable pairing during connection (not supported on macOS)
    client = BleakClient(address, pair=True, timeout=30.0)
    
    async with client:
        print(f"Connected and paired with {client.name}")
        
        # Perform operations that might require authentication
        secured_char_uuid = "12345678-1234-5678-9012-123456789abc"
        try:
            data = await client.read_gatt_char(secured_char_uuid)
            print(f"Secured data: {data.hex()}")
        except Exception as e:
            print(f"Could not read secured characteristic: {e}")

# Only run on platforms that support explicit pairing
import platform
if platform.system() != "Darwin":  # Not macOS
    asyncio.run(connect_with_pairing())

Types

# Buffer type for write operations
Buffer = Union[bytes, bytearray, memoryview]

# Platform-specific client arguments
class WinRTClientArgs(TypedDict, total=False):
    address_type: Literal["public", "random"]
    use_cached_services: bool

class CBStartNotifyArgs(TypedDict, total=False):
    notification_discriminator: Optional[NotificationDiscriminator]

# Notification discriminator for CoreBluetooth
NotificationDiscriminator = Callable[[bytes], bool]

# Callback types
DisconnectedCallback = Callable[[BleakClient], None]
NotificationCallback = Callable[
    [BleakGATTCharacteristic, bytearray], 
    Union[None, Awaitable[None]]
]

Install with Tessl CLI

npx tessl i tessl/pypi-bleak