Use a decorator for callbacks

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

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}")

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")

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
+    async def play_audio(transport, audio_generator):
-    # chunks of audio to play sequentially. the "mic" object is a handle
+        async for audio in audio_generator:
-    # we can use to inspect and control the queue if we need to. in this
+            transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio))
-    # case we will pipe into this queue from the tts service
+
+    # 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?
+    # Register an event handler so we can play the audio when the participant joins.
-    transport.set_audio_generator(audio_generator)
+    @transport.event_handler("on_participant_joined")
-    try:
+    def on_participant_joined(transport, participant):
-        transport.run()
+        print("participant joined", participant)
-    except KeyboardInterrupt:
+        asyncio.run(play_audio(transport, audio_generator))
-        print("Keyboard interrupt")
-    finally:
         transport.stop()
 
+    transport.run()
+
 
 if __name__ == "__main__":
     asyncio.run(main("https://moishe.daily.co/Lettvins"))