From 90ad2a4e8185f1f35fb5c8f55e9f278e8c81e43e Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 8 Feb 2026 14:44:48 -0500 Subject: [PATCH] Remove SequentialMergePipeline --- .../pipeline/to_be_updated/merge_pipeline.py | 53 ------------------- 1 file changed, 53 deletions(-) delete mode 100644 src/pipecat/pipeline/to_be_updated/merge_pipeline.py diff --git a/src/pipecat/pipeline/to_be_updated/merge_pipeline.py b/src/pipecat/pipeline/to_be_updated/merge_pipeline.py deleted file mode 100644 index 0254b6309..000000000 --- a/src/pipecat/pipeline/to_be_updated/merge_pipeline.py +++ /dev/null @@ -1,53 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Sequential pipeline merging for Pipecat. - -This module provides a pipeline implementation that sequentially merges -the output from multiple pipelines, processing them one after another -in a specified order. -""" - -from typing import List - -from pipecat.frames.frames import EndFrame, EndPipeFrame -from pipecat.pipeline.pipeline import Pipeline - - -class SequentialMergePipeline(Pipeline): - """Pipeline that sequentially merges output from multiple pipelines. - - This pipeline merges the sink queues from a list of pipelines by processing - frames from each pipeline's sink sequentially in the order specified. Each - pipeline runs to completion before the next one begins processing. - """ - - def __init__(self, pipelines: List[Pipeline]): - """Initialize the sequential merge pipeline. - - Args: - pipelines: List of pipelines to merge sequentially. Pipelines will - be processed in the order they appear in this list. - """ - super().__init__([]) - self.pipelines = pipelines - - async def run_pipeline(self): - """Run all pipelines sequentially and merge their output. - - Processes each pipeline in order, consuming all frames from each - pipeline's sink until an EndFrame or EndPipeFrame is encountered, - then moves to the next pipeline. After all pipelines complete, - sends a final EndFrame to signal completion. - """ - for idx, pipeline in enumerate(self.pipelines): - while True: - frame = await pipeline.sink.get() - if isinstance(frame, EndFrame) or isinstance(frame, EndPipeFrame): - break - await self.sink.put(frame) - - await self.sink.put(EndFrame())