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

no longer necessary to call AIService super().start/stop/cancel(frame)

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-12-12 14:45:20 -08:00
parent 10f854aeba
commit 3f3a853d71
10 changed files with 16 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGELOG.md
View File

@@ -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.

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

@@ -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

3
src/pipecat/services/assemblyai.py
View File

@@ -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]:

3
src/pipecat/services/azure.py
View File

@@ -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()

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

@@ -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):

3
src/pipecat/services/deepgram.py
View File

@@ -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]:

3
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

@@ -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()
#

3
src/pipecat/services/gladia.py
View File

@@ -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]:

3
src/pipecat/services/openai_realtime_beta/openai.py
View File

@@ -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()
#

3
src/pipecat/services/riva.py
View File

@@ -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):