From e5ee3e3720b4cb4f3247e3e74679f05afc1fd112 Mon Sep 17 00:00:00 2001 From: Chad Bailey Date: Mon, 31 Mar 2025 20:34:50 +0000 Subject: [PATCH] dynamic params --- pyproject.toml | 1 + .../transports/services/pipecat_cloud.py | 182 ++++++++++++++++++ 2 files changed, 183 insertions(+) create mode 100644 src/pipecat/transports/services/pipecat_cloud.py diff --git a/pyproject.toml b/pyproject.toml index ecf4d5272..2fbeee223 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,6 +74,7 @@ openai = [ "websockets~=13.1" ] openpipe = [ "openpipe~=4.48.0" ] openrouter = [] perplexity = [] +pipecatcloud = ["pipecatcloud" ] playht = [ "pyht~=0.1.12", "websockets~=13.1" ] qwen = [] rime = [ "websockets~=13.1" ] diff --git a/src/pipecat/transports/services/pipecat_cloud.py b/src/pipecat/transports/services/pipecat_cloud.py new file mode 100644 index 000000000..cb58b725d --- /dev/null +++ b/src/pipecat/transports/services/pipecat_cloud.py @@ -0,0 +1,182 @@ +import logging +from typing import Optional, Union + +from fastapi import WebSocket +from pipecatcloud.agent import DailySessionArguments, WebsocketSessionArguments +from pydantic import BaseModel, ConfigDict, create_model + +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.network.fastapi_websocket import ( + FastAPIWebsocketParams, + FastAPIWebsocketTransport, +) +from pipecat.transports.network.small_webrtc import ( + SmallWebRTCConnection, + SmallWebRTCTransport, +) +from pipecat.transports.services.daily import DailyParams, DailyTransport + +logger = logging.getLogger(__name__) + + +def _get_model_fields(model_class): + """Get all fields from a Pydantic model class.""" + return { + field_name: (field_info.annotation, field_info.default) + for field_name, field_info in model_class.model_fields.items() + } + + +# Dynamically create PipecatCloudParams by combining fields from all parameter classes +PipecatCloudParams = create_model( + "PipecatCloudParams", + __config__=ConfigDict(arbitrary_types_allowed=True), + __doc__="""Parameters for PipecatCloudTransport. + + This class combines parameters from all transport types. When a specific transport + is created, it will extract only the parameters relevant to that transport type. + + The fields are automatically inherited from: + - TransportParams + - FastAPIWebsocketParams + - DailyParams + """, + **{ + **_get_model_fields(TransportParams), + **_get_model_fields(FastAPIWebsocketParams), + **_get_model_fields(DailyParams), + }, +) + + +def _extract_matching_fields(source_obj, target_class): + """Extract fields from source_obj that match the fields in target_class.""" + target_fields = set(target_class.model_fields.keys()) + matching_fields = { + field: getattr(source_obj, field) for field in target_fields if hasattr(source_obj, field) + } + return target_class(**matching_fields) + + +# Add the conversion methods to PipecatCloudParams +def to_transport_params(self) -> TransportParams: + """Convert to TransportParams.""" + return _extract_matching_fields(self, TransportParams) + + +def to_websocket_params(self) -> FastAPIWebsocketParams: + """Convert to FastAPIWebsocketParams.""" + return _extract_matching_fields(self, FastAPIWebsocketParams) + + +def to_daily_params(self) -> DailyParams: + """Convert to DailyParams.""" + params = _extract_matching_fields(self, DailyParams) + # Also copy over the base transport params + transport_params = self.to_transport_params() + for key, value in transport_params.model_dump().items(): + setattr(params, key, value) + return params + + +# Add the conversion methods to the dynamically created class +setattr(PipecatCloudParams, "to_transport_params", to_transport_params) +setattr(PipecatCloudParams, "to_websocket_params", to_websocket_params) +setattr(PipecatCloudParams, "to_daily_params", to_daily_params) + + +class WebRTCSessionArguments: + """Arguments for creating a SmallWebRTCTransport session.""" + + def __init__( + self, + webrtc_connection: SmallWebRTCConnection, + ): + self.webrtc_connection = webrtc_connection + + +SessionArguments = Union[WebsocketSessionArguments, DailySessionArguments, WebRTCSessionArguments] + + +class PipecatCloudTransport(BaseTransport): + """A transport that wraps FastAPIWebsocketTransport, SmallWebRTCTransport, and DailyTransport. + + This transport will instantiate one of the three underlying transports based on the + session arguments provided to the constructor. + + Args: + session_args: Arguments for creating the session. The type of arguments determines + which transport will be used. + params: Configuration parameters for the transport. Parameters will be extracted + based on the session arguments type. + input_name: Optional name for the input transport. + output_name: Optional name for the output transport. + """ + + def __init__( + self, + session_args: SessionArguments, + params: Optional[PipecatCloudParams] = None, + *, + input_name: Optional[str] = None, + output_name: Optional[str] = None, + ): + super().__init__(input_name=input_name, output_name=output_name) + logger.debug(f"SessionArguments: {session_args}") + + params = params or PipecatCloudParams() + + # Create the appropriate transport based on session arguments type + if isinstance(session_args, WebsocketSessionArguments): + logger.info("Using FastAPIWebsocketTransport") + websocket_params = params.to_websocket_params() + self._transport = FastAPIWebsocketTransport( + session_args.websocket, + websocket_params, + input_name=input_name, + output_name=output_name, + ) + elif isinstance(session_args, DailySessionArguments): + logger.info("Using DailyTransport") + daily_params = params.to_daily_params() + self._transport = DailyTransport( + session_args.room_url, + session_args.token, + session_args.bot_name, + params=daily_params, + input_name=input_name, + output_name=output_name, + ) + elif isinstance(session_args, WebRTCSessionArguments): + logger.info("Using SmallWebRTCTransport") + transport_params = params.to_transport_params() + self._transport = SmallWebRTCTransport( + session_args.webrtc_connection, + transport_params, + input_name=input_name, + output_name=output_name, + ) + else: + raise ValueError(f"Unsupported session arguments type: {type(session_args)}") + + async def start(self, frame): + """Start the transport.""" + await self._transport.start(frame) + + async def stop(self, frame): + """Stop the transport.""" + await self._transport.stop(frame) + + async def cancel(self, frame): + """Cancel the transport.""" + await self._transport.cancel(frame) + + @property + def input(self): + """Get the input transport.""" + return self._transport.input + + @property + def output(self): + """Get the output transport.""" + return self._transport.output