Working vision example

src/dailyai/pipeline/frames.py

@@ -187,8 +187,8 @@ class VideoImageFrame(Frame):
     participantId: str
     image: bytes
 
-    def __str__(self):
+    # def __str__(self):
-        return f"{self.__class__.__name__}, participantId: {self.participantId}, image size: {len(self.image)} B"
+    #     return f"{self.__class__.__name__}, participantId: {self.participantId}, image size: {len(self.image)} B"
 
 
 @dataclass()
@@ -196,5 +196,5 @@ class VisionFrame(Frame):
     prompt: str
     image: bytes
 
-    def __str__(self):
+    # def __str__(self):
-        return f"{self.__class__.__name__}, prompt: {self.prompt}, image size: {len(self.image)} B"
+    #     return f"{self.__class__.__name__}, prompt: {self.prompt}, image size: {len(self.image)} B"

src/dailyai/services/ai_services.py

@@ -148,6 +148,8 @@ class VisionService(AIService):
         if isinstance(frame, VisionFrame):
             async for frame in self.run_vision(frame.prompt, frame.image):
                 yield frame
+        else:
+            yield frame
 
 
 class FrameLogger(AIService):

src/dailyai/services/daily_transport_service.py

@@ -230,14 +230,12 @@ class DailyTransportService(BaseTransportService, EventHandler):
         self.client.release()
 
     def _handle_video_frame(self, participant_id, video_frame):
-        # TODO-CB: What about multiple participants?
         if (not participant_id in self._participant_frame_times) or (time.time() > self._participant_frame_times[participant_id] + 1.0/self._receive_video_fps):
-            print(f"### sending frame now")
             self._participant_frame_times[participant_id] = time.time()
-            asyncio.run_coroutine_threadsafe(
+
+            future = asyncio.run_coroutine_threadsafe(
                 self.receive_queue.put(
-                    VideoImageFrame(participant_id, video_frame)), self._loop
+                    VideoImageFrame(participant_id, video_frame)), self._loop)
-            )
 
     def on_first_other_participant_joined(self):
         pass

src/dailyai/services/elevenlabs_ai_service.py

@@ -35,6 +35,7 @@ class ElevenLabsTTSService(TTSService):
             "xi-api-key": self._api_key,
             "Content-Type": "application/json",
         }
+
         async with self._aiohttp_session.post(
             url, json=payload, headers=headers, params=querystring
         ) as r:

src/dailyai/services/open_ai_services.py

@@ -71,7 +71,24 @@ class OpenAIVisionService(VisionService):
         self._client = AsyncOpenAI(api_key=api_key)
 
     async def run_vision(self, prompt: str, image: bytes):
-        base64_image = base64.b64encode(image).decode('utf-8')
+        IMAGE_WIDTH = image.width
+        IMAGE_HEIGHT = image.height
+        COLOR_FORMAT = image.color_format
+        a_image = Image.frombytes(
+            'RGBA', (IMAGE_WIDTH, IMAGE_HEIGHT), image.buffer)
+        new_image = a_image.convert('RGB')
+
+        # Uncomment these lines to write the frame to a jpg in the same directory.
+        # current_path = os.getcwd()
+        # image_path = os.path.join(current_path, "image.jpg")
+        # image.save(image_path, format="JPEG")
+
+        jpeg_buffer = io.BytesIO()
+
+        new_image.save(jpeg_buffer, format='JPEG')
+
+        jpeg_bytes = jpeg_buffer.getvalue()
+        base64_image = base64.b64encode(jpeg_bytes).decode('utf-8')
         messages = [
             {
                 "role": "user",
@@ -94,5 +111,7 @@ class OpenAIVisionService(VisionService):
             )
         )
         async for chunk in chunks:
-            print(f"!!! chunk: {chunk}")
+            if len(chunk.choices) == 0:
-            yield TextFrame(chunk)
+                continue
+            if chunk.choices[0].delta.content:
+                yield TextFrame(chunk.choices[0].delta.content)

src/examples/foundational/12-describe-video.py

@@ -10,6 +10,7 @@ from dailyai.pipeline.frame_processor import FrameProcessor
 from dailyai.services.daily_transport_service import DailyTransportService
 from dailyai.services.elevenlabs_ai_service import ElevenLabsTTSService
 from dailyai.services.open_ai_services import OpenAILLMService, OpenAIVisionService
+from dailyai.services.deepgram_ai_services import DeepgramTTSService
 from dailyai.services.ai_services import FrameLogger
 from dailyai.pipeline.aggregators import (
     LLMAssistantContextAggregator,
@@ -59,10 +60,7 @@ async def main(room_url: str, token):
         llm = OpenAILLMService(
             api_key=os.getenv("OPENAI_CHATGPT_API_KEY"),
             model="gpt-4-turbo-preview")
-        fl = FrameLogger("!!! before VIFP")
+
-        fl2 = FrameLogger("Outer")
-        fl3 = FrameLogger("### Before VS")
-        fl4 = FrameLogger("$$$ After VS")
         messages = [
             {
                 "role": "system",
@@ -80,13 +78,9 @@ async def main(room_url: str, token):
         vifp = VideoImageFrameProcessor()
         pipeline = Pipeline(
             processors=[
-                fl,
                 vifp,
-                fl3,
                 vs,
-                fl4,
                 llm,
-                fl2,
                 tts,
                 tma_out,
             ],

src/examples/starter-apps/chatbot.py

@@ -124,7 +124,6 @@ async def main(room_url: str, token):
 
         @transport.event_handler("on_first_other_participant_joined")
         async def on_first_other_participant_joined(transport):
-            print(f"!!! in here, pipeline.source is {pipeline.source}")
             await pipeline.queue_frames([LLMMessagesQueueFrame(messages)])
 
         async def run_conversation():