From 712ab97f88e0f364522608f31a628b011c7e7c9f Mon Sep 17 00:00:00 2001
From: Moishe Lettvin <moishel@gmail.com>
Date: Mon, 8 Jan 2024 11:27:58 -0500
Subject: [PATCH] Support coroutine callbacks

---
 .../services/daily_transport_service.py       | 27 ++++++++++++++++---
 .../theoretical-to-real/01-say-one-thing.py   | 16 +++++------
 2 files changed, 30 insertions(+), 13 deletions(-)

diff --git a/src/dailyai/services/daily_transport_service.py b/src/dailyai/services/daily_transport_service.py
index ed1c22416..426a8014f 100644
--- a/src/dailyai/services/daily_transport_service.py
+++ b/src/dailyai/services/daily_transport_service.py
@@ -1,3 +1,4 @@
+import asyncio
 import inspect
 import logging
 import time
@@ -42,13 +43,29 @@ class DailyTransportService(EventHandler):
         self.mic_sample_rate = 16000
         self.camera_enabled = False
 
+        self.camera_thread = None
+        self.frame_consumer_thread = None
+
         self.logger: logging.Logger = logging.getLogger("dailyai")
 
         self.event_handlers = {}
 
-    def monkeypatch(self, event_name, *args):
+        try:
+            self.loop = asyncio.get_running_loop()
+        except RuntimeError:
+            self.loop = None
+
+    def patch_method(self, event_name, *args):
         for handler in self.event_handlers[event_name]:
-            handler(*args)
+            if inspect.iscoroutinefunction(handler):
+                if self.loop:
+                    future = asyncio.run_coroutine_threadsafe(handler(*args), self.loop)
+                    #concurrent.futures.wait(future)
+                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.")
+                asyncio.run(handler(*args))
+            else:
+                handler(*args)
 
     def add_event_handler(self, event_name: str, handler):
         if not event_name.startswith("on_"):
@@ -60,7 +77,7 @@ class DailyTransportService(EventHandler):
 
         if not event_name in self.event_handlers:
             self.event_handlers[event_name] = [getattr(self, event_name), types.MethodType(handler, self)]
-            setattr(self, event_name, partial(self.monkeypatch, event_name))
+            setattr(self, event_name, partial(self.patch_method, event_name))
         else:
             self.event_handlers[event_name].append(types.MethodType(handler, self))
 
@@ -191,6 +208,9 @@ class DailyTransportService(EventHandler):
                 }
             )
 
+    def on_call_state_updated(self, state):
+        pass
+
     def on_participant_joined(self, participant):
         pass
 
@@ -224,7 +244,6 @@ class DailyTransportService(EventHandler):
                 time.sleep(1.0 / 8)  # 8 fps
         except Exception as e:
             self.logger.error(f"Exception {e} in camera thread.")
-        print("exiting run_camera thread")
 
     def frame_consumer(self):
         self.logger.info("🎬 Starting frame consumer thread")
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 b87483ac8..22a671542 100644
--- a/src/samples/theoretical-to-real/01-say-one-thing.py
+++ b/src/samples/theoretical-to-real/01-say-one-thing.py
@@ -1,4 +1,5 @@
 import asyncio
+import time
 from typing import AsyncGenerator
 
 from dailyai.output_queue import OutputQueueFrame, FrameType
@@ -14,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
+    meeting_duration_minutes = 1
     transport = DailyTransportService(
         room_url,
         None,
@@ -26,20 +27,17 @@ async def main(room_url):
     # similarly, create a tts service
     tts = AzureTTSService()
 
-    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")
 
     # 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()
+    async def on_participant_joined(transport, participant):
+        if participant["info"]["isLocal"]:
+            return
+        async for audio in audio_generator:
+            transport.output_queue.put(OutputQueueFrame(FrameType.AUDIO_FRAME, audio))
 
     transport.run()