diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index 8836fcd6d..583234ae4 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -9,7 +9,7 @@ import json from pydantic import BaseModel -from pipecat.frames.frames import AudioRawFrame, Frame +from pipecat.frames.frames import AudioRawFrame, Frame, StartInterruptionFrame from pipecat.serializers.base_serializer import FrameSerializer from pipecat.utils.audio import ulaw_to_pcm, pcm_to_ulaw @@ -28,22 +28,25 @@ class TwilioFrameSerializer(FrameSerializer): self._params = params def serialize(self, frame: Frame) -> str | bytes | None: - if not isinstance(frame, AudioRawFrame): - return None + if isinstance(frame, AudioRawFrame): + data = frame.audio - data = frame.audio - - serialized_data = pcm_to_ulaw(data, frame.sample_rate, self._params.twilio_sample_rate) - payload = base64.b64encode(serialized_data).decode("utf-8") - answer = { - "event": "media", - "streamSid": self._stream_sid, - "media": { - "payload": payload + serialized_data = pcm_to_ulaw( + data, frame.sample_rate, self._params.twilio_sample_rate) + payload = base64.b64encode(serialized_data).decode("utf-8") + answer = { + "event": "media", + "streamSid": self._stream_sid, + "media": { + "payload": payload + } } - } - return json.dumps(answer) + return json.dumps(answer) + + if isinstance(frame, StartInterruptionFrame): + answer = {"event": "clear", "streamSid": self._stream_sid} + return json.dumps(answer) def deserialize(self, data: str | bytes) -> Frame | None: message = json.loads(data) diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index 914870114..16c5e81fc 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -12,8 +12,8 @@ import wave from typing import Awaitable, Callable from pydantic.main import BaseModel -from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, StartFrame -from pipecat.processors.frame_processor import FrameProcessor +from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, Frame, StartFrame, StartInterruptionFrame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.serializers.base_serializer import FrameSerializer from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport @@ -93,11 +93,18 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): self._params = params self._websocket_audio_buffer = bytes() + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartInterruptionFrame): + await self._write_frame(frame) + async def write_raw_audio_frames(self, frames: bytes): self._websocket_audio_buffer += frames while len(self._websocket_audio_buffer) >= self._params.audio_frame_size: frame = AudioRawFrame( - audio=self._websocket_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 +128,13 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): if payload and self._websocket.client_state == WebSocketState.CONNECTED: await self._websocket.send_text(payload) - self._websocket_audio_buffer = self._websocket_audio_buffer[self._params.audio_frame_size:] + self._websocket_audio_buffer = self._websocket_audio_buffer[ + self._params.audio_frame_size:] + + async def _write_frame(self, frame: Frame): + payload = self._params.serializer.serialize(frame) + if payload and self._websocket.client_state == WebSocketState.CONNECTED: + await self._websocket.send_text(payload) class FastAPIWebsocketTransport(BaseTransport):