wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-12-12 12:58:45 -08:00
parent 2de0737056
commit 3c3fd67d96
57 changed files with 56 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/pipecat/pipeline/parallel_pipeline.py
View File

@@ -28,8 +28,6 @@ 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):
@@ -51,8 +49,6 @@ 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)
@@ -120,8 +116,6 @@ 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()

6
src/pipecat/pipeline/pipeline.py
View File

@@ -17,8 +17,6 @@ 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)
@@ -32,8 +30,6 @@ 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)
@@ -74,8 +70,6 @@ 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:

6
src/pipecat/pipeline/sync_parallel_pipeline.py
View File

@@ -31,8 +31,6 @@ 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)
@@ -46,8 +44,6 @@ 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)
@@ -103,8 +99,6 @@ 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

4
src/pipecat/pipeline/task.py
View File

@@ -45,8 +45,6 @@ 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)
@@ -75,8 +73,6 @@ 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)

2
src/pipecat/processors/aggregators/gated.py
View File

@@ -56,8 +56,6 @@ 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)

2
src/pipecat/processors/aggregators/gated_openai_llm_context.py
View File

@@ -24,8 +24,6 @@ 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()

4
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -86,8 +86,6 @@ 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):
@@ -240,8 +238,6 @@ 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):

2
src/pipecat/processors/aggregators/sentence.py
View File

@@ -33,8 +33,6 @@ 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

2
src/pipecat/processors/aggregators/user_response.py
View File

@@ -85,8 +85,6 @@ 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):

2
src/pipecat/processors/aggregators/vision_image_frame.py
View File

@@ -31,8 +31,6 @@ 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):

2
src/pipecat/processors/async_generator.py
View File

@@ -24,8 +24,6 @@ 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)):

2
src/pipecat/processors/audio/audio_buffer_processor.py
View File

@@ -68,8 +68,6 @@ 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)

2
src/pipecat/processors/audio/vad/silero.py
View File

@@ -39,8 +39,6 @@ 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:

2
src/pipecat/processors/filters/frame_filter.py
View File

@@ -26,7 +26,5 @@ 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)

2
src/pipecat/processors/filters/function_filter.py
View File

@@ -29,8 +29,6 @@ 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:

1
src/pipecat/processors/filters/identity_filter.py
View File

@@ -26,5 +26,4 @@ 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)

2
src/pipecat/processors/filters/wake_check_filter.py
View File

@@ -45,8 +45,6 @@ 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)

2
src/pipecat/processors/filters/wake_notifier_filter.py
View File

@@ -32,8 +32,6 @@ 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()

54
src/pipecat/processors/frame_processor.py
View File

@@ -161,6 +161,13 @@ 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,
@@ -175,32 +182,13 @@ 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):
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
pass
async def push_error(self, error: ErrorFrame):
await self.push_frame(error, FrameDirection.UPSTREAM)
@@ -228,6 +216,28 @@ 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
#
@@ -289,7 +299,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:

2
src/pipecat/processors/frameworks/langchain.py
View File

@@ -36,8 +36,6 @@ 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.

16
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -380,8 +380,6 @@ 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)):
@@ -415,8 +413,6 @@ 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)):
@@ -446,8 +442,6 @@ 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):
@@ -473,8 +467,6 @@ 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):
@@ -496,8 +488,6 @@ 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):
@@ -514,8 +504,6 @@ 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):
@@ -532,8 +520,6 @@ 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):
@@ -642,8 +628,6 @@ 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

2
src/pipecat/processors/gstreamer/pipeline_source.py
View File

@@ -66,8 +66,6 @@ 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

2
src/pipecat/processors/idle_frame_processor.py
View File

@@ -35,8 +35,6 @@ 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

2
src/pipecat/processors/text_transformer.py
View File

@@ -27,8 +27,6 @@ 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):

2
src/pipecat/processors/user_idle_processor.py
View File

@@ -43,8 +43,6 @@ 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()

2
src/pipecat/services/ai_services.py
View File

@@ -110,8 +110,6 @@ 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):

1
src/pipecat/services/simli.py
View File

@@ -92,7 +92,6 @@ 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()

2
src/pipecat/transports/base_input.py
View File

@@ -79,8 +79,6 @@ 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

2
src/pipecat/transports/base_output.py
View File

@@ -108,8 +108,6 @@ 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

2
src/pipecat/utils/test_frame_processor.py
View File

@@ -13,8 +13,6 @@ 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?