Revert "no longer necessary to call super().process_frame(frame, direction)"

src/pipecat/pipeline/parallel_pipeline.py

@@ -28,6 +28,8 @@ class Source(FrameProcessor):
         self._push_frame_func = push_frame_func
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         match direction:
             case FrameDirection.UPSTREAM:
                 if isinstance(frame, SystemFrame):
@@ -49,6 +51,8 @@ class Sink(FrameProcessor):
         self._push_frame_func = push_frame_func
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         match direction:
             case FrameDirection.UPSTREAM:
                 await self.push_frame(frame, direction)
@@ -116,6 +120,8 @@ class ParallelPipeline(BasePipeline):
         self._down_task = loop.create_task(self._process_down_queue())
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if isinstance(frame, StartFrame):
             await self._start_tasks()
 

src/pipecat/pipeline/pipeline.py

@@ -17,6 +17,8 @@ class PipelineSource(FrameProcessor):
         self._upstream_push_frame = upstream_push_frame
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         match direction:
             case FrameDirection.UPSTREAM:
                 await self._upstream_push_frame(frame, direction)
@@ -30,6 +32,8 @@ class PipelineSink(FrameProcessor):
         self._downstream_push_frame = downstream_push_frame
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         match direction:
             case FrameDirection.UPSTREAM:
                 await self.push_frame(frame, direction)
@@ -70,6 +74,8 @@ class Pipeline(BasePipeline):
         await self._cleanup_processors()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if direction == FrameDirection.DOWNSTREAM:
             await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM)
         elif direction == FrameDirection.UPSTREAM:

src/pipecat/pipeline/sync_parallel_pipeline.py

@@ -31,6 +31,8 @@ class Source(FrameProcessor):
         self._up_queue = upstream_queue
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         match direction:
             case FrameDirection.UPSTREAM:
                 await self._up_queue.put(frame)
@@ -44,6 +46,8 @@ class Sink(FrameProcessor):
         self._down_queue = downstream_queue
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         match direction:
             case FrameDirection.UPSTREAM:
                 await self.push_frame(frame, direction)
@@ -99,6 +103,8 @@ class SyncParallelPipeline(BasePipeline):
     #
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         # The last processor of each pipeline needs to be synchronous otherwise
         # this element won't work. Since, we know it should be synchronous we
         # push a SyncFrame. Since frames are ordered we know this frame will be

src/pipecat/pipeline/task.py

@@ -45,6 +45,8 @@ class Source(FrameProcessor):
         self._up_queue = up_queue
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         match direction:
             case FrameDirection.UPSTREAM:
                 await self._handle_upstream_frame(frame)
@@ -73,6 +75,8 @@ class Sink(FrameProcessor):
         self._down_queue = down_queue
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         # We really just want to know when the EndFrame reached the sink.
         if isinstance(frame, EndFrame):
             await self._down_queue.put(frame)

src/pipecat/processors/aggregators/gated.py

@@ -56,6 +56,8 @@ class GatedAggregator(FrameProcessor):
         self._accumulator: List[Tuple[Frame, FrameDirection]] = []
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         # We must not block system frames.
         if isinstance(frame, SystemFrame):
             await self.push_frame(frame, direction)

src/pipecat/processors/aggregators/gated_openai_llm_context.py

@@ -24,6 +24,8 @@ class GatedOpenAILLMContextAggregator(FrameProcessor):
         self._last_context_frame = None
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if isinstance(frame, StartFrame):
             await self.push_frame(frame)
             await self._start()

src/pipecat/processors/aggregators/llm_response.py

@@ -86,6 +86,8 @@ class LLMResponseAggregator(FrameProcessor):
     # and T2 would be dropped.
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         send_aggregation = False
 
         if isinstance(frame, self._start_frame):
@@ -238,6 +240,8 @@ class LLMFullResponseAggregator(FrameProcessor):
         self._aggregation = ""
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if isinstance(frame, TextFrame):
             self._aggregation += frame.text
         elif isinstance(frame, LLMFullResponseEndFrame):

src/pipecat/processors/aggregators/sentence.py

@@ -33,6 +33,8 @@ class SentenceAggregator(FrameProcessor):
         self._aggregation = ""
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         # We ignore interim description at this point.
         if isinstance(frame, InterimTranscriptionFrame):
             return

src/pipecat/processors/aggregators/user_response.py

@@ -85,6 +85,8 @@ class ResponseAggregator(FrameProcessor):
     # and T2 would be dropped.
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         send_aggregation = False
 
         if isinstance(frame, self._start_frame):

src/pipecat/processors/aggregators/vision_image_frame.py

@@ -31,6 +31,8 @@ class VisionImageFrameAggregator(FrameProcessor):
         self._describe_text = None
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if isinstance(frame, TextFrame):
             self._describe_text = frame.text
         elif isinstance(frame, InputImageRawFrame):

src/pipecat/processors/async_generator.py

@@ -24,6 +24,8 @@ class AsyncGeneratorProcessor(FrameProcessor):
         self._data_queue = asyncio.Queue()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         await self.push_frame(frame, direction)
 
         if isinstance(frame, (CancelFrame, EndFrame)):

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -68,6 +68,8 @@ class AudioBufferProcessor(FrameProcessor):
         self._bot_audio_buffer = bytearray()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         # Include all audio from the user.
         if isinstance(frame, InputAudioRawFrame):
             resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate)

src/pipecat/processors/audio/vad/silero.py

@@ -39,6 +39,8 @@ class SileroVAD(FrameProcessor):
     #
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if isinstance(frame, AudioRawFrame):
             await self._analyze_audio(frame)
             if self._audio_passthrough:

src/pipecat/processors/filters/frame_filter.py

@@ -26,5 +26,7 @@ class FrameFilter(FrameProcessor):
         return isinstance(frame, ControlFrame) or isinstance(frame, SystemFrame)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if self._should_passthrough_frame(frame):
             await self.push_frame(frame, direction)

src/pipecat/processors/filters/function_filter.py

@@ -29,6 +29,8 @@ class FunctionFilter(FrameProcessor):
         return isinstance(frame, SystemFrame) or direction != self._direction
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         passthrough = self._should_passthrough_frame(frame, direction)
         allowed = await self._filter(frame)
         if passthrough or allowed:

src/pipecat/processors/filters/identity_filter.py

@@ -26,4 +26,5 @@ class IdentityFilter(FrameProcessor):
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process an incoming frame by passing it through unchanged."""
+        await super().process_frame(frame, direction)
         await self.push_frame(frame, direction)

src/pipecat/processors/filters/wake_check_filter.py

@@ -45,6 +45,8 @@ class WakeCheckFilter(FrameProcessor):
             self._wake_patterns.append(pattern)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         try:
             if isinstance(frame, TranscriptionFrame):
                 p = self._participant_states.get(frame.user_id)

src/pipecat/processors/filters/wake_notifier_filter.py

@@ -32,6 +32,8 @@ class WakeNotifierFilter(FrameProcessor):
         self._filter = filter
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if isinstance(frame, self._types) and await self._filter(frame):
             await self._notifier.notify()
 

src/pipecat/processors/frame_processor.py

@@ -161,13 +161,6 @@ class FrameProcessor:
     def get_clock(self) -> BaseClock:
         return self._clock
 
-    async def pause_processing_frames(self):
-        self.__should_block_frames = True
-
-    async def resume_processing_frames(self):
-        self.__input_event.set()
-        self.__should_block_frames = False
-
     async def queue_frame(
         self,
         frame: Frame,
@@ -182,13 +175,32 @@ class FrameProcessor:
 
         if isinstance(frame, SystemFrame):
             # We don't want to queue system frames.
-            await self._process_frame(frame, direction)
+            await self.process_frame(frame, direction)
         else:
             # We queue everything else.
             await self.__input_queue.put((frame, direction, callback))
 
+    async def pause_processing_frames(self):
+        self.__should_block_frames = True
+
+    async def resume_processing_frames(self):
+        self.__input_event.set()
+        self.__should_block_frames = False
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        pass
+        if isinstance(frame, StartFrame):
+            self._clock = frame.clock
+            self._allow_interruptions = frame.allow_interruptions
+            self._enable_metrics = frame.enable_metrics
+            self._enable_usage_metrics = frame.enable_usage_metrics
+            self._report_only_initial_ttfb = frame.report_only_initial_ttfb
+        elif isinstance(frame, StartInterruptionFrame):
+            await self._start_interruption()
+            await self.stop_all_metrics()
+        elif isinstance(frame, StopInterruptionFrame):
+            self._should_report_ttfb = True
+        elif isinstance(frame, CancelFrame):
+            self._cancelling = True
 
     async def push_error(self, error: ErrorFrame):
         await self.push_frame(error, FrameDirection.UPSTREAM)
@@ -216,28 +228,6 @@ class FrameProcessor:
             raise Exception(f"Event handler {event_name} already registered")
         self._event_handlers[event_name] = []
 
-    #
-    # Frame processing
-    #
-
-    async def _process_frame(self, frame: Frame, direction: FrameDirection):
-        if isinstance(frame, StartFrame):
-            self._clock = frame.clock
-            self._allow_interruptions = frame.allow_interruptions
-            self._enable_metrics = frame.enable_metrics
-            self._enable_usage_metrics = frame.enable_usage_metrics
-            self._report_only_initial_ttfb = frame.report_only_initial_ttfb
-        elif isinstance(frame, StartInterruptionFrame):
-            await self._start_interruption()
-            await self.stop_all_metrics()
-        elif isinstance(frame, StopInterruptionFrame):
-            self._should_report_ttfb = True
-        elif isinstance(frame, CancelFrame):
-            self._cancelling = True
-
-        # Call subclass.
-        await self.process_frame(frame, direction)
-
     #
     # Handle interruptions
     #
@@ -299,7 +289,7 @@ class FrameProcessor:
                 (frame, direction, callback) = await self.__input_queue.get()
 
                 # Process the frame.
-                await self._process_frame(frame, direction)
+                await self.process_frame(frame, direction)
 
                 # If this frame has an associated callback, call it now.
                 if callback:

src/pipecat/processors/frameworks/langchain.py

@@ -36,6 +36,8 @@ class LangchainProcessor(FrameProcessor):
         self._participant_id = participant_id
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if isinstance(frame, LLMMessagesFrame):
             # Messages are accumulated on the context as a list of messages.
             # The last one by the human is the one we want to send to the LLM.

src/pipecat/processors/frameworks/rtvi.py

@@ -380,6 +380,8 @@ class RTVISpeakingProcessor(RTVIFrameProcessor):
         super().__init__(**kwargs)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         await self.push_frame(frame, direction)
 
         if isinstance(frame, (UserStartedSpeakingFrame, UserStoppedSpeakingFrame)):
@@ -413,6 +415,8 @@ class RTVIUserTranscriptionProcessor(RTVIFrameProcessor):
         super().__init__(**kwargs)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         await self.push_frame(frame, direction)
 
         if isinstance(frame, (TranscriptionFrame, InterimTranscriptionFrame)):
@@ -442,6 +446,8 @@ class RTVIUserLLMTextProcessor(RTVIFrameProcessor):
         super().__init__(**kwargs)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         await self.push_frame(frame, direction)
 
         if isinstance(frame, OpenAILLMContextFrame):
@@ -467,6 +473,8 @@ class RTVIBotTranscriptionProcessor(RTVIFrameProcessor):
         self._aggregation = ""
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         await self.push_frame(frame, direction)
 
         if isinstance(frame, UserStartedSpeakingFrame):
@@ -488,6 +496,8 @@ class RTVIBotLLMProcessor(RTVIFrameProcessor):
         super().__init__(**kwargs)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         await self.push_frame(frame, direction)
 
         if isinstance(frame, LLMFullResponseStartFrame):
@@ -504,6 +514,8 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor):
         super().__init__(**kwargs)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         await self.push_frame(frame, direction)
 
         if isinstance(frame, TTSStartedFrame):
@@ -520,6 +532,8 @@ class RTVIMetricsProcessor(RTVIFrameProcessor):
         super().__init__(**kwargs)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         await self.push_frame(frame, direction)
 
         if isinstance(frame, MetricsFrame):
@@ -628,6 +642,8 @@ class RTVIProcessor(FrameProcessor):
         await self._push_transport_message(message, exclude_none=False)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         # Specific system frames
         if isinstance(frame, StartFrame):
             # Push StartFrame before start(), because we want StartFrame to be

src/pipecat/processors/gstreamer/pipeline_source.py

@@ -66,6 +66,8 @@ class GStreamerPipelineSource(FrameProcessor):
         bus.connect("message", self._on_gstreamer_message)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         # Specific system frames
         if isinstance(frame, StartFrame):
             # Push StartFrame before start(), because we want StartFrame to be

src/pipecat/processors/idle_frame_processor.py

@@ -35,6 +35,8 @@ class IdleFrameProcessor(FrameProcessor):
         self._create_idle_task()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         await self.push_frame(frame, direction)
 
         # If we are not waiting for any specific frame set the event, otherwise

src/pipecat/processors/text_transformer.py

@@ -27,6 +27,8 @@ class StatelessTextTransformer(FrameProcessor):
         self._transform_fn = transform_fn
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if isinstance(frame, TextFrame):
             result = self._transform_fn(frame.text)
             if isinstance(result, Coroutine):

src/pipecat/processors/user_idle_processor.py

@@ -43,6 +43,8 @@ class UserIdleProcessor(FrameProcessor):
         await self._idle_task
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         # Check for end frames before processing
         if isinstance(frame, (EndFrame, CancelFrame)):
             await self._stop()

src/pipecat/services/ai_services.py

@@ -110,6 +110,8 @@ class AIService(FrameProcessor):
                 logger.warning(f"Unknown setting for {self.name} service: {key}")
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         if isinstance(frame, StartFrame):
             await self.start(frame)
         elif isinstance(frame, CancelFrame):

src/pipecat/services/simli.py

@@ -92,6 +92,7 @@ class SimliVideoService(FrameProcessor):
             pass
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
         if isinstance(frame, StartFrame):
             await self.push_frame(frame, direction)
             await self._start_connection()

src/pipecat/transports/base_input.py

@@ -79,6 +79,8 @@ class BaseInputTransport(FrameProcessor):
     #
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         # Specific system frames
         if isinstance(frame, StartFrame):
             # Push StartFrame before start(), because we want StartFrame to be

src/pipecat/transports/base_output.py

@@ -120,6 +120,8 @@ class BaseOutputTransport(FrameProcessor):
     #
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
         #
         # System frames (like StartInterruptionFrame) are pushed
         # immediately. Other frames require order so they are put in the sink

src/pipecat/utils/test_frame_processor.py

@@ -13,6 +13,8 @@ class TestFrameProcessor(FrameProcessor):
         super().__init__()
 
     async def process_frame(self, frame, direction):
+        await super().process_frame(frame, direction)
+
         if not self.test_frames[
             0
         ]:  # then we've run out of required frames but the generator is still going?