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

parallel_pipeline: fix system frames again

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-12-10 18:12:33 -08:00
parent 9e6fabf110
commit 7b6bbc29ed
1 changed files with 9 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -27,11 +27,8 @@ class Source(FrameProcessor):
match direction:
case FrameDirection.UPSTREAM:
# We don't want to queue system frames as they would be
# processed by a separate task.
if isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
else:
# SystemFrames are pushed directly from ParallelPipeline.
if not isinstance(frame, SystemFrame):
await self._up_queue.put(frame)
case FrameDirection.DOWNSTREAM:
await self.push_frame(frame, direction)
@@ -49,11 +46,8 @@ class Sink(FrameProcessor):
case FrameDirection.UPSTREAM:
await self.push_frame(frame, direction)
case FrameDirection.DOWNSTREAM:
# We don't want to queue system frames as they would be
# processed by a separate task.
if isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
else:
# SystemFrames are pushed directly from ParallelPipeline.
if not isinstance(frame, SystemFrame):
await self._down_queue.put(frame)
@@ -123,11 +117,13 @@ class ParallelPipeline(BasePipeline):
await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sinks])
elif direction == FrameDirection.DOWNSTREAM:
# If we get a downstream frame we process it in each source.
# TODO(aleix): We are creating task for each frame. For real-time
# video/audio this might be too slow. We should use an already
# created task instead.
await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sources])
# If we have a SystemFrame we will push it from this task. Note that the
# connected sinks and sources ignore SystemFrames.
if isinstance(frame, SystemFrame):
await self.push_frame(frame, direction)
# If we get an EndFrame we stop our queue processing tasks and wait on
# all the pipelines to finish.
if isinstance(frame, (CancelFrame, EndFrame)):