diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d885a2dd..6b89d51a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,17 @@ All notable changes to **pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- Added new `AzureSTTService`. This allows you to use Azure Speech-To-Text. + +### Other + +- Updated `07f-interruptible-azure.py` to use `AzureLLMService`, + `AzureSTTService` and `AzureTTSService`. + ## [0.0.31] - 2024-06-13 ### Performance diff --git a/README.md b/README.md index 51f14f390..bd4569f16 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,7 @@ pip install "pipecat-ai[option,...]" Your project may or may not need these, so they're made available as optional requirements. Here is a list: -- **AI services**: `anthropic`, `azure`, `deepgram`, `google`, `fal`, `moondream`, `openai`, `playht`, `silero`, `whisper` +- **AI services**: `anthropic`, `azure`, `deepgram`, `google`, `fal`, `moondream`, `openai`, `openpipe`, `playht`, `silero`, `whisper` - **Transports**: `local`, `websocket`, `daily` ## Code examples diff --git a/examples/foundational/07d-interruptible-cartesia.py b/examples/foundational/07d-interruptible-cartesia.py index 283baa49a..8c4a16f02 100644 --- a/examples/foundational/07d-interruptible-cartesia.py +++ b/examples/foundational/07d-interruptible-cartesia.py @@ -5,7 +5,6 @@ # import asyncio -import aiohttp import os import sys @@ -33,62 +32,61 @@ logger.add(sys.stderr, level="DEBUG") async def main(room_url: str, token): - async with aiohttp.ClientSession() as session: - transport = DailyTransport( - room_url, - token, - "Respond bot", - DailyParams( - audio_out_enabled=True, - audio_out_sample_rate=44100, - transcription_enabled=True, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer() - ) + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + audio_out_sample_rate=44100, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() ) + ) - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - voice_name="British Lady", - output_format="pcm_44100" - ) + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_name="British Lady", + output_format="pcm_44100" + ) - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") - messages = [ - { - "role": "system", - "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", - }, - ] + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] - tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages) + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) - pipeline = Pipeline([ - transport.input(), # Transport user input - tma_in, # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - tma_out # Assistant spoken responses - ]) + pipeline = Pipeline([ + transport.input(), # Transport user input + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses + ]) - task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - # Kick off the conversation. - messages.append( - {"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([LLMMessagesFrame(messages)]) + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) - runner = PipelineRunner() + runner = PipelineRunner() - await runner.run(task) + await runner.run(task) if __name__ == "__main__": diff --git a/examples/foundational/07e-interruptible-playht.py b/examples/foundational/07e-interruptible-playht.py index c6c062e52..a29c640c7 100644 --- a/examples/foundational/07e-interruptible-playht.py +++ b/examples/foundational/07e-interruptible-playht.py @@ -5,7 +5,6 @@ # import asyncio -import aiohttp import os import sys @@ -19,7 +18,6 @@ from pipecat.services.playht import PlayHTTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer -from pipecat.processors.logger import FrameLogger from runner import configure @@ -33,62 +31,61 @@ logger.add(sys.stderr, level="DEBUG") async def main(room_url: str, token): - async with aiohttp.ClientSession() as session: - transport = DailyTransport( - room_url, - token, - "Respond bot", - DailyParams( - audio_out_enabled=True, - audio_out_sample_rate=16000, - transcription_enabled=True, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer() - ) + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + audio_out_sample_rate=16000, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() ) + ) - tts = PlayHTTTSService( - user_id=os.getenv("PLAYHT_USER_ID"), - api_key=os.getenv("PLAYHT_API_KEY"), - voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json", - ) + tts = PlayHTTTSService( + user_id=os.getenv("PLAYHT_USER_ID"), + api_key=os.getenv("PLAYHT_API_KEY"), + voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json", + ) - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") - messages = [ - { - "role": "system", - "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", - }, - ] + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] - tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages) + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) - pipeline = Pipeline([ - transport.input(), # Transport user input - tma_in, # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - tma_out # Assistant spoken responses - ]) + pipeline = Pipeline([ + transport.input(), # Transport user input + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses + ]) - task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - # Kick off the conversation. - messages.append( - {"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([LLMMessagesFrame(messages)]) + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) - runner = PipelineRunner() + runner = PipelineRunner() - await runner.run(task) + await runner.run(task) if __name__ == "__main__": diff --git a/examples/foundational/07f-interruptible-azure-tts.py b/examples/foundational/07f-interruptible-azure-tts.py deleted file mode 100644 index 1770a3213..000000000 --- a/examples/foundational/07f-interruptible-azure-tts.py +++ /dev/null @@ -1,95 +0,0 @@ -# -# Copyright (c) 2024, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import asyncio -import aiohttp -import os -import sys - -from pipecat.frames.frames import LLMMessagesFrame -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.llm_response import ( - LLMAssistantResponseAggregator, LLMUserResponseAggregator) -from pipecat.services.azure import AzureTTSService -from pipecat.services.openai import OpenAILLMService -from pipecat.transports.services.daily import DailyParams, DailyTransport -from pipecat.vad.silero import SileroVADAnalyzer - - -from runner import configure - -from loguru import logger - -from dotenv import load_dotenv -load_dotenv(override=True) - -logger.remove(0) -logger.add(sys.stderr, level="DEBUG") - - -async def main(room_url: str, token): - async with aiohttp.ClientSession() as session: - transport = DailyTransport( - room_url, - token, - "Respond bot", - DailyParams( - audio_out_enabled=True, - audio_out_sample_rate=16000, - transcription_enabled=True, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer() - ) - ) - - tts = AzureTTSService( - api_key=os.getenv("AZURE_SPEECH_API_KEY"), - region=os.getenv("AZURE_SPEECH_REGION"), - ) - - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") - - messages = [ - { - "role": "system", - "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", - }, - ] - - tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages) - - pipeline = Pipeline([ - transport.input(), # Transport user input - tma_in, # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - tma_out # Assistant spoken responses - ]) - - task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) - - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - # Kick off the conversation. - messages.append( - {"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([LLMMessagesFrame(messages)]) - - runner = PipelineRunner() - - await runner.run(task) - - -if __name__ == "__main__": - (url, token) = configure() - asyncio.run(main(url, token)) diff --git a/examples/foundational/07f-interruptible-azure.py b/examples/foundational/07f-interruptible-azure.py new file mode 100644 index 000000000..31b588488 --- /dev/null +++ b/examples/foundational/07f-interruptible-azure.py @@ -0,0 +1,100 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantResponseAggregator, LLMUserResponseAggregator) +from pipecat.services.azure import AzureLLMService, AzureSTTService, AzureTTSService +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(room_url: str, token): + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + audio_out_sample_rate=16000, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + ) + ) + + stt = AzureSTTService( + api_key=os.getenv("AZURE_SPEECH_API_KEY"), + region=os.getenv("AZURE_SPEECH_REGION"), + ) + + tts = AzureTTSService( + api_key=os.getenv("AZURE_SPEECH_API_KEY"), + region=os.getenv("AZURE_SPEECH_REGION"), + ) + + llm = AzureLLMService( + api_key=os.getenv("AZURE_CHATGPT_API_KEY"), + endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"), + model=os.getenv("AZURE_CHATGPT_MODEL"), + ) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + + pipeline = Pipeline([ + transport.input(), # Transport user input + stt, # STT + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token)) diff --git a/examples/foundational/07g-interruptible-openai-tts.py b/examples/foundational/07g-interruptible-openai-tts.py index 2a45b63d8..7552273c6 100644 --- a/examples/foundational/07g-interruptible-openai-tts.py +++ b/examples/foundational/07g-interruptible-openai-tts.py @@ -5,7 +5,6 @@ # import asyncio -import aiohttp import os import sys @@ -32,61 +31,60 @@ logger.add(sys.stderr, level="DEBUG") async def main(room_url: str, token): - async with aiohttp.ClientSession() as session: - transport = DailyTransport( - room_url, - token, - "Respond bot", - DailyParams( - audio_out_enabled=True, - audio_out_sample_rate=24000, - transcription_enabled=True, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer() - ) + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + audio_out_sample_rate=24000, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() ) + ) - tts = OpenAITTSService( - api_key=os.getenv("OPENAI_API_KEY"), - voice="alloy" - ) + tts = OpenAITTSService( + api_key=os.getenv("OPENAI_API_KEY"), + voice="alloy" + ) - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") - messages = [ - { - "role": "system", - "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", - }, - ] + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] - tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages) + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) - pipeline = Pipeline([ - transport.input(), # Transport user input - tma_in, # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - tma_out # Assistant spoken responses - ]) + pipeline = Pipeline([ + transport.input(), # Transport user input + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses + ]) - task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - # Kick off the conversation. - messages.append( - {"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([LLMMessagesFrame(messages)]) + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) - runner = PipelineRunner() + runner = PipelineRunner() - await runner.run(task) + await runner.run(task) if __name__ == "__main__": diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index 2821bb7b7..e9423d8f1 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -7,26 +7,30 @@ import aiohttp import asyncio import io +import time from PIL import Image from typing import AsyncGenerator -from openai import AsyncAzureOpenAI - -from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, URLImageRawFrame -from pipecat.services.ai_services import TTSService, ImageGenService +from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, ErrorFrame, Frame, StartFrame, SystemFrame, TranscriptionFrame, URLImageRawFrame +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_services import AIService, TTSService, ImageGenService from pipecat.services.openai import BaseOpenAILLMService from loguru import logger # See .env.example for Azure configuration needed try: + from openai import AsyncAzureOpenAI from azure.cognitiveservices.speech import ( - SpeechSynthesizer, SpeechConfig, + SpeechRecognizer, + SpeechSynthesizer, ResultReason, CancellationReason, ) + from azure.cognitiveservices.speech.audio import AudioStreamFormat, PushAudioInputStream + from azure.cognitiveservices.speech.dialog import AudioConfig except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( @@ -34,14 +38,35 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") +class AzureLLMService(BaseOpenAILLMService): + def __init__( + self, + *, + api_key: str, + endpoint: str, + model: str, + api_version: str = "2023-12-01-preview"): + # Initialize variables before calling parent __init__() because that + # will call create_client() and we need those values there. + self._endpoint = endpoint + self._api_version = api_version + super().__init__(api_key=api_key, model=model) + + def create_client(self, api_key=None, base_url=None, **kwargs): + return AsyncAzureOpenAI( + api_key=api_key, + azure_endpoint=self._endpoint, + api_version=self._api_version, + ) + + class AzureTTSService(TTSService): def __init__(self, *, api_key: str, region: str, voice="en-US-SaraNeural", **kwargs): super().__init__(**kwargs) - self.speech_config = SpeechConfig(subscription=api_key, region=region) - self.speech_synthesizer = SpeechSynthesizer( - speech_config=self.speech_config, audio_config=None - ) + speech_config = SpeechConfig(subscription=api_key, region=region) + self._speech_synthesizer = SpeechSynthesizer(speech_config=speech_config, audio_config=None) + self._voice = voice def can_generate_metrics(self) -> bool: @@ -62,7 +87,7 @@ class AzureTTSService(TTSService): f"{text}" " ") - result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml)) + result = await asyncio.to_thread(self._speech_synthesizer.speak_ssml, (ssml)) if result.reason == ResultReason.SynthesizingAudioCompleted: await self.stop_ttfb_metrics() @@ -75,26 +100,73 @@ class AzureTTSService(TTSService): logger.error(f"{self} error: {cancellation_details.error_details}") -class AzureLLMService(BaseOpenAILLMService): +class AzureSTTService(AIService): def __init__( self, *, api_key: str, - endpoint: str, - model: str, - api_version: str = "2023-12-01-preview"): - # Initialize variables before calling parent __init__() because that - # will call create_client() and we need those values there. - self._endpoint = endpoint - self._api_version = api_version - super().__init__(api_key=api_key, model=model) + region: str, + language="en-US", + sample_rate=16000, + channels=1, + **kwargs): + super().__init__(**kwargs) - def create_client(self, api_key=None, base_url=None): - return AsyncAzureOpenAI( - api_key=api_key, - azure_endpoint=self._endpoint, - api_version=self._api_version, - ) + speech_config = SpeechConfig(subscription=api_key, region=region) + speech_config.speech_recognition_language = language + + stream_format = AudioStreamFormat(samples_per_second=sample_rate, channels=channels) + self._audio_stream = PushAudioInputStream(stream_format) + + audio_config = AudioConfig(stream=self._audio_stream) + self._speech_recognizer = SpeechRecognizer( + speech_config=speech_config, audio_config=audio_config) + self._speech_recognizer.recognized.connect(self._on_handle_recognized) + + self._create_push_task() + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, SystemFrame): + await self.push_frame(frame, direction) + elif isinstance(frame, AudioRawFrame): + self._audio_stream.write(frame.audio) + else: + await self._push_queue.put((frame, direction)) + + async def start(self, frame: StartFrame): + self._speech_recognizer.start_continuous_recognition_async() + + async def stop(self, frame: EndFrame): + self._speech_recognizer.stop_continuous_recognition_async() + await self._push_queue.put((frame, FrameDirection.DOWNSTREAM)) + await self._push_frame_task + + async def cancel(self, frame: CancelFrame): + self._speech_recognizer.stop_continuous_recognition_async() + self._push_frame_task.cancel() + + def _create_push_task(self): + self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler()) + self._push_queue = asyncio.Queue() + + async def _push_frame_task_handler(self): + running = True + while running: + try: + (frame, direction) = await self._push_queue.get() + await self.push_frame(frame, direction) + running = not isinstance(frame, EndFrame) + except asyncio.CancelledError: + break + + def _on_handle_recognized(self, event): + if event.result.reason == ResultReason.RecognizedSpeech and len(event.result.text) > 0: + direction = FrameDirection.DOWNSTREAM + frame = TranscriptionFrame(event.result.text, "", int(time.time_ns() / 1000000)) + asyncio.run_coroutine_threadsafe( + self._push_queue.put((frame, direction)), self.get_event_loop()) class AzureImageGenServiceREST(ImageGenService):