如何在不放弃主线程的情况下将 CoreBluetooth 用于 Python
How can I use CoreBluetooth for Python without giving up the main thread
我正在尝试实现一个通用 BLE 接口,它将 运行 on OS/X 并与 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 专家交谈,发现 Dispatch Queues 可能是解决方案。我进一步挖掘,发现 PyObjC 包的作者最近发布了一个 v4.1,它增加了对迄今为止缺失的调度队列的支持。
我已经阅读了 Apple 开发人员文档几个小时了,我知道可以创建 Dispatch Source 对象来监视某些系统事件(例如我感兴趣的 BLE 外设事件)并且配置它们涉及创建并分配一个 Dispatch Queue,它是调用我的 CBCentralManager 委托回调方法的 class。我仍然遗漏的一块拼图是如何将 Dispatch Source/Queue 东西连接到 AppHelper.runConsoleEventLoop
,后者调用 Foundation.NSRunLoop.currentRunLoop()
。如果我将对 AppHelper
的调用放在一个单独的线程上,我如何告诉它使用哪个 Dispatch Source/Queue 来获取事件信息?
所以我终于想通了。如果你想 运行 在一个单独的线程上进行事件循环,这样你就不会失去对主线程的控制,你必须创建一个新的调度队列并用它初始化你的 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()
我正在尝试实现一个通用 BLE 接口,它将 运行 on OS/X 并与 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 专家交谈,发现 Dispatch Queues 可能是解决方案。我进一步挖掘,发现 PyObjC 包的作者最近发布了一个 v4.1,它增加了对迄今为止缺失的调度队列的支持。
我已经阅读了 Apple 开发人员文档几个小时了,我知道可以创建 Dispatch Source 对象来监视某些系统事件(例如我感兴趣的 BLE 外设事件)并且配置它们涉及创建并分配一个 Dispatch Queue,它是调用我的 CBCentralManager 委托回调方法的 class。我仍然遗漏的一块拼图是如何将 Dispatch Source/Queue 东西连接到 AppHelper.runConsoleEventLoop
,后者调用 Foundation.NSRunLoop.currentRunLoop()
。如果我将对 AppHelper
的调用放在一个单独的线程上,我如何告诉它使用哪个 Dispatch Source/Queue 来获取事件信息?
所以我终于想通了。如果你想 运行 在一个单独的线程上进行事件循环,这样你就不会失去对主线程的控制,你必须创建一个新的调度队列并用它初始化你的 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()