From 3f3a853d715f5d602a69b9028a1c09d79da70e02 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Dec 2024 14:45:20 -0800 Subject: [PATCH] no longer necessary to call AIService super().start/stop/cancel(frame) --- CHANGELOG.md | 4 ++++ src/pipecat/services/ai_services.py | 18 ++++++++++++------ src/pipecat/services/assemblyai.py | 3 --- src/pipecat/services/azure.py | 3 --- src/pipecat/services/canonical.py | 2 -- src/pipecat/services/deepgram.py | 3 --- .../services/gemini_multimodal_live/gemini.py | 3 --- src/pipecat/services/gladia.py | 3 --- .../services/openai_realtime_beta/openai.py | 3 --- src/pipecat/services/riva.py | 3 --- 10 files changed, 16 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c4a005280..fdc45c33f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- It's no longer necessary to call `super().start/stop/cancel(frame)` if you + subclass and implement `AIService.start/stop/cancel()`. This is all now done + internally and will avoid possible issues if you forget to add it. + - It's no longer necessary to call `super().process_frame(frame, direction)` if you subclass and implement `FrameProcessor.process_frame()`. This is all now done internally and will avoid possible issues if you forget to add it. diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index e324d413c..5f867246f 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -111,11 +111,11 @@ class AIService(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): - await self.start(frame) + await self._start(frame) elif isinstance(frame, CancelFrame): - await self.cancel(frame) + await self._cancel(frame) elif isinstance(frame, EndFrame): - await self.stop(frame) + await self._stop(frame) async def process_generator(self, generator: AsyncGenerator[Frame | None, None]): async for f in generator: @@ -125,6 +125,15 @@ class AIService(FrameProcessor): else: await self.push_frame(f) + async def _start(self, frame: StartFrame): + await self.start(frame) + + async def _stop(self, frame: EndFrame): + await self.stop(frame) + + async def _cancel(self, frame: CancelFrame): + await self.cancel(frame) + class LLMService(AIService): """This class is a no-op but serves as a base class for LLM services.""" @@ -248,19 +257,16 @@ class TTSService(AIService): pass async def start(self, frame: StartFrame): - await super().start(frame) if self._push_stop_frames: self._stop_frame_task = self.get_event_loop().create_task(self._stop_frame_handler()) async def stop(self, frame: EndFrame): - await super().stop(frame) if self._stop_frame_task: self._stop_frame_task.cancel() await self._stop_frame_task self._stop_frame_task = None async def cancel(self, frame: CancelFrame): - await super().cancel(frame) if self._stop_frame_task: self._stop_frame_task.cancel() await self._stop_frame_task diff --git a/src/pipecat/services/assemblyai.py b/src/pipecat/services/assemblyai.py index 36a7e92ff..577df7d76 100644 --- a/src/pipecat/services/assemblyai.py +++ b/src/pipecat/services/assemblyai.py @@ -61,15 +61,12 @@ class AssemblyAISTTService(STTService): self._settings["language"] = language async def start(self, frame: StartFrame): - await super().start(frame) await self._connect() async def stop(self, frame: EndFrame): - await super().stop(frame) await self._disconnect() async def cancel(self, frame: CancelFrame): - await super().cancel(frame) await self._disconnect() async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index a95ff7d3c..01ab22940 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -676,16 +676,13 @@ class AzureSTTService(STTService): yield None async def start(self, frame: StartFrame): - await super().start(frame) self._speech_recognizer.start_continuous_recognition_async() async def stop(self, frame: EndFrame): - await super().stop(frame) self._speech_recognizer.stop_continuous_recognition_async() self._audio_stream.close() async def cancel(self, frame: CancelFrame): - await super().cancel(frame) self._speech_recognizer.stop_continuous_recognition_async() self._audio_stream.close() diff --git a/src/pipecat/services/canonical.py b/src/pipecat/services/canonical.py index 265cc1b1b..2986d8b94 100644 --- a/src/pipecat/services/canonical.py +++ b/src/pipecat/services/canonical.py @@ -84,11 +84,9 @@ class CanonicalMetricsService(AIService): self._output_dir = output_dir async def stop(self, frame: EndFrame): - await super().stop(frame) await self._process_audio() async def cancel(self, frame: CancelFrame): - await super().cancel(frame) await self._process_audio() async def process_frame(self, frame: Frame, direction: FrameDirection): diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 6578f2873..89064dab8 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -176,15 +176,12 @@ class DeepgramSTTService(STTService): await self._connect() async def start(self, frame: StartFrame): - await super().start(frame) await self._connect() async def stop(self, frame: EndFrame): - await super().stop(frame) await self._disconnect() async def cancel(self, frame: CancelFrame): - await super().cancel(frame) await self._disconnect() async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index bf433054a..efdeea329 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -229,15 +229,12 @@ class GeminiMultimodalLiveLLMService(LLMService): # async def start(self, frame: StartFrame): - await super().start(frame) await self._connect() async def stop(self, frame: EndFrame): - await super().stop(frame) await self._disconnect() async def cancel(self, frame: CancelFrame): - await super().cancel(frame) await self._disconnect() # diff --git a/src/pipecat/services/gladia.py b/src/pipecat/services/gladia.py index 8909c4bb2..adc261222 100644 --- a/src/pipecat/services/gladia.py +++ b/src/pipecat/services/gladia.py @@ -177,18 +177,15 @@ class GladiaSTTService(STTService): return language_to_gladia_language(language) async def start(self, frame: StartFrame): - await super().start(frame) response = await self._setup_gladia() self._websocket = await websockets.connect(response["url"]) self._receive_task = self.get_event_loop().create_task(self._receive_task_handler()) async def stop(self, frame: EndFrame): - await super().stop(frame) await self._send_stop_recording() await self._websocket.close() async def cancel(self, frame: CancelFrame): - await super().cancel(frame) await self._websocket.close() async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index ac492a205..e8e2e50dd 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -112,15 +112,12 @@ class OpenAIRealtimeBetaLLMService(LLMService): # async def start(self, frame: StartFrame): - await super().start(frame) await self._connect() async def stop(self, frame: EndFrame): - await super().stop(frame) await self._disconnect() async def cancel(self, frame: CancelFrame): - await super().cancel(frame) await self._disconnect() # diff --git a/src/pipecat/services/riva.py b/src/pipecat/services/riva.py index 6be722d49..7d14cb583 100644 --- a/src/pipecat/services/riva.py +++ b/src/pipecat/services/riva.py @@ -187,17 +187,14 @@ class ParakeetSTTService(STTService): return False async def start(self, frame: StartFrame): - await super().start(frame) self._thread_task = self.get_event_loop().create_task(self._thread_task_handler()) self._response_task = self.get_event_loop().create_task(self._response_task_handler()) self._response_queue = asyncio.Queue() async def stop(self, frame: EndFrame): - await super().stop(frame) await self._stop_tasks() async def cancel(self, frame: CancelFrame): - await super().cancel(frame) await self._stop_tasks() async def _stop_tasks(self):