如何将 microbit 与 BLE 连接并监听按钮按下事件?

Posted

技术标签:

【中文标题】如何将 microbit 与 BLE 连接并监听按钮按下事件?【英文标题】:How do I connect microbit with BLE and listen for button press events? 【发布时间】:2020-12-24 06:01:18 【问题描述】:

11/28/2021 编辑:

如果您需要使用低功耗蓝牙将 microbit 连接到计算机,并在单击按钮时执行操作。直接跳转并关注下面@ukBaz的回答。

注意:该解决方案可以在 GNU/Linux 上完美运行,但在 Windows 上可能效果不佳。

以下是帖子的原始问题。我不会编辑它来隐藏我的错误。


总结:我有一个连接到 rpi-zero 的 microbit。我对 microbit 进行了编码,当按下A button 时,它将通过uart.write 将数据发送到 rpi-zero。

在这个测试中,microbit 将uart.write("Test"),写一个“测试”字到rpi-zero。

我的最终目标是使用 rpi-zero 的 BLE 功能充当控制设备,并通过 microbit 按钮发送指令。

我发现这个GATT Server Code 是用python 为rpi 编写的。它运行起来完全没有问题。

下面的代码将用于监听microbit uart服务并检查接收到的数据是否为"Test"

import serial

serialPort = serial.Serial(port = "/dev/ttyACM0", baudrate=115200, bytesize=8, timeout=0.5, stopbits=serial.STOPBITS_ONE)

serialString = " "

(serialPort.in_waiting > 0)

while True:

        serialString = serialPort.readline()

        if serialString == b'Test':
            print("Yes")
        else:
            print("F")

但真正的问题是当我尝试将此循环代码实现到 GATT 服务器代码中时。

我似乎无法理解如何将此值传递给self.send_tx

此外,GATT 服务器代码中似乎已经存在一个全局循环。所以我尝试使用线程同时运行这两个函数,但是当我添加self.send_tx("Test") 时,它只会抛出错误Self is not defined

对不起,我对编码一窍不通,有人知道可能的解决方法吗?谢谢

这是完整的代码:

import sys
import threading
import dbus, dbus.mainloop.glib
import serial
from gi.repository import GLib
from example_advertisement import Advertisement
from example_advertisement import register_ad_cb, register_ad_error_cb
from example_gatt_server import Service, Characteristic
from example_gatt_server import register_app_cb, register_app_error_cb

BLUEZ_SERVICE_NAME =           'org.bluez'
DBUS_OM_IFACE =                'org.freedesktop.DBus.ObjectManager'
LE_ADVERTISING_MANAGER_IFACE = 'org.bluez.LEAdvertisingManager1'
GATT_MANAGER_IFACE =           'org.bluez.GattManager1'
GATT_CHRC_IFACE =              'org.bluez.GattCharacteristic1'
UART_SERVICE_UUID =            '6e400001-b5a3-f393-e0a9-e50e24dcca9e'
UART_RX_CHARACTERISTIC_UUID =  '6e400002-b5a3-f393-e0a9-e50e24dcca9e'
UART_TX_CHARACTERISTIC_UUID =  '6e400003-b5a3-f393-e0a9-e50e24dcca9e'
LOCAL_NAME =                   'rpi-gatt-server'
mainloop = None

serialPort = serial.Serial(port = "/dev/ttyACM0", baudrate=115200, bytesize=8, timeout=0.8, stopbits=serial.STOPBITS_ONE)

serialString = " "

(serialPort.in_waiting > 0)

class TxCharacteristic(Characteristic):
    def __init__(self, bus, index, service):
        Characteristic.__init__(self, bus, index, UART_TX_CHARACTERISTIC_UUID,
                                ['notify'], service)
        self.notifying = False
        GLib.io_add_watch(sys.stdin, GLib.IO_IN, self.on_console_input)

    def on_console_input(self, fd, condition):
        s = fd.readline()
        if s.isspace():
            pass
        else:
            self.send_tx(s)
        return True

    def send_tx(self, s):
        if not self.notifying:
            return
        value = []
        for c in s:
            value.append(dbus.Byte(c.encode()))
        self.PropertiesChanged(GATT_CHRC_IFACE, 'Value': value, [])

    def StartNotify(self):
        if self.notifying:
            print("yes")
            return
        self.notifying = True

    def StopNotify(self):
        if not self.notifying:
            print("no")
            return
        self.notifying = False

class RxCharacteristic(Characteristic):
    def __init__(self, bus, index, service):
        Characteristic.__init__(self, bus, index, UART_RX_CHARACTERISTIC_UUID,
                                ['write'], service)

    def WriteValue(self, value, options):
        print('remote: '.format(bytearray(value).decode()))

class UartService(Service):
    def __init__(self, bus, index):
        Service.__init__(self, bus, index, UART_SERVICE_UUID, True)
        self.add_characteristic(TxCharacteristic(bus, 0, self))
        self.add_characteristic(RxCharacteristic(bus, 1, self))

class Application(dbus.service.Object):
    def __init__(self, bus):
        self.path = '/'
        self.services = []
        dbus.service.Object.__init__(self, bus, self.path)

    def get_path(self):
        return dbus.ObjectPath(self.path)

    def add_service(self, service):
        self.services.append(service)

    @dbus.service.method(DBUS_OM_IFACE, out_signature='aoasasv')
    def GetManagedObjects(self):
        response = 
        for service in self.services:
            response[service.get_path()] = service.get_properties()
            chrcs = service.get_characteristics()
            for chrc in chrcs:
                response[chrc.get_path()] = chrc.get_properties()
        return response

class UartApplication(Application):
    def __init__(self, bus):
        Application.__init__(self, bus)
        self.add_service(UartService(bus, 0))

class UartAdvertisement(Advertisement):
    def __init__(self, bus, index):
        Advertisement.__init__(self, bus, index, 'peripheral')
        self.add_service_uuid(UART_SERVICE_UUID)
        self.add_local_name(LOCAL_NAME)
        self.include_tx_power = True

def find_adapter(bus):
    remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
                               DBUS_OM_IFACE)
    objects = remote_om.GetManagedObjects()
    for o, props in objects.items():
        if LE_ADVERTISING_MANAGER_IFACE in props and GATT_MANAGER_IFACE in props:
            return o
        print('Skip adapter:', o)
    return None

def check():
    while True:
        serialString = serialPort.readline()
        if serialString == b'Test':
            print("Okay, Test")
            self.send_tx("Test")
        else:
            print("No")

def main():
    global mainloop
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    bus = dbus.SystemBus()
    adapter = find_adapter(bus)
    if not adapter:
        print('BLE adapter not found')
        return

    service_manager = dbus.Interface(
                                bus.get_object(BLUEZ_SERVICE_NAME, adapter),
                                GATT_MANAGER_IFACE)
    ad_manager = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter),
                                LE_ADVERTISING_MANAGER_IFACE)

    app = UartApplication(bus)
    adv = UartAdvertisement(bus, 0)

    mainloop = GLib.MainLoop()

    service_manager.RegisterApplication(app.get_path(), ,
                                        reply_handler=register_app_cb,
                                        error_handler=register_app_error_cb)
    ad_manager.RegisterAdvertisement(adv.get_path(), ,
                                     reply_handler=register_ad_cb,
                                     error_handler=register_ad_error_cb)

    try:
        mainloop.run()
    except KeyboardInterrupt:
        adv.Release()

if __name__ == '__main__':
    p1 = threading.Thread(target=main)
    p2 = threading.Thread(target=check)
    p1.start()
    p2.start()

【问题讨论】:

【参考方案1】:

我认为这可能是XY problem。我所了解的是,您想通过低功耗蓝牙 (BLE) 从 micro:bit 向 RPi 发送按钮按下。如果是这种情况,那么有一种更有效的方法来做到这一点。

在 micro:bit 的Bluetooth Profile 中有一个按钮服务和一个可以使用的按钮状态特性。它们的 UUID 是:

BTN_SRV = 'E95D9882-251D-470A-A062-FA1922DFA9A8'
BTN_A_STATE = 'E95DDA90-251D-470A-A062-FA1922DFA9A8'

您需要在 micro:bit 上设置 GATT 服务器。最有效的方法是使用https://makecode.microbit.org/#editor 创建以下内容:

如果左侧菜单中没有蓝牙,请单击屏幕右上角的齿轮,选择Extensions,然后选择Bluetooth 替换Radio 扩展名。

使用用于 D-Bus 绑定的 pydbus 库可以简化 RPi 上的 GATT 客户端代码。

使用蓝牙,与其拥有while 循环并不断轮询micro:bit,不如拥有一个订阅来自(在这种情况下)Button A 特征的通知的事件循环更有效。每次按下或释放 micro:bit 按钮 A 时都会调用我命名为 btn_handler 的函数。

下面的代码不会在 micro:bit 和 RPi 之间进行初始配对。由于配对是一次性配置步骤,因此我手动进行。

这里是 RPi 的示例 python 代码,它响应 micro:bit 上的按钮 A 被按下...

from time import sleep
import pydbus
from gi.repository import GLib

DEVICE_ADDR = 'DE:82:35:E7:CE:BE' #  micro:bit address
BTN_A_STATE = 'E95DDA90-251D-470A-A062-FA1922DFA9A8'

# DBus object paths
BLUEZ_SERVICE = 'org.bluez'
ADAPTER_PATH = '/org/bluez/hci0'
device_path = f"ADAPTER_PATH/dev_DEVICE_ADDR.replace(':', '_')"

# setup dbus
bus = pydbus.SystemBus()
mngr = bus.get(BLUEZ_SERVICE, '/')
adapter = bus.get(BLUEZ_SERVICE, ADAPTER_PATH) 
device = bus.get(BLUEZ_SERVICE, device_path)

device.Connect()

while not device.ServicesResolved:
    sleep(0.5)

def get_characteristic_path(dev_path, uuid):
    """Look up DBus path for characteristic UUID"""
    mng_objs = mngr.GetManagedObjects()
    for path in mng_objs:
        chr_uuid = mng_objs[path].get('org.bluez.GattCharacteristic1', ).get('UUID')
        if path.startswith(dev_path) and chr_uuid == uuid.casefold():
           return path

# Characteristic DBus information
btn_a_path = get_characteristic_path(device._path, BTN_A_STATE)
btn_a = bus.get(BLUEZ_SERVICE, btn_a_path)
# Read button A without event loop notifications
print(btn_a.ReadValue())

# Enable eventloop for notifications
def btn_handler(iface, prop_changed, prop_removed):
    """Notify event handler for button press"""
    if 'Value' in prop_changed:
        new_value = prop_changed['Value']
        print(f"Button A state: new_value")
        print(f'As byte: bytes(new_value)')
        print(f'As bytearray: bytearray(new_value)')
        print(f'As int: int(new_value[0])')
        print(f'As bool: bool(new_value[0])')

mainloop = GLib.MainLoop()
btn_a.onPropertiesChanged = btn_handler
btn_a.StartNotify()
try:
    mainloop.run()
except KeyboardInterrupt:
    mainloop.quit()
    btn_a.StopNotify()
    device.Disconnect()

【讨论】:

哇,这样一个清晰而完美的答案。但遗憾的是,我的 microbit 上的 BLE 模块损坏了,所以我只能在上面使用 micropython(USB 模式)。 我希望我没有发布这篇文章。然而仅仅一年后,回顾你的答案。我当时就觉得自己好傻。我同意你的看法,这显然是一个 XY 问题。我刚刚在我的新 microbit 上进行了测试,您的解决方案运行良好,尽管 BlueZ 将受限于 GNU/Linux 操作系统。

以上是关于如何将 microbit 与 BLE 连接并监听按钮按下事件?的主要内容,如果未能解决你的问题,请参考以下文章

是否有可能(并且有意义)让蓝牙 iOT 设备在外围模式下同时监听 BLE 和蓝牙经典连接

如何在 Android 中获取当前连接的 BLE 外围设备

如何将 BLE 活动继续到下一个视图控制器

SQLAlchemy ORM如何将查询与表连接起来

GATT连接后如何与BLE设备配对

Microbit MicroPython 介绍