Merge pull request #1926 from pipecat-ai/aleix/pause-base-input-transport

handle StopFrame in base input transport and stop pushing frames

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `BaseInputTransport` now handles `StopFrame`. When a `StopFrame` is received
+  the transport will pause sending frames downstream until a new `StartFrame` is
+  received. This allows the transport to be reused (keeping the same connection)
+  in a different pipeline.
+
 - Updated AssemblyAI STT service to support their latest streaming
   speech-to-text model with improved transcription latency and endpointing.
 

examples/phone-chatbot/voicemail_detection.py

@@ -279,6 +279,7 @@ async def main(
     voicemail_detection_pipeline_task = PipelineTask(
         voicemail_detection_pipeline,
         params=PipelineParams(allow_interruptions=True),
+        check_dangling_tasks=False,
     )
 
     # ------------ EVENT HANDLERS ------------

src/pipecat/transports/base_input.py

@@ -24,9 +24,11 @@ from pipecat.frames.frames import (
     FilterUpdateSettingsFrame,
     Frame,
     InputAudioRawFrame,
+    InputImageRawFrame,
     MetricsFrame,
     StartFrame,
     StartInterruptionFrame,
+    StopFrame,
     StopInterruptionFrame,
     SystemFrame,
     UserStartedSpeakingFrame,
@@ -57,6 +59,11 @@ class BaseInputTransport(FrameProcessor):
         # if passthrough is enabled.
         self._audio_task = None
 
+        # If the transport is stopped with `StopFrame` we might still be
+        # receiving frames from the transport but we really don't want to push
+        # them downstream until we get another `StartFrame`.
+        self._paused = False
+
         if self._params.vad_enabled:
             import warnings
 
@@ -117,6 +124,8 @@ class BaseInputTransport(FrameProcessor):
         return self._params.turn_analyzer
 
     async def start(self, frame: StartFrame):
+        self._paused = False
+
         self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
 
         # Configure VAD analyzer.
@@ -133,28 +142,33 @@ class BaseInputTransport(FrameProcessor):
 
     async def stop(self, frame: EndFrame):
         # Cancel and wait for the audio input task to finish.
-        if self._audio_task and self._params.audio_in_enabled:
+        await self._cancel_audio_task()
-            await self.cancel_task(self._audio_task)
-            self._audio_task = None
         # Stop audio filter.
         if self._params.audio_in_filter:
             await self._params.audio_in_filter.stop()
 
+    async def pause(self, frame: StopFrame):
+        self._paused = True
+        # Cancel task so we clear the queue
+        await self._cancel_audio_task()
+        # Retart the task
+        self._create_audio_task()
+
     async def cancel(self, frame: CancelFrame):
         # Cancel and wait for the audio input task to finish.
-        if self._audio_task and self._params.audio_in_enabled:
+        await self._cancel_audio_task()
-            await self.cancel_task(self._audio_task)
-            self._audio_task = None
 
     async def set_transport_ready(self, frame: StartFrame):
         """To be called when the transport is ready to stream."""
         # Create audio input queue and task if needed.
-        if not self._audio_task and self._params.audio_in_enabled:
+        self._create_audio_task()
-            self._audio_in_queue = asyncio.Queue()
+
-            self._audio_task = self.create_task(self._audio_task_handler())
+    async def push_video_frame(self, frame: InputImageRawFrame):
+        if self._params.video_in_enabled and not self._paused:
+            await self.push_frame(frame)
 
     async def push_audio_frame(self, frame: InputAudioRawFrame):
-        if self._params.audio_in_enabled:
+        if self._params.audio_in_enabled and not self._paused:
             await self._audio_in_queue.put(frame)
 
     #
@@ -190,6 +204,9 @@ class BaseInputTransport(FrameProcessor):
             # finish and the task finishes when EndFrame is processed.
             await self.push_frame(frame, direction)
             await self.stop(frame)
+        elif isinstance(frame, StopFrame):
+            await self.push_frame(frame, direction)
+            await self.pause(frame)
         elif isinstance(frame, VADParamsUpdateFrame):
             if self.vad_analyzer:
                 self.vad_analyzer.set_params(frame.params)
@@ -231,6 +248,16 @@ class BaseInputTransport(FrameProcessor):
     # Audio input
     #
 
+    def _create_audio_task(self):
+        if not self._audio_task and self._params.audio_in_enabled:
+            self._audio_in_queue = asyncio.Queue()
+            self._audio_task = self.create_task(self._audio_task_handler())
+
+    async def _cancel_audio_task(self):
+        if self._audio_task:
+            await self.cancel_task(self._audio_task)
+            self._audio_task = None
+
     async def _vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState:
         state = VADState.QUIET
         if self.vad_analyzer:

src/pipecat/transports/network/fastapi_websocket.py

@@ -122,8 +122,17 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
         self._receive_task = None
         self._monitor_websocket_task = None
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         await self._client.setup(frame)
         if self._params.serializer:
             await self._params.serializer.setup(frame)
@@ -204,8 +213,17 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
         self._send_interval = 0
         self._next_send_time = 0
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         await self._client.setup(frame)
         if self._params.serializer:
             await self._params.serializer.setup(frame)

src/pipecat/transports/network/small_webrtc.py

@@ -19,7 +19,6 @@ from pipecat.frames.frames import (
     EndFrame,
     Frame,
     InputAudioRawFrame,
-    InputImageRawFrame,
     OutputAudioRawFrame,
     OutputImageRawFrame,
     SpriteFrame,
@@ -232,7 +231,8 @@ class SmallWebRTCClient:
             frame_array = frame.to_ndarray(format=format_name)
             frame_rgb = self._convert_frame(frame_array, format_name)
 
-            image_frame = InputImageRawFrame(
+            image_frame = UserImageRawFrame(
+                user_id=self._webrtc_connection.pc_id,
                 image=frame_rgb.tobytes(),
                 size=(frame.width, frame.height),
                 format="RGB",
@@ -377,6 +377,9 @@ class SmallWebRTCInputTransport(BaseInputTransport):
         self._receive_video_task = None
         self._image_requests = {}
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -385,6 +388,12 @@ class SmallWebRTCInputTransport(BaseInputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         await self._client.setup(self._params, frame)
         await self._client.connect()
         if not self._receive_audio_task and self._params.audio_in_enabled:
@@ -424,7 +433,7 @@ class SmallWebRTCInputTransport(BaseInputTransport):
         try:
             async for video_frame in self._client.read_video_frame():
                 if video_frame:
-                    await self.push_frame(video_frame)
+                    await self.push_video_frame(video_frame)
 
                     # Check if there are any pending image requests and create UserImageRawFrame
                     if self._image_requests:
@@ -438,7 +447,7 @@ class SmallWebRTCInputTransport(BaseInputTransport):
                                 format=video_frame.format,
                             )
                             # Push the frame to the pipeline
-                            await self.push_frame(image_frame)
+                            await self.push_video_frame(image_frame)
                             # Remove from pending requests
                             del self._image_requests[req_id]
 
@@ -480,8 +489,17 @@ class SmallWebRTCOutputTransport(BaseOutputTransport):
         self._client = client
         self._params = params
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         await self._client.setup(self._params, frame)
         await self._client.connect()
         await self.set_transport_ready(frame)

src/pipecat/transports/network/websocket_client.py

@@ -132,12 +132,21 @@ class WebsocketClientInputTransport(BaseInputTransport):
         self._session = session
         self._params = params
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def setup(self, setup: FrameProcessorSetup):
         await super().setup(setup)
         await self._session.setup(setup.task_manager)
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         if self._params.serializer:
             await self._params.serializer.setup(frame)
         await self._session.connect()
@@ -188,12 +197,21 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
         self._send_interval = 0
         self._next_send_time = 0
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def setup(self, setup: FrameProcessorSetup):
         await super().setup(setup)
         await self._session.setup(setup.task_manager)
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         self._send_interval = (self.audio_chunk_size / self.sample_rate) / 2
         if self._params.serializer:
             await self._params.serializer.setup(frame)

src/pipecat/transports/network/websocket_server.py

@@ -78,8 +78,17 @@ class WebsocketServerInputTransport(BaseInputTransport):
 
         self._stop_server_event = asyncio.Event()
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         if self._params.serializer:
             await self._params.serializer.setup(frame)
         if not self._server_task:
@@ -190,6 +199,9 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
         self._send_interval = 0
         self._next_send_time = 0
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def set_client_connection(self, websocket: Optional[websockets.WebSocketServerProtocol]):
         if self._websocket:
             await self._websocket.close()
@@ -198,6 +210,12 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         if self._params.serializer:
             await self._params.serializer.setup(frame)
         self._send_interval = (self.audio_chunk_size / self.sample_rate) / 2

src/pipecat/transports/services/daily.py

@@ -1006,14 +1006,14 @@ class DailyInputTransport(BaseInputTransport):
         await self._transport.cleanup()
 
     async def start(self, frame: StartFrame):
+        # Parent start.
+        await super().start(frame)
+
         if self._initialized:
             return
 
         self._initialized = True
 
-        # Parent start.
-        await super().start(frame)
-
         # Setup client.
         await self._client.start(frame)
 
@@ -1148,7 +1148,7 @@ class DailyInputTransport(BaseInputTransport):
                 format=video_frame.color_format,
             )
             frame.transport_source = video_source
-            await self.push_frame(frame)
+            await self.push_video_frame(frame)
             self._video_renderers[participant_id][video_source]["timestamp"] = curr_time
 
 
@@ -1183,14 +1183,14 @@ class DailyOutputTransport(BaseOutputTransport):
         await self._transport.cleanup()
 
     async def start(self, frame: StartFrame):
+        # Parent start.
+        await super().start(frame)
+
         if self._initialized:
             return
 
         self._initialized = True
 
-        # Parent start.
-        await super().start(frame)
-
         # Setup client.
         await self._client.start(frame)
 

src/pipecat/transports/services/livekit.py

@@ -363,6 +363,11 @@ class LiveKitInputTransport(BaseInputTransport):
         self._audio_in_task = None
         self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer
         self._resampler = create_default_resampler()
+        if self._initialized:
+            return
+
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
 
     @property
     def vad_analyzer(self) -> Optional[VADAnalyzer]:
@@ -370,6 +375,12 @@ class LiveKitInputTransport(BaseInputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         await self._client.start(frame)
         await self._client.connect()
         if not self._audio_in_task and self._params.audio_in_enabled:
@@ -447,8 +458,17 @@ class LiveKitOutputTransport(BaseOutputTransport):
         self._transport = transport
         self._client = client
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         await self._client.start(frame)
         await self._client.connect()
         await self.set_transport_ready(frame)

src/pipecat/transports/services/tavus.py

@@ -18,7 +18,6 @@ from pipecat.frames.frames import (
     Frame,
     InputAudioRawFrame,
     OutputAudioRawFrame,
-    OutputImageRawFrame,
     StartFrame,
     StartInterruptionFrame,
     TransportMessageFrame,
@@ -317,6 +316,9 @@ class TavusInputTransport(BaseInputTransport):
         self._params = params
         self._resampler = create_default_resampler()
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def setup(self, setup: FrameProcessorSetup):
         await super().setup(setup)
         await self._client.setup(setup)
@@ -327,6 +329,12 @@ class TavusInputTransport(BaseInputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         await self._client.start(frame)
         await self.set_transport_ready(frame)
 
@@ -375,6 +383,9 @@ class TavusOutputTransport(BaseOutputTransport):
         self._start_time = None
         self._current_idx_str: Optional[str] = None
 
+        # Whether we have seen a StartFrame already.
+        self._initialized = False
+
     async def setup(self, setup: FrameProcessorSetup):
         await super().setup(setup)
         await self._client.setup(setup)
@@ -385,6 +396,12 @@ class TavusOutputTransport(BaseOutputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+
+        if self._initialized:
+            return
+
+        self._initialized = True
+
         await self._client.start(frame)
         await self.set_transport_ready(frame)