diff --git a/CHANGELOG.md b/CHANGELOG.md index d77524a06..0a50cc8ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue where local SmartTurn was not being ran in a separate thread. +### Deprecated + +- `DailyTransportMessageFrame` and `DailyTransportMessageUrgentFrame` are + deprecated, use `DailyOutputTransportMessageFrame` and + `DailyOutputTransportMessageUrgentFrame` respectively instead. + +- `LiveKitTransportMessageFrame` and `LiveKitTransportMessageUrgentFrame` are + deprecated, use `LiveKitOutputTransportMessageFrame` and + `LiveKitOutputTransportMessageUrgentFrame` respectively instead. + +- `TransportMessageFrame` and `TransportMessageUrgentFrame` are deprecated, use + `OutputTransportMessageFrame` and `OutputTransportMessageUrgentFrame` + respectively instead. + +- `InputTransportMessageUrgentFrame` is deprecated, use + `InputTransportMessageFrame` instead. + ## [0.0.86] - 2025-09-24 ### Added diff --git a/examples/foundational/16-gpu-container-local-bot.py b/examples/foundational/16-gpu-container-local-bot.py index 89316e63c..90ff271c6 100644 --- a/examples/foundational/16-gpu-container-local-bot.py +++ b/examples/foundational/16-gpu-container-local-bot.py @@ -26,7 +26,11 @@ from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.tts import DeepgramTTSService from pipecat.services.openai.llm import OpenAILLMService from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.transports.daily.transport import DailyParams, DailyTransportMessageFrame +from pipecat.transports.daily.transport import ( + DailyOutputTransportMessageFrame, + DailyOutputTransportMessageUrgentFrame, + DailyParams, +) from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams load_dotenv(override=True) @@ -128,14 +132,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.debug(f"Received latency ping app message: {message}") ts = message["latency-ping"]["ts"] # Send immediately - transport.output().send_message( - DailyTransportMessageFrame( + await task.queue_frame( + DailyOutputTransportMessageUrgentFrame( message={"latency-pong-msg-handler": {"ts": ts}}, participant_id=sender ) ) # And push to the pipeline for the Daily transport.output to send await task.queue_frame( - DailyTransportMessageFrame( + DailyOutputTransportMessageFrame( message={"latency-pong-pipeline-delivery": {"ts": ts}}, participant_id=sender, ) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index ffef7c99c..2a9651b83 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -672,7 +672,7 @@ class TTSSpeakFrame(DataFrame): @dataclass -class TransportMessageFrame(DataFrame): +class OutputTransportMessageFrame(DataFrame): """Frame containing transport-specific message data. Parameters: @@ -685,6 +685,32 @@ class TransportMessageFrame(DataFrame): return f"{self.name}(message: {self.message})" +@dataclass +class TransportMessageFrame(OutputTransportMessageFrame): + """Frame containing transport-specific message data. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `OutputTransportMessageFrame`. + + Parameters: + message: The transport message payload. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "TransportMessageFrame is deprecated and will be removed in a future version. " + "Instead, use OutputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + @dataclass class DTMFFrame: """Base class for DTMF (Dual-Tone Multi-Frequency) keypad frames. @@ -1092,8 +1118,8 @@ class STTMuteFrame(SystemFrame): @dataclass -class TransportMessageUrgentFrame(SystemFrame): - """Frame for urgent transport messages that need immediate processing. +class InputTransportMessageFrame(SystemFrame): + """Frame for transport messages received from external sources. Parameters: message: The urgent transport message payload. @@ -1106,20 +1132,69 @@ class TransportMessageUrgentFrame(SystemFrame): @dataclass -class InputTransportMessageUrgentFrame(TransportMessageUrgentFrame): +class InputTransportMessageUrgentFrame(InputTransportMessageFrame): """Frame for transport messages received from external sources. - This frame wraps incoming transport messages to distinguish them from outgoing - urgent transport messages (TransportMessageUrgentFrame), preventing infinite - message loops in the transport layer. It inherits the message payload from - TransportMessageFrame while marking the message as having been received - rather than generated locally. + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `InputTransportMessageFrame`. - Used by transport implementations to properly handle bidirectional message - flow without creating feedback loops. + Parameters: + message: The urgent transport message payload. """ - pass + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "InputTransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use InputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + +@dataclass +class OutputTransportMessageUrgentFrame(SystemFrame): + """Frame for urgent transport messages that need to be sent immediately. + + Parameters: + message: The urgent transport message payload. + """ + + message: Any + + def __str__(self): + return f"{self.name}(message: {self.message})" + + +@dataclass +class TransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): + """Frame for urgent transport messages that need to be sent immediately. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `OutputTransportMessageUrgentFrame`. + + Parameters: + message: The urgent transport message payload. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "TransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use OutputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) @dataclass diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 9389d819d..4d1f8aeae 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -42,6 +42,7 @@ from pipecat.frames.frames import ( Frame, FunctionCallResultFrame, InputAudioRawFrame, + InputTransportMessageUrgentFrame, InterimTranscriptionFrame, LLMConfigureOutputFrame, LLMContextFrame, @@ -50,10 +51,10 @@ from pipecat.frames.frames import ( LLMMessagesAppendFrame, LLMTextFrame, MetricsFrame, + OutputTransportMessageUrgentFrame, StartFrame, SystemFrame, TranscriptionFrame, - TransportMessageUrgentFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -1345,7 +1346,9 @@ class RTVIProcessor(FrameProcessor): async def push_transport_message(self, model: BaseModel, exclude_none: bool = True): """Push a transport message frame.""" - frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none)) + frame = OutputTransportMessageUrgentFrame( + message=model.model_dump(exclude_none=exclude_none) + ) await self.push_frame(frame) async def handle_message(self, message: RTVIMessage): @@ -1418,7 +1421,7 @@ class RTVIProcessor(FrameProcessor): elif isinstance(frame, ErrorFrame): await self._send_error_frame(frame) await self.push_frame(frame, direction) - elif isinstance(frame, TransportMessageUrgentFrame): + elif isinstance(frame, InputTransportMessageUrgentFrame): await self._handle_transport_message(frame) # All other system frames elif isinstance(frame, SystemFrame): @@ -1481,7 +1484,7 @@ class RTVIProcessor(FrameProcessor): await self._handle_message(message) self._message_queue.task_done() - async def _handle_transport_message(self, frame: TransportMessageUrgentFrame): + async def _handle_transport_message(self, frame: InputTransportMessageUrgentFrame): """Handle an incoming transport message frame.""" try: transport_message = frame.message diff --git a/src/pipecat/processors/logger.py b/src/pipecat/processors/logger.py index acc85c40b..887ce71fe 100644 --- a/src/pipecat/processors/logger.py +++ b/src/pipecat/processors/logger.py @@ -15,7 +15,7 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, OutputAudioRawFrame, - TransportMessageFrame, + UserSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -36,9 +36,9 @@ class FrameLogger(FrameProcessor): color: Optional[str] = None, ignored_frame_types: Tuple[Type[Frame], ...] = ( BotSpeakingFrame, + UserSpeakingFrame, InputAudioRawFrame, OutputAudioRawFrame, - TransportMessageFrame, ), ): """Initialize the frame logger. diff --git a/src/pipecat/serializers/exotel.py b/src/pipecat/serializers/exotel.py index 1a5859211..3922e5a17 100644 --- a/src/pipecat/serializers/exotel.py +++ b/src/pipecat/serializers/exotel.py @@ -21,9 +21,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputDTMFFrame, InterruptionFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -121,7 +121,7 @@ class ExotelFrameSerializer(FrameSerializer): } return json.dumps(answer) - elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): return json.dumps(frame.message) return None diff --git a/src/pipecat/serializers/plivo.py b/src/pipecat/serializers/plivo.py index 519e0893a..51ae11151 100644 --- a/src/pipecat/serializers/plivo.py +++ b/src/pipecat/serializers/plivo.py @@ -23,9 +23,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputDTMFFrame, InterruptionFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -148,7 +148,7 @@ class PlivoFrameSerializer(FrameSerializer): } return json.dumps(answer) - elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): return json.dumps(frame.message) # Return None for unhandled frames diff --git a/src/pipecat/serializers/protobuf.py b/src/pipecat/serializers/protobuf.py index 867fa0674..803de0575 100644 --- a/src/pipecat/serializers/protobuf.py +++ b/src/pipecat/serializers/protobuf.py @@ -15,11 +15,12 @@ import pipecat.frames.protobufs.frames_pb2 as frame_protos from pipecat.frames.frames import ( Frame, InputAudioRawFrame, + InputTransportMessageFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, TextFrame, TranscriptionFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -82,7 +83,7 @@ class ProtobufFrameSerializer(FrameSerializer): Serialized frame as bytes, or None if frame type is not serializable. """ # Wrapping this messages as a JSONFrame to send - if isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): + if isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): frame = MessageFrame( data=json.dumps(frame.message), ) @@ -134,11 +135,11 @@ class ProtobufFrameSerializer(FrameSerializer): if "pts" in args_dict: del args_dict["pts"] - # Special handling for MessageFrame -> TransportMessageUrgentFrame + # Special handling for MessageFrame -> OutputTransportMessageUrgentFrame if class_name == MessageFrame: try: msg = json.loads(args_dict["data"]) - instance = TransportMessageUrgentFrame(message=msg) + instance = InputTransportMessageFrame(message=msg) logger.debug(f"ProtobufFrameSerializer: Transport message {instance}") except Exception as e: logger.error(f"Error parsing MessageFrame data: {e}") diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index aaeb07ec3..d61b29295 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -23,9 +23,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputDTMFFrame, InterruptionFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -175,7 +175,7 @@ class TwilioFrameSerializer(FrameSerializer): } return json.dumps(answer) - elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): return json.dumps(frame.message) # Return None for unhandled frames diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 9d1b2b3bb..9d4630100 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -29,20 +29,19 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, - InputTransportMessageUrgentFrame, InterruptionFrame, MixerControlFrame, OutputAudioRawFrame, OutputDTMFFrame, OutputDTMFUrgentFrame, OutputImageRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, OutputTransportReadyFrame, SpeechOutputAudioRawFrame, SpriteFrame, StartFrame, SystemFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, TTSAudioRawFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -178,7 +177,9 @@ class BaseOutputTransport(FrameProcessor): # Sending a frame indicating that the output transport is ready and able to receive frames. await self.push_frame(OutputTransportReadyFrame(), FrameDirection.UPSTREAM) - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message. Args: @@ -307,9 +308,7 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, InterruptionFrame): await self.push_frame(frame, direction) await self._handle_frame(frame) - elif isinstance(frame, TransportMessageUrgentFrame) and not isinstance( - frame, InputTransportMessageUrgentFrame - ): + elif isinstance(frame, OutputTransportMessageUrgentFrame): await self.send_message(frame) elif isinstance(frame, OutputDTMFUrgentFrame): await self.write_dtmf(frame) @@ -646,7 +645,7 @@ class BaseOutputTransport(FrameProcessor): await self._set_video_image(frame) elif isinstance(frame, SpriteFrame): await self._set_video_images(frame.images) - elif isinstance(frame, TransportMessageFrame): + elif isinstance(frame, OutputTransportMessageFrame): await self._transport.send_message(frame) elif isinstance(frame, OutputDTMFFrame): await self._transport.write_dtmf(frame) diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index f61a799e8..7f6b21ee2 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -30,15 +30,15 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, InputAudioRawFrame, - InputTransportMessageUrgentFrame, + InputTransportMessageFrame, InterimTranscriptionFrame, OutputAudioRawFrame, OutputImageRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, SpriteFrame, StartFrame, TranscriptionFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, UserAudioRawFrame, UserImageRawFrame, UserImageRequestFrame, @@ -74,7 +74,7 @@ VAD_RESET_PERIOD_MS = 2000 @dataclass -class DailyTransportMessageFrame(TransportMessageFrame): +class DailyOutputTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in Daily calls. Parameters: @@ -85,7 +85,7 @@ class DailyTransportMessageFrame(TransportMessageFrame): @dataclass -class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame): +class DailyOutputTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): """Frame for urgent transport messages in Daily calls. Parameters: @@ -96,7 +96,59 @@ class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame): @dataclass -class DailyInputTransportMessageUrgentFrame(InputTransportMessageUrgentFrame): +class DailyTransportMessageFrame(DailyOutputTransportMessageFrame): + """Frame for transport messages in Daily calls. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `DailyOutputTransportMessageFrame`. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "DailyTransportMessageFrame is deprecated and will be removed in a future version. " + "Instead, use DailyOutputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + +@dataclass +class DailyTransportMessageUrgentFrame(DailyOutputTransportMessageUrgentFrame): + """Frame for urgent transport messages in Daily calls. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `DailyOutputTransportMessageUrgentFrame`. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "DailyTransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use DailyOutputTransportMessageUrgentFrame.", + DeprecationWarning, + stacklevel=2, + ) + + +@dataclass +class DailyInputTransportMessageFrame(InputTransportMessageFrame): """Frame for input urgent transport messages in Daily calls. Parameters: @@ -106,6 +158,31 @@ class DailyInputTransportMessageUrgentFrame(InputTransportMessageUrgentFrame): participant_id: Optional[str] = None +class DailyInputTransportMessageUrgentFrame(DailyInputTransportMessageFrame): + """Frame for input urgent transport messages in Daily calls. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `DailyInputTransportMessageFrame`. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "DailyInputTransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use DailyInputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + @dataclass class DailyUpdateRemoteParticipantsFrame(ControlFrame): """Frame to update remote participants in Daily calls. @@ -474,7 +551,9 @@ class DailyTransportClient(EventHandler): """ return self._out_sample_rate - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send an application message to participants. Args: @@ -484,7 +563,9 @@ class DailyTransportClient(EventHandler): return participant_id = None - if isinstance(frame, (DailyTransportMessageFrame, DailyTransportMessageUrgentFrame)): + if isinstance( + frame, (DailyOutputTransportMessageFrame, DailyOutputTransportMessageUrgentFrame) + ): participant_id = frame.participant_id future = self._get_event_loop().create_future() @@ -1621,7 +1702,7 @@ class DailyInputTransport(BaseInputTransport): message: The message data to send. sender: ID of the message sender. """ - frame = DailyInputTransportMessageUrgentFrame(message=message, participant_id=sender) + frame = DailyInputTransportMessageFrame(message=message, participant_id=sender) await self.push_frame(frame) # @@ -1843,7 +1924,9 @@ class DailyOutputTransport(BaseOutputTransport): if isinstance(frame, DailyUpdateRemoteParticipantsFrame): await self._client.update_remote_participants(frame.remote_participants) - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message to participants. Args: diff --git a/src/pipecat/transports/livekit/transport.py b/src/pipecat/transports/livekit/transport.py index 334d48ecb..f1b8d9088 100644 --- a/src/pipecat/transports/livekit/transport.py +++ b/src/pipecat/transports/livekit/transport.py @@ -29,9 +29,9 @@ from pipecat.frames.frames import ( OutputAudioRawFrame, OutputDTMFFrame, OutputDTMFUrgentFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, UserAudioRawFrame, UserImageRawFrame, ) @@ -68,7 +68,7 @@ DTMF_CODE_MAP = { @dataclass -class LiveKitTransportMessageFrame(TransportMessageFrame): +class LiveKitOutputTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in LiveKit rooms. Parameters: @@ -79,7 +79,7 @@ class LiveKitTransportMessageFrame(TransportMessageFrame): @dataclass -class LiveKitTransportMessageUrgentFrame(TransportMessageUrgentFrame): +class LiveKitOutputTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): """Frame for urgent transport messages in LiveKit rooms. Parameters: @@ -89,6 +89,50 @@ class LiveKitTransportMessageUrgentFrame(TransportMessageUrgentFrame): participant_id: Optional[str] = None +@dataclass +class LiveKitTransportMessageFrame(LiveKitOutputTransportMessageFrame): + """Frame for transport messages in LiveKit rooms. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "LiveKitTransportMessageFrame is deprecated and will be removed in a future version. " + "Instead, use LiveKitOutputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + +@dataclass +class LiveKitTransportMessageUrgentFrame(LiveKitOutputTransportMessageUrgentFrame): + """Frame for urgent transport messages in LiveKit rooms. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "LiveKitTransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use LiveKitOutputTransportMessageUrgentFrame.", + DeprecationWarning, + stacklevel=2, + ) + + class LiveKitParams(TransportParams): """Configuration parameters for LiveKit transport. @@ -677,7 +721,7 @@ class LiveKitInputTransport(BaseInputTransport): message: The message data to send. sender: ID of the message sender. """ - frame = LiveKitTransportMessageUrgentFrame(message=message, participant_id=sender) + frame = LiveKitOutputTransportMessageUrgentFrame(message=message, participant_id=sender) await self.push_frame(frame) async def _audio_in_task_handler(self): @@ -836,7 +880,9 @@ class LiveKitOutputTransport(BaseOutputTransport): await super().cleanup() await self._transport.cleanup() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message to participants. Args: @@ -846,7 +892,9 @@ class LiveKitOutputTransport(BaseOutputTransport): if isinstance(message, dict): # fix message encoding for dict-like messages, e.g. RTVI messages. message = json.dumps(message, ensure_ascii=False) - if isinstance(frame, (LiveKitTransportMessageFrame, LiveKitTransportMessageUrgentFrame)): + if isinstance( + frame, (LiveKitOutputTransportMessageFrame, LiveKitOutputTransportMessageUrgentFrame) + ): await self._client.send_data(message.encode(), frame.participant_id) else: await self._client.send_data(message.encode()) @@ -1105,7 +1153,9 @@ class LiveKitTransport(BaseTransport): participant_id: Optional specific participant to send to. """ if self._output: - frame = LiveKitTransportMessageFrame(message=message, participant_id=participant_id) + frame = LiveKitOutputTransportMessageFrame( + message=message, participant_id=participant_id + ) await self._output.send_message(frame) async def send_message_urgent(self, message: str, participant_id: Optional[str] = None): @@ -1116,7 +1166,7 @@ class LiveKitTransport(BaseTransport): participant_id: Optional specific participant to send to. """ if self._output: - frame = LiveKitTransportMessageUrgentFrame( + frame = LiveKitOutputTransportMessageUrgentFrame( message=message, participant_id=participant_id ) await self._output.send_message(frame) diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index 4b2437be1..cdefa976f 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/src/pipecat/transports/smallwebrtc/transport.py @@ -26,13 +26,13 @@ from pipecat.frames.frames import ( EndFrame, Frame, InputAudioRawFrame, - InputTransportMessageUrgentFrame, + InputTransportMessageFrame, OutputAudioRawFrame, OutputImageRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, SpriteFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, UserImageRawFrame, UserImageRequestFrame, ) @@ -461,7 +461,9 @@ class SmallWebRTCClient: await self._webrtc_connection.disconnect() await self._handle_peer_disconnected() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send an application message through the WebRTC connection. Args: @@ -683,7 +685,7 @@ class SmallWebRTCInputTransport(BaseInputTransport): message: The application message to process. """ logger.debug(f"Received app message inside SmallWebRTCInputTransport {message}") - frame = InputTransportMessageUrgentFrame(message=message) + frame = InputTransportMessageFrame(message=message) await self.push_frame(frame) # Add this method similar to DailyInputTransport.request_participant_image @@ -820,7 +822,9 @@ class SmallWebRTCOutputTransport(BaseOutputTransport): await super().cancel(frame) await self._client.disconnect() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message through the WebRTC connection. Args: diff --git a/src/pipecat/transports/tavus/transport.py b/src/pipecat/transports/tavus/transport.py index f6da826fd..40ffeeed7 100644 --- a/src/pipecat/transports/tavus/transport.py +++ b/src/pipecat/transports/tavus/transport.py @@ -27,9 +27,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InterruptionFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup from pipecat.transports.base_input import BaseInputTransport @@ -345,7 +345,9 @@ class TavusTransportClient: participant_id, callback, audio_source, sample_rate, callback_interval_ms ) - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a message to participants. Args: @@ -373,7 +375,7 @@ class TavusTransportClient: async def send_interrupt_message(self) -> None: """Send an interrupt message to the conversation.""" - transport_frame = TransportMessageUrgentFrame( + transport_frame = OutputTransportMessageUrgentFrame( message={ "message_type": "conversation", "event_type": "conversation.interrupt", @@ -605,7 +607,9 @@ class TavusOutputTransport(BaseOutputTransport): await super().cancel(frame) await self._client.stop() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a message to participants. Args: diff --git a/src/pipecat/transports/websocket/client.py b/src/pipecat/transports/websocket/client.py index c18c0d22a..966b78e00 100644 --- a/src/pipecat/transports/websocket/client.py +++ b/src/pipecat/transports/websocket/client.py @@ -28,9 +28,9 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.processors.frame_processor import FrameProcessorSetup from pipecat.serializers.base_serializer import FrameSerializer @@ -385,7 +385,9 @@ class WebsocketClientOutputTransport(BaseOutputTransport): await super().cleanup() await self._transport.cleanup() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message through the WebSocket. Args: diff --git a/src/pipecat/transports/websocket/fastapi.py b/src/pipecat/transports/websocket/fastapi.py index 72654204a..28d08abcd 100644 --- a/src/pipecat/transports/websocket/fastapi.py +++ b/src/pipecat/transports/websocket/fastapi.py @@ -28,9 +28,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InterruptionFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -402,7 +402,9 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): await self._write_frame(frame) self._next_send_time = 0 - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message frame. Args: diff --git a/src/pipecat/transports/websocket/server.py b/src/pipecat/transports/websocket/server.py index 095407422..94743ce66 100644 --- a/src/pipecat/transports/websocket/server.py +++ b/src/pipecat/transports/websocket/server.py @@ -27,9 +27,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InterruptionFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.serializers.base_serializer import FrameSerializer @@ -338,7 +338,9 @@ class WebsocketServerOutputTransport(BaseOutputTransport): await self._write_frame(frame) self._next_send_time = 0 - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message frame to the client. Args: diff --git a/tests/test_frame_processor.py b/tests/test_frame_processor.py index f514c94b0..d0072e5fb 100644 --- a/tests/test_frame_processor.py +++ b/tests/test_frame_processor.py @@ -11,8 +11,8 @@ from pipecat.frames.frames import ( EndFrame, Frame, InterruptionFrame, + OutputTransportMessageUrgentFrame, TextFrame, - TransportMessageUrgentFrame, ) from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.filters.identity_filter import IdentityFilter @@ -81,7 +81,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): if isinstance(frame, TextFrame): await self.push_interruption_task_frame_and_wait() - await self.push_frame(TransportMessageUrgentFrame(message=frame.text)) + await self.push_frame(OutputTransportMessageUrgentFrame(message=frame.text)) else: await self.push_frame(frame, direction) @@ -101,7 +101,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ InterruptionFrame, InterruptionFrame, - TransportMessageUrgentFrame, + OutputTransportMessageUrgentFrame, EndFrame, ] await run_test(