qi)dZddlZddlmZddlmZddlmZddlm Z ddl m Z m Z m Z mZddlmZdd lmZdd lmZmZmZeGd d e ZGd deZGddeZGddeZy)aSynchronous parallel pipeline implementation for concurrent frame processing. This module provides a pipeline that processes frames through multiple parallel pipelines simultaneously, synchronizing their output to maintain frame ordering and prevent duplicate processing. N) dataclass)chain)List)logger) ControlFrameEndFrameFrame SystemFrame) BasePipeline)Pipeline)FrameDirectionFrameProcessorFrameProcessorSetupceZdZdZy) SyncFramezControl frame used to synchronize parallel pipeline processing. This frame is sent through parallel pipelines to determine when the internal pipelines have finished processing a batch of frames. N)__name__ __module__ __qualname____doc__Y/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/pipeline/sync_parallel_pipeline.pyrrs   rrcPeZdZdZdej ffd Zdedeffd Z xZ S)SyncParallelPipelineSourcezSource processor for synchronous parallel pipeline processing. Routes frames to parallel pipelines and collects upstream responses for synchronization purposes. upstream_queuec4t|d||_y)zInitialize the sync parallel pipeline source. Args: upstream_queue: Queue for collecting upstream frames from the pipeline. Tenable_direct_modeN)super__init__ _up_queue)selfr __class__s rr z#SyncParallelPipelineSource.__init__-s D1'rframe directioncKt|||d{|xtjk(r%|jj |d{ytj k(r|j||d{yy7k757 wzProcess frames and route them based on direction. Args: frame: The frame to process. direction: The direction of frame flow. N)r process_framer UPSTREAMr!put DOWNSTREAM push_framer"r$r%r#s rr(z(SyncParallelPipelineSource.process_frame6sug#E9555(((nn((///**ooeY777+ 607s3B B7B B,B <B=B B B  rrrrasyncioQueuer r r r( __classcell__r#s@rrr&s0 (w}}( 8 8> 8 8rrcPeZdZdZdej ffd Zdedeffd Z xZ S)SyncParallelPipelineSinkzSink processor for synchronous parallel pipeline processing. Collects downstream frames from parallel pipelines and routes upstream frames back through the pipeline. downstream_queuec4t|d||_y)zInitialize the sync parallel pipeline sink. Args: downstream_queue: Queue for collecting downstream frames from the pipeline. TrN)rr _down_queue)r"r5r#s rr z!SyncParallelPipelineSink.__init__Ms D1+rr$r%cKt|||d{|xtjk(r|j ||d{ytj k(r$|j j|d{yy7k7>7 wr')rr(r r)r,r+r7r*r-s rr(z&SyncParallelPipelineSink.process_frameVswg#E9555(((ooeY777**&&**5111+ 681s3B B.B B5B <B=B B B r.r2s@rr4r4Fs0 ,, 2 2> 2 2rr4ceZdZdZfdZedZededfdZdee fdZ de ffd Z fd Z d ed effd ZxZS)SyncParallelPipelinea5Pipeline that processes frames through multiple parallel pipelines synchronously. Creates multiple parallel processing paths that all receive the same input frames and produces synchronized output. Each parallel path is a separate pipeline that processes frames independently, with synchronization points to ensure consistent ordering and prevent duplicate frame processing. The pipeline uses SyncFrame control frames to coordinate between parallel paths and ensure all paths have completed processing before moving to the next frame. ct|t|dk(r tdg|_g|_g|_tj|_ tj|_ tjd|d|D]}t|tstd|dtj}tj}t!|}t#|}|j j%||d|jj%||dt'|||}|j j%|tjd |dy ) aInitialize the synchronous parallel pipeline. Args: *args: Variable number of processor lists, each representing a parallel pipeline path. Each argument should be a list of FrameProcessor instances. Raises: Exception: If no arguments are provided. TypeError: If any argument is not a list of processors. rz0SyncParallelPipeline needs at least one argumentz Creating z pipelineszSyncParallelPipeline argument z is not a list) processorqueue)sourcesinkzFinished creating N)rr len Exception_sinks_sources _pipelinesr/r0r!r7rdebug isinstancelist TypeErrorrr4appendr ) r"args processorsup_queue down_queuer>r?pipeliner#s rr zSyncParallelPipeline.__init__rs0  t9>NP P   "==? yj12 -Jj$/"@ N [\\}}H J/9F+J7D MM v !K L KK  THE F 6EH OO " "8 ,# -&  )$z:;rc|jS)a?Return the list of sub-processors contained within this processor. Only compound processors (e.g. pipelines and parallel pipelines) have sub-processors. Non-compound processors will return an empty list. Returns: The list of sub-processors if this is a compound processor. )rDr"s rrKzSyncParallelPipeline.processorssrreturnrc|jS)aReturn the list of entry processors for this processor. Entry processors are the first processors in a compound processor (e.g. pipelines, parallel pipelines). Note that pipelines can also be an entry processor as pipelines are processors themselves. Non-compound processors will simply return an empty list. Returns: The list of entry processors. )rCrPs rentry_processorsz%SyncParallelPipeline.entry_processorss}}rc`ttjd|jDS)zCollect processors that can generate metrics from all parallel pipelines. Returns: List of frame processors that support metrics collection from all parallel paths. c3<K|]}|jyw)N)processors_with_metrics).0ps r z?SyncParallelPipeline.processors_with_metrics..s'](A(A(C']s)rGr from_iterablerDrPs rrVz,SyncParallelPipeline.processors_with_metricss$ E''']T__']]^^rsetupcKt||d{tj|jDcgc]}|j|c}d{y7Fcc}w7 w)zSet up the parallel pipeline and all contained processors. Args: setup: Configuration for frame processor setup. N)rr[r/gatherrD)r"r[rXr#s rr[zSyncParallelPipeline.setupsO gmE"""nntG!qwwu~GHHH #GHs,A&A!A&AA&A$A&A&cKt|d{tj|jDcgc]}|jc}d{y7Ecc}w7 w)z.wait_for_syncsJK(ILE))%; ; ;%+x!89"'))+- i+x)@A$..333([(4KL(nnY777)*/))+$5 )[(4KL  --ik9EEE"'))+- $Y :$..333OO%&+iik 1I%Y : <.38$5E-3 1s F(F-F(F.F(?F/F(/F0'F(FF(3F(F F(*F"+(F(F$'F(;F&<F(F(F(F(F(F( F("F($F(&F()rr(r/r0r r r)r]rBr!r+rCr7setemptyrcidr,addrd)r"r$r%rgsseen_idsr#s rr(z"SyncParallelPipeline.process_framesg#E9555 2$]] 238 2ES 22 // /..NRkkZ-4>>5)DZ  .33 3..PTP]P]^1-4#3#3UIF^  5..&&(..,,..Exxx'ooe^-D-DEEE UXX& NN $ $ & ..&&(5""((***..00Exxx'ooe^-F-FGGG UXX&    & & ( ""((*k 6H[  _  /E 1GsJ$JAJ$,J J$J5J$J!J$(J)AJ$.J/5J$$J%AJ$9AJ$:J ;5J$0J"1AJ$J$ J$J$J$J$ J$"J$)rrrrr propertyrKrrSrrVrr[r_r r r(r1r2s@rr:r:fs +E)E)rr:)rr/ dataclassesr itertoolsrtypingrlogururpipecat.frames.framesrrr r pipecat.pipeline.base_pipeliner pipecat.pipeline.pipeliner "pipecat.processors.frame_processorr rrrrr4r:rrrrwsp!LL7.bb     88@2~2@r)<r)r