如何注册以通知BLE外设的变化?

Question

我有一个具有单一服务和单一特性的BLE外围设备(Arduino nano)。该特性持有一个8位的单位，当开关打开时设置为1，当开关关闭时设置为0。该特性支持READ和NOTIFY。

使用nRF Connect，我可以看到NOTIFY部分正在工作，因为当交换机状态发生变化时，值正在更新。

但我真正想做的是使用一个树莓派作为中心设备，使用Adafruit CircuitPython BLE。

按照Adafruit CircuitPython BLE存储库中的示例，我创建了下面的简单程序:

observe_ble_switch.py

#!/usr/bin/env python3
import asyncio
import time
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]
while ble_device_connection.connected:
print("status %r" % switch_service.read_status())
time.sleep(0.3)
else:
print("switch service not available")
if __name__ == "__main__":
asyncio.run(main())

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')
switch_characteristic = Uint8Characteristic(
uuid=VendorUUID("8158b2fe-94e4-4ff5-a99d-9a7980e998d7")
)
def __init__(self, service=None):
super().__init__(service=service)
self.status = self.switch_characteristic
def read_status(self):
return self.status

我遇到的问题是，read_status()将始终返回BLE开关的状态是当程序第一次运行时。它不会收到BLE开关后续状态变化的通知。我的想法是，我缺少的是注册BLE开关以通知更改。我正在努力寻找例子或参考来做到这一点。

谢谢。

Answer 1

感谢ukBaz指出萧瑟。下面的代码使用了非常好用的萧伯纳。

import asyncio
from bleak import BleakScanner
from bleak.backends.bluezdbus.client import BleakClientBlueZDBus
device_name = "My Arduino"
switch_status_char_uuid = "8158b2fe-94e4-4ff5-a99d-9a7980e998d7"

def notification_handler(sender, data):
print("Switch is active: {}".format(bool(data[0])))

async def run():
client = None
external_heartbeat_received = False
device = await BleakScanner.find_device_by_filter(
lambda d, ad: d.name and d.name.lower() == device_name.lower()
)
if device is None:
print("{} not found".format(device_name))
else:
print("{} found".format(device))
client = BleakClientBlueZDBus(device)
try:
timer = 60  # seconds
while timer != 0 or external_heartbeat_received:
if not client.is_connected:
if await client.connect():
print("Connected to {}".format(device_name))
await client.start_notify(switch_status_char_uuid, notification_handler)
await asyncio.sleep(1)
timer -= 1
# If timer expired and we received a heartbeat, restart timer and carry on.
if timer == 0:
if external_heartbeat_received:
timer = 60
external_heartbeat_received = False
except:
print("Connected to {} failed".format(device_name))
if client is not None and client.is_connected:
await client.disconnect()
print("Disconnected from {}".format(device_name))
loop = asyncio.get_event_loop()
loop.run_until_complete(run())