如何实现BLE无Notify特征值循环读取及并行代码调用？

针对你遇到的BLE特征无Notify标志、无法使用通知机制的问题，以下是两种方案的具体实现，解决异步与同步代码冲突的核心问题：

方案一：循环读取并缓存特征值 #

通过异步任务持续读取特征值并更新线程安全的缓存，其他并行代码可随时获取最新数据。

修改后的完整代码 #

import asyncio
from bleak import BleakClient, BleakScanner
import threading
from typing import Optional

class Connection:
    client: Optional[BleakClient] = None

    def __init__(
        self,
        loop: asyncio.AbstractEventLoop,
        read_characteristic: str,
        write_characteristic: str,
        read_interval: float = 1.0
    ):
        self.loop = loop
        self.read_characteristic = read_characteristic
        self.write_characteristic = write_characteristic
        self.read_interval = read_interval

        self.connected = False
        self.connected_device = None
        self._data = bytearray(b'')
        self._data_lock = threading.Lock()  # 线程安全锁，兼容同步/异步访问

    @property
    def data(self):
        # 线程安全的属性读取
        with self._data_lock:
            return self._data.copy()

    @data.setter
    def data(self, value):
        with self._data_lock:
            self._data = value

    async def cleanup(self):
        if self.client:
            await self.client.disconnect()
            self.connected = False

    async def manager(self):
        print("Starting connection manager.")
        while True:
            if self.client:
                await self.connect()
            else:
                await self.select_device()
                await asyncio.sleep(5.0)

    async def connect(self):
        if self.connected:
            return
        try:
            print('try connecting')
            await self.client.connect()
            self.connected = self.client.is_connected
            if self.connected:
                print(f"Connected to {self.connected_device.name}")
                # 启动独立的循环读取任务
                asyncio.create_task(self._continuous_read())
                # 维持连接检测
                while self.connected:
                    if not self.client.is_connected:
                        self.connected = False
                        print("Device disconnected.")
                        break
                    await asyncio.sleep(1.0)
            else:
                print(f"Failed to connect to {self.connected_device.name}")
        except Exception as e:
            print(f"Connection error: {e}")
            self.connected = False
            await asyncio.sleep(5.0)

    async def _continuous_read(self):
        """内部循环读取任务"""
        while self.connected:
            try:
                new_data = await self.client.read_gatt_char(self.read_characteristic)
                self.data = new_data
                print('[continuous_read] DATA:', self.data)
            except Exception as e:
                print(f"Read error: {e}")
                self.connected = False
                break
            await asyncio.sleep(self.read_interval)

    async def read_char(self):
        """单次读取方法，兼容方案二"""
        if not self.connected or not self.client:
            raise ConnectionError("Device not connected")
        self.data = await self.client.read_gatt_char(self.read_characteristic)
        print('[read_char] DATA:', self.data)
        return self.data

    async def reset_ah(self):
        ah_reset_value = bytearray([0xee, 0xff, 0xc0, 0x0b])
        await self.client.write_gatt_char(self.write_characteristic, ah_reset_value)

    async def reset_wh(self):
        wh_reset_value = bytearray([0xee, 0xff, 0xc0, 0x0c])
        await self.client.write_gatt_char(self.write_characteristic, wh_reset_value)

    async def select_device(self):
        print("Bluetooh LE hardware warming up...")
        await asyncio.sleep(2.0)
        devices = await BleakScanner.discover()

        print("Please select device: ")
        response = -1
        while response == -1:
            for i, device in enumerate(devices):
                print(f"{i}: {device.name}")
                if device.name == 'DCPM':
                    response = i
            if response == -1:
                devices = await BleakScanner.discover()

        print("Select device: ", response)
        print(f"Connecting to {devices[response].name}")
        self.connected_device = devices[response]
        self.client = BleakClient(devices[response].address, loop=self.loop)

# 并行任务示例：获取缓存数据
async def parallel_task(connection: Connection):
    while True:
        if connection.connected:
            current_data = connection.data
            print('[parallel_task] Got data:', current_data)
            # 在此添加你的业务处理逻辑
        await asyncio.sleep(2.0)

async def main():
    loop = asyncio.get_event_loop()
    connection = Connection(loop, read_characteristic, write_characteristic)
    # 启动连接管理器
    asyncio.create_task(connection.manager())
    # 启动并行数据处理任务
    asyncio.create_task(parallel_task(connection))
    # 保持事件循环运行
    while True:
        await asyncio.sleep(1.0)

read_characteristic = "7a95ce01-0ea8-1bcc-71a2-fc7539b81c9c"
write_characteristic = "7a95ce03-0ea8-1bcc-71a2-fc7539b81c9c"

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main())
    except KeyboardInterrupt:
        print("\nUser stopped program.")
    finally:
        print("Disconnecting...")
        loop.run_until_complete(connection.cleanup())

关键修改说明 #

• 新增_continuous_read异步任务，连接成功后自动循环读取特征值，通过带锁的data属性更新缓存，确保线程安全。
• 用@property实现线程安全的数据访问，同步代码也可通过connection.data安全获取最新值。
• 分离连接逻辑与读取逻辑，connect方法仅负责维持连接状态，避免阻塞读取任务。

方案二：按需调用读取方法 #

直接调用read_char方法获取实时特征值，需在异步上下文执行，确保设备已连接。

使用示例 #

# 按需读取并处理数据的异步任务
async def process_data_on_demand(connection: Connection):
    while True:
        user_input = input("Enter 'read' to get data, 'exit' to quit: ")
        if user_input.lower() == 'exit':
            break
        elif user_input.lower() == 'read':
            try:
                data = await connection.read_char()
                # 解析字节数据示例
                parsed_value = int.from_bytes(data, byteorder='little')
                print(f"Parsed value: {parsed_value}")
                # 在此添加后续处理逻辑
            except ConnectionError as e:
                print(e)
        await asyncio.sleep(0.1)

核心注意事项 #

• 所有BLE操作必须在异步函数内执行，同步代码不能直接调用异步方法，可通过loop.call_soon_threadsafe或loop.run_until_complete包装调用。
• 调用前需检查connection.connected状态，避免未连接时触发异常。

异步同步冲突解决要点 #

• 所有BLE相关操作（连接、读写）必须放在异步上下文，同步代码需通过事件循环调度异步任务。
• 共享数据访问必须加锁，避免多任务竞争导致的数据异常。
• 长期运行的任务（连接管理、循环读取）拆分为独立异步任务，通过事件循环并行调度，避免阻塞主逻辑。

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