diff --git a/CHANGELOG.md b/CHANGELOG.md index 70716937b..c7868b186 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- `TransportMessageFrame` and `TransportMessageUrgentFrame` are deprecated, use + `OutputTransportMessageFrame` and `OutputTransportMessageUrgentFrame` + respectively. + - `InputTransportMessageUrgentFrame` is deprecated, use `InputTransportMessageFrame` instead. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index f075a6896..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. @@ -1091,20 +1117,6 @@ class STTMuteFrame(SystemFrame): mute: bool -@dataclass -class TransportMessageUrgentFrame(SystemFrame): - """Frame for urgent transport messages that need immediate processing. - - Parameters: - message: The urgent transport message payload. - """ - - message: Any - - def __str__(self): - return f"{self.name}(message: {self.message})" - - @dataclass class InputTransportMessageFrame(SystemFrame): """Frame for transport messages received from external sources. @@ -1145,6 +1157,46 @@ class InputTransportMessageUrgentFrame(InputTransportMessageFrame): ) +@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 class UserImageRequestFrame(SystemFrame): """Frame requesting an image from a specific user. diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 3202d66b0..4d1f8aeae 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -51,10 +51,10 @@ from pipecat.frames.frames import ( LLMMessagesAppendFrame, LLMTextFrame, MetricsFrame, + OutputTransportMessageUrgentFrame, StartFrame, SystemFrame, TranscriptionFrame, - TransportMessageUrgentFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -1346,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): 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 8a70bc7eb..803de0575 100644 --- a/src/pipecat/serializers/protobuf.py +++ b/src/pipecat/serializers/protobuf.py @@ -17,10 +17,10 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputTransportMessageFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, TextFrame, TranscriptionFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -83,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), ) @@ -135,7 +135,7 @@ 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"]) 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 e51d444e3..9d4630100 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -35,13 +35,13 @@ from pipecat.frames.frames import ( OutputDTMFFrame, OutputDTMFUrgentFrame, OutputImageRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, OutputTransportReadyFrame, SpeechOutputAudioRawFrame, SpriteFrame, StartFrame, SystemFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, TTSAudioRawFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -177,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: @@ -306,7 +308,7 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, InterruptionFrame): await self.push_frame(frame, direction) await self._handle_frame(frame) - elif isinstance(frame, TransportMessageUrgentFrame): + elif isinstance(frame, OutputTransportMessageUrgentFrame): await self.send_message(frame) elif isinstance(frame, OutputDTMFUrgentFrame): await self.write_dtmf(frame) @@ -643,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 1ffb1b231..787216341 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -34,11 +34,11 @@ from pipecat.frames.frames import ( 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 DailyTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in Daily calls. Parameters: @@ -85,7 +85,7 @@ class DailyTransportMessageFrame(TransportMessageFrame): @dataclass -class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame): +class DailyTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): """Frame for urgent transport messages in Daily calls. Parameters: @@ -499,7 +499,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: @@ -1868,7 +1870,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..cb06464f0 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 LiveKitTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in LiveKit rooms. Parameters: @@ -79,7 +79,7 @@ class LiveKitTransportMessageFrame(TransportMessageFrame): @dataclass -class LiveKitTransportMessageUrgentFrame(TransportMessageUrgentFrame): +class LiveKitTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): """Frame for urgent transport messages in LiveKit rooms. Parameters: @@ -836,7 +836,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: diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index 17b492d77..cdefa976f 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/src/pipecat/transports/smallwebrtc/transport.py @@ -29,10 +29,10 @@ from pipecat.frames.frames import ( 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: @@ -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(