diff --git a/src/dailyai/output_queue.py b/src/dailyai/output_queue.py index 11b36b95e..bfd1760d4 100644 --- a/src/dailyai/output_queue.py +++ b/src/dailyai/output_queue.py @@ -11,4 +11,4 @@ class FrameType(Enum): @dataclass(frozen=True) class OutputQueueFrame: frame_type: FrameType - frame_data: bytes + frame_data: bytes | None diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py index c73c988d6..ed1c22416 100644 --- a/src/dailyai/services/daily_transport_service.py +++ b/src/dailyai/services/daily_transport_service.py @@ -64,6 +64,13 @@ class DailyTransportService(EventHandler): else: self.event_handlers[event_name].append(types.MethodType(handler, self)) + def event_handler(self, event_name: str): + def decorator(handler): + self.add_event_handler(event_name, handler) + return handler + + return decorator + def configure_daily(self): Daily.init() self.client = CallClient(event_handler=self) @@ -141,7 +148,7 @@ class DailyTransportService(EventHandler): try: participant_count: int = len(self.client.participants()) self.logger.info(f"{participant_count} participants in room") - while time.time() < self.expiration and not self.participant_left: + while time.time() < self.expiration and not self.participant_left and not self.stop_threads.is_set(): # all handling of incoming transcriptions happens in on_transcription_message time.sleep(1) except Exception as e: @@ -149,14 +156,14 @@ class DailyTransportService(EventHandler): finally: self.client.leave() - def stop(self): - self.stop_threads.set() if self.camera_thread and self.camera_thread.is_alive(): self.camera_thread.join() if self.frame_consumer_thread and self.frame_consumer_thread.is_alive(): self.output_queue.put(OutputQueueFrame(FrameType.END_STREAM, None)) self.frame_consumer_thread.join() - self.client.leave() + + def stop(self): + self.stop_threads.set() def call_joined(self, join_data, client_error): self.logger.info(f"Call_joined: {join_data}, {client_error}") diff --git a/src/dailyai/services/foo.py b/src/dailyai/services/foo.py new file mode 100644 index 000000000..cb7a097dd --- /dev/null +++ b/src/dailyai/services/foo.py @@ -0,0 +1,10 @@ +def hello(p): + def decorator(f): + print("running f", p) + return f + + return decorator + +@hello("world") +def foo(): + print("hi") diff --git a/src/samples/theoretical-to-real/01-say-one-thing.py b/src/samples/theoretical-to-real/01-say-one-thing.py index 514d401b9..b87483ac8 100644 --- a/src/samples/theoretical-to-real/01-say-one-thing.py +++ b/src/samples/theoretical-to-real/01-say-one-thing.py @@ -5,29 +5,6 @@ from dailyai.output_queue import OutputQueueFrame, FrameType from dailyai.services.daily_transport_service import DailyTransportService from dailyai.services.azure_ai_services import AzureTTSService -class Sample01Transport(DailyTransportService): - def __init__( - self, - room_url: str, - token: str | None, - bot_name: str, - duration: float = 10, - ): - super().__init__(room_url, token, bot_name, duration) - - def set_audio_generator(self, audio_generator) -> None: - self.audio_generator = audio_generator - - async def play_audio(self): - print("playing audio", self.audio_generator) - async for audio in self.audio_generator: - print("putting frame on queue") - self.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio)) - - def on_participant_joined(self, participant): - super().on_participant_joined(participant) - asyncio.run(self.play_audio()) - async def main(room_url): # create a transport service object using environment variables for # the transport service's API key, room url, and any other configuration. @@ -38,7 +15,7 @@ async def main(room_url): # the abstract transport service APIs presumably can map pretty closely # to the daily-python basic API meeting_duration_minutes = 4 - transport = Sample01Transport( + transport = DailyTransportService( room_url, None, "Say One Thing", @@ -49,21 +26,23 @@ async def main(room_url): # similarly, create a tts service tts = AzureTTSService() - # ask the transport to create a local audio "device"/queue for - # chunks of audio to play sequentially. the "mic" object is a handle - # we can use to inspect and control the queue if we need to. in this - # case we will pipe into this queue from the tts service + async def play_audio(transport, audio_generator): + async for audio in audio_generator: + transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio)) + + # Get the generator for the audio. This will start running in the background, + # and when we ask the generator for its items, we'll get what it's generated. audio_generator: AsyncGenerator[bytes, None] = tts.run_tts("hello world") - # Should this just happen when we create the object? - transport.set_audio_generator(audio_generator) - try: - transport.run() - except KeyboardInterrupt: - print("Keyboard interrupt") - finally: + # Register an event handler so we can play the audio when the participant joins. + @transport.event_handler("on_participant_joined") + def on_participant_joined(transport, participant): + print("participant joined", participant) + asyncio.run(play_audio(transport, audio_generator)) transport.stop() + transport.run() + if __name__ == "__main__": asyncio.run(main("https://moishe.daily.co/Lettvins"))