fixed bot thread loop

app/auth.py

@@ -8,6 +8,7 @@ import os
 
 load_dotenv()
 
+
 def get_meeting_token(room_name, daily_api_key, token_expiry):
     api_path = os.getenv('DAILY_API_PATH') or 'https://api.daily.co/v1'
 
@@ -18,6 +19,8 @@ def get_meeting_token(room_name, daily_api_key, token_expiry):
                         json={'properties': {'room_name': room_name, 'is_owner': True, 'exp': token_expiry}})
     if res.status_code != 200:
         return json.dumps({'error': 'Unable to create meeting token', 'detail': res.text}), 500
+    
+    #@TODO handle errors here
     meeting_token = res.json()['token']
     return meeting_token
 

app/bot.py

@@ -1,8 +1,11 @@
 import argparse
+import queue
+import time
+import threading
 import logging
 import os
-import time
-from threading import Thread
+
+from PIL import Image
 from typing import Any, Mapping
 
 from daily import EventHandler, CallClient, Daily
@@ -13,131 +16,112 @@ from auth import get_meeting_token, get_room_name
 
 load_dotenv()
 
-class DailyLLM(EventHandler):
+class DailyVision(EventHandler):
     def __init__(
         self,
-        room_url=os.getenv("DAILY_URL"),
-        token=os.getenv("DAILY_TOKEN"),
-        bot_name="TestBot",
+        room_url,
+        room_name,
+        expiration,
+        bot_name="Daily Bot",
     ):
-        duration = os.getenv("BOT_MAX_DURATION")
-        if not duration:
-            duration = 300
-        else:
-            duration = int(duration)
-        self.expiration = time.time() + duration
-
-        # room + bot details
-        self.room_url = room_url
-        room_name = get_room_name(room_url)
-        if token:
-            self.token = token
-        else:
-            self.token = get_meeting_token(
-                room_name, os.getenv("DAILY_API_KEY"), self.expiration
-            )
-        self.bot_name = bot_name
-
-        self.finished_talking_at = None
-
-        FORMAT = f"%(asctime)s {room_name} %(message)s"
+        self.__client = CallClient(event_handler=self)
+        self.__pipeline = None
+        self.__camera = None
+        self.__time = time.time()
+        self.__queue = queue.Queue()
+        self.__app_quit = False
+        self.__bot_name = bot_name
+        self.__room_url = room_url
+        self.__room_name = room_name
+        self.__expiration = expiration
+
+        # Configure logger
+        FORMAT = f"%(asctime)s {self.__room_url} %(message)s"
         logging.basicConfig(format=FORMAT)
         self.logger = logging.getLogger("bot-instance")
         self.logger.setLevel(logging.DEBUG)
 
-        self.logger.info(f"Joining as {self.bot_name}")
-        self.logger.info(f"Joining room {self.room_url}")
-        self.logger.info(
-            f"expiration: {datetime.utcfromtimestamp(self.expiration).strftime('%Y-%m-%d %H:%M:%S')}"
-        )
-        self.logger.info(
-            f"now: {datetime.utcfromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')}"
-        )
-
-        self.my_participant_id = None
-
-        self.logger.info("configuring daily")
-        self.configure_daily()
-
-        self.stop_threads = False
-        self.image = None
+        self.logger.info(f"Expiration timer set to: {self.__expiration}")
 
-        #self.logger.info("starting camera thread")
-        #self.camera_thread = Thread(target=self.run_camera)
-        #self.camera_thread.start()
+        # Start thread
+        self.__thread = threading.Thread(target = self.process_frames)
+        self.__thread.start()
 
-        self.participant_left = False
-        self.last_fragment_at = None
+    def run(self, meeting_url, token):
+        # Join
+        self.logger.info(f"Connecting to room {meeting_url} as {self.__bot_name}")
+        self.__client.set_user_name(self.__bot_name)
+        self.__client.join(meeting_url, token, completion=self.on_joined)
 
-        try:
-            participant_count = len(self.client.participants())
-            self.logger.info(f"{participant_count} participants in room")
-            while time.time() < self.expiration and not self.participant_left:
-                time.sleep(1)
-        except Exception as e:
-            self.logger.error(f"Exception {e}")
-        finally:
-            self.client.leave()
+        #self.__participant_id = self.client.participants()["local"]["id"]
 
-        self.stop_threads = True
-        self.logger.info("Shutting down")
-        #self.camera_thread.join()
-        #self.logger.info("camera thread stopped")
-        #self.logger.info("Services closed.")
+        # Keep-alive on thread
+        self.__thread.join()
+ 
+    def leave(self):
+        self.logger.info(f"Leaving...")
+        self.__app_quit = True
+        self.__thread.join()
+        self.__client.leave()
 
-    def configure_daily(self):
-        Daily.init()
-        self.client = CallClient(event_handler=self)
-
-        self.mic = Daily.create_microphone_device("mic", sample_rate=16000, channels=1)
-        self.speaker = Daily.create_speaker_device(
-            "speaker", sample_rate=16000, channels=1
-        )
-        self.camera = Daily.create_camera_device(
-            "camera", width=1024, height=1024, color_format="RGB"
-        )
+    def on_joined(self, join_data, client_error):
+        self.logger.info(f"call_joined: {join_data}, {client_error}")
 
-        Daily.select_speaker_device("speaker")
+    def on_participant_joined(self, participant):
+        self.logger.info(f"Participant {participant['id']} joined, analyzing frames...")
+        self.__client.set_video_renderer(participant["id"], self.on_video_frame) 
 
-        self.client.set_user_name(self.bot_name)
-        self.client.join(self.room_url, completion=self.call_joined)
-        #self.client.join(self.room_url, self.token, completion=self.call_joined)
+        # Say hello
+        self.wave()
 
-        self.client.update_inputs(
-            {
+    def setup_camera(self, video_frame):
+        if not self.__camera:
+            self.__camera = Daily.create_camera_device("camera",
+                                                       width = video_frame.width,
+                                                       height = video_frame.height,
+                                                       color_format="RGB")
+            self.__client.update_inputs({
                 "camera": {
                     "isEnabled": True,
                     "settings": {
-                        "deviceId": "camera",
-                        "frameRate": 5,
-                    },
-                },
-                "microphone": {"isEnabled": True, "settings": {"deviceId": "mic"}},
-            }
-        )
-
-        self.my_participant_id = self.client.participants()["local"]["id"]
-
-    def call_joined(self, join_data, client_error):
-        self.logger.info(f"call_joined: {join_data}, {client_error}")
-
-    def on_participant_joined(self, participant):
-        self.logger.info(f"on_participant_joined: {participant}")
-        #self.client.send_app_message({"event": "story-id", "storyID": self.story_id})
-        self.wave()
-        time.sleep(2)
-
-    def on_participant_left(self, participant, reason):
-        if len(self.client.participants()) < 2:
-            self.logger.info("participant left")
-            self.participant_left = True
-
-    def wave(self):
-        self.client.send_app_message(
+                        "deviceId": "camera"
+                    }
+                }
+            })
+
+    def process_frames(self):
+        while not self.__app_quit:
+            # Check expiry timer
+            if time.time() > self.__expiration:
+                self.logger.info(f"Expiration timer exceeded. Exiting...")
+                self.__app_quit = True
+                return
+            try:
+                video_frame = self.__queue.get(timeout=5)
+
+                if video_frame:
+                    image = Image.frombytes("RGBA", (video_frame.width, video_frame.height), video_frame.buffer)
+                    result = self.__pipeline(image)
+
+                    pil = Image.fromarray(result.render()[0], mode="RGB").tobytes()
+
+                    self.__camera.write_frame(pil)
+            except queue.Empty:
+                pass
+
+    def on_video_frame(self, participant_id, video_frame):
+        # Process ~15 frames per second (considering incoming frames at 30fps).
+        if time.time() - self.__time > 0.05:
+            self.__time = time.time()
+            self.setup_camera(video_frame)
+            self.__queue.put(video_frame)    
+
+    def wave(self, emoji="👋"):
+        self.__client.send_app_message(
             {
                 "event": "sync-emoji-reaction",
                 "reaction": {
-                    "emoji": "👋",
+                    "emoji": emoji,
                     "room": "main-room",
                     "sessionId": "bot",
                     "id": time.time(),
@@ -145,17 +129,39 @@ class DailyLLM(EventHandler):
             }
         )
 
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description="Daily LLM bot")
-    parser.add_argument("-u", "--url", type=str, help="URL of the Daily room")
-    parser.add_argument("-t", "--token", type=str, help="Token for Daily API")
-    parser.add_argument("-b", "--bot-name", type=str, help="Name of the bot")
-
+def main():
+    parser = argparse.ArgumentParser(description="Daily Bot")
+    # Required args
+    parser.add_argument("-u", "--url", required=True, type=str, help="URL of the Daily room")
+    parser.add_argument("-k", "--api_key", required=True, type=str, help="Daily API key")
+    # Optional args
+    parser.add_argument("-t", "--private", type=bool, help="Is this room private?", default=True)
+    parser.add_argument("-n", "--bot-name", type=str, help="Name of the bot", default="Daily Bot")
+    parser.add_argument("-e", "--expiration", type=int, help="Duration of bot", default=os.getenv("BOT_MAX_DURATION", 300))
     args = parser.parse_args()
-
-    url = args.url or os.getenv("DAILY_URL")
-    bot_name = args.bot_name or "TestBot"
-    token = args.token or None
-
-    app = DailyLLM(url, token, bot_name)
+    
+    Daily.init()
+
+    expiration = time.time() + args.expiration
+    room_name = get_room_name(args.url)
+
+    # Retrieve a meeting token, if not provided 
+    #@TODO do room lookup to check privacy
+    if args.private:
+        token = get_meeting_token(room_name, args.api_key, expiration)
+
+    app = DailyVision(args.url, room_name, expiration, args.bot_name)
+    
+    try :
+        app.run(args.url, token)
+    except KeyboardInterrupt:
+        print("Ctrl-C detected. Exiting!")
+    finally:
+        print("Bot loop completed. Exiting")
+        app.leave()
+
+    # Let leave finish
+    time.sleep(2)
+
+if __name__ == '__main__':
+    main()

requirements.txt

@@ -5,4 +5,5 @@ imageio
 requests
 fastapi
 uvicorn[standard]
-requests
+requests
+pillow

server.py

@@ -9,7 +9,7 @@ import requests
 import subprocess
 import time
 
-from app.auth import get_meeting_token
+from app.auth import get_room_name
 
 load_dotenv()
 
@@ -27,13 +27,9 @@ app.add_middleware(
 app.mount("/static", StaticFiles(directory="frontend/out", html=True), name="static")
 
 def _start_bot(bot_path, args=None):
-    daily_api_key = os.getenv("DAILY_API_KEY") or ""
+    '''
     api_path = os.getenv("DAILY_API_PATH") or "https://api.daily.co/v1"
 
-    timeout = int(os.getenv("ROOM_TIMEOUT") or os.getenv("BOT_MAX_DURATION") or 300)
-    exp = time.time() + timeout
-
-    '''
     res = requests.post(
         f"{api_path}/rooms",
         headers={"Authorization": f"Bearer {daily_api_key}"},
@@ -59,10 +55,11 @@ def _start_bot(bot_path, args=None):
             500,
         )
     '''
-    room_url = os.getenv("DAILY_ROOM_URL") #res.json()["url"]
-    room_name = os.getenv("DAILY_ROOM_NAME") #res.json()["name"]
 
-    meeting_token = get_meeting_token(room_url, daily_api_key, exp)
+    daily_room_url = os.getenv("DAILY_ROOM_URL") #res.json()["url"]
+    daily_room_name = get_room_name(daily_room_url)
+    daily_api_path = os.get("DAILY_API_PATH")
+    daily_api_key = os.getenv("DAILY_API_KEY")
 
     if args:
         extra_args = " ".join([f'-{x[0]} "{x[1]}"' for x in args])
@@ -71,7 +68,7 @@ def _start_bot(bot_path, args=None):
 
     proc = subprocess.Popen(
         [
-            f"python3 {bot_path} -u {room_url} -t {meeting_token} {extra_args}"
+            f"python3 {bot_path} -u {daily_room_url} -k {daily_api_key} {extra_args}"
         ],
         shell=True,
         bufsize=1,
@@ -83,14 +80,15 @@ def _start_bot(bot_path, args=None):
         time.sleep(0.1)
         attempts += 1
         res = requests.get(
-            f"{api_path}/rooms/{room_name}/get-session-data",
+            f"{daily_api_path}/rooms/{daily_room_name}/get-session-data",
             headers={"Authorization": f"Bearer {daily_api_key}"},
         )
         if res.status_code == 200:
             break
-    print(f"Took {attempts} attempts to join room {room_name}")
+    print(f"Took {attempts} attempts to join room {daily_room_name}")
+
+    return JSONResponse({"room_url": daily_room_url})
 
-    return JSONResponse({"room_url": room_url, "token": meeting_token})
 
 @app.post("/start")
 async def start():