Keep connection, write

[rayshen0411]

• bleak version:0.20.2
• Python version:3.11
• Operating System:Windows 11
• BlueZ version (bluetoothctl -v) in case of Linux:

Description

1. If the device is connected, how to connect to connected/paired device again without discover?
2. How can I keep the connection, and write char data in the different session?
   Example, Run continuous 3 test cases
   Test case 1 : write data
   then Test case 2 : write data
   then Test case 3 : write data
   write data without disconnection

What I Did

    if device is None:
        raise BleakError(f"A device with address {ble_address} could not be found.")

    print("connecting to device...")

    async with BleakClient(device) as client:
        #x = client.is_connected
        await client.connect()
        print("Success! ", f"Connected to {device.address}")
        #print("Connected: {0}".format(x))
        # paired = await client.pair(protection_level=2)
        # print(f"Paired: {paired}")

        for service in client.services:
            if debug:
                print("[Service] %s", service)
            for char in service.characteristics:
                if "read" in char.properties:
                    try:
                        value = await client.read_gatt_char(char.uuid)
                        if debug:
                            print(
                                "  [Characteristic] {0}: (Handle: {1}) ({2}) | Name: {3}, Value: {4} ".format(
                                    char.uuid,
                                    char.handle,
                                    ",".join(char.properties),
                                    char.description,
                                    value,
                                )
                            )
                    except Exception as e:
                        print(
                            "  [Characteristic] %s (%s), Error: %s",
                            char,
                            ",".join(char.properties),
                            e,
                        )
                else:
                    if debug:
                        print(
                            "  [Characteristic] {0}: (Handle: {1}) ({2}) | Name: {3}, Value: {4} ".format(
                                char.uuid,
                                char.handle,
                                ",".join(char.properties),
                                char.description,
                                value,
                            )
                        )

                for descriptor in char.descriptors:
                    try:
                        value = await client.read_gatt_descriptor(descriptor.handle)
                        if debug:
                            print("    [Descriptor] {0}: (Handle: {1}) | Value: {2} ".format(
                                descriptor.uuid, descriptor.handle, bytes(value)))
                    except Exception as e:
                        print("    [Descriptor] %s, Error: %s", descriptor, e)

                if char.uuid == write_uuid:
                    await write_data(client, write_uuid, read_uuid, key_press_list, key_release_list, interval)

        print("disconnecting...")
    await client.disconnect()
    print("disconnected!")

Logs

Include any relevant logs here.

Logs are essential to understand what is going on behind the scenes.

See https://bleak.readthedocs.io/en/latest/troubleshooting.html for information on how to collect debug logs.

If you receive an OS error (WinError, BleakDBusError, NSError, etc.), Wireshark logs of Bluetooth packets are required to understand the issue!

[dlech]

1. If the device is connected, how to connect to connected/paired device again without discover?

This is not currently possible. Give #367 a 👍 if you would like to see this feature implemented.

2. How can I keep the connection, and write char data in the different session?

You can avoid disconnection by not allowing the context manager to exit. The easiest way to do this is to wait for an asyncio.Event.

[rayshen0411]

this solution does not work, it's for linux, not window...

[rayshen0411]

Hi sir

for question 1, I'll check it.
for question 2, I understand that to avoid disconnection by keeping context manager to exist, but
my scenario is for test case 1, send A data, test case 2 send B data and test case 3 send C data,
all test case is one python file

[dlech]

all test case is one python file

You can handle each connection in a separate asyncio task.

[rayshen0411]

Yes, but the connection will not keep it. test every case must reconnect again.

[rayshen0411]

Here is the running task to write gatt char. data
try:
key_press_list = [Controller.snapshot_press()]
key_release_list = [Controller.button_release()]

    loop = asyncio.get_event_loop()
    # controller key press
    loop.run_until_complete(ble_utils.controller_action(address, True, key_press_list, key_release_list, 3))
except KeyboardInterrupt:
    quit()

[rayshen0411]

Here is the ble connection and write gatt char. data process

async with BleakClient(device) as client:
    x = client.is_connected
    #await client.connect()
    #print("Success! ", f"Connected to {device.address}")
    print("Connected: {0}".format(x))
    # paired = await client.pair(protection_level=2)
    # print(f"Paired: {paired}")

    for service in client.services:
        if debug:
            print("[Service] %s", service)
        for char in service.characteristics:
            if "read" in char.properties:
                try:
                    value = await client.read_gatt_char(char.uuid)
                    if debug:
                        print(
                            "  [Characteristic] {0}: (Handle: {1}) ({2}) | Name: {3}, Value: {4} ".format(
                                char.uuid,
                                char.handle,
                                ",".join(char.properties),
                                char.description,
                                value,
                            )
                        )
                except Exception as e:
                    print(
                        "  [Characteristic] %s (%s), Error: %s",
                        char,
                        ",".join(char.properties),
                        e,
                    )
            else:
                if debug:
                    print(
                        "  [Characteristic] {0}: (Handle: {1}) ({2}) | Name: {3}, Value: {4} ".format(
                            char.uuid,
                            char.handle,
                            ",".join(char.properties),
                            char.description,
                            value,
                        )
                    )

            for descriptor in char.descriptors:
                try:
                    value = await client.read_gatt_descriptor(descriptor.handle)
                    if debug:
                        print("    [Descriptor] {0}: (Handle: {1}) | Value: {2} ".format(
                            descriptor.uuid, descriptor.handle, bytes(value)))
                except Exception as e:
                    print("    [Descriptor] %s, Error: %s", descriptor, e)

            if char.uuid == write_uuid:
                await write_button_data(client, write_uuid, read_uuid, key_press_list, key_release_list, interval)

    print("disconnecting...")
await client.disconnect()
print("disconnected!")

[rayshen0411]

both codes are in 1 python file

[dlech]

This is not the full program.

[rayshen0411]

It's full program

key_press_list = []
key_release_list = []

async def adv_scan(scan_duration: int, service_uuid: str, device_name: str):
    # scan for devices
    scanner = BleakScanner(filters={"UUIDs": [service_uuid], "DuplicateData": False})

    def detection_callback(dev, advertisement_data):
        print("detected:", dev, " – RSSI:", dev.rssi, " – adv data: ", advertisement_data)

    scanner.register_detection_callback(detection_callback)
    await scanner.start()
    await asyncio.sleep(scan_duration)
    await scanner.stop()
    scanned_devices = await scanner.get_discovered_devices()
    print(f'Device discovery filtered output:')
    for device in scanned_devices:
        dev_dump(device)
        devices_dict[device.address] = []
        devices_dict[device.address].append(device.name)
        devices_dict[device.address].append(device.metadata["uuids"])

        if device.name == device_name:
            devices_addr_list.append(device.address)

async def write_button_data(client: BleakClient, uuid_write: str, uuid_read: str,
                            key_press_list: list, key_release_list: list, interval: int):
    if len(key_press_list) != len(key_release_list):
        raise ValueError("press and release must be a pair")

    for i in range(0, len(key_press_list)):
        print("write button press data...")
        await client.write_gatt_char(uuid_write, Controller.set_cbih())
        print_hex(Controller.set_cbih())
        await client.write_gatt_char(uuid_write, key_press_list[i])
        print_hex(key_press_list[i])

        key_press_status = await client.read_gatt_char(uuid_read)
        if Status.response_status(key_press_status[7]) == Status.COMMAND_SUCCESSFUL:
            print("start notify...")
            await client.start_notify(uuid_read, notification_handler)
            await asyncio.sleep(3.0)
            await client.stop_notify(uuid_read)

        print("write button release data...")
        await client.write_gatt_char(uuid_write, Controller.set_cbih())
        print_hex(Controller.set_cbih())
        await client.write_gatt_char(uuid_write, key_release_list[i])
        print_hex(key_release_list[i])

        key_release_status = await client.read_gatt_char(uuid_read)
        if Status.response_status(key_release_status[7]) == Status.COMMAND_SUCCESSFUL:
            print("start notify...")
            await client.start_notify(uuid_read, notification_handler)
            await asyncio.sleep(3.0)
            await client.stop_notify(uuid_read)

        await asyncio.sleep(interval)

async def controller_action(ble_address=None, debug=False,
                            key_press_list=None, key_release_list=None, interval=0):
    device = await BleakScanner.find_device_by_address(ble_address)

    if device is None:
        raise BleakError(f"A device with address {ble_address} could not be found.")

    print("connecting to device...")

    async with BleakClient(device) as client:
        x = client.is_connected
        #await client.connect()
        #print("Success! ", f"Connected to {device.address}")
        print("Connected: {0}".format(x))
        # paired = await client.pair(protection_level=2)
        # print(f"Paired: {paired}")

        for service in client.services:
            if debug:
                print("[Service] %s", service)
            for char in service.characteristics:
                if "read" in char.properties:
                    try:
                        value = await client.read_gatt_char(char.uuid)
                        if debug:
                            print(
                                "  [Characteristic] {0}: (Handle: {1}) ({2}) | Name: {3}, Value: {4} ".format(
                                    char.uuid,
                                    char.handle,
                                    ",".join(char.properties),
                                    char.description,
                                    value,
                                )
                            )
                    except Exception as e:
                        print(
                            "  [Characteristic] %s (%s), Error: %s",
                            char,
                            ",".join(char.properties),
                            e,
                        )
                else:
                    if debug:
                        print(
                            "  [Characteristic] {0}: (Handle: {1}) ({2}) | Name: {3}, Value: {4} ".format(
                                char.uuid,
                                char.handle,
                                ",".join(char.properties),
                                char.description,
                                value,
                            )
                        )

                for descriptor in char.descriptors:
                    try:
                        value = await client.read_gatt_descriptor(descriptor.handle)
                        if debug:
                            print("    [Descriptor] {0}: (Handle: {1}) | Value: {2} ".format(
                                descriptor.uuid, descriptor.handle, bytes(value)))
                    except Exception as e:
                        print("    [Descriptor] %s, Error: %s", descriptor, e)

                if char.uuid == write_uuid:
                    await write_button_data(client, write_uuid, read_uuid, key_press_list, key_release_list, interval)

        print("disconnecting...")
    await client.disconnect()
    print("disconnected!")

if __name__ == "__main__":
    # scan device
    loop = asyncio.get_event_loop()
    loop.run_until_complete(ble_utils.adv_scan(3, uuids.Controller.uuid128_dict[uuids.Controller.SERVICE_UUID], "simulator"))
    address = ble_utils.get_ble_address()

    try:
        key_press_list = [Controller.snapshot_press()]
        key_release_list = [Controller.button_release()]

        loop = asyncio.get_event_loop()
        # controller key press
        loop.run_until_complete(ble_utils.controller_action(address, True, key_press_list, key_release_list, 3))
    except KeyboardInterrupt:
        quit()

[rayshen0411]

so looks like bleak can not do that what I mentioned of 2 questions currently?