如何让BleakClient实例跨代码块持久化以适配REST接口？

问题描述 #

我写了一个基于bleak低功耗蓝牙客户端的Python命令行脚本，现在想把它拆成REST端点。为此需要替换async with BleakClient(d, timeout=10) as client:的写法，让client成为全局变量，这样其他REST端点也能复用，也就是连接设备的调用结束后client还能保留。

尝试用client = await BleakClient(d, timeout=10)时抛出异常：object BleakClient can't be used in 'await' expression。

附截断后的脚本（ConnectAndReadChar()可正常运行，ConnectUUT()在上述代码行触发TypeError）：

import asyncio
import traceback
from bleak import BleakScanner, BleakClient
import sys
import time

global client        

async def ConnectAndReadChar(MAC_ID):
    d = await BleakScanner.find_device_by_address(MAC_ID, timeout=20)            
    time.sleep(0.5)
    while True:
        try:
            if d.address == MAC_ID:
                async with BleakClient(d, timeout=10) as client:
                    time.sleep(0.5)
                    res = await client.pair(protection_level=None)
                    print(res)

                    x = client.is_connected
                    print("Connected: {0}".format(x))

                    svc = client.services.get_service("Service ID")

                    thisCharacteristic = svc.get_characteristic("Char ID")
                    value = bytes(await client.read_gatt_char(thisCharacteristic))
                    print("\tValue: {0} ".format(value))

                    await client.disconnect()
                    await client.unpair()
            return
        except:
            e = sys.exc_info()[0]
            print(e)


async def ConnectUUT(MAC_ID):
    global client
    d = await BleakScanner.find_device_by_address(MAC_ID, timeout=20)            
    while True:
        try:
            if d.address == MAC_ID:
                # async with BleakClient(d, timeout=10) as client:
                client = await BleakClient(d, timeout=10)
                if client is not None:
                    time.sleep(0.5)
                    res = await client.pair(protection_level=None)
                    print(res)

                    x = client.is_connected
                    print("Connected: {0}".format(x))
            return
        except:
            e = sys.exc_info()[0]
            print(e)

async def ReadFrames():
    global client

    try:
        svc = client.services.get_service("Service ID")
        thisCharacteristic = svc.get_characteristic("Char ID")
        return
    except:
        e = sys.exc_info()[0]
        print(e)

async def DisconnectUUT():
    global client

    try:
        await client.disconnect()
        await client.unpair()
    except:
        e = sys.exc_info()[0]
        print(e)
    return

async def main():
    await ConnectAndReadChar("MAC ID goes here")

    await ConnectUUT("MAC ID goes here")
    await ReadFrames()
    await DisconnectUUT()
        
if __name__ == "__main__":
    asyncio.run(main())

问题根源 #

BleakClient不是可等待对象，不能直接用await实例化。async with语法能工作是因为它内部自动调用了__aenter__方法完成连接，而直接实例化后需要手动调用connect()异步方法来建立连接。

修正后的代码 #

修复全局客户端连接逻辑 #

修改ConnectUUT函数，同时优化异常处理和状态检查：

import asyncio
import traceback
from bleak import BleakScanner, BleakClient
import sys

# 全局客户端实例
client = None        

async def ConnectAndReadChar(MAC_ID):
    d = await BleakScanner.find_device_by_address(MAC_ID, timeout=20)            
    await asyncio.sleep(0.5)  # 替换time.sleep避免阻塞事件循环
    while True:
        try:
            if d.address == MAC_ID:
                async with BleakClient(d, timeout=10) as temp_client:
                    await asyncio.sleep(0.5)
                    res = await temp_client.pair(protection_level=None)
                    print(res)

                    x = temp_client.is_connected
                    print("已连接: {0}".format(x))

                    svc = temp_client.services.get_service("Service ID")
                    thisCharacteristic = svc.get_characteristic("Char ID")
                    value = bytes(await temp_client.read_gatt_char(thisCharacteristic))
                    print("\t值: {0} ".format(value))

                    await temp_client.disconnect()
                    await temp_client.unpair()
            return
        except Exception as e:
            print(type(e), str(e))
            traceback.print_exc()


async def ConnectUUT(MAC_ID):
    global client
    d = await BleakScanner.find_device_by_address(MAC_ID, timeout=20)            
    while True:
        try:
            if d.address == MAC_ID:
                # 正确流程：实例化 → 异步连接
                client = BleakClient(d, timeout=10)
                await client.connect()  # 关键：手动调用异步连接方法
                if client.is_connected:
                    await asyncio.sleep(0.5)
                    res = await client.pair(protection_level=None)
                    print(res)

                    x = client.is_connected
                    print("已连接: {0}".format(x))
            return
        except Exception as e:
            print(type(e), str(e))
            traceback.print_exc()

async def ReadFrames():
    global client

    try:
        if not client or not client.is_connected:
            print("错误：客户端未连接")
            return None
        svc = client.services.get_service("Service ID")
        thisCharacteristic = svc.get_characteristic("Char ID")
        value = bytes(await client.read_gatt_char(thisCharacteristic))
        print("\t读取到的值: {0} ".format(value))
        return value
    except Exception as e:
        print(type(e), str(e))
        traceback.print_exc()

async def DisconnectUUT():
    global client

    try:
        if client and client.is_connected:
            await client.disconnect()
            await client.unpair()
            print("已断开连接并解除配对")
        client = None  # 重置全局变量
    except Exception as e:
        print(type(e), str(e))
        traceback.print_exc()
    return

async def main():
    # await ConnectAndReadChar("此处填入MAC地址")

    await ConnectUUT("此处填入MAC地址")
    await ReadFrames()
    await DisconnectUUT()
        
if __name__ == "__main__":
    asyncio.run(main())

REST端点适配示例（FastAPI） #

如果要改成REST服务，推荐用FastAPI（天然支持异步），示例如下：

from fastapi import FastAPI
import asyncio
from bleak import BleakScanner, BleakClient

app = FastAPI()
client = None

@app.post("/connect/{mac_id}")
async def connect_device(mac_id: str):
    """连接指定MAC地址的蓝牙设备"""
    global client
    if client and client.is_connected:
        return {"status": "error", "message": "已有设备处于连接状态"}
    
    device = await BleakScanner.find_device_by_address(mac_id, timeout=20)
    if not device or device.address != mac_id:
        return {"status": "error", "message": "未找到目标设备"}
    
    client = BleakClient(device, timeout=10)
    await client.connect()
    await client.pair(protection_level=None)
    
    return {"status": "success", "connected": client.is_connected}

@app.get("/read-characteristic")
async def read_characteristic():
    """读取指定特征值"""
    global client
    if not client or not client.is_connected:
        return {"status": "error", "message": "客户端未连接设备"}
    
    try:
        service = client.services.get_service("Service ID")
        characteristic = service.get_characteristic("Char ID")
        value = bytes(await client.read_gatt_char(characteristic)).hex()
        return {"status": "success", "value": value}
    except Exception as e:
        return {"status": "error", "message": str(e)}

@app.post("/disconnect")
async def disconnect_device():
    """断开当前连接的设备"""
    global client
    if client and client.is_connected:
        await client.disconnect()
        await client.unpair()
        client = None
        return {"status": "success", "message": "已断开连接并解除配对"}
    
    return {"status": "error", "message": "无已连接的设备"}

关键注意事项 #

• 全局client必须做好状态检查，避免在未连接时执行操作
• 异步代码中替换time.sleep()为await asyncio.sleep()，防止阻塞事件循环
• REST服务中要处理并发请求，可添加锁机制确保同一时间只有一个设备连接

内容的提问来源于stack exchange，提问作者Hamilton Woods