wip: telestrator

src/dailyai/pipeline/frames.py

@@ -191,6 +191,10 @@ class VideoImageFrame(Frame):
     #     return f"{self.__class__.__name__}, participantId: {self.participantId}, image size: {len(self.image)} B"
 
 
+class TelestratorImageFrame(ImageFrame):
+    pass
+
+
 @dataclass()
 class VisionFrame(Frame):
     prompt: str

src/dailyai/services/base_transport_service.py

@@ -24,7 +24,8 @@ from dailyai.pipeline.frames import (
     TextFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
-    RequestVideoImageFrame
+    RequestVideoImageFrame,
+    TelestratorImageFrame
 )
 from dailyai.pipeline.pipeline import Pipeline
 from dailyai.services.ai_services import TTSService
@@ -456,6 +457,12 @@ class BaseTransportService:
                                     self.write_frame_to_mic(
                                         bytes(b[:truncated_length]))
                                     b = b[truncated_length:]
+                            elif isinstance(frame, TelestratorImageFrame):
+                                self._set_image(frame.image)
+                                asyncio.run_coroutine_threadsafe(
+                                    self.receive_queue.put(frame),
+                                    self._loop,
+                                )
                             elif isinstance(frame, ImageFrame):
                                 self._set_image(frame.image)
                             elif isinstance(frame, SpriteFrame):

src/dailyai/services/daily_transport_service.py

@@ -12,7 +12,8 @@ from typing import Any
 from dailyai.pipeline.frames import (
     ReceivedAppMessageFrame,
     TranscriptionQueueFrame,
-    VideoImageFrame
+    VideoImageFrame,
+    TelestratorImageFrame
 )
 
 from threading import Event

src/dailyai/services/fal_ai_services.py

@@ -53,4 +53,7 @@ class FalImageGenService(ImageGenService):
         async with self._aiohttp_session.get(image_url) as response:
             image_stream = io.BytesIO(await response.content.read())
             image = Image.open(image_stream)
-            return (image_url, image.tobytes())
+            image_bytes = image.tobytes()
+            print(f"!!! fal image tobytes is:")
+            print(image)
+            return (image_url, image_bytes)

src/dailyai/services/open_ai_services.py

@@ -14,6 +14,7 @@ from openai.types.chat import (
     ChatCompletionMessageParam,
 )
 
+from daily import VideoFrame
 from dailyai.services.ai_services import LLMService, ImageGenService, VisionService
 from dailyai.services.openai_api_llm_service import BaseOpenAILLMService
 from dailyai.pipeline.frames import TextFrame
@@ -71,17 +72,22 @@ class OpenAIVisionService(VisionService):
         self._client = AsyncOpenAI(api_key=api_key)
 
     async def run_vision(self, prompt: str, image: bytes):
-        IMAGE_WIDTH = image.width
+        if isinstance(image, VideoFrame):
-        IMAGE_HEIGHT = image.height
+            # Then it's from a daily video frame
-        COLOR_FORMAT = image.color_format
+            print("### processing daily video frame for recognition")
-        a_image = Image.frombytes(
+            IMAGE_WIDTH = image.width
-            'RGBA', (IMAGE_WIDTH, IMAGE_HEIGHT), image.buffer)
+            IMAGE_HEIGHT = image.height
-        new_image = a_image.convert('RGB')
+            COLOR_FORMAT = image.color_format
-
+            a_image = Image.frombytes(
-        # Uncomment these lines to write the frame to a jpg in the same directory.
+                'RGBA', (IMAGE_WIDTH, IMAGE_HEIGHT), image.buffer)
-        # current_path = os.getcwd()
+            new_image = a_image.convert('RGB')
-        # image_path = os.path.join(current_path, "image.jpg")
+        else:
-        # image.save(image_path, format="JPEG")
+            # handle it as a byte stream from image gen
+            new_image = Image.frombytes('RGB', (1024, 1024), image)
+            # 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()
 
@@ -89,6 +95,7 @@ class OpenAIVisionService(VisionService):
 
         jpeg_bytes = jpeg_buffer.getvalue()
         base64_image = base64.b64encode(jpeg_bytes).decode('utf-8')
+
         messages = [
             {
                 "role": "user",

src/examples/starter-apps/telestrator/describer.py

@@ -0,0 +1,100 @@
+import asyncio
+import aiohttp
+import logging
+import os
+from typing import AsyncGenerator
+
+from dailyai.pipeline.frames import Frame, LLMMessagesQueueFrame, RequestVideoImageFrame, LLMResponseEndFrame
+from dailyai.pipeline.pipeline import Pipeline
+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.fal_ai_services import FalImageGenService
+from dailyai.services.deepgram_ai_services import DeepgramTTSService
+from dailyai.services.ai_services import FrameLogger
+from dailyai.pipeline.aggregators import (
+    LLMAssistantContextAggregator,
+    LLMUserContextAggregator,
+)
+from dailyai.pipeline.frames import VideoImageFrame, VisionFrame
+from examples.support.runner import configure
+
+logging.basicConfig(format=f"%(levelno)s %(asctime)s %(message)s")
+logger = logging.getLogger("dailyai")
+logger.setLevel(logging.DEBUG)
+
+
+class VideoImageFrameProcessor(FrameProcessor):
+    def __init__(self):
+        pass
+
+    async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
+        if isinstance(frame, VideoImageFrame):
+            yield VisionFrame("Describe the image in one sentence.", frame.image)
+        else:
+            yield frame
+
+
+class ImageRefresher(FrameProcessor):
+    async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
+        if isinstance(frame, LLMResponseEndFrame):
+            yield RequestVideoImageFrame(participantId=None)
+            yield frame
+        else:
+            yield frame
+
+
+async def main(room_url: str, token):
+    async with aiohttp.ClientSession() as session:
+        transport = DailyTransportService(
+            room_url,
+            token,
+            "Respond bot",
+            duration_minutes=5,
+            start_transcription=True,
+            mic_enabled=True,
+            mic_sample_rate=16000,
+            camera_enabled=True,
+            camera_width=1024,
+            camera_height=1024,
+            vad_enabled=False,
+            receive_video=True,
+            receive_video_fps=0
+        )
+
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
+        )
+
+        llm = OpenAILLMService(
+            api_key=os.getenv("OPENAI_CHATGPT_API_KEY"),
+            model="gpt-4-turbo-preview")
+
+        vs = OpenAIVisionService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY"))
+        vifp = VideoImageFrameProcessor()
+        ir = ImageRefresher()
+
+        pipeline = Pipeline(
+            processors=[
+                vifp,
+                vs,
+                tts,
+                ir,
+            ],
+        )
+
+        @transport.event_handler("on_first_other_participant_joined")
+        async def on_first_other_participant_joined(transport):
+            await pipeline.queue_frames([RequestVideoImageFrame(participantId=None)])
+
+        transport.transcription_settings["extra"]["endpointing"] = True
+        transport.transcription_settings["extra"]["punctuate"] = True
+        await transport.run(pipeline)
+
+
+if __name__ == "__main__":
+    (url, token) = configure()
+    asyncio.run(main(url, token))

src/examples/starter-apps/telestrator/illustrator.py

@@ -0,0 +1,112 @@
+import asyncio
+import aiohttp
+import logging
+import os
+from typing import AsyncGenerator
+
+from dailyai.pipeline.frames import Frame, LLMMessagesQueueFrame, RequestVideoImageFrame, LLMResponseEndFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, TranscriptionQueueFrame, TextFrame
+from dailyai.pipeline.pipeline import Pipeline
+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.fal_ai_services import FalImageGenService
+from dailyai.services.deepgram_ai_services import DeepgramTTSService
+from dailyai.services.ai_services import FrameLogger
+from dailyai.pipeline.aggregators import (
+    LLMAssistantContextAggregator,
+    LLMUserContextAggregator,
+)
+from dailyai.pipeline.frames import VideoImageFrame, VisionFrame
+from examples.support.runner import configure
+
+logging.basicConfig(format=f"%(levelno)s %(asctime)s %(message)s")
+logger = logging.getLogger("dailyai")
+logger.setLevel(logging.DEBUG)
+
+
+class VADAggregator(FrameProcessor):
+    def __init__(self):
+        self.aggregating = False
+        self.aggregation = ""
+
+    async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
+        if isinstance(frame, UserStartedSpeakingFrame):
+            self.aggregating = True
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            self.aggregating = False
+            # Sometimes VAD triggers quickly on and off. If we don't get any transcription,
+            # it creates empty LLM message queue frames
+            if len(self.aggregation) > 0:
+                yield TextFrame(self.aggregation)
+
+                self.aggregation = ""
+                yield frame
+        elif isinstance(frame, TranscriptionQueueFrame) and self.aggregating:
+            self.aggregation += f" {frame.text}"
+        else:
+            yield frame
+
+
+async def main(room_url: str, token):
+    async with aiohttp.ClientSession() as session:
+        transport = DailyTransportService(
+            room_url,
+            token,
+            "Respond bot",
+            duration_minutes=5,
+            start_transcription=True,
+            mic_enabled=True,
+            mic_sample_rate=16000,
+            camera_enabled=True,
+            camera_width=1024,
+            camera_height=1024,
+            vad_enabled=True,
+            receive_video=True,
+            receive_video_fps=0,
+            vad_timeout_s=1.0
+        )
+
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
+        )
+
+        llm = OpenAILLMService(
+            api_key=os.getenv("OPENAI_CHATGPT_API_KEY"),
+            model="gpt-4-turbo-preview")
+
+        vs = OpenAIVisionService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY"))
+        vad = VADAggregator()
+        img = FalImageGenService(
+            image_size="1024x1024",
+            aiohttp_session=session,
+            key_id=os.getenv("FAL_KEY_ID"),
+            key_secret=os.getenv("FAL_KEY_SECRET"),
+        )
+        fl = FrameLogger("!!! Start")
+        fl2 = FrameLogger("!!! AFTER VAD")
+        fl3 = FrameLogger("!!! After img")
+        pipeline = Pipeline(
+            processors=[
+                fl,
+                vad,
+                fl2,
+                img,
+                fl3
+            ],
+        )
+
+        @transport.event_handler("on_first_other_participant_joined")
+        async def on_first_other_participant_joined(transport):
+            await pipeline.queue_frames([RequestVideoImageFrame(participantId=None)])
+
+        transport.transcription_settings["extra"]["endpointing"] = True
+        transport.transcription_settings["extra"]["punctuate"] = True
+        await transport.run(pipeline)
+
+
+if __name__ == "__main__":
+    (url, token) = configure()
+    asyncio.run(main(url, token))

src/examples/starter-apps/telestrator/telestrator.py

@@ -0,0 +1,120 @@
+import asyncio
+import aiohttp
+import logging
+import os
+from typing import AsyncGenerator
+
+from dailyai.pipeline.frames import Frame, LLMMessagesQueueFrame, RequestVideoImageFrame, LLMResponseEndFrame, TelestratorImageFrame, ImageFrame
+from dailyai.pipeline.pipeline import Pipeline
+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.fal_ai_services import FalImageGenService
+from dailyai.services.deepgram_ai_services import DeepgramTTSService
+from dailyai.services.ai_services import FrameLogger
+from dailyai.pipeline.aggregators import (
+    LLMAssistantContextAggregator,
+    LLMUserContextAggregator,
+    LLMFullResponseAggregator
+)
+from dailyai.pipeline.frames import VideoImageFrame, VisionFrame
+from examples.support.runner import configure
+
+logging.basicConfig(format=f"%(levelno)s %(asctime)s %(message)s")
+logger = logging.getLogger("dailyai")
+logger.setLevel(logging.DEBUG)
+
+
+class VideoImageFrameProcessor(FrameProcessor):
+    def __init__(self):
+        pass
+
+    async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
+        if isinstance(frame, VideoImageFrame) or isinstance(frame, TelestratorImageFrame):
+            yield VisionFrame("Describe the image in one sentence, in the style of David Attenborough.", frame.image)
+        else:
+            yield frame
+
+
+class ImageRefresher(FrameProcessor):
+    async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
+        if isinstance(frame, LLMResponseEndFrame):
+            yield RequestVideoImageFrame(participantId=None)
+            yield frame
+        else:
+            yield frame
+
+
+class TelestratorImageWrapper(FrameProcessor):
+    async def process_frame(self, frame: Frame) -> AsyncGenerator[Frame, None]:
+        if isinstance(frame, ImageFrame):
+            yield TelestratorImageFrame(None, frame.image)
+        else:
+            yield frame
+
+
+async def main(room_url: str, token):
+    async with aiohttp.ClientSession() as session:
+        transport = DailyTransportService(
+            room_url,
+            token,
+            "Respond bot",
+            duration_minutes=5,
+            start_transcription=True,
+            mic_enabled=True,
+            mic_sample_rate=16000,
+            camera_enabled=True,
+            camera_width=1024,
+            camera_height=1024,
+            vad_enabled=False,
+            receive_video=True,
+            receive_video_fps=0
+        )
+
+        tts = ElevenLabsTTSService(
+            aiohttp_session=session,
+            api_key=os.getenv("ELEVENLABS_API_KEY"),
+            voice_id=os.getenv("ELEVENLABS_VOICE_ID"),
+        )
+
+        llm = OpenAILLMService(
+            api_key=os.getenv("OPENAI_CHATGPT_API_KEY"),
+            model="gpt-4-turbo-preview")
+
+        vs = OpenAIVisionService(api_key=os.getenv("OPENAI_CHATGPT_API_KEY"))
+        vifp = VideoImageFrameProcessor()
+        ir = ImageRefresher()
+        img = FalImageGenService(
+            image_size="1024x1024",
+            aiohttp_session=session,
+            key_id=os.getenv("FAL_KEY_ID"),
+            key_secret=os.getenv("FAL_KEY_SECRET"),
+        )
+        tiw = TelestratorImageWrapper()
+        lfra = LLMFullResponseAggregator()
+        fl1 = FrameLogger("!!! About to image gen")
+        pipeline = Pipeline(
+            processors=[
+                vifp,
+                vs,
+                tts,
+                lfra,
+                fl1,
+                img,
+                tiw,
+            ],
+        )
+
+        @transport.event_handler("on_first_other_participant_joined")
+        async def on_first_other_participant_joined(transport):
+            await pipeline.queue_frames([RequestVideoImageFrame(participantId=None)])
+
+        transport.transcription_settings["extra"]["endpointing"] = True
+        transport.transcription_settings["extra"]["punctuate"] = True
+        await transport.run(pipeline)
+
+
+if __name__ == "__main__":
+    (url, token) = configure()
+    asyncio.run(main(url, token))