transport(output): improve audio and image handling for video use cases

CHANGELOG.md

@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added new `GStreamerPipelineSource`. This processor can generate image or
+  audio frames from a GStreamer pipeline (e.g. reading an MP4 file, and RTP
+  stream or anything supported by GStreamer).
+
+- Added `TransportParams.audio_out_is_live`. This flag is False by default and
+  it is useful to indicate we should not synchronize audio with sporadic images.
+
 - Added new `BotStartedSpeakingFrame` and `BotStoppedSpeakingFrame` control
   frames. These frames are pushed upstream and they should wrap
   `BotSpeakingFrame`.

src/pipecat/frames/frames.py

@@ -101,7 +101,7 @@ class UserImageRawFrame(ImageRawFrame):
 class SpriteFrame(Frame):
     """An animated sprite. Will be shown by the transport if the transport's
     camera is enabled. Will play at the framerate specified in the transport's
-    `fps` constructor parameter.
+    `camera_out_framerate` constructor parameter.
 
     """
     images: List[ImageRawFrame]

src/pipecat/transports/base_output.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import itertools
+import time
 
 from PIL import Image
 from typing import List
@@ -51,6 +52,7 @@ class BaseOutputTransport(FrameProcessor):
         audio_bytes_10ms = int(self._params.audio_out_sample_rate / 100) * \
             self._params.audio_out_channels * 2
         self._audio_chunk_size = audio_bytes_10ms * 2
+        self._audio_buffer = bytearray()
 
         self._stopped_event = asyncio.Event()
 
@@ -68,6 +70,10 @@ class BaseOutputTransport(FrameProcessor):
         if self._params.camera_out_enabled:
             self._camera_out_queue = asyncio.Queue()
             self._camera_out_task = self.get_event_loop().create_task(self._camera_out_task_handler())
+        # Create audio output queue and task if needed.
+        if self._params.audio_out_enabled and self._params.audio_out_is_live:
+            self._audio_out_queue = asyncio.Queue()
+            self._audio_out_task = self.get_event_loop().create_task(self._audio_out_task_handler())
 
     async def stop(self, frame: EndFrame):
         # Cancel and wait for the camera output task to finish.
@@ -75,6 +81,11 @@ class BaseOutputTransport(FrameProcessor):
             self._camera_out_task.cancel()
             await self._camera_out_task
 
+        # Cancel and wait for the audio output task to finish.
+        if self._params.audio_out_enabled and self._params.audio_out_is_live:
+            self._audio_out_task.cancel()
+            await self._audio_out_task
+
         # Wait for the push frame and sink tasks to finish. They will finish when
         # the EndFrame is actually processed.
         await self._push_frame_task
@@ -138,6 +149,8 @@ class BaseOutputTransport(FrameProcessor):
         # Other frames.
         elif isinstance(frame, AudioRawFrame):
             await self._handle_audio(frame)
+        elif isinstance(frame, ImageRawFrame) or isinstance(frame, SpriteFrame):
+            await self._handle_image(frame)
         else:
             await self._sink_queue.put(frame)
 
@@ -156,11 +169,27 @@ class BaseOutputTransport(FrameProcessor):
             self._create_push_task()
 
     async def _handle_audio(self, frame: AudioRawFrame):
-        audio = frame.audio
+        if not self._params.audio_out_enabled:
-        for i in range(0, len(audio), self._audio_chunk_size):
+            return
-            chunk = AudioRawFrame(audio[i: i + self._audio_chunk_size],
+
-                                  sample_rate=frame.sample_rate, num_channels=frame.num_channels)
+        if self._params.audio_out_is_live:
-            await self._sink_queue.put(chunk)
+            await self._audio_out_queue.put(frame)
+        else:
+            self._audio_buffer.extend(frame.audio)
+            while len(self._audio_buffer) >= self._audio_chunk_size:
+                chunk = AudioRawFrame(bytes(self._audio_buffer[:self._audio_chunk_size]),
+                                      sample_rate=frame.sample_rate, num_channels=frame.num_channels)
+                await self._sink_queue.put(chunk)
+                self._audio_buffer = self._audio_buffer[self._audio_chunk_size:]
+
+    async def _handle_image(self, frame: ImageRawFrame | SpriteFrame):
+        if not self._params.camera_out_enabled:
+            return
+
+        if self._params.camera_out_is_live:
+            await self._camera_out_queue.put(frame)
+        else:
+            await self._sink_queue.put(frame)
 
     def _create_sink_task(self):
         loop = self.get_event_loop()
@@ -168,19 +197,16 @@ class BaseOutputTransport(FrameProcessor):
         self._sink_task = loop.create_task(self._sink_task_handler())
 
     async def _sink_task_handler(self):
-        # Audio accumlation buffer
-        buffer = bytearray()
-
         running = True
         while running:
             try:
                 frame = await self._sink_queue.get()
-                if isinstance(frame, AudioRawFrame) and self._params.audio_out_enabled:
+                if isinstance(frame, AudioRawFrame):
-                    buffer.extend(frame.audio)
+                    await self.write_raw_audio_frames(frame.audio)
-                    buffer = await self._maybe_send_audio(buffer)
+                    await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
-                elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled:
+                elif isinstance(frame, ImageRawFrame):
                     await self._set_camera_image(frame)
-                elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled:
+                elif isinstance(frame, SpriteFrame):
                     await self._set_camera_images(frame.images)
                 elif isinstance(frame, TransportMessageFrame):
                     await self.send_message(frame)
@@ -247,21 +273,19 @@ class BaseOutputTransport(FrameProcessor):
         await self.write_frame_to_camera(frame)
 
     async def _set_camera_image(self, image: ImageRawFrame):
-        if self._params.camera_out_is_live:
+        self._camera_images = itertools.cycle([image])
-            await self._camera_out_queue.put(image)
-        else:
-            self._camera_images = itertools.cycle([image])
 
     async def _set_camera_images(self, images: List[ImageRawFrame]):
         self._camera_images = itertools.cycle(images)
 
     async def _camera_out_task_handler(self):
+        self._camera_out_start_time = None
+        self._camera_out_frame_index = 0
+        self._camera_out_frame_duration = 1 / self._params.camera_out_framerate
         while True:
             try:
                 if self._params.camera_out_is_live:
-                    image = await self._camera_out_queue.get()
+                    await self._camera_out_is_live_handler()
-                    await self._draw_image(image)
-                    self._camera_out_queue.task_done()
                 elif self._camera_images:
                     image = next(self._camera_images)
                     await self._draw_image(image)
@@ -273,6 +297,26 @@ class BaseOutputTransport(FrameProcessor):
             except Exception as e:
                 logger.exception(f"{self} error writing to camera: {e}")
 
+    async def _camera_out_is_live_handler(self):
+        image = await self._camera_out_queue.get()
+
+        # We get the start time as soon as we get the first image.
+        if not self._camera_out_start_time:
+            self._camera_out_start_time = time.time()
+
+        # Calculate how much time we need to wait before rendering next image.
+        render_time = self._camera_out_start_time + \
+            self._camera_out_frame_index * self._camera_out_frame_duration
+        time_until_render = render_time - time.time()
+        if time_until_render > 0:
+            await asyncio.sleep(time_until_render)
+        self._camera_out_frame_index += 1
+
+        # Render image
+        await self._draw_image(image)
+
+        self._camera_out_queue.task_done()
+
     #
     # Audio out
     #
@@ -280,9 +324,13 @@ class BaseOutputTransport(FrameProcessor):
     async def send_audio(self, frame: AudioRawFrame):
         await self.process_frame(frame, FrameDirection.DOWNSTREAM)
 
-    async def _maybe_send_audio(self, buffer: bytearray) -> bytearray:
+    async def _audio_out_task_handler(self):
-        if len(buffer) >= self._audio_chunk_size:
+        while True:
-            await self.write_raw_audio_frames(bytes(buffer[:self._audio_chunk_size]))
+            try:
-            buffer = buffer[self._audio_chunk_size:]
+                frame = await self._audio_out_queue.get()
-            await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
+                await self.write_raw_audio_frames(frame.audio)
-        return buffer
+                await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.exception(f"{self} error writing to camera: {e}")

src/pipecat/transports/base_transport.py

@@ -29,6 +29,7 @@ class TransportParams(BaseModel):
     camera_out_framerate: int = 30
     camera_out_color_format: str = "RGB"
     audio_out_enabled: bool = False
+    audio_out_is_live: bool = False
     audio_out_sample_rate: int = 16000
     audio_out_channels: int = 1
     audio_in_enabled: bool = False