如何在 python 中使用 pygatt 从 BLE 设备获取通知?

How to get notifications from BLE Device using pygatt in python?

我正在使用 python 开发一个 Linux 应用程序,它将连接到我的 BLE 设备并通过通知特性获取数据。我正在使用 pygatt 进行 BLE 通信。我可以成功连接并绑定到设备并读取 from/write 到特性。即使我可以订阅通知特性,但问题是,我的 BLE 设备是一台自定义机器,里面有 4 个计数器,每次计数器的一个数据发生变化时,它都会设置相应的通知标志,因此,onDataChanged-像方法一样,我可以从读取特性中读取计数器的数据。在 Python 使用 pygatt 时,我可以通过以下方式订阅通知特性:

class_name.device.subscribe(uuid.UUID(notify_characteristic),callback=notifyBle)

notifyBle 是:

def notifyBle(self,handle,data):
    read_data = class_name.device.char_read(uuid.UUID(read_characteristic))
    print(read_data)

当我 运行 程序时,首先我扫描设备并连接到我的设备并与之绑定,然后我发现特征并列出它们。一切顺利。列出特性后,我写写特性来清除通知标志,也成功了。最后订阅notify characteristic成功了

完成所有这些过程后,我物理上增加了设备的计数器(设备上有用于增加计数器的按钮)。当我按下按钮时,程序转到 notifyBle 方法并给出错误,即:

Exception in thread Thread-3:
Traceback (most recent call last):
  File "/usr/lib/python3.5/threading.py", line 914, in _bootstrap_inner
    self.run()
  File "/usr/local/lib/python3.5/dist-packages/pygatt/backends/gatttool/gatttool.py", line 137, in run
    event["callback"](event)
  File "/usr/local/lib/python3.5/dist-packages/pygatt/backends/gatttool/gatttool.py", line 479, in _handle_notification_string
    self._connected_device.receive_notification(handle, values)
  File "/usr/local/lib/python3.5/dist-packages/pygatt/device.py", line 226, in receive_notification
    callback(handle, value)
  File "/home/acd/Masaüstü/python_workspace/ble.py", line 54, in notifyBle
    read_data = bleFunctions.dev.char_read(uuid.UUID(bleFunctions.read_characteristic))
  File "/usr/local/lib/python3.5/dist-packages/pygatt/backends/gatttool/device.py", line 17, in wrapper
    return func(self, *args, **kwargs)
  File "/usr/local/lib/python3.5/dist-packages/pygatt/backends/gatttool/device.py", line 40, in char_read
    return self._backend.char_read(self, uuid, *args, **kwargs)
  File "/usr/local/lib/python3.5/dist-packages/pygatt/backends/gatttool/gatttool.py", line 53, in wrapper
    return func(self, *args, **kwargs)
  File "/usr/local/lib/python3.5/dist-packages/pygatt/backends/gatttool/gatttool.py", line 519, in char_read
    self.sendline('char-read-uuid %s' % uuid)
  File "/usr/lib/python3.5/contextlib.py", line 66, in __exit__
    next(self.gen)
  File "/usr/local/lib/python3.5/dist-packages/pygatt/backends/gatttool/gatttool.py", line 180, in event
    self.wait(event, timeout)
  File "/usr/local/lib/python3.5/dist-packages/pygatt/backends/gatttool/gatttool.py", line 154, in wait
    raise NotificationTimeout()
pygatt.exceptions.NotificationTimeout

如有任何帮助,我们将不胜感激。

PS:我在 Android 和 Windows UWP 中编写了完全相同的程序。使用 python,我的目标是 运行 在 raspberry pi 3.

PSS:我正在使用 raspberry pi 3 并安装了 Ubuntu Mate 用于在 python.

中开发此程序

首先,创建活动 class,如下所示,

class Event:
    def __init__(self):
        self.handlers = set()

    def handle(self, handler):
        self.handlers.add(handler)
        return self

    def unhandle(self, handler):
        try:
            self.handlers.remove(handler)
        except:
            raise ValueError("Handler is not handling this event, so cannot unhandle it.")
        return self

    def fire(self, *args, **kargs):
        for handler in self.handlers:
            handler(*args, **kargs)

    def getHandlerCount(self):
        return len(self.handlers)

    __iadd__ = handle
    __isub__ = unhandle
    __call__ = fire
    __len__  = getHandlerCount

那么, 创建一个 ble class

import pygatt
from eventclass import Event

class myBle:
    ADDRESS_TYPE = pygatt.BLEAddressType.random
    read_characteristic = "0000xxxx-0000-1000-8000-00805f9b34fb"
    write_characteristic = "0000xxxx-0000-1000-8000-00805f9b34fb"
    notify_characteristic = "0000xxxxx-0000-1000-8000-00805f9b34fb"
    def __init__(self,device):
        self.device = device
        self.valueChanged = Event()
        self.checkdata = False

    def alert(self):
         self.valueChanged(self.checkdata)

    def write(self,data):
        self.device.write_char(self.write_characteristic,binascii.unhexlify(data))

    def notify(self,handle,data):
        self.checkdata = True

    def read(self):
        if(self.checkdata):
            self.read_data = self.device.char_read(uuid.UUID(self.read_characteristic))
            self.write(bytearray(b'\x10\x00'))
            self.checkdata = False
            return self.read_data
    def discover(self):
        return self.device.discover_characteristics().keys()

当您收到通知时,您会将布尔值设置为 true 并且警报方法将通知布尔值为 changed.You 将使用

监听警报方法
def triggerEvent(checkdata):
   print(str(checkdata))

ble = myBle(device)
ble.valueChanged += triggerEvent
ble.alert()

可以通过triggerEvent方法调用read方法获取特征值