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

Question

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

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

from PyObjCTools import AppHelper

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

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

我尝试运行事件循环:

• 来自不同的线程 ---> 未调用委托(delegate)回调
• 来自子流程 ---> 未调用委托(delegate)回调
• 来自 fork 的子进程 ---> 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 委托(delegate)回调？是否有其他版本可以调用而不必在主线程上运行？我在网上读到一些关于它与 GUI 相关的内容，因此必须在主线程上运行，但我的 python 应用程序没有任何 GUI 元素。是否有我可以使用的与 GUI 不太相关的标志或 API？

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

更新:

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

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

Answer 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()