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

transports: don't queue system frames

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-06-10 20:56:01 -07:00
parent ce2ebd3198
commit 6554479d39
2 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -83,9 +83,9 @@ class BaseInputTransport(FrameProcessor):
await super().process_frame(frame, direction)
if isinstance(frame, CancelFrame):
await self.stop()
# We don't queue a CancelFrame since we want to stop ASAP.
await self.push_frame(frame, direction)
await self.stop()
elif isinstance(frame, StartFrame):
await self.start(frame)
await self._internal_push_frame(frame, direction)

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

@@ -28,6 +28,7 @@ from pipecat.frames.frames import (
ImageRawFrame,
StartInterruptionFrame,
StopInterruptionFrame,
SystemFrame,
TransportMessageFrame)
from pipecat.transports.base_transport import TransportParams
@@ -53,6 +54,7 @@ class BaseOutputTransport(FrameProcessor):
if self._params.camera_out_enabled:
self._camera_out_queue = queue.Queue()
self._sink_queue = queue.Queue()
self._sink_thread = None
self._stopped_event = asyncio.Event()
self._is_interrupted = threading.Event()
@@ -106,7 +108,8 @@ class BaseOutputTransport(FrameProcessor):
if self._params.camera_out_enabled:
await self._camera_out_thread
await self._sink_thread
if self._sink_thread:
await self._sink_thread
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -121,11 +124,13 @@ class BaseOutputTransport(FrameProcessor):
self._sink_queue.put_nowait(frame)
# EndFrame is managed in the queue handler.
elif isinstance(frame, CancelFrame):
await self.push_frame(frame, direction)
await self.stop()
elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame):
await self.push_frame(frame, direction)
elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame):
await self._handle_interruptions(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
else:
self._sink_queue.put_nowait(frame)