qidZddlmZddlmZmZddlmZddlm Z m Z m Z m Z ddl mZddlmZmZmZddlmZmZmZGd d eZy ) aParallel pipeline implementation for concurrent frame processing. This module provides a parallel pipeline that processes frames through multiple sub-pipelines concurrently, with coordination for system frames and proper handling of pipeline lifecycle events. )chain)DictList)logger) CancelFrameEndFrameFrame StartFrame) BasePipeline)Pipeline PipelineSinkPipelineSource)FrameDirectionFrameProcessorFrameProcessorSetupceZdZdZfdZedZededfdZdee fdZ de ffd Z fd Z d ed effd Zd ed efdZd ed efdZdZxZS)ParallelPipelinea~Pipeline that processes frames through multiple sub-pipelines concurrently. Creates multiple parallel processing branches from the provided processor lists, coordinating frame flow and ensuring proper synchronization of lifecycle events like EndFrames. Each branch runs independently while system frames are handled specially to maintain pipeline coordination. cFt|t|dk(r tdg|_t |_i|_d|_g|_ tjd|d|D]}t|tstd|dt|j}t|j |d| }t#|j$|d | }t'||| }|jj)|tjd |dy )a6Initialize the parallel pipeline with processor lists. Args: *args: Variable number of processor lists, each becoming a parallel branch. Raises: Exception: If no processor lists are provided. TypeError: If any argument is not a list of processors. rz,ParallelPipeline needs at least one argumentFz Creating z pipelineszParallelPipeline argument z is not a listz::Source)namez::Sink)sourcesinkzFinished creating N)super__init__len Exception _pipelinesset _seen_ids_frame_counter_synchronizing_buffered_framesrdebug isinstancelist TypeErrorr_parallel_push_framer _pipeline_sink_push_framer append)selfargs processors num_pipelinesrrpipeline __class__s T/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/pipeline/parallel_pipeline.pyrzParallelPipeline.__init__"s  t9>JL L.0$)DF yj12 -Jj$/" >vVTaSbEcdD 6EH OO " "8 , -"  )$z:;c|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. rr)s r/r+zParallelPipeline.processorsSsr0returnrc|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. r2r3s r/entry_processorsz!ParallelPipeline.entry_processors_sr0c`ttjd|jDS)zCollect processors that can generate metrics from all parallel branches. Returns: List of frame processors that support metrics collection from all branches. c3<K|]}|jyw)N)processors_with_metrics).0ps r/ z;ParallelPipeline.processors_with_metrics..ss'](A(A(C']s)r$r from_iterablerr3s r/r9z(ParallelPipeline.processors_with_metricsms$ E''']T__']]^^r0setupcKt||d{|jD]}|j|d{y7/7 w)zSet up the parallel pipeline and all its branches. Args: setup: Configuration for frame processor setup. Raises: TypeError: If any processor list argument is not actually a list. N)rr>r)r)r>r;r.s r/r>zParallelPipeline.setupusHgmE""" !A''%.  ! # sA A'A AA A cKt|d{|jD]}|jd{y7.7 w)z4Clean up the parallel pipeline and all its branches.N)rcleanupr)r)r;r.s r/rAzParallelPipeline.cleanupsBgo A))+    sAA&AAAAframe directioncKt|||d{t|ttt frct |j|j|j<d|_ |jd{|jd{|jD]}|j||d{y77J747 w)zProcess frames through all parallel branches with lifecycle coordination. Args: frame: The frame to process. direction: The direction of frame flow. NT)r process_framer#r rrrrridr pause_processing_system_framespause_processing_frames queue_frame)r)rBrCr;r.s r/rEzParallelPipeline.process_framesg#E9555 ej(K@ A,/,@D   )"&D 557 7 7..0 0 0 2A--y1 1 1 2 6 8 0 2sFCCA%C=C>CC (C>C ?CC C CcK|j|jvrj|jj|j|jr|jj ||fy|j ||d{yy7w)a Push frames while avoiding duplicates using frame ID tracking. During lifecycle frame synchronization, non-lifecycle frames are buffered to prevent them from escaping the parallel pipeline before all branches have finished processing the lifecycle frame. N)rFraddr r!r( push_frame)r)rBrCs r/r&z%ParallelPipeline._parallel_push_framesn 884>> ) NN  uxx (""%%,,eY-?@ooeY777 * 8sA=B?BBc4Kt|tttfr|jj |j d}|dkDr:|j|j xxdzcc<|j|j }|dk(rjd|_|j||d{|jd{|jd{|jd{yy|j||d{y7i7S7=7'7 w)NrF) r#r rrrgetrFr r&_flush_buffered_framesresume_processing_system_framesresume_processing_frames)r)rBrC frame_counters r/r'z*ParallelPipeline._pipeline_sink_push_frames ej(K@ A //33EHHa@Mq ##EHH-2- $ 3 3EHH = !&+#//yAAA11333::<<<33555 "++E9= = = B3<5 =sZB"D$D%D<D=DDD,D-DD DDDDDc~K|j}g|_|D]\}}|j||d{!y7w)zGFlush frames that were buffered during lifecycle frame synchronization.N)r!rL)r)framesrBrCs r/rPz'ParallelPipeline._flush_buffered_framessD&& " & 4 E9//%3 3 3 4 3s 1=;=)__name__ __module__ __qualname____doc__rpropertyr+rr6rr9rr>rAr rrEr&r'rP __classcell__)r.s@r/rrs+2( 8 8. 8>U>~>&4r0rN)rY itertoolsrtypingrrlogururpipecat.frames.framesrrr r pipecat.pipeline.base_pipeliner pipecat.pipeline.pipeliner r r"pipecat.processors.frame_processorrrrrr0r/rds8JJ7LLbbi4|i4r0