bleak_retry_connector_ble_connection_with_retries_and_caching.py

1import asyncio
2from bleak import BleakScanner
3from bleak_retry_connector import establish_connection
4
5async def run():
6    device_name_or_address = "XX:XX:XX:XX:XX:XX"
7    device = await BleakScanner.find_device_by_address(device_name_or_address)
8    
9    if not device:
10        print(f"Could not find device with address {device_name_or_address}")
11        return
12
13    # establish_connection handles retries and maintains the connection
14    client = await establish_connection(
15        client_class=None, # Uses default BleakClient or platform specific one
16        device=device,
17        name=device.name or device.address,
18        disconnected_callback=lambda client: print("Disconnected"),
19    )
20
21    try:
22        print(f"Connected: {client.is_connected}")
23        # Perform BLE operations here
24    finally:
25        await client.disconnect()
26
27if __name__ == "__main__":
28    asyncio.run(run())