Bleak 使用指北

发现任意蓝牙设备

import asyncio
from bleak import discover

async def discover_bluetooth():
    devices = await discover(timeout=5.0)
    for d in devices:
        print(d)

asyncio.run(discover_bluetooth())

或者

import asyncio
from bleak import BleakScanner

async def discover_bluetooth():
    devices = await BleakScanner.discover()
    for d in devices:
        print(d)
        
asyncio.run(discover_bluetooth())

通过地址发现指定设备

import asyncio
import platform
from bleak import BleakScanner

ADDRESS = (
    '08:B9:5F:5F:6F:47'
    if platform.system() == 'Linux'
    else '074A9728-EC7E-5E5A-5DDB-DF54AA486B90'
)

async def discover_address(address):
    device = await BleakScanner.find_device_by_address(address)
    print(device)

asyncio.run(discover_address(ADDRESS))

连接设备并显示服务

import asyncio
import platform
from bleak import BleakClient, BleakScanner
from bleak.exc import BleakError

ADDRESS = (
    '08:B9:5F:5F:6F:47'
    if platform.system() == 'Linux'
    else '074A9728-EC7E-5E5A-5DDB-DF54AA486B90'
)

async def connect(address):
    device = await BleakScanner.find_device_by_address(address, timeout=20.0)
    if not device:
        raise BleakError(f"A device with address {address} could not be found.")
    async with BleakClient(device) as client:
        services = await client.get_services()
        print("Services:")
        for service in services:
            print(service)

asyncio.run(connect(ADDRESS))

连接设备并显示服务以及特征值

import asyncio
import platform
from bleak import BleakClient

async def connect_char(address):
    async with BleakClient(address) as client:
        print(f"Connected: {client.is_connected}")

        for service in client.services:
            print(f"[Service] {service}")
            for char in service.characteristics:
                if "read" in char.properties:
                    try:
                        value = bytes(await client.read_gatt_char(char.uuid))
                        print(
                            f"\t[Characteristic] {char} ({','.join(char.properties)}), Value: {value}"
                        )
                    except Exception as e:
                        print(
                            f"\t[Characteristic] {char} ({','.join(char.properties)}), Value: {e}"
                        )

                else:
                    value = None
                    print(
                        f"\t[Characteristic] {char} ({','.join(char.properties)}), Value: {value}"
                    )

                for descriptor in char.descriptors:
                    try:
                        value = bytes(
                            await client.read_gatt_descriptor(descriptor.handle)
                        )
                        print(f"\t\t[Descriptor] {descriptor}) | Value: {value}")
                    except Exception as e:
                        print(f"\t\t[Descriptor] {descriptor}) | Value: {e}")

asyncio.run(connect_char(ADDRESS))

扫描 Beacon 广播

参考：

• scan_iterator
• BlueMaestroAPI
• 扫描蓝牙设备并获取原始广播数据的功能实现

import asyncio
from bleak import BleakScanner

async def discover_ad():
    async with BleakScanner() as scanner:
        print("Scanning...")

        print(f"advertisement packets:")
        async for bd, ad in scanner.advertisement_data():
            print(f"{bd!r} with {ad!r}")

asyncio.run(discover_ad())

或者

import binascii
import asyncio
from bleak import BleakScanner
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData

def callback(device: BLEDevice, advertisement_data: AdvertisementData):
    manufacturer_data = advertisement_data.manufacturer_data
    if not manufacturer_data:
        return
    for key, value in manufacturer_data.items():
        # manufacturer_data服务下原始数据，bytes转16进制再转字符串
        value_string = binascii.hexlify(value).decode("utf-8")
        print(f'{key} :{value_string}')

    # for key in manufacturer_data.keys():
    # for value in manufacturer_data.values():

    service_uuid = advertisement_data.service_uuids
    # 空 list 本身等同于 False
    if service_uuid:
        print(service_uuid)

    # if not service_uuid:
    #     print('List Is Empty')

    local_name = advertisement_data.local_name
    if local_name is not None:
        print(local_name)

    service_data = advertisement_data.service_data
    if not service_data:
        return

    for key, value in service_data.items():
        # service_data服务下原始数据，bytes转16进制再转字符串
        value_string = binascii.hexlify(value).decode("utf-8")
        print(f'{key} :{value_string}')

async def discover_ad():
    scanner = BleakScanner()
    scanner.register_detection_callback(callback)
    await scanner.start()
    await asyncio.sleep(100.0)
    await scanner.stop()

asyncio.run(discover_ad())