如何在不放弃主线程的情况下为 Python 使用 CoreBluetooth

Posted 2023-03-05

【中文标题】如何在不放弃主线程的情况下为 Python 使用 CoreBluetooth

【英文标题】：How can I use CoreBluetooth for Python without giving up the main thread

【发布时间】：2018-02-24 00:57:22

【问题描述】：

我正在尝试实现将在 OS/X 上运行并与 BLE 外围设备通信的通用 BLE 接口。外围设备非常复杂：可以查询、发送数百个不同的命令、提供通知等。我需要能够连接到它、发送命令、读取响应、获取更新等。

我拥有我需要的所有代码，但我对一件事感到沮丧：从我可以在网上找到的有限信息来看，似乎调用 CoreBluetooth 的委托回调的唯一方法是运行：

from PyObjCTools import AppHelper

# [functional CoreBluetooth code that scans for peripherals]
AppHelper.runConsoleEventLoop()

问题是AppHelper.runConsoleEventLoop 阻止了主线程继续运行，所以我无法执行代码与外围设备交互。

我尝试过运行事件循环：

来自不同的线程 ---> 未调用委托回调 来自子进程 ---> 未调用委托回调 来自一个分叉的孩子 ---> Python 崩溃并显示错误消息：The process has forked and you cannot use this CoreFoundation functionality safely. You MUST exec(). 来自multiprocessing.Pool(1).apply_async(f) ---> Python 崩溃并显示错误消息：The process has forked and you cannot use this CoreFoundation functionality safely. You MUST exec().

都没有成功。

我不明白AppHelper.runConsoleEventLoop 的性质。为什么需要运行它才能调用 CoreBluetooth 委托回调？是否有其他可以调用的版本不必在主线程上运行？我在网上阅读了一些关于它与 GUI 相关的内容，因此必须在主线程上运行，但我的 python 应用程序没有任何 GUI 元素。是否有我可以使用的不太关心 GUI 的标志或 API？

任何帮助将不胜感激。感谢您的宝贵时间！

更新：

我与工作中的 ios/CoreBluetooth 专家交谈，发现调度队列可能是解决方案。我进一步挖掘，发现 PyObjC 包的作者最近发布了一个 v4.1，它增加了对迄今为止缺少的调度队列的支持。

我已经阅读 Apple 开发人员文档几个小时了，我知道可以创建监控某些系统事件（例如我感兴趣的 BLE 外围设备事件）的 Dispatch Source 对象，并且配置它们涉及创建和分配一个调度队列，它是调用我的 CBCentralManager 委托回调方法的类。我仍然缺少的一个难题是如何将调度源/队列的东西连接到AppHelper.runConsoleEventLoop，它调用Foundation.NSRunLoop.currentRunLoop()。如果我在单独的线程上调用AppHelper，我如何告诉它使用哪个调度源/队列来获取事件信息？

【参考方案1】：

所以我终于想通了。如果你想在一个单独的线程上运行一个事件循环，这样你就不会失去对主线程的控制，你必须创建一个新的调度队列并用它初始化你的 CBCentralManager。

import CoreBluetooth
import libdispatch


class CentralManager(object):
    def __init__(self):
        central_manager = CoreBluetooth.CBCentralManager.alloc()
        dispatch_queue = libdispatch.dispatch_queue_create('<queue name>', None)
        central_manager.initWithDelegate_queue_options_(delegate, dispatch_queue, None)


    def do_the_things(args):
        # scan, connect, send messages, w/e


class EventLoopThread(threading.Thread):
    def __init__(self):
        super(EventLoopThread, self).__init__()
        self.setDaemon(True)
        self.should_stop = False


    def run(self):
        logging.info('Starting event loop on background thread')
        AppHelper.runConsoleEventLoop(installInterrupt=True)


    def stop(self):
        logging.info('Stop the event loop')
        AppHelper.stopEventLoop()


event_loop_thread = EventLoopThread()
event_loop_thread.start()
central_device = BLECentralDevice(service_uuid_list)
central_device.do_the_things('woo hoo')
event_loop_thread.stop()