Merge branch 'main' into pr/add-lmnt

src/pipecat/serializers/livekit.py

@@ -0,0 +1,46 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import ctypes
+import pickle
+
+from pipecat.frames.frames import AudioRawFrame, Frame
+from pipecat.serializers.base_serializer import FrameSerializer
+
+from loguru import logger
+
+try:
+    from livekit.rtc import AudioFrame
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use LiveKit, you need to `pip install pipecat-ai[livekit]`.")
+    raise Exception(f"Missing module: {e}")
+
+
+class LivekitFrameSerializer(FrameSerializer):
+    SERIALIZABLE_TYPES = {
+        AudioRawFrame: "audio",
+    }
+
+    def serialize(self, frame: Frame) -> str | bytes | None:
+        if not isinstance(frame, AudioRawFrame):
+            return None
+        audio_frame = AudioFrame(
+            data=frame.audio,
+            sample_rate=frame.sample_rate,
+            num_channels=frame.num_channels,
+            samples_per_channel=len(frame.audio) // ctypes.sizeof(ctypes.c_int16),
+        )
+        return pickle.dumps(audio_frame)
+
+    def deserialize(self, data: str | bytes) -> Frame | None:
+        audio_frame: AudioFrame = pickle.loads(data)['frame']
+        return AudioRawFrame(
+            audio=bytes(audio_frame.data),
+            sample_rate=audio_frame.sample_rate,
+            num_channels=audio_frame.num_channels,
+        )

src/pipecat/transports/network/fastapi_websocket.py

@@ -91,13 +91,13 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
 
         self._websocket = websocket
         self._params = params
-        self._audio_buffer = bytes()
+        self._websocket_audio_buffer = bytes()
 
     async def write_raw_audio_frames(self, frames: bytes):
-        self._audio_buffer += frames
-        while len(self._audio_buffer) >= self._params.audio_frame_size:
+        self._websocket_audio_buffer += frames
+        while len(self._websocket_audio_buffer) >= self._params.audio_frame_size:
             frame = AudioRawFrame(
-                audio=self._audio_buffer[:self._params.audio_frame_size],
+                audio=self._websocket_audio_buffer[:self._params.audio_frame_size],
                 sample_rate=self._params.audio_out_sample_rate,
                 num_channels=self._params.audio_out_channels
             )
@@ -121,7 +121,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
             if payload and self._websocket.client_state == WebSocketState.CONNECTED:
                 await self._websocket.send_text(payload)
 
-            self._audio_buffer = self._audio_buffer[self._params.audio_frame_size:]
+            self._websocket_audio_buffer = self._websocket_audio_buffer[self._params.audio_frame_size:]
 
 
 class FastAPIWebsocketTransport(BaseTransport):