在Python中使用asyncio存储回调的值

huangapple go评论92阅读模式
英文:

Store value of a callback in Python using asyncio

问题

我正在尝试在项目中使用Deepgram流式语音识别。我可以使用他们的快速入门演示代码将转录文本流式传输到控制台,但文本是从回调函数内部打印的。我想将转录文本的各个片段从回调函数中提取出来,放入单个字符串(或数组或其他什么东西)中,以便在打印之前格式化更长的转录片段。

似乎类似于https://stackoverflow.com/a/66279927,但我认为由于asyncio(或我没有理解的其他原因),我的情况需要不同处理。

这段代码有效,但只是将转录文本的每个小片段倾倒到控制台:

from deepgram import Deepgram
import asyncio
import aiohttp

DEEPGRAM_API_KEY = '****'
URL = 'http://stream.live.vc.bbcmedia.co.uk/bbc_world_service'

async def main():
  deepgram = Deepgram(DEEPGRAM_API_KEY)

  # 创建到Deepgram的WebSocket连接
  deepgramLive = await deepgram.transcription.live({ 'language': 'en-US' })

  # 监听连接关闭事件
  deepgramLive.registerHandler(deepgramLive.event.CLOSE, lambda c: print(f'Connection closed with code {c}.'))

  # 监听从Deepgram接收到的任何转录,并将它们写入控制台
  deepgramLive.registerHandler(deepgramLive.event.TRANSCRIPT_RECEIVED, print_transcript) # 使用比print_transcript更复杂/持久的内容会引发'raise AttributeError(name) from None'错误

  # 监听连接打开事件并将来自URL的流式音频发送到Deepgram
  async with aiohttp.ClientSession() as session:
    async with session.get(URL) as audio:
      while True:
        data = await audio.content.readany()
        deepgramLive.send(data)

        # 在此处处理转录片段?

        if not data:
            break
  await deepgramLive.finish()


def print_transcript(json_data):
   print(json_data['channel']['alternatives'][0]['transcript'])


asyncio.run(main())

我尝试过使用一个具有__call__方法的类,如其他问题中所示,并尝试过使用asyncio.Queue,但我似乎漏掉了一些东西。

英文:

I am trying to use Deepgram streaming speech recognition for a project. I can stream the transcribed text to the console using their quickstart demo code, but the text is printed from within a callback function. I would like to get the individual chunks of transcribed text out of the callback into a single string (or an array or whatever) so I can format longer pieces of the transcription before printing it.

Seems like a similar problem as [https://stackoverflow.com/a/66279927](this question), but I think my situation needs to be treated differently due to asyncio (or something else I am not understanding)

This works, but just dumps each little piece of transcribed text to the console:

from deepgram import Deepgram
import asyncio
import aiohttp

DEEPGRAM_API_KEY = '****'
URL = 'http://stream.live.vc.bbcmedia.co.uk/bbc_world_service'

async def main():
  deepgram = Deepgram(DEEPGRAM_API_KEY)

  # Create a websocket connection to Deepgram
  deepgramLive = await deepgram.transcription.live({ 'language': 'en-US' })

  # Listen for the connection to close
  deepgramLive.registerHandler(deepgramLive.event.CLOSE, lambda c: print(f'Connection closed with code {c}.'))

  # Listen for any transcripts received from Deepgram and write them to the console
  deepgramLive.registerHandler(deepgramLive.event.TRANSCRIPT_RECEIVED, print_transcript) # using anything more complex/persistent than print_transcript here throws 'raise AttributeError(name) from None' error

  # Listen for the connection to open and send streaming audio from the URL to Deepgram
  async with aiohttp.ClientSession() as session:
    async with session.get(URL) as audio:
      while True:
        data = await audio.content.readany()
        deepgramLive.send(data)

        # do more with the transcribed chunks here?

        if not data:
            break
  await deepgramLive.finish()


def print_transcript(json_data):
   print(json_data['channel']['alternatives'][0]['transcript'])


asyncio.run(main())

I tried using a class with a __call__ method as in the other question and I tried messing with asyncio.Queue, but I'm missing something.

答案1

得分: 1

他们的Python文档非常糟糕,所以我们必须检查源代码。但是似乎LiveTranscription.register_handler方法期望handler参数的类型为在这里定义的EventHandler。这只是一个可以接受任何类型的一个参数并返回None或等效的协程函数的函数。

这仍然非常糟糕,因为我们根本不知道这个处理程序通常会接收什么类型的对象。但根据_你的_代码中的print_transcript函数,你似乎期望一个字典(或类似的东西)。

如果你想存储这些对象而不仅仅是打印和丢弃它们,你有许多选择。其中一种方法是编写一个处理程序函数,它接受某种数据结构(例如列表)作为_额外的_参数,并将这些对象存储在该数据结构中,而不是将它们打印出来,然后在将部分初始化的函数传递给register_handler之前,使用functools.partialmain函数中预绑定这样的存储对象。

类似于这样:

from functools import partial
from typing import Any

def store_data(data: Any, storage: list[Any]) -> None:
    storage.append(data)

async def main() -> None:
    ...
    storage = []
    handler = partial(store_data, storage=storage)
    deepgram_live.register_handler(deepgram_live.event.TRANSCRIPT_RECEIVED, handler)

另一种几乎等效的选择是在main函数内部定义处理程序函数,并在该main函数的范围内为其提供访问存储对象的权限:

from typing import Any

async def main() -> None:
    ...
    storage = []

    def store_data(data: Any) -> None:
        storage.append(data)
    
    deepgram_live.register_handler(deepgram_live.event.TRANSCRIPT_RECEIVED, store_data)

如果你愿意,你确实可以使用asyncio.Queue而不是简单的list,但如何使处理程序函数_访问_该队列对象的原理仍然相同。

我不使用Deepgram,所以我没有测试过这个,但至少从我从糟糕的文档、源代码和你的示例中了解到的信息,我认为这应该可以工作。

英文:

Their Python documentation is horrendous, so we have to check the source code. But it seems the LiveTranscription.register_handler method expects the handler argument to be of type EventHandler as defined here. That is just a function that can be called with one argument of any type and that returns None or an equivalent coroutine function.

This is still very badly typed because we have absolutely no idea what type of object this handler will receive in general. But judging from your code with that print_transcript function, you seem to be expecting a dictionary (or something similar).

If you want to store those objects rather than just printing and discarding, you have many options. One would be to write a handler function that takes some sort of data structure (a list for example) as an additional argument and stores those objects in that data structure instead of printing them, then use functools.partial in your main function to pre-bind such a storage object to that function argument before passing the partially initialized function to register_handler.

Something like this:

from functools import partial
from typing import Any


def store_data(data: Any, storage: list[Any]) -> None:
    storage.append(data)


async def main() -> None:
    ...
    storage = []
    handler = partial(store_data, storage=storage)
    deepgram_live.register_handler(deepgram_live.event.TRANSCRIPT_RECEIVED, handler)

Another almost equivalent option would be to define that handler function inside the main function and provide it access to a storage object from within that main function's scope:

from typing import Any

async def main() -> None:
    ...
    storage = []

    def store_data(data: Any) -> None:
        storage.append(data)
    
    deepgram_live.register_handler(deepgram_live.event.TRANSCRIPT_RECEIVED, store_data)

You could indeed use an asyncio.Queue instead of a simple list if you want, but the principles of how you make the handler function access that queue object are still the same.

I don't use Deepgram, so I have not tested this, but at least from what I could gather from the poor documentation, the source, and your example, I think this should work.

huangapple
  • 本文由 发表于 2023年7月6日 12:17:21
  • 转载请务必保留本文链接:https://go.coder-hub.com/76625455.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定