add dynamic cameras and youtube chat

README.md

@@ -1,6 +1,6 @@
 ---
 title: The Emergent Show
-emoji: 🌃
+emoji: 📺
 colorFrom: green
 colorTo: purple
 sdk: gradio

app.py

@@ -11,8 +11,10 @@ import random
 import threading
 from timing import TimeManager
 from validity import is_valid_rpm_url
+from yt_chat import StreamChatHost
+from prompts import system_prompt_tv_crew_guest, system_prompt_tv_crew_chat
 
-# Remove these to upload on Spaces
+# Remove these before uploading on Spaces
 # import dotenv
 
 # dotenv.load_dotenv()
@@ -27,20 +29,34 @@ client_guard = OpenAI(base_url=guard_api_base, api_key=guard_api_key)
 unreal_orchestrator_url = os.getenv("ORCHESTRATOR_URL")
 api_key_unreal = os.getenv("API_KEY_UNREAL")
 
-show_state = {"current_guest": None, "time_since_last_guest_message": 0}
+
+show_state = {
+    "current_guest": None,
+    "time_since_last_guest_message": 0,
+}
 turn_limit = 7
 
 host = Host(client, show_state)
-tv_crew = TVCrew(client)
+
+tv_crew = TVCrew(client, system_prompt_tv_crew_guest)
+tv_crew_chat = TVCrew(client, system_prompt_tv_crew_chat)
+
 audience = Audience(client)
 guardian = Guardian(client_guard)
 
 timeManager = TimeManager(host, show_state)
+streamChatHost = StreamChatHost(client, tv_crew_chat, guardian)
 
-t = threading.Thread(target=timeManager.guest_time_limit, daemon=True)
-t.start()
+thread_time_manager = threading.Thread(target=timeManager.guest_time_limit, daemon=True)
+thread_time_manager.start()
 
-in_construction = True
+thread_stream_chat = threading.Thread(
+    target=streamChatHost.chat_interaction_loop, daemon=True
+)
+thread_stream_chat.start()
+
+
+in_construction = False
 
 
 def join_show(avatar_url):
@@ -77,6 +93,7 @@ def join_show(avatar_url):
     host.clear_context()
     audience.clear_context()
     tv_crew.clear_context()
+
     return f"You have joined the show. Your guest_name: {name}"
 
 
@@ -96,7 +113,6 @@ def speak(guest_name, text):
         return "Join the show first! And then use the guest_name you get"
 
     headers = {"Content-Type": "application/json", "x-api-key": api_key_unreal}
-
     is_safe, categories = guardian.moderate_message(text)
     if not is_safe:
         requests.post(url=f"{unreal_orchestrator_url}/wrapup", json={}, headers=headers)
@@ -114,21 +130,19 @@ def speak(guest_name, text):
     # **************************************************************************
 
     # Get TV suggestion
-    image_url, current_tv_image_caption = None, None
+    image_base64, current_tv_image_caption = None, None
 
     # Update TV 50% of the time
-    if random.random() < 0.7:
-        image_url, current_tv_image_caption = tv_crew.suggest_image()
-        if image_url:
+    if random.random() < 0.5:
+        image_base64, current_tv_image_caption = tv_crew.suggest_image()
+        if image_base64:
             # Let host know that Television has been updated
             tv_info_suffix = (
                 f"\n[Crew Updated Television to: {current_tv_image_caption}]\n"
             )
             text = text + tv_info_suffix
-
-    # Pass in image_url which will be displayed just after the guest has finished speaking
-    if image_url:
-        payload["image_url"] = image_url
+            # Pass in image_url which will be displayed just after the guest has finished speaking
+            payload["base64"] = image_base64
 
     # **************************************************************************
 
@@ -171,6 +185,9 @@ def speak(guest_name, text):
 
     if shouldWrapUp:
         requests.post(url=f"{unreal_orchestrator_url}/wrapup", json={}, headers=headers)
+        tv_crew.clear_context()
+        audience.clear_context()
+        host.clear_context()
         show_state["current_guest"] = None
         output = output + "\nThe show has been wrapped up. Thank you for joining!"
 
@@ -182,9 +199,7 @@ with gr.Blocks() as demo:
     gr.Markdown("# The Emergent Show 🍻")
     gr.Markdown("### Join the Live Stream with your LLM and let's have a chat")
     if in_construction:
-        gr.Markdown(
-            "# 🚧 Under construction. Please check back later! 🚧"
-        )
+        gr.Markdown("# 🚧 Under construction. Please check back later! 🚧")
     with gr.Tab("Join"):
         gr.HTML(
             """
@@ -253,7 +268,7 @@ with gr.Blocks() as demo:
            left:0%;
            width:100%;
            height:100%;"
-src="https://www.youtube.com/embed/VhvLqnE1k5M?si=WSmI4Ypo8Bm3JbEO" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" referrerpolicy="strict-origin-when-cross-origin" allowfullscreen></iframe>
+src="https://www.youtube.com/embed/xBVJIJQg1FY?si=tsCm1DdajNCP45ho" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" referrerpolicy="strict-origin-when-cross-origin" allowfullscreen></iframe>
 </div>
 """
         )

guard.py

@@ -1,4 +1,6 @@
 
+
+
 class Guardian:
     def __init__(self, client):
         self.model = "Qwen/Qwen3Guard-Gen-4B"
@@ -8,12 +10,16 @@ class Guardian:
         messages = [
             {"role": "user", "content": prompt}
         ]
-        chat_completion = self.client.chat.completions.create(
-            messages=messages,
-            model=self.model)
-        response = chat_completion.choices[0].message.content
-        safety, categories = response.split("\n")
-        safety = safety.replace("Safety:", "").strip()
-        categories = categories.replace("Categories:", "").strip()
-        is_safe = True if "Safe" in safety else False
-        return is_safe, categories
+        try:
+            chat_completion = self.client.chat.completions.create(
+                messages=messages,
+                model=self.model)
+            response = chat_completion.choices[0].message.content
+            safety, categories = response.split("\n")
+            safety = safety.replace("Safety:", "").strip()
+            categories = categories.replace("Categories:", "").strip()
+            is_safe = True if "Safe" in safety else False
+            return is_safe, categories
+        except Exception as e:
+            print(f"Error in Guardian moderate_message: {e}")
+            return False, "Guard down!"

host.py

@@ -19,7 +19,6 @@ Keep your responses short, like two or three sentences, and be witty."""
         self.show_state["time_since_last_guest_message"] = 0
         shouldWrapUp = False
         if len(self.messages) == 1:
-            print(self.messages)
             msg_prefix = f"<Production Instructions>\nA guest entered the show, now they will talk to you\n</Production Instructions>\n"
             user_message = msg_prefix + user_message + "\n"
 

image_api.py

@@ -6,6 +6,7 @@ load_dotenv()
 base_url = "https://api.pexels.com/v1"
 api_key = os.getenv("API_KEY_PEXELS")
 
+
 def get_random_image(topic: str) -> tuple[str, str] | tuple[None, None]:
     # Returns the url of a random image on a topic
     params = {"query": topic, "per_page": 1}
@@ -18,8 +19,6 @@ def get_random_image(topic: str) -> tuple[str, str] | tuple[None, None]:
         photo = response_json["photos"][0]
         landscape_url = photo["src"]["landscape"]
         alt = photo["alt"]
-        print("Found landscape URL:")
-        print(landscape_url)
     except:
         landscape_url, alt = None, None
     return landscape_url, alt

prompts.py

@@ -0,0 +1,15 @@
+# TV Crew
+
+system_prompt_tv_crew_guest = """You are a crew member of a Live Talk show who has the responsibility to update Television Images based on the conversation going on between host and the guest.
+You will be provided the history of conversation, and you have to output in json, a specific keyword like "green forest" or "haunted mansion" etc. which can be used to fetch related image and show them on the Television.
+Format: {"image_keyword": "Desired keyword"}
+If you think guest is talking about previous image or an image isn't necessary for the conversation then just output none
+{"image_keyword": "none"}
+"""
+
+system_prompt_tv_crew_chat = """You are a crew member of a Live Talk show who has the responsibility to update Television Images based on the conversation going on between host and live chat.
+You will be provided the history of conversation, and you have to output in json, a specific keyword like "green forest" or "haunted mansion" etc. which can be used to fetch related image and show them on the Television.
+Format: {"image_keyword": "Desired keyword"}
+If you think people are talking about previous image or an image isn't necessary for the conversation then just output none
+{"image_keyword": "none"}
+"""

requirements.txt

@@ -1 +1,2 @@
-openai
+openai
+google-api-python-client

timing.py

@@ -13,9 +13,6 @@ class TimeManager:
     def __init__(self, host: Host, show_state: dict):
         self.show_state = show_state
         self.single_response_time_limit = 90  # seconds
-        self.unreal_orchestrator_url = (
-            unreal_orchestrator_url  # Replace with actual URL
-        )
         self.headers = {"Content-Type": "application/json", "x-api-key": api_key_unreal}
         self.host = host
 

tv_crew.py

@@ -1,6 +1,8 @@
 from pydantic import BaseModel
 import json
 from image_api import get_random_image
+import base64
+import requests
 
 
 class TVCrewOutput(BaseModel):
@@ -8,14 +10,9 @@ class TVCrewOutput(BaseModel):
 
 
 class TVCrew:
-    def __init__(self, client):
+    def __init__(self, client, system_prompt: str):
         self.model = "google/gemma-3-12b-it"
-        self.system_prompt = """You are a crew member of a Live Talk show who has the responsibility to update Television Images based on the conversation going on between host and the guest.
-        You will be provided the history of conversation, and you have to output in json, a specific keyword like "green forest" or "haunted mansion" etc. which can be used to fetch related image and show them on the Television.
-        Format: {"image_keyword": "Desired keyword"}
-        If you think guest is talking about previous image or an image isn't necessary for the conversation then just output none
-        {"image_keyword": "none"}
-        """
+        self.system_prompt = system_prompt
         self.context = ""
         self.client = client
 
@@ -23,14 +20,10 @@ class TVCrew:
         """
         Finds a image related to conversation and returns the url and caption
         """
-        # print("*************************** TV Crew ***************************")
-        # print(self.context)
-        # print("***************************************************************")
         messages_tv_crew = [
             {"role": "system", "content": self.system_prompt},
             {"role": "user", "content": self.context},
         ]
-        print("Crew is on it...")
         response_tv_crew = self.client.chat.completions.parse(
             model=self.model,
             messages=messages_tv_crew,
@@ -46,8 +39,9 @@ class TVCrew:
         if image_url:
             caption = alt if len(alt) > 0 else image_keyword
             self.add_context(f"TV Shows: {caption}\n")
-
-            return image_url, caption
+            img_bytes = requests.get(image_url).content
+            image_base64 = base64.b64encode(img_bytes).decode("utf-8")
+            return image_base64, caption
         return None, None
 
     def add_context(self, content):
@@ -55,4 +49,4 @@ class TVCrew:
         self.context += content
 
     def clear_context(self):
-        self.context = ""
+        self.context = ""

yt_chat.py

@@ -0,0 +1,142 @@
+import os
+# from dotenv import load_dotenv
+from guard import Guardian
+from tv_crew import TVCrew
+import requests
+import time
+import googleapiclient.discovery
+
+# load_dotenv()
+unreal_orchestrator_url = os.getenv("ORCHESTRATOR_URL")
+api_key_unreal = os.getenv("API_KEY_UNREAL")
+
+API_KEY = os.getenv("YOUTUBE_API_KEY")
+VIDEO_ID="xBVJIJQg1FY"
+POLL_INTERVAL = 10  # seconds
+
+
+class StreamChatHost:
+    def __init__(self, client, tv_crew: TVCrew, guardian: Guardian):
+        self.client = client
+        self.guardian = guardian
+        self.tv_crew = tv_crew
+        self.model = "deepseek/deepseek-v3.2-exp"
+        system_prompt = """You are the Host of a Talk Show, where users come and have a chat.
+There is a TV Display near you which the crew members use to display images based on your conversation. 
+The TV is not in control of the guest. It is controlled by crew members and is only for graphic uplifting of the conversation.
+You will be told when TV shows something, otherwise, don't talk about it.
+Keep your responses short, like two or three sentences, and be witty."""
+        self.system_message = {"role": "system", "content": system_prompt}
+
+        self.messages = [self.system_message]
+        self.headers = {"Content-Type": "application/json", "x-api-key": api_key_unreal}
+        self.youtube = googleapiclient.discovery.build(
+            "youtube", "v3", developerKey=API_KEY
+        )
+        self.chat_id = self.get_live_chat_id(self.youtube, VIDEO_ID)
+        self.next_page_token = None
+
+    def chat_interaction_loop(self):
+        """Checks if the Host is in lobby, then pulls chat messages from the Live Stream. The Host then replies to the message."""
+        while True:
+            time.sleep(POLL_INTERVAL)
+            request = requests.get(
+                f"{unreal_orchestrator_url}/status", headers=self.headers
+            )
+            in_lobby = request.json().get("in_lobby", False)
+            if not in_lobby:
+                continue
+            if len(self.messages) > 20:
+                self.messages = [self.system_message]
+                self.tv_crew.clear_context()
+            try:
+                request = self.youtube.liveChatMessages().list(
+                    liveChatId=self.chat_id,
+                    part="snippet,authorDetails",
+                    maxResults=20,
+                    pageToken=self.next_page_token,
+                )
+                response = request.execute()
+                user_messages = []
+                for item in response.get("items", []):
+                    author_name = item["authorDetails"]["displayName"]
+                    message_text = item["snippet"]["displayMessage"]
+                    user_messages.append(f"{author_name}: {message_text}")
+
+                self.next_page_token = response.get("nextPageToken")
+                if len(user_messages) == 0:
+                    continue
+            except Exception as e:
+                continue
+            # **************************************************************
+
+            user_message = "[YouTube Chat]\n"
+            for msg in user_messages:
+                if len(msg) <= 500:
+                    user_message += f"{msg}\n"
+            is_safe, _ = self.guardian.moderate_message(user_message)
+
+            if not is_safe:
+                continue
+
+            self.tv_crew.add_context(user_message)
+            image_base64, caption = self.tv_crew.suggest_image()
+            if image_base64:
+                is_safe, _ = self.guardian.moderate_message(caption)
+                if not is_safe:
+                    image_base64 = None
+                    caption = None
+                else:
+                    user_message += f"[Crew Updated Television to: {caption}]\n"
+
+            self.messages.append({"role": "user", "content": user_message})
+            try:
+                response = self.client.responses.create(
+                    model=self.model, input=self.messages
+                )
+                host_response = response.output_text
+                self.messages.append({"role": "assistant", "content": host_response})
+            except Exception as e:
+                print(f"Error generating host response: {e}")
+                continue
+
+            self.tv_crew.add_context(f"Host: {host_response}\n")
+            # Just to make sure, that we're in the lobby
+            request = requests.get(
+                f"{unreal_orchestrator_url}/status", headers=self.headers
+            )
+            in_lobby = request.json().get("in_lobby", False)
+            if not in_lobby:
+                continue
+
+            if image_base64:
+                payload_tv = {"base64": image_base64}
+                response = requests.post(
+                    url=f"{unreal_orchestrator_url}/television",
+                    json=payload_tv,
+                    headers=self.headers,
+                )
+
+            payload = {"text": host_response, "is_host": True}
+            response = requests.post(
+                url=f"{unreal_orchestrator_url}/tts",
+                json=payload,
+                headers=self.headers,
+            )
+
+    def get_live_chat_id(self, youtube, video_id):
+        try:
+            request = youtube.videos().list(part="liveStreamingDetails", id=video_id)
+            response = request.execute()
+            if not response["items"]:
+                print("No video found with the given ID.")
+                return None
+
+            details = response["items"][0]["liveStreamingDetails"]
+            if not details:
+                print("No live streaming details found for this video.")
+                return None
+            return details.get("activeLiveChatId", None)
+        except Exception as e:
+            print(f"An error occurred: {e}")
+            return None