使用Adafruit CircuitPython BLE如何注册监听BLE外设状态变更通知？

问题原因 #

你的代码存在两个核心问题：

1. 特征值定义未声明NOTIFY属性，Adafruit BLE库不会主动为该特征注册通知监听，本地缓存的特征值不会自动更新，所以每次读的都是初始连接时的取值
2. async函数内使用阻塞的time.sleep()，会卡住BLE底层的事件处理逻辑，就算有通知上报也无法被处理

修复步骤 #

1. 修改switch_service.py #

在特征值定义时声明支持的属性，代码如下：

from adafruit_ble.uuid import VendorUUID
from adafruit_ble import Service
from adafruit_ble.characteristics.int import Uint8Characteristic

class SwitchService(Service):
    uuid = VendorUUID('8158b2fd-94e4-4ff5-a99d-9a7980e998d7')

    # 新增properties参数声明支持READ和NOTIFY
    switch_characteristic = Uint8Characteristic(
        uuid=VendorUUID("8158b2fe-94e4-4ff5-a99d-9a7980e998d7"),
        properties=(Uint8Characteristic.READ | Uint8Characteristic.NOTIFY)
    )

    def __init__(self, service=None):
        super().__init__(service=service)
        self.status = self.switch_characteristic

2. 修改observe_ble_switch.py #

替换主动轮询逻辑为通知监听，把阻塞sleep换成异步sleep：

#!/usr/bin/env python3
import asyncio
from switch_service import SwitchService

from adafruit_ble import BLERadio
from adafruit_ble.advertising.standard import Advertisement

ble = BLERadio()

async def main():
    device_name = "My Arduino"
    device_found = False
    ble_device_connection = None

    print("Scanning for %r" % device_name)

    while not device_found:
        print("...")
        for adv in ble.start_scan(Advertisement, timeout=5):
            name = adv.complete_name
            if not name:
                continue
            if name.strip("\x00") == device_name:
                ble.stop_scan()
                device_found = True
                print("%r found!" % name)
                ble_device_connection = ble.connect(adv)
                break

    if ble_device_connection and ble_device_connection.connected:
        print("Connected to %r!" % name)
        if SwitchService in ble_device_connection:
            print("switch service available")
            switch_service = ble_device_connection[SwitchService]
            # 启用通知监听，异步迭代特征值变更
            async for status_val in switch_service.switch_characteristic:
                print(f"status {status_val}")
                # 如果需要额外的间隔处理可以用asyncio.sleep，不需要可以删掉
                await asyncio.sleep(0.3)
        else:
            print("switch service not available")

if __name__ == "__main__":
    asyncio.run(main())

验证逻辑 #

修改后程序会自动订阅特征值的NOTIFY事件，外设状态变更时会主动上报给树莓派，不需要主动轮询就能拿到最新状态，事件驱动的方式也更节省资源。

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