将带有回调的Python函数转换为可等待的异步
问题描述
我希望在异步上下文中使用PyAudio
库,但是该库的主入口点只有基于回调的API:
import pyaudio
def callback(in_data, frame_count, time_info, status):
# Do something with data
pa = pyaudio.PyAudio()
self.stream = self.pa.open(
stream_callback=callback
)
我希望如何使用它是这样的:
pa = SOME_ASYNC_COROUTINE()
async def listen():
async for block in pa:
# Do something with block
问题是,我不确定如何将此回调语法转换为在触发回调时完成的将来语法。在JavaScript中,我会使用promise.promisify()
,但是Python似乎没有这样的东西。
解决方案
等效的promisify
不适用于此用例,原因有两个:
- PyAudio的异步API没有使用异步事件循环-文档规定回调是从后台线程调用的。这需要采取预防措施才能与异步IO正确通信。
- 回调不能以单个未来建模,因为它被多次调用,而未来只能有一个结果。相反,必须将其转换为异步迭代器,如示例代码所示。
以下是一种可能的实现:
def make_iter():
loop = asyncio.get_event_loop()
queue = asyncio.Queue()
def put(*args):
loop.call_soon_threadsafe(queue.put_nowait, args)
async def get():
while True:
yield await queue.get()
return get(), put
make_iter
返回对async for
,该协程将在等待下一个值时挂起:
async def main():
stream_get, stream_put = make_iter()
stream = pa.open(stream_callback=stream_put)
stream.start_stream()
async for in_data, frame_count, time_info, status in stream_get:
# ...
asyncio.get_event_loop().run_until_complete(main())
请注意,根据documentation,回调还必须返回有意义的值、帧的元组和布尔标志。通过更改fill
功能以也从异步侧接收数据,可以将这一点合并到设计中。不包括该实现,因为如果不了解域,它可能没有多大意义。
相关文章