qidZddlmZmZmZmZddlmZddlm Z ddl m Z m Z m Z Gdde ZGdd e ZGd d e Zy ) aPipeline implementation for connecting and managing frame processors. This module provides the main Pipeline class that connects frame processors in sequence and manages frame flow between them, along with helper classes for pipeline source and sink operations. )Callable CoroutineListOptional)Frame) BasePipeline)FrameDirectionFrameProcessorFrameProcessorSetupcJeZdZdZdeeegefffd Zdedeffd Z xZ S)PipelineSourcezSource processor that forwards frames to an upstream handler. This processor acts as the entry point for a pipeline, forwarding downstream frames to the next processor and upstream frames to a provided upstream handler function. upstream_push_framec 6t|dddi|||_y)zInitialize the pipeline source. Args: upstream_push_frame: Coroutine function to handle upstream frames. **kwargs: Additional arguments passed to parent class. enable_direct_modeTN)super__init___upstream_push_frame)selfrkwargs __class__s K/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/pipeline/pipeline.pyrzPipelineSource.__init__s" ;D;F;$7!frame directioncKt|||d{|xtjk(r|j ||d{ytj k(r|j ||d{yy7b757 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 DOWNSTREAM push_framerrrrs rrzPipelineSource.process_frame'stg#E9555(((//yAAA**ooeY777+ 6B73BA:.BA<,B3A>4B<B>B __name__ __module__ __qualname____doc__rrr rrr __classcell__rs@rr r s;8He^5Li5W,X8 8 8> 8 8rr cJeZdZdZdeeegefffd Zdedeffd Z xZ S) PipelineSinkzSink processor that forwards frames to a downstream handler. This processor acts as the exit point for a pipeline, forwarding upstream frames to the previous processor and downstream frames to a provided downstream handler function. downstream_push_framec 6t|dddi|||_y)zInitialize the pipeline sink. Args: downstream_push_frame: Coroutine function to handle downstream frames. **kwargs: Additional arguments passed to parent class. rTNr)rr_downstream_push_frame)rr-rrs rrzPipelineSink.__init__?s" ;D;F;&;#rrrcKt|||d{|xtjk(r|j ||d{ytj k(r|j ||d{yy7b757 wr)rrr rr!r r/r"s rrzPipelineSink.process_frameKssg#E9555(((ooeY777**11%CCC+ 68Cr#r$r*s@rr,r,7s@ <%-un.Ey.P%Q < D D> D Drr,ceZdZdZddddeedeedeeffdZedZ ed ed fd Z d Z d e ffd Z fdZdedeffd Zd e fdZdZdZxZS)Pipelinea#Main pipeline implementation that connects frame processors in sequence. Creates a linear chain of frame processors with automatic source and sink processors for external frame handling. Manages processor lifecycle and provides metrics collection from contained processors. N)sourcesink processorsr3r4ct|d|xst|j|d|_|xst |j|d|_|jg|z|j gz|_|jy)aInitialize the pipeline with a list of processors. Args: processors: List of frame processors to connect in sequence. source: An optional pipeline source processor. sink: An optional pipeline sink processor. T)rz::Source)namez::SinkN) rrr r!_sourcer,_sink _processors_link_processors)rr5r3r4rs rrzPipeline.__init__cs D1XhFW!X P\$//4&P 26,,*1LPTPZPZ|1[ 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. )r:rs rr5zPipeline.processorssrreturnr c|jgS)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. )r8r=s rentry_processorszPipeline.entry_processorss ~rcg}|jD]B}|jr|j||j|j D|S)aReturn processors that can generate metrics. Recursively collects all processors that support metrics generation, including those from nested pipelines. Returns: List of frame processors that can generate metrics. )r5can_generate_metricsappendextendprocessors_with_metrics)rservicesps rrEz Pipeline.processors_with_metricssP 9A%%'" OOA557 8 9rsetupcvKt||d{|j|d{y77w)zSet up the pipeline and all contained processors. Args: setup: Configuration for frame processor setup. N)rrH_setup_processors)rrHrs rrHzPipeline.setups8 gmE"""$$U+++ #+s959799crKt|d{|jd{y77w)z3Clean up the pipeline and all contained processors.N)rcleanup_cleanup_processors)rrs rrLzPipeline.cleanups2go&&((( (s737577rrcbKt|||d{|tjk(r3|jj |tjd{y|tj k(r3|jj |tj d{yy77N7 w)zProcess frames by routing them through the pipeline. Args: frame: The frame to process. direction: The direction of frame flow. N)rrr r r8 queue_framerr9r"s rrzPipeline.process_framesg#E9555 11 1,,**5.2K2KL L L .11 1**((0G0GH H H2 6 M Hs5B/B)AB/B+AB/"B-#B/+B/-B/cdK|jD]}|j|d{y7w)z&Set up all processors in the pipeline.N)r:rH)rrHrGs rrJzPipeline._setup_processorss/!! !A''%.  ! s $0.0cbK|jD]}|jd{y7w)z(Clean up all processors in the pipeline.N)r:rL)rrGs rrMzPipeline._cleanup_processorss-!! A))+    s #/-/cp|jd}|jddD]}|j||}y)z5Link all processors in sequence and set their parent.rN)r:link)rprevcurrs rr;zPipeline._link_processorss>"$$QR( D IIdOD r)r%r&r'r(rr rrpropertyr5r@rEr rHrLrr rrJrMr;r)r*s@rr2r2[s,0)-  ( (  ~&  8     $'7"8   ,!4,) I I> I!-@!  rr2N)r(typingrrrrpipecat.frames.framesrpipecat.pipeline.base_pipeliner"pipecat.processors.frame_processorr r r r r,r2rrrr\sH76'7bb8^8D!D>!DHy|yr