From 8eef21db6eda58ea7077eb064dc0271c47b332f6 Mon Sep 17 00:00:00 2001 From: Rafal Skorski Date: Thu, 23 Jan 2025 15:39:46 +0100 Subject: [PATCH] Adding telnyx serializer --- examples/telnyx-chatbot/bot.py | 9 ++- examples/telnyx-chatbot/server.py | 5 +- src/pipecat/audio/utils.py | 22 ++++++- src/pipecat/serializers/telnyx.py | 96 +++++++++++++++++++++++++++++++ 4 files changed, 125 insertions(+), 7 deletions(-) create mode 100644 src/pipecat/serializers/telnyx.py diff --git a/examples/telnyx-chatbot/bot.py b/examples/telnyx-chatbot/bot.py index af6a80501..771147e0f 100644 --- a/examples/telnyx-chatbot/bot.py +++ b/examples/telnyx-chatbot/bot.py @@ -16,7 +16,9 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.serializers.twilio import TwilioFrameSerializer + +from pipecat.serializers.telnyx import TelnyxFrameSerializer + from pipecat.services.elevenlabs import ElevenLabsTTSService, Language from pipecat.services.deepgram import DeepgramSTTService from pipecat.services.openai import OpenAILLMService @@ -25,13 +27,14 @@ from pipecat.transports.network.fastapi_websocket import ( FastAPIWebsocketTransport, ) + load_dotenv(override=True) logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def run_bot(websocket_client, stream_sid): +async def run_bot(websocket_client, stream_id, encoding): transport = FastAPIWebsocketTransport( websocket=websocket_client, params=FastAPIWebsocketParams( @@ -40,7 +43,7 @@ async def run_bot(websocket_client, stream_sid): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), vad_audio_passthrough=True, - serializer=TwilioFrameSerializer(stream_sid), + serializer=TelnyxFrameSerializer(stream_id, encoding), ), ) diff --git a/examples/telnyx-chatbot/server.py b/examples/telnyx-chatbot/server.py index c76e2719a..27dafdae6 100644 --- a/examples/telnyx-chatbot/server.py +++ b/examples/telnyx-chatbot/server.py @@ -30,9 +30,10 @@ async def websocket_endpoint(websocket: WebSocket): await start_data.__anext__() call_data = json.loads(await start_data.__anext__()) print(call_data, flush=True) - stream_sid = call_data["stream_id"] + stream_id = call_data["stream_id"] + encoding = call_data["start"]["media_format"]["encoding"] print("WebSocket connection accepted") - await run_bot(websocket, stream_sid) + await run_bot(websocket, stream_id, encoding) if __name__ == "__main__": diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index 143f05c28..8f4807f1c 100644 --- a/src/pipecat/audio/utils.py +++ b/src/pipecat/audio/utils.py @@ -81,10 +81,9 @@ def ulaw_to_pcm(ulaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int): # Resample out_pcm_bytes = resample_audio(in_pcm_bytes, in_sample_rate, out_sample_rate) - + return out_pcm_bytes - def pcm_to_ulaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int): # Resample in_pcm_bytes = resample_audio(pcm_bytes, in_sample_rate, out_sample_rate) @@ -93,3 +92,22 @@ def pcm_to_ulaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int): ulaw_bytes = audioop.lin2ulaw(in_pcm_bytes, 2) return ulaw_bytes + +def alaw_to_pcm(alaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int) -> bytes: + + # Convert a-law to PCM + in_pcm_bytes = audioop.alaw2lin(alaw_bytes, 2) + + # Resample + out_pcm_bytes = resample_audio(in_pcm_bytes, in_sample_rate, out_sample_rate) + + return out_pcm_bytes + +def pcm_to_alaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int): + # Resample + in_pcm_bytes = resample_audio(pcm_bytes, in_sample_rate, out_sample_rate) + + # Convert PCM to μ-law + alaw_bytes = audioop.lin2alaw(in_pcm_bytes, 2) + + return alaw_bytes \ No newline at end of file diff --git a/src/pipecat/serializers/telnyx.py b/src/pipecat/serializers/telnyx.py new file mode 100644 index 000000000..4535c730a --- /dev/null +++ b/src/pipecat/serializers/telnyx.py @@ -0,0 +1,96 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import base64 +import json + +from pydantic import BaseModel + +from pipecat.audio.utils import pcm_to_ulaw, ulaw_to_pcm, pcm_to_alaw, alaw_to_pcm +from pipecat.frames.frames import ( + AudioRawFrame, + Frame, + InputAudioRawFrame, + InputDTMFFrame, + KeypadEntry, + StartInterruptionFrame, +) +from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType + + +class TelnyxFrameSerializer(FrameSerializer): + class InputParams(BaseModel): + telnyx_sample_rate: int = 8000 + sample_rate: int = 16000 + encoding: str = "PCMU" + + def __init__(self, stream_id: str, encoding: str, params: InputParams = InputParams()): + self._stream_id = stream_id + params.encoding = encoding + self._params = params + + @property + def type(self) -> FrameSerializerType: + return FrameSerializerType.TEXT + + def serialize(self, frame: Frame) -> str | bytes | None: + if isinstance(frame, AudioRawFrame): + data = frame.audio + + if self._params.encoding == "PCMU": + serialized_data = pcm_to_ulaw(data, frame.sample_rate, self._params.telnyx_sample_rate) + elif self._params.encoding == "PCMA": + serialized_data = pcm_to_alaw(data, frame.sample_rate, self._params.telnyx_sample_rate) + else: + raise ValueError(f"Unsupported encoding: {self._params.encoding}") + + payload = base64.b64encode(serialized_data).decode("utf-8") + answer = { + "event": "media", + "media": {"payload": payload}, + } + + return json.dumps(answer) + + if isinstance(frame, StartInterruptionFrame): + answer = {"event": "clear"} + return json.dumps(answer) + + def deserialize(self, data: str | bytes) -> Frame | None: + message = json.loads(data) + + if message["event"] == "start": + print(f"Start received encoding:{message['start']['media_format']['encoding']}") + self._params.encoding = message["start"]["media_format"]["encoding"] + if message["event"] == "media": + payload_base64 = message["media"]["payload"] + payload = base64.b64decode(payload_base64) + + if self._params.encoding == "PCMU": + deserialized_data = ulaw_to_pcm( + payload, self._params.telnyx_sample_rate, self._params.sample_rate + ) + elif self._params.encoding == "PCMA": + deserialized_data = alaw_to_pcm( + payload, self._params.telnyx_sample_rate, self._params.sample_rate + ) + else: + raise ValueError(f"Unsupported encoding: {self._params.encoding}") + + audio_frame = InputAudioRawFrame( + audio=deserialized_data, num_channels=1, sample_rate=self._params.sample_rate + ) + return audio_frame + elif message["event"] == "dtmf": + digit = message.get("dtmf", {}).get("digit") + + try: + return InputDTMFFrame(KeypadEntry(digit)) + except ValueError as e: + # Handle case where string doesn't match any enum value + return None + else: + return None