esp-idf/tools/esp_prov/transport/ble_cli.py

# SPDX-FileCopyrightText: 2018-2023 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
#

import platform

from utils import hex_str_to_bytes, str_to_bytes

fallback = True


# Check if required packages are installed
# else fallback to console mode
try:
    import bleak
    fallback = False
except ImportError:
    pass


# --------------------------------------------------------------------

def device_sort(device):
    return device[0].address


class BLE_Bleak_Client:
    def __init__(self):
        self.adapter = None
        self.adapter_props = None
        self.characteristics = dict()
        self.chrc_names = None
        self.device = None
        self.devname = None
        self.iface = None
        self.nu_lookup = None
        self.services = None
        self.srv_uuid_adv = None
        self.srv_uuid_fallback = None

    async def connect(self, devname, iface, chrc_names, fallback_srv_uuid):
        self.devname = devname
        self.srv_uuid_fallback = fallback_srv_uuid
        self.chrc_names = [name.lower() for name in chrc_names]
        self.iface = iface

        print('Discovering...')
        try:
            discovery = await bleak.BleakScanner.discover(return_adv=True)
            devices = list(discovery.values())
        except bleak.exc.BleakDBusError as e:
            if str(e) == '[org.bluez.Error.NotReady] Resource Not Ready':
                raise RuntimeError('Bluetooth is not ready. Maybe try `bluetoothctl power on`?')
            raise

        found_device = None

        if self.devname is None:
            if len(devices) == 0:
                print('No devices found!')
                exit(1)

            while True:
                devices.sort(key=device_sort)
                print('==== BLE Discovery results ====')
                print('{0: >4} {1: <33} {2: <12}'.format(
                    'S.N.', 'Name', 'Address'))
                for i, _ in enumerate(devices):
                    print('[{0: >2}] {1: <33} {2: <12}'.format(i + 1, devices[i][0].name or 'Unknown', devices[i][0].address))

                while True:
                    try:
                        select = int(input('Select device by number (0 to rescan) : '))
                        if select < 0 or select > len(devices):
                            raise ValueError
                        break
                    except ValueError:
                        print('Invalid input! Retry')

                if select != 0:
                    break

                discovery = await bleak.BleakScanner.discover(return_adv=True)
                devices = list(discovery.values())

            self.devname = devices[select - 1][0].name
            found_device = devices[select - 1]
        else:
            for d in devices:
                if d[0].name == self.devname:
                    found_device = d

        if not found_device:
            raise RuntimeError('Device not found')

        uuids = found_device[1].service_uuids
        # There should be 1 service UUID in advertising data
        # If bluez had cached an old version of the advertisement data
        # the list of uuids may be incorrect, in which case connection
        # or service discovery may fail the first time. If that happens
        # the cache will be refreshed before next retry
        if len(uuids) == 1:
            self.srv_uuid_adv = uuids[0]

        print('Connecting...')
        self.device = bleak.BleakClient(found_device[0].address)
        await self.device.connect()
        # must be paired on Windows to access characteristics;
        # cannot be paired on Mac
        if platform.system() == 'Windows':
            await self.device.pair()

        print('Getting Services...')
        services = self.device.services

        service = services[self.srv_uuid_adv] or services[self.srv_uuid_fallback]
        if not service:
            await self.device.disconnect()
            self.device = None
            raise RuntimeError('Provisioning service not found')

        nu_lookup = dict()
        for characteristic in service.characteristics:
            for descriptor in characteristic.descriptors:
                if descriptor.uuid[4:8] != '2901':
                    continue
                readval = await self.device.read_gatt_descriptor(descriptor.handle)
                found_name = ''.join(chr(b) for b in readval).lower()
                nu_lookup[found_name] = characteristic.uuid
                self.characteristics[characteristic.uuid] = characteristic

        match_found = True
        for name in self.chrc_names:
            if name not in nu_lookup:
                # Endpoint name not present
                match_found = False
                break

        # Create lookup table only if all endpoint names found
        self.nu_lookup = [None, nu_lookup][match_found]

        return True

    def get_nu_lookup(self):
        return self.nu_lookup

    def has_characteristic(self, uuid):
        print('checking for characteristic ' + uuid)
        if uuid in self.characteristics:
            return True
        return False

    async def disconnect(self):
        if self.device:
            print('Disconnecting...')
            if platform.system() == 'Windows':
                await self.device.unpair()
            await self.device.disconnect()
            self.device = None
            self.nu_lookup = None
            self.characteristics = dict()

    async def send_data(self, characteristic_uuid, data):
        await self.device.write_gatt_char(characteristic_uuid, bytearray(data.encode('latin-1')), True)
        readval = await self.device.read_gatt_char(characteristic_uuid)
        return ''.join(chr(b) for b in readval)
# --------------------------------------------------------------------


# Console based BLE client for Cross Platform support
class BLE_Console_Client:
    async def connect(self, devname, iface, chrc_names, fallback_srv_uuid):
        print('BLE client is running in console mode')
        print('\tThis could be due to your platform not being supported or dependencies not being met')
        print('\tPlease ensure all pre-requisites are met to run the full fledged client')
        print('BLECLI >> Please connect to BLE device `' + devname + '` manually using your tool of choice')
        resp = input('BLECLI >> Was the device connected successfully? [y/n] ')
        if resp != 'Y' and resp != 'y':
            return False
        print('BLECLI >> List available attributes of the connected device')
        resp = input("BLECLI >> Is the service UUID '" + fallback_srv_uuid + "' listed among available attributes? [y/n] ")
        if resp != 'Y' and resp != 'y':
            return False
        return True

    def get_nu_lookup(self):
        return None

    def has_characteristic(self, uuid):
        resp = input("BLECLI >> Is the characteristic UUID '" + uuid + "' listed among available attributes? [y/n] ")
        if resp != 'Y' and resp != 'y':
            return False
        return True

    async def disconnect(self):
        pass

    async def send_data(self, characteristic_uuid, data):
        print("BLECLI >> Write following data to characteristic with UUID '" + characteristic_uuid + "' :")
        print('\t>> ' + str_to_bytes(data).hex())
        print('BLECLI >> Enter data read from characteristic (in hex) :')
        resp = input('\t<< ')
        return hex_str_to_bytes(resp)


# --------------------------------------------------------------------


# Function to get client instance depending upon platform
def get_client():
    if fallback:
        return BLE_Console_Client()
    return BLE_Bleak_Client()