diff --git a/CHANGELOG.md b/CHANGELOG.md index 099783d7d..7c734120f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,10 +11,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `GoogleHttpTTSService` which uses Google's HTTP TTS API. -- Added `TavusTransport`, a new transport implementation compatible with any - Pipecat pipeline. When using the `TavusTransport`the Pipecat bot will +- Added `TavusTransport`, a new transport implementation compatible with any + Pipecat pipeline. When using the `TavusTransport`the Pipecat bot will connect in the same room as the Tavus Avatar and the user. +- Added `PlivoFrameSerializer` to support Plivo calls. A full running example + has also been added to `examples/plivo-chatbot`. + - Added `UserBotLatencyLogObserver`. This is an observer that logs the latency between when the user stops speaking and when the bot starts speaking. This gives you an initial idea on how quickly the AI services respond. @@ -36,8 +39,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 correspond to the `StartFrame`, `StopFrame`, `EndFrame` and `CancelFrame` respectively. -- Added additional languages to `LmntTTSService`. Languages include: `hi`, `id`, - `it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`. +- Added additional languages to `LmntTTSService`. Languages include: `hi`, + `id`, `it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`. - Added a `model` parameter to the `LmntTTSService` constructor, allowing switching between LMNT models. @@ -75,8 +78,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ``` By default, Pipecat has implemented service decorators to trace execution of - STT, LLM, and TTS services. You can enable tracing by setting `enable_tracing` - to `True` in the PipelineTask. + STT, LLM, and TTS services. You can enable tracing by setting + `enable_tracing` to `True` in the PipelineTask. - Added `TurnTrackingObserver`, which tracks the start and end of a user/bot turn pair and emits events `on_turn_started` and `on_turn_stopped` @@ -86,13 +89,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- Updated `GoogleTTSService` to use Google's streaming TTS API. The default voice also updated to `en-US-Chirp3-HD-Charon`. +- Updated `GoogleTTSService` to use Google's streaming TTS API. The default + voice also updated to `en-US-Chirp3-HD-Charon`. -- ⚠️Refactored the `TavusVideoService`, so it acts like a proxy, sending audio to - Tavus and receiving both audio and video. This will make `TavusVideoService` usable - with any Pipecat pipeline and with any transport. This is a **breaking change**, - check the `examples/foundational/21a-tavus-layer-small-webrtc.py` to see how to - use it. +- ⚠️ Refactored the `TavusVideoService`, so it acts like a proxy, sending audio + to Tavus and receiving both audio and video. This will make + `TavusVideoService` usable with any Pipecat pipeline and with any transport. + This is a **breaking change**, check the + `examples/foundational/21a-tavus-layer-small-webrtc.py` to see how to use it. - `DailyTransport` now uses custom microphone audio tracks instead of virtual microphones. Now, multiple Daily transports can be used in the same process. diff --git a/examples/plivo-chatbot/.gitignore b/examples/plivo-chatbot/.gitignore new file mode 100644 index 000000000..2bc1403d1 --- /dev/null +++ b/examples/plivo-chatbot/.gitignore @@ -0,0 +1,161 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/#use-with-ide +.pdm.toml + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ +runpod.toml diff --git a/examples/plivo-chatbot/Dockerfile b/examples/plivo-chatbot/Dockerfile new file mode 100644 index 000000000..4f7a1960c --- /dev/null +++ b/examples/plivo-chatbot/Dockerfile @@ -0,0 +1,20 @@ +# Use an official Python runtime as a parent image +FROM python:3.10-bullseye + +# Set the working directory in the container +WORKDIR /plivo-chatbot + +# Copy the requirements file into the container +COPY requirements.txt . + +# Install any needed packages specified in requirements.txt +RUN pip install --no-cache-dir -r requirements.txt + +# Copy the current directory contents into the container +COPY . . + +# Expose the desired port +EXPOSE 8765 + +# Run the application +CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8765"] diff --git a/examples/plivo-chatbot/README.md b/examples/plivo-chatbot/README.md new file mode 100644 index 000000000..f8d12e4b9 --- /dev/null +++ b/examples/plivo-chatbot/README.md @@ -0,0 +1,128 @@ +# Plivo Chatbot + +This project is a FastAPI-based chatbot that integrates with Plivo to handle WebSocket connections and provide real-time communication. The project includes endpoints for starting a call and handling WebSocket connections. + +## Table of Contents + +- [Features](#features) +- [Requirements](#requirements) +- [Installation](#installation) +- [Configure Plivo URLs](#configure-plivo-urls) +- [Running the Application](#running-the-application) +- [Usage](#usage) + +## Features + +- **FastAPI**: A modern, fast (high-performance), web framework for building APIs with Python 3.6+. +- **WebSocket Support**: Real-time communication using WebSockets. +- **CORS Middleware**: Allowing cross-origin requests for testing. +- **Dockerized**: Easily deployable using Docker. + +## Requirements + +- Python 3.10 +- Docker (for containerized deployment) +- ngrok (for tunneling) +- Plivo Account + +## Installation + +1. **Set up a virtual environment** (optional but recommended): + + ```sh + python -m venv venv + source venv/bin/activate # On Windows, use `venv\Scripts\activate` + ``` + +2. **Install dependencies**: + + ```sh + pip install -r requirements.txt + ``` + +3. **Create .env**: + Copy the example environment file and update with your settings: + + ```sh + cp env.example .env + ``` + +4. **Install ngrok**: + Follow the instructions on the [ngrok website](https://ngrok.com/download) to download and install ngrok. + +## Configure Plivo URLs + +1. **Start ngrok**: + In a new terminal, start ngrok to tunnel the local server: + + ```sh + ngrok http 8765 + ``` + +2. **Update the Plivo Application**: + + - Go to your Plivo console and navigate to Voice > Applications > XML + - Select "Add New Application" or edit an existing one + - Set the Primary Answer URL to your ngrok URL (e.g., https:///) + - Ensure the Answer Method is set to POST + - Save the application + - Configure your number to use the newly created (or updated) application + - Phone Numbers > Active > Your number + - Select Application Type: XML Application + - Plivo Application: Your application + - Click "Update" to save + +3. **Configure streams.xml**: + + - Copy the template file to create your local version: + ```sh + cp templates/streams.xml.template templates/streams.xml + ``` + - In `templates/streams.xml`, replace `` with your ngrok URL (without `https://`) + - The final URL should look like: `wss://abc123.ngrok.io/ws` + +4. **Assign the Application to a Plivo Number**: + - Go to Phone Numbers > Your Numbers in the Plivo console + - Edit your Plivo number + - Select the application you created/updated in the previous step + - Save the configuration + +## Running the Application + +Choose one of these two methods to run the application: + +### Using Python (Option 1) + +**Run the FastAPI application**: + +```sh +# Make sure you're in the project directory and your virtual environment is activated +python server.py +``` + +### Using Docker (Option 2) + +1. **Build the Docker image**: + + ```sh + docker build -t plivo-chatbot . + ``` + +2. **Run the Docker container**: + ```sh + docker run -it --rm -p 8765:8765 plivo-chatbot + ``` + +The server will start on port 8765. Keep this running while you test with Plivo. + +## Usage + +To start a call, simply make a call to your configured Plivo phone number. The Answer URL will direct the call to your FastAPI application, which will handle it accordingly. + +## Key Differences from Twilio + +- Plivo uses `streamId` instead of `streamSid` +- Plivo uses `callId` instead of `callSid` +- Plivo uses `` element instead of `` +- Plivo's Stream element has `bidirectional`, `keepCallAlive`, and `contentType` attributes +- Plivo API authentication uses Auth ID and Auth Token (similar to Twilio's Account SID and Auth Token) diff --git a/examples/plivo-chatbot/bot.py b/examples/plivo-chatbot/bot.py new file mode 100644 index 000000000..5ce4be0c6 --- /dev/null +++ b/examples/plivo-chatbot/bot.py @@ -0,0 +1,111 @@ +# +# Copyright (c) 2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import os +import sys +from typing import Optional + +from dotenv import load_dotenv +from fastapi import WebSocket +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +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.plivo import PlivoFrameSerializer +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.network.fastapi_websocket import ( + FastAPIWebsocketParams, + FastAPIWebsocketTransport, +) + +load_dotenv() +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def run_bot(websocket_client: WebSocket, stream_id: str, call_id: Optional[str]): + logger.info(f"Starting bot for stream: {stream_id}") + + serializer = PlivoFrameSerializer( + stream_id=stream_id, + call_id=call_id, + auth_id=os.getenv("PLIVO_AUTH_ID"), + auth_token=os.getenv("PLIVO_AUTH_TOKEN"), + ) + + transport = FastAPIWebsocketTransport( + websocket=websocket_client, + params=FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + add_wav_header=False, + vad_analyzer=SileroVADAnalyzer(), + serializer=serializer, + ), + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady + ) + + messages = [ + { + "role": "system", + "content": "You are an elementary teacher in an audio call. Your output will be converted to audio so don't include special characters in your answers. Respond to what the student said in a short short sentence.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Websocket input from client + stt, # Speech-To-Text + context_aggregator.user(), + llm, # LLM + tts, # Text-To-Speech + transport.output(), # Websocket output to client + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=8000, + audio_out_sample_rate=8000, + allow_interruptions=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + await task.cancel() + + # We use `handle_sigint=False` because `uvicorn` is controlling keyboard + # interruptions. We use `force_gc=True` to force garbage collection after + # the runner finishes running a task which could be useful for long running + # applications with multiple clients connecting. + runner = PipelineRunner(handle_sigint=False, force_gc=True) + + await runner.run(task) diff --git a/examples/plivo-chatbot/env.example b/examples/plivo-chatbot/env.example new file mode 100644 index 000000000..7e9278758 --- /dev/null +++ b/examples/plivo-chatbot/env.example @@ -0,0 +1,5 @@ +OPENAI_API_KEY= +DEEPGRAM_API_KEY= +CARTESIA_API_KEY= +PLIVO_AUTH_ID= +PLIVO_AUTH_TOKEN= \ No newline at end of file diff --git a/examples/plivo-chatbot/requirements.txt b/examples/plivo-chatbot/requirements.txt new file mode 100644 index 000000000..ff134bae3 --- /dev/null +++ b/examples/plivo-chatbot/requirements.txt @@ -0,0 +1,5 @@ +pipecat-ai[cartesia,openai,silero,deepgram] +fastapi +uvicorn +python-dotenv +loguru \ No newline at end of file diff --git a/examples/plivo-chatbot/server.py b/examples/plivo-chatbot/server.py new file mode 100644 index 000000000..b6d6ebd10 --- /dev/null +++ b/examples/plivo-chatbot/server.py @@ -0,0 +1,59 @@ +# +# Copyright (c) 2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import json + +import uvicorn +from bot import run_bot +from fastapi import FastAPI, WebSocket +from fastapi.middleware.cors import CORSMiddleware +from loguru import logger +from starlette.responses import HTMLResponse + +app = FastAPI() + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], # Allow all origins for testing + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +@app.post("/") +async def start_call(): + print("POST Plivo XML") + return HTMLResponse(content=open("templates/streams.xml").read(), media_type="application/xml") + + +@app.websocket("/ws") +async def websocket_endpoint(websocket: WebSocket): + await websocket.accept() + + # Plivo sends a start event when the stream begins + start_data = websocket.iter_text() + start_message = json.loads(await start_data.__anext__()) + + print("Received start message:", start_message, flush=True) + + # Extract stream_id and call_id from the start event + start_info = start_message.get("start", {}) + stream_id = start_info.get("streamId") + call_id = start_info.get("callId") + + if not stream_id: + logger.error("No streamId found in start message") + await websocket.close() + return + + print(f"WebSocket connection accepted for stream: {stream_id}, call: {call_id}") + await run_bot(websocket, stream_id, call_id) + + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=8765) diff --git a/examples/plivo-chatbot/templates/streams.xml.template b/examples/plivo-chatbot/templates/streams.xml.template new file mode 100644 index 000000000..8f657672d --- /dev/null +++ b/examples/plivo-chatbot/templates/streams.xml.template @@ -0,0 +1,4 @@ + + + wss:///ws + \ No newline at end of file diff --git a/src/pipecat/serializers/plivo.py b/src/pipecat/serializers/plivo.py new file mode 100644 index 000000000..7fcd951d4 --- /dev/null +++ b/src/pipecat/serializers/plivo.py @@ -0,0 +1,252 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import base64 +import json +from typing import Optional + +from loguru import logger +from pydantic import BaseModel + +from pipecat.audio.utils import create_default_resampler, pcm_to_ulaw, ulaw_to_pcm +from pipecat.frames.frames import ( + AudioRawFrame, + CancelFrame, + EndFrame, + Frame, + InputAudioRawFrame, + InputDTMFFrame, + KeypadEntry, + StartFrame, + StartInterruptionFrame, + TransportMessageFrame, + TransportMessageUrgentFrame, +) +from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType + + +class PlivoFrameSerializer(FrameSerializer): + """Serializer for Plivo Audio Streaming WebSocket protocol. + + This serializer handles converting between Pipecat frames and Plivo's WebSocket + audio streaming protocol. It supports audio conversion, DTMF events, and automatic + call termination. + + When auto_hang_up is enabled (default), the serializer will automatically terminate + the Plivo call when an EndFrame or CancelFrame is processed, but requires Plivo + credentials to be provided. + + Attributes: + _stream_id: The Plivo Stream ID. + _call_id: The associated Plivo Call ID. + _auth_id: Plivo auth ID for API access. + _auth_token: Plivo authentication token for API access. + _params: Configuration parameters. + _plivo_sample_rate: Sample rate used by Plivo (typically 8kHz). + _sample_rate: Input sample rate for the pipeline. + _resampler: Audio resampler for format conversion. + """ + + class InputParams(BaseModel): + """Configuration parameters for PlivoFrameSerializer. + + Attributes: + plivo_sample_rate: Sample rate used by Plivo, defaults to 8000 Hz. + sample_rate: Optional override for pipeline input sample rate. + auto_hang_up: Whether to automatically terminate call on EndFrame. + """ + + plivo_sample_rate: int = 8000 + sample_rate: Optional[int] = None + auto_hang_up: bool = True + + def __init__( + self, + stream_id: str, + call_id: Optional[str] = None, + auth_id: Optional[str] = None, + auth_token: Optional[str] = None, + params: Optional[InputParams] = None, + ): + """Initialize the PlivoFrameSerializer. + + Args: + stream_id: The Plivo Stream ID. + call_id: The associated Plivo Call ID (optional, but required for auto hang-up). + auth_id: Plivo auth ID (required for auto hang-up). + auth_token: Plivo auth token (required for auto hang-up). + params: Configuration parameters. + """ + self._stream_id = stream_id + self._call_id = call_id + self._auth_id = auth_id + self._auth_token = auth_token + self._params = params or PlivoFrameSerializer.InputParams() + + self._plivo_sample_rate = self._params.plivo_sample_rate + self._sample_rate = 0 # Pipeline input rate + + self._resampler = create_default_resampler() + self._hangup_attempted = False + + @property + def type(self) -> FrameSerializerType: + """Gets the serializer type. + + Returns: + The serializer type, either TEXT or BINARY. + """ + return FrameSerializerType.TEXT + + async def setup(self, frame: StartFrame): + """Sets up the serializer with pipeline configuration. + + Args: + frame: The StartFrame containing pipeline configuration. + """ + self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate + + async def serialize(self, frame: Frame) -> str | bytes | None: + """Serializes a Pipecat frame to Plivo WebSocket format. + + Handles conversion of various frame types to Plivo WebSocket messages. + For EndFrames, initiates call termination if auto_hang_up is enabled. + + Args: + frame: The Pipecat frame to serialize. + + Returns: + Serialized data as string or bytes, or None if the frame isn't handled. + """ + if ( + self._params.auto_hang_up + and not self._hangup_attempted + and isinstance(frame, (EndFrame, CancelFrame)) + ): + self._hangup_attempted = True + await self._hang_up_call() + return None + elif isinstance(frame, StartInterruptionFrame): + answer = {"event": "clearAudio", "streamId": self._stream_id} + return json.dumps(answer) + elif isinstance(frame, AudioRawFrame): + data = frame.audio + + # Output: Convert PCM at frame's rate to 8kHz μ-law for Plivo + serialized_data = await pcm_to_ulaw( + data, frame.sample_rate, self._plivo_sample_rate, self._resampler + ) + payload = base64.b64encode(serialized_data).decode("utf-8") + answer = { + "event": "playAudio", + "media": { + "contentType": "audio/x-mulaw", + "sampleRate": self._plivo_sample_rate, + "payload": payload, + }, + "streamId": self._stream_id, + } + + return json.dumps(answer) + elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): + return json.dumps(frame.message) + + # Return None for unhandled frames + return None + + async def _hang_up_call(self): + """Hang up the Plivo call using Plivo's REST API.""" + try: + import aiohttp + + auth_id = self._auth_id + auth_token = self._auth_token + call_id = self._call_id + + if not call_id or not auth_id or not auth_token: + missing = [] + if not call_id: + missing.append("call_id") + if not auth_id: + missing.append("auth_id") + if not auth_token: + missing.append("auth_token") + + logger.warning( + f"Cannot hang up Plivo call: missing required parameters: {', '.join(missing)}" + ) + return + + # Plivo API endpoint for hanging up calls + endpoint = f"https://api.plivo.com/v1/Account/{auth_id}/Call/{call_id}/" + + # Create basic auth from auth_id and auth_token + auth = aiohttp.BasicAuth(auth_id, auth_token) + + # Make the DELETE request to hang up the call + async with aiohttp.ClientSession() as session: + async with session.delete(endpoint, auth=auth) as response: + if response.status == 204: # Plivo returns 204 for successful hangup + logger.debug(f"Successfully terminated Plivo call {call_id}") + elif response.status == 404: # Call already ended + logger.debug(f"Plivo call {call_id} already terminated") + else: + # Get the error details for better debugging + error_text = await response.text() + logger.error( + f"Failed to terminate Plivo call {call_id}: " + f"Status {response.status}, Response: {error_text}" + ) + + except Exception as e: + logger.exception(f"Failed to hang up Plivo call: {e}") + + async def deserialize(self, data: str | bytes) -> Frame | None: + """Deserializes Plivo WebSocket data to Pipecat frames. + + Handles conversion of Plivo media events to appropriate Pipecat frames. + + Args: + data: The raw WebSocket data from Plivo. + + Returns: + A Pipecat frame corresponding to the Plivo event, or None if unhandled. + """ + try: + message = json.loads(data) + except json.JSONDecodeError: + logger.warning(f"Failed to parse JSON message: {data}") + return None + + if message.get("event") == "media": + media = message.get("media", {}) + payload_base64 = media.get("payload") + + if not payload_base64: + return None + + payload = base64.b64decode(payload_base64) + + # Input: Convert Plivo's 8kHz μ-law to PCM at pipeline input rate + deserialized_data = await ulaw_to_pcm( + payload, self._plivo_sample_rate, self._sample_rate, self._resampler + ) + audio_frame = InputAudioRawFrame( + audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate + ) + return audio_frame + elif message.get("event") == "dtmf": + dtmf_data = message.get("dtmf", {}) + digit = dtmf_data.get("digit") + if digit: + try: + return InputDTMFFrame(KeypadEntry(digit)) + except ValueError: + # Handle case where string doesn't match any enum value + logger.warning(f"Invalid DTMF digit received: {digit}") + return None + else: + return None