diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index 85fe4a212..8d66714c4 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -58,14 +58,17 @@ class DailyTransportService(EventHandler): self.loop = None def patch_method(self, event_name, *args, **kwargs): - for handler in self.event_handlers[event_name]: - if inspect.iscoroutinefunction(handler): - if self.loop: - asyncio.run_coroutine_threadsafe(handler(*args, **kwargs), self.loop) + try: + for handler in self.event_handlers[event_name]: + if inspect.iscoroutinefunction(handler): + if self.loop: + asyncio.run_coroutine_threadsafe(handler(*args, **kwargs), self.loop) + else: + raise Exception("No event loop to run coroutine. In order to use async event handlers, you must run the DailyTransportService in an asyncio event loop.") else: - raise Exception("No event loop to run coroutine. In order to use async event handlers, you must run the DailyTransportService in an asyncio event loop.") - else: - handler(*args, **kwargs) + handler(*args, **kwargs) + except Exception as e: + self.logger.error(f"Exception in event handler {event_name}: {e}") def add_event_handler(self, event_name: str, handler): if not event_name.startswith("on_"): @@ -106,6 +109,14 @@ class DailyTransportService(EventHandler): "speaker", sample_rate=16000, channels=1 ) + self.image: bytes | None = None + self.camera_thread = Thread(target=self.run_camera, daemon=True) + self.camera_thread.start() + + self.logger.info("Starting frame consumer thread") + self.frame_consumer_thread = Thread(target=self.frame_consumer, daemon=True) + self.frame_consumer_thread.start() + Daily.select_speaker_device("speaker") self.client.set_user_name(self.bot_name) @@ -150,8 +161,27 @@ class DailyTransportService(EventHandler): } ) + if self.token: + self.client.start_transcription( + { + "language": "en", + "tier": "nova", + "model": "2-conversationalai", + "profanity_filter": True, + "redact": False, + "extra": { + "endpointing": True, + "punctuate": False, + }, + }, + self.transcription_callback, + ) + self.my_participant_id = self.client.participants()["local"]["id"] + def transcription_callback(self, foo, bar): + print("transcription callback", foo, bar) + async def run(self) -> None: self.configure_daily() @@ -181,29 +211,6 @@ class DailyTransportService(EventHandler): def call_joined(self, join_data, client_error): self.logger.info(f"Call_joined: {join_data}, {client_error}") - self.image: bytes | None = None - self.camera_thread = Thread(target=self.run_camera, daemon=True) - self.camera_thread.start() - - self.logger.info("Starting frame consumer thread") - self.frame_consumer_thread = Thread(target=self.frame_consumer, daemon=True) - self.frame_consumer_thread.start() - - if self.token: - self.client.start_transcription( - { - "language": "en", - "tier": "nova", - "model": "2-conversationalai", - "profanity_filter": True, - "redact": False, - "extra": { - "endpointing": True, - "punctuate": False, - }, - } - ) - def on_call_state_updated(self, state): pass @@ -219,15 +226,19 @@ class DailyTransportService(EventHandler): pass def on_transcription_message(self, message): + print("on transcription message", message) pass def on_transcription_stopped(self, stopped_by, stopped_by_error): + print("on transcription stopped", stopped_by, stopped_by_error) pass def on_transcription_error(self, message): + print("transcription error", message) pass def on_transcription_started(self, status): + print("transcription started", status) pass def set_image(self, image: bytes): diff --git a/src/samples/theoretical-to-real/05-queued.py b/src/samples/theoretical-to-real/05-queued.py index aa6da0257..5c61f8fe5 100644 --- a/src/samples/theoretical-to-real/05-queued.py +++ b/src/samples/theoretical-to-real/05-queued.py @@ -5,10 +5,9 @@ from asyncio.queues import Queue import re from dailyai.output_queue import OutputQueueFrame, FrameType -from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService -from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService -from dailyai.services.open_ai_services import OpenAILLMService, OpenAIImageGenService +from dailyai.services.azure_ai_services import AzureLLMService from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService +from dailyai.services.open_ai_services import OpenAIImageGenService from dailyai.services.daily_transport_service import DailyTransportService async def main(room_url): @@ -89,6 +88,9 @@ async def main(room_url): if participant["id"] == transport.my_participant_id: return + # This will play the months in the order they're completed. The benefit + # is we'll have as little delay as possible before the first month, and + # likely no delay between months, but the months won't display in order. for month_data_task in asyncio.as_completed(month_tasks): data = await month_data_task transport.output_queue.put( diff --git a/src/samples/theoretical-to-real/06-listen-and-respond.py b/src/samples/theoretical-to-real/06-listen-and-respond.py index 149439a65..cede0cb72 100644 --- a/src/samples/theoretical-to-real/06-listen-and-respond.py +++ b/src/samples/theoretical-to-real/06-listen-and-respond.py @@ -1,6 +1,8 @@ import argparse import asyncio import requests +import time +import urllib.parse from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureLLMService, AzureTTSService @@ -14,7 +16,7 @@ async def main(room_url:str, token): transport = DailyTransportService( room_url, token, - "Say Two Things Bot", + "Respond bot", 1, ) transport.mic_enabled = True @@ -24,34 +26,37 @@ async def main(room_url:str, token): llm = AzureLLMService() tts = AzureTTSService() + transcribed_message = "" + transcription_timeout = None + """ @transport.event_handler("on_participant_joined") async def on_joined(transport, participant): if participant["id"] == transport.my_participant_id: return - # queue two pieces of speech: one specified as a text literal, - # and one generated by an llm. We'll kick off the llm first, and let - # it generate a response while we're speaking the literal string. - llm_response_task = asyncio.create_task(llm.run_llm( - [{"role": "system", "content": "tell the user a joke about llamas"}] - )) - - async for audio_chunk in tts.run_tts("My friend the LLM is now going to tell a joke about llamas."): + async for audio_chunk in tts.run_tts("If you say something, I will respond."): transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio_chunk)) - llm_response = await llm_response_task - async for audio_chunk in tts.run_tts(llm_response): - transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio_chunk)) - - # wait for the output queue to be empty, then leave the meeting - transport.output_queue.join() - transport.stop() - @transport.event_handler("on_transcription_message") async def on_transcription_message(transport, message) -> None: - print("message received", message) + nonlocal transcribed_message + nonlocal transcription_timeout + print(message) + if message["session_id"] != transport.my_participant_id: + transcribed_message += message['text'] - await transport.run() + print("message received", transcribed_message) + + @transport.event_handler("on_transcription_error") + def on_transcription_error(transport, status) -> None: + print("transcription error", status) + + @transport.event_handler("on_transcription_started") + def on_transcription_started(transport, status) -> None: + print("transcription started", status) + """ + #await transport.run() + transport.run() if __name__ == "__main__":