qi/,dZddlZddlZddlZddlmZddlmZm Z m Z m Z m Z m Z mZmZmZmZddlmZddlmZmZmZddlmZddlmZdd lmZdd lmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-dd l.m/Z/m0Z0dd l1m2Z2m3Z3dd l4m5Z5ddl6m7Z7ddl8m9Z9ddl:m;Z;mZ>m?Z?m@Z@ddlAmBZBddlCmDZDddlEmFZFmGZGmHZHddlImJZJmKZKmLZLddlMmNZNmOZOmPZPddlQmRZRddlSmTZTddlUmVZVdZWeWdzZXdZYdZZedZ[Gdd e2Z\Gd!d"eZ]Gd#d$e;Z^y)%aPipeline task implementation for managing frame processing pipelines. This module provides the main PipelineTask class that orchestrates pipeline execution, frame routing, lifecycle management, and monitoring capabilities including heartbeats, idle detection, and observer integration. N)Path) Any AsyncIterableDictIterableListOptionalSetTupleTypeTypeVar)logger) BaseModel ConfigDictField)BaseInterruptionStrategy) BaseClock) SystemClock)BotSpeakingFrame CancelFrameCancelTaskFrameEndFrame EndTaskFrame ErrorFrameFrameHeartbeatFrameInterruptionFrameInterruptionTaskFrame MetricsFrame StartFrame StopFrame StopTaskFrameUserSpeakingFrame)ProcessingMetricsDataTTFBMetricsData) BaseObserver FramePushed)TurnTrackingObserver)UserBotLatencyObserver) BasePipeline)BasePipelineTaskPipelineTaskParams)Pipeline PipelineSinkPipelineSource) TaskObserver)LLMUserContextAggregator)FrameDirectionFrameProcessorFrameProcessorSetup) RTVIObserverRTVIObserverParams RTVIProcessor)BaseTaskManager TaskManagerTaskManagerParams)is_tracing_available)TracingContext)TurnTraceObserverg? i,g4@Tc\eZdZdZdej deeedfffd Z de fdZ xZ S)IdleFrameObserverzIdle timeout observer. This observer waits for specific frames being generated in the pipeline. If the frames are generated the given asyncio event is set. If the event is not set it means the pipeline is probably idle. idle_eventidle_timeout_frames.c\t|||_||_t |_y)zInitialize the observer. Args: idle_event: The event to set if the idle timeout frames are being pushed. idle_timeout_frames: A tuple with the frames that should set the event when received N)super__init__ _idle_event_idle_timeout_framesset_processed_frames)selfrBrC __class__s G/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/pipeline/task.pyrFzIdleFrameObserver.__init__Os* %$7!!$datacZK|jj|jvry|jj|jjt |jt s t |j|j r|jjyyw)z~Callback executed when a frame is pushed in the pipeline. Args: data: The frame push event data. N) frameidrJadd isinstancer rHrGrI)rKrOs rM on_push_framezIdleFrameObserver.on_push_frame[sv ::==D22 2  ""4::==1 djj* -DJJHaHa1b    "2csB)B+) __name__ __module__ __qualname____doc__asyncioEventr r rrFr'rU __classcell__rLs@rMrArAFs= 'gmm '%PTUZP[]`P`Ja ' # #rNrAc eZdZUdZedZdZeed<dZ e ed<dZ e ed<d Z eed <d Z eed <d Zeed <eZeed <eeZeeed<eeZeeed<d Zeed<dZeed<eeZeeefed<y)PipelineParamsaConfiguration parameters for pipeline execution. These parameters are usually passed to all frame processors through StartFrame. For other generic pipeline task parameters use PipelineTask constructor arguments instead. Parameters: allow_interruptions: Whether to allow pipeline interruptions. .. deprecated:: 0.0.99 Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead. audio_in_sample_rate: Input audio sample rate in Hz. audio_out_sample_rate: Output audio sample rate in Hz. enable_heartbeats: Whether to enable heartbeat monitoring. enable_metrics: Whether to enable metrics collection. enable_usage_metrics: Whether to enable usage metrics. heartbeats_period_secs: Period between heartbeats in seconds. interruption_strategies: [deprecated] Strategies for bot interruption behavior. .. deprecated:: 0.0.99 Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead. observers: [deprecated] Use `observers` arg in `PipelineTask` class. .. deprecated:: 0.0.58 Use the `observers` argument in the `PipelineTask` class instead. report_only_initial_ttfb: Whether to report only initial time to first byte. send_initial_empty_metrics: Whether to send initial empty metrics. start_metadata: Additional metadata for pipeline start. T)arbitrary_types_allowedallow_interruptionsi>audio_in_sample_ratei]audio_out_sample_rateFenable_heartbeatsenable_metricsenable_usage_metricsheartbeats_period_secs)default_factoryinterruption_strategies observersreport_only_initial_ttfbsend_initial_empty_metricsstart_metadataN) rVrWrXrYr model_configrabool__annotations__rbintrcrdrerfHEARTBEAT_SECSrgfloatrlistrirrrjr&rkrldictrmrstrrrNrMr_r_ksBd;L $$ %#%!&3&#t# ND !&$&$2E2>CTX>YT":;Y$)$$?ItL!?%*d*'++%*4%@NDcN@rNr_c$eZdZdZdddeddddddeefeddddddede e de e d e d e d e d e ed e ede de de deeedfde e de eede ede ede ef"fdZede fdZedefdZede efdZede efdZedefdZ edeeedffdZ!edeeedffd Z"d!effd" Z#d#efd$Z$d#efd%Z%d&eeedffd'Z&d&eeedffd(Z'd&eeedffd)Z(d&eeedffd*Z)de fd+Z*d,Z+dd-d.e efd/Z,de-fd0Z.e/j`fd1ed2e/fd3Z1e/j`fd4e2ee3ezd2e/fd5Z4dd-d.e efd6Z5d7Z6d8Z7d9Z8d:Z9d;Z:d<Z;deZ>d1efd?Z?d@Z@de-fdAZAdBe fdCZBdDZCd1ed2e/fdEZDd1ed2e/fdFZEdGZFdHZGdIZHde fdJZIdKZJdLZKdMZLdeMeeNffdNZOdOePdPeeQde eQfdQZRxZSS)R PipelineTaska/ Manages the execution of a pipeline, handling frame processing and task lifecycle. This class orchestrates pipeline execution with comprehensive monitoring, event handling, and lifecycle management. It provides event handlers for various pipeline states and frame types, idle detection, heartbeat monitoring, and observer integration. Event handlers available: - on_frame_reached_upstream: Called when upstream frames reach the source - on_frame_reached_downstream: Called when downstream frames reach the sink - on_idle_timeout: Called when pipeline is idle beyond timeout threshold - on_pipeline_started: Called when pipeline starts with StartFrame - on_pipeline_stopped: [deprecated] Called when pipeline stops with StopFrame .. deprecated:: 0.0.86 Use `on_pipeline_finished` instead. - on_pipeline_ended: [deprecated] Called when pipeline ends with EndFrame .. deprecated:: 0.0.86 Use `on_pipeline_finished` instead. - on_pipeline_cancelled: [deprecated] Called when pipeline is cancelled with CancelFrame .. deprecated:: 0.0.86 Use `on_pipeline_finished` instead. - on_pipeline_finished: Called after the pipeline has reached any terminal state. This includes: - StopFrame: pipeline was stopped (processors keep connections open) - EndFrame: pipeline ended normally - CancelFrame: pipeline was cancelled Use this event for cleanup, logging, or post-processing tasks. Users can inspect the frame if they need to handle specific cases. - on_pipeline_error: Called when an error occurs with ErrorFrame Example:: @task.event_handler("on_frame_reached_upstream") async def on_frame_reached_upstream(task, frame): ... @task.event_handler("on_idle_timeout") async def on_pipeline_idle_timeout(task): ... @task.event_handler("on_pipeline_started") async def on_pipeline_started(task, frame): ... @task.event_handler("on_pipeline_finished") async def on_pipeline_finished(task, frame): ... @task.event_handler("on_pipeline_error") async def on_pipeline_error(task, frame): ... NTF)paramsadditional_span_attributescancel_on_idle_timeoutcancel_timeout_secscheck_dangling_tasksclockconversation_idenable_tracingenable_turn_tracking enable_rtvirCidle_timeout_secsrjrtvi_processorrtvi_observer_params task_managerpipelinerzr{r|r}r~rrrrrrC.rrjrrrc t||xs t|_|xsi|_||_||_||_|xs t|_ ||_ | xr t|_ | |_ | |_|jjrZddl}|j#5|j%d|j'dt(ddd|jj}|xsg}d|_d|_d|_d|_|jr*t3|_|j5|j*|jr|j*rt7|_t9|_|j5|j,t;|j*|j,|j|j|j0|_|j5|j.d|_d|_|xs tA|_!tEjF|_$d|_%tEjF|_&d|_'d|_(d|_)d}|jU|tV}tYd|D}|r|st[j\|dn|s|rt[j\|d nf|r"|r t[j^|d ||_)nB| r@|xs tW|_)|j5|jRja| d }|jRr(|jbjed dtVfd}tEjf|_4d|_5|jr(tm|jh| }|j5|tEjf|_7tEjf|_8tEjf|_9tu|jv|d}ty|jz|d|_>|r|jR|gn|g}t|||j||_@t||jB|_Bt|_Dt|_E|jd|jd|jd|jd|jd|jd|jd|jd|jdy#1swYtxYw)asInitialize the PipelineTask. Args: pipeline: The pipeline to execute. params: Configuration parameters for the pipeline. additional_span_attributes: Optional dictionary of attributes to propagate as OpenTelemetry conversation span attributes. cancel_on_idle_timeout: Whether the pipeline task should be cancelled if the idle timeout is reached. cancel_timeout_secs: Timeout (in seconds) to wait for cancellation to happen cleanly. check_dangling_tasks: Whether to check for processors' tasks finishing properly. clock: Clock implementation for timing operations. conversation_id: Optional custom ID for the conversation. enable_rtvi: Whether to automatically add RTVI support to the pipeline. enable_tracing: Whether to enable tracing. enable_turn_tracking: Whether to enable turn tracking. idle_timeout_frames: A tuple with the frames that should trigger an idle timeout if not received within `idle_timeout_seconds`. idle_timeout_secs: Timeout (in seconds) to consider pipeline idle or None. If a pipeline is idle the pipeline task will be cancelled automatically. observers: List of observers for monitoring pipeline execution. rtvi_observer_params: The RTVI observer parameter to use if RTVI is enabled. rtvi_processor: The RTVI processor to add if RTVI is enabled. task_manager: Optional task manager for handling asyncio tasks. rNalwayszGField 'observers' is deprecated, use the 'observers' parameter instead.)latency_trackerrr{tracing_contextFc3<K|]}t|tywN)rTr5).0os rM z(PipelineTask.__init__..Os%UajL&A%UszZ: RTVIProcessor found in pipeline but no RTVIObserver in observers. Make sure to add both.zZ: RTVIObserver found in observers but no RTVIProcessor in pipeline. Make sure to add both.z|: RTVIProcessor and RTVIObserver found, skipping default ones. They are both added by default, no need to add them yourself.)rzTon_client_readyrtvic@K|jd{y7wr) set_bot_ready)rs rMrz.PipelineTask.__init__..on_client_readyhs((***s )rBrCz::Source)namez::Sink)sourcesink)rjron_frame_reached_upstreamon_frame_reached_downstreamon_idle_timeouton_pipeline_startedon_pipeline_stoppedon_pipeline_endedon_pipeline_cancelledon_pipeline_finishedon_pipeline_error)GrErFr__params_additional_span_attributes_cancel_on_idle_timeout_cancel_timeout_secs_check_dangling_tasksr_clock_conversation_idr;_enable_tracing_enable_turn_tracking_idle_timeout_secsrjwarningscatch_warnings simplefilterwarnDeprecationWarning_turn_tracking_observer_user_bot_latency_observer_turn_trace_observer_tracing_contextr(appendr<r)r= _finished _cancelledr9 _task_managerrZQueue _push_queue_process_push_task_heartbeat_queue_heartbeat_push_task_heartbeat_monitor_task_rtvi_find_processorr7anyrerrorwarningcreate_rtvi_observerr event_handlerr[rG_idle_monitor_taskrA_pipeline_start_event_pipeline_end_event_pipeline_finished_eventr/_source_push_framer._sink_push_frame_sinkr- _pipeliner0 _observerrI_reached_upstream_types_reached_downstream_types_register_event_handler)rKrrzr{r|r}r~rrrrrrCrrjrrrr prepend_rtvi external_rtviexternal_observer_foundridle_frame_observerr processorsrLs rMrFzPipelineTask.__init__s` 1!1 +E+K('=$$7!%9",{} /-H2F2H%9""3 << ! ! ((* %%h/ ]&   ..IO GK$LP'AE!:>  % %+?+AD (   T99 :   D$@$@$2$4D !.D.FD +   T<< =(9,, $ ? ? $ 5 5+/+K+K $ 5 5 )D %   T66 7*:[]#==?:>!( <@!?C$  ,,X}E "%%U9%U"U !8 LL&)) #: LL&)) 6 NN&PP 'DJ ':=?DJ   TZZ<  " ""3++$7#    0 1&-]]_"$+==? )0 %   7 7h>OP!$"7"7foN 05& $$%@A $$%BC $$%67 $$%:; $$%:; $$%89 $$%<= $$%;< $$%89S  s ((UU returnc|jS)zxGet the pipeline parameters for this task. Returns: The pipeline parameters configuration. )rrKs rMrzzPipelineTask.paramss||rNc|jS)zGet the full pipeline managed by this pipeline task. This will also include any internal processors added by the pipeline task. Returns: The pipeline managed by the pipeline task. )rrs rMrzPipelineTask.pipeline~~rNc|jS)zGet the turn tracking observer if enabled. Returns: The turn tracking observer instance or None if not enabled. )rrs rMturn_tracking_observerz#PipelineTask.turn_tracking_observers+++rNc|jS)zGet the turn trace observer if enabled. Returns: The turn trace observer instance or None if not enabled. )rrs rMturn_trace_observerz PipelineTask.turn_trace_observers(((rNcN|jst|d|jS)zGet the RTVI processor if RTVI is enabled. Returns: The RTVI processor added to the pipeline when RTVI is enabled. z RTVI is not enabled.)r Exceptionrs rMrzPipelineTask.rtvis(zztf$9:; ;zzrNc,t|jS)zGet the currently configured upstream frame type filters. Returns: Tuple of frame types that trigger the on_frame_reached_upstream event. )tuplerrs rMreached_upstream_typesz#PipelineTask.reached_upstream_typessT1122rNc,t|jS)zGet the currently configured downstream frame type filters. Returns: Tuple of frame types that trigger the on_frame_reached_downstream event. )rrrs rMreached_downstream_typesz%PipelineTask.reached_downstream_typessT3344rN event_namec|dvrHddl}|j5|jd|jd|dtdddt ||S#1swYxYw)zDecorator for registering event handlers. Args: event_name: The name of the event to handle. Returns: The decorator function that registers the handler. )rrrrNrzEvent 'z4' is deprecated, use 'on_pipeline_finished' instead.)rrrrrrEr)rKrrrLs rMrzPipelineTask.event_handlersn ^ ^ ((* %%h/ j\)]^& w$Z00  s ,AA&observerc:|jj|y)zAdd an observer to monitor pipeline execution. Args: observer: The observer to add to the pipeline monitoring. N)r add_observerrKrs rMrzPipelineTask.add_observers ##H-rNcVK|jj|d{y7w)zRemove an observer from pipeline monitoring. Args: observer: The observer to remove from pipeline monitoring. N)rremove_observerrs rMrzPipelineTask.remove_observers nn,,X666s )')typesc$t||_y)zSet which frame types trigger the on_frame_reached_upstream event. Args: types: Tuple of frame types to monitor for upstream events. N)rIrrKrs rMset_reached_upstream_filterz(PipelineTask.set_reached_upstream_filters (+5z$rNc$t||_y)zSet which frame types trigger the on_frame_reached_downstream event. Args: types: Tuple of frame types to monitor for downstream events. N)rIrrs rMset_reached_downstream_filterz*PipelineTask.set_reached_downstream_filters *-U&rNc:|jj|y)zAdd frame types to trigger the on_frame_reached_upstream event. Args: types: Tuple of frame types to add to upstream monitoring. N)rupdaters rMadd_reached_upstream_filterz(PipelineTask.add_reached_upstream_filters $$++E2rNc:|jj|y)zAdd frame types to trigger the on_frame_reached_downstream event. Args: types: Tuple of frame types to add to downstream monitoring. N)rrrs rMadd_reached_downstream_filterz*PipelineTask.add_reached_downstream_filter&s &&--e4rNc|jS)zCheck if the pipeline task has finished execution. This indicates whether the tasks has finished, meaninig all processors have stopped. Returns: True if all processors have stopped and the task is complete. )rrs rM has_finishedzPipelineTask.has_finished.rrNcKtjd|d|jtd{y7w)zSchedule the pipeline to stop after processing all queued frames. Sends an EndFrame to gracefully terminate the pipeline once all current processing is complete. zTask z scheduled to stop when doneN)rdebug queue_framerrs rMstop_when_donezPipelineTask.stop_when_done9s4  uTF">?@xz***s6A>Areasonrc^K|js|j|d{yy7w)zRequest the running pipeline to cancel. Args: reason: Optional reason to indicate why the pipeline is being cancelled. rN)r_cancelrKrs rMcancelzPipelineTask.cancelBs+ ~~,,f,- - - -s "-+-cZK|jry|j|d{|jd{ |jd{ t jd|d|jd{|jr|jd|_ t jd|dy777w#tj $rMt jd|d|jd{7|jd{7wxYw7#t jd|d|jd{7|jr|jd|_ t jd|dwxYww)zStart and manage the pipeline execution until completion or cancellation. Args: params: Configuration parameters for pipeline execution. NzPipeline task z got cancelled from outside...z is finishing...Tz has finished) r_setup _create_tasks_wait_for_pipeline_finishedrZCancelledErrorrrr _cancel_tasksr_print_dangling_tasksr)rKrzs rMrunzPipelineTask.runKst     kk&!!!   """ ?224 4 4 LL>$/?@ A$$& & &))**,!DN LL>$}= >; " # 5%%  LL>$/MN O,,. 224 4 4   ' LL>$/?@ A$$& & &))**,!DN LL>$}= >s&F+C F+CF+CCC,F+ D6 AF+F+C?D3DD3+D.,D33D86F+8-F(%E(&AF((F+rQ directioncK|tjk(r$|jj|d{y|jj ||d{y7*7w)aXQueue a single frame to be pushed through the pipeline. Downstream frames are pushed from the beginning of the pipeline. Upstream frames are pushed from the end of the pipeline. Args: frame: The frame to be processed. direction: The direction to push the frame. Defaults to downstream. N)r2 DOWNSTREAMrputrrrKrQrs rMrzPipelineTask.queue_frametsS 11 1""&&u- - -**(( : : : . :s!2A#A$A#A!A#!A#framescKt|tr&|23d{}|j||d{$t|tr"|D]}|j||d{yy7R7:6y7w)ayQueue multiple frames to be pushed through the pipeline. Downstream frames are pushed from the beginning of the pipeline. Upstream frames are pushed from the end of the pipeline. Args: frames: An iterable or async iterable of frames to be processed. direction: The direction to push the frames. Defaults to downstream. N)rTrrr)rKrrrQs rM queue_frameszPipelineTask.queue_framess fm ,% 9 9e&&ui888  ) 9&&ui888 9* 98 &9sAA4A0A,A0A4A./A4#A2$A4,A0.A40A4cK|jsCtjd|d|_|jt |d{yy7w)zInternal cancellation logic for the pipeline task. Args: reason: Optional reason to indicate why the pipeline is being cancelled. zCancelling pipeline task TrN)rrrrrrs rMrzPipelineTask._cancelsM  LL4TF; <"DO"";f#=> > > ?sA A A AcK|jj|j|d|_|jj d{|jS7w)z/Create and start all pipeline processing tasks.z::_process_push_queueN)r create_task_process_push_queuerrstartrs rMrzPipelineTask._create_taskss^"&"4"4"@"@  $ $ &4&0E(F# nn""$$$&&& %sAA%A#A%c|jjrr|je|jj |j |d|_|jj |j |d|_yyy)zHStart heartbeat tasks if heartbeats are enabled and not already running.Nz::_heartbeat_push_handlerz::_heartbeat_monitor_handler)rrdrrr_heartbeat_push_handler_heartbeat_monitor_handlerrrs rM_maybe_start_heartbeat_tasksz)PipelineTask._maybe_start_heartbeat_taskss << ) )d.G.G.O(,(:(:(F(F,,.4&8Q0R)D %,0+=+=+I+I//1dV;W3X,D ( /P )rNc|jr3|jj|j|d|_yy)z9Start idle monitoring task if idle timeout is configured.z::_idle_monitor_handlerN)rrr_idle_monitor_handlerrrs rM_maybe_start_idle_taskz#PipelineTask._maybe_start_idle_tasks?  " "&*&8&8&D&D**,6M.N'D # #rNc@K|jjd{|jr4|jj |jd{d|_|j d{|j d{y7u7>7!7 w)z"Cancel all running pipeline tasks.N)rstoprr cancel_task_maybe_cancel_heartbeat_tasks_maybe_cancel_idle_taskrs rMr zPipelineTask._cancel_taskssnn!!###  " "$$001H1HI I I&*D #00222**,,, $ J 3,sEBB8BBB8B9BBBBBBcDK|jjsy|jr4|jj |jd{d|_|j r5|jj |j d{d|_yy7M7w)z+Cancel heartbeat tasks if they are running.N)rrdrrr$rrs rMr%z*PipelineTask._maybe_cancel_heartbeat_taskss||--   $ $$$001J1JK K K(,D %  ' '$$001M1MN N N+/D ( ( L Os$A B B?B BB B cK|jr5|jj|jd{d|_yy7 w)z-Cancel idle monitoring task if it is running.N)rrr$rs rMr&z$PipelineTask._maybe_cancel_idle_tasks?  " "$$001H1HI I I&*D # # Is5AAAc|jj}g}|D]N}|jt|jd|jt |jdPt |S)zDCreate an initial metrics frame with zero values for all processors.g) processorvalue)rO)rprocessors_with_metricsrr%rr$r)rKrrOps rM_initial_metrics_framez#PipelineTask._initial_metrics_framesf^^;;=  LA KK!&&D E KK-cJ K L&&rNcKtj|d|d|jjd{|jj tj|d|dy7:w)zDWait for the specified start frame to reach the end of the pipeline.z: Starting. Waiting for $ to reach the end of the pipeline...N: z8 reached the end of the pipeline, pipeline is now ready.)rrrwaitclear)rKrQs rM_wait_for_pipeline_startz%PipelineTask._wait_for_pipeline_startsj v5eW<`ab((--/// ""((* vRw&^_` 0s9A8A6;A8c|Kfd}tjddttr|d{n=jj d{tjddjj jjy7w7Vw)z>Wait for the specified frame to reach the end of the pipeline.cK tjjjjd{t j ddjdd{jdd{y7T#tj$rt jddYgwxYw7U7=#jdd{7jdd{7wxYww)Ntimeoutr1z! reached the end of the pipeline.z: timeout waiting for z= to reach the end of the pipeline (being blocked somewhere?).rr) rZwait_forrr2rrr TimeoutErrorr_call_event_handler)rQrKsrMwait_for_cancelz.wait_for_cancels N&&,,113T=V=V vRw.OPQ ../FNNN../EuMMM'' f25'9vw  OM../FNNN../EuMMMsD >..((* * *   GD2H$I  % % > > @ ..((* * *   + +s4B'B!-B'B#AB'B%B'#B'%B'c K|jj|jt|jj |jj |jj|jj|j|jj|jj|jj|j }|j|_|j j#|d{|j%|d{|jjrG|jj&r1|j j#|j)d{d}d}|r|j*j-d{}|j j#|d{t/|t0t2t4fr|j7|d{t/|t0t2t4f }t/|t4 }|j*j9|r|j;|d{y7^7H7777t7w)a8Process frames from the push queue and send them through the pipeline. This is the task that runs the pipeline for the first time by sending a StartFrame and by pushing any other frames queued by the user. It runs until the tasks is cancelled or stopped (e.g. with an EndFrame). ) rarbrcrerrfrkrirNT)rrr!r rrarbrcrerrfrkrir_create_start_metadatametadatarrr4rlr.rgetrTrrr!r= task_donerJ)rK start_framerunningrErQs rMrz PipelineTask._process_push_queue7s  ##% $ @ @!%!B!B"&,,"D"D<<66//!%!B!B%)\\%J%J$(LL$H$H 11   $::< nn((555++K888 << & &4<<+R+R..,,T-H-H-JK K K**..00E..,,U3 3 3%+x!CD11%888$U[(I,NOOG#-eY#??     & & (mm,---% 6 9 L 1 38 .sDJI2J-I5.AJ I8 'J2I:3"JI<3J I> A JJ,J-J5J8J:J<J>JJcZKt|t|jr|jd|d{t|trHt j |d||jt|jd{yt|trHt j |d||jt|jd{yt|trugF G"";ell#CD D D } - LLD6!-@A A A z ***+>F F F{{ 7w?@&&{}555$'?wGH++ O B E 0 B F6s5H+HAH+HAH+&H 'A H+2H#3AH+H% *H+3H'4AH+8H)9"H+H+ H+#H+%H+'H+)H+cKt|t|jr|jd|d{t|trg|jd|d{|j j d{|j|jjyt|trO|jd|d{|jd|d{|jjyt|trO|jd|d{|jd|d{|jjyt|tr|jjyt|tr$|jj!|d{yy77s7T77777w)a3Process frames coming downstream from the pipeline. This tasks process frames coming downstream from the pipeline. For example, heartbeat frames or an EndFrame which would indicate all processors have handled the EndFrame and therefore we can exit the task cleanly. rNrrrr)rTrrr;r rrrrrIrrr!rrrrrs rMrzPipelineTask._sink_push_frames eU4#A#AB C**+H%P P P eZ (**+@%H H H..446 6 6  - - /  & & * * , x (**+>F F F**+A5I I I  $ $ ( ( * y )**+@%H H H**+A5I I I  $ $ ( ( * { +  $ $ ( ( * ~ .''++E2 2 2/+ Q I 6 G I I I 3s5G%G)G%!G"!G%GAG%GG%2G3AG%7G8G%G!A8G% G# G%G%G%G%G%G%!G%#G%cK |jjt|jj d{t j |jjd{w767w)z+Push heartbeat frames at regular intervals.) timestampN) rrrrget_timerZsleeprrgrs rMrz$PipelineTask._heartbeat_push_handlers`..,,^dkkFZFZF\-]^ ^ ^-- C CD D D  _ Ds$AA?A;0A?5A=6A?=A?cKt} tj|jj |d{}|j j |jz dz }tj|d|d|jj7d#tj$rtj|d|dY3wxYww)azMonitor heartbeat frames for processing time and timeout detection. This task monitors heartbeat frames. If a heartbeat frame has not been received for a long period a warning will be logged. It also logs the time that a heartbeat frame takes to processes, that is how long it takes for the heartbeat frame to traverse all the pipeline. r7Niʚ;z: heartbeat frame processed in z secondsz-: heartbeat frame not received for more than ) HEARTBEAT_MONITOR_SECSrZr9rrNrrWrVrtracerOr:r)rK wait_timerQ process_times rMrz'PipelineTask._heartbeat_monitor_handlers+  %..t/D/D/H/H/JT]^^ $ 4 4 6 HMY  v%D\NRZ[\%%//1 ^'' fI)T\] s9C2B$B"A#B$!C"B$$.CCCCc<Kd}|r` tj|jj|jd{|jj |r_yy7##tj $r|jd{7}Y5wxYww)aMonitor pipeline activity and detect idle conditions. Tracks frame activity and triggers idle timeout events when the pipeline hasn't received relevant frames within the timeout period. Note: Heartbeats are excluded from idle detection. Tr7N)rZr9rGr2rr3r:_idle_timeout_detected)rKrQs rMr z"PipelineTask._idle_monitor_handlers >&&t'7'7'<'<'>H_H_```  &&(`'' > $ ; ; === >sLB>>#fL6Z[! S]?  S v%H2aSQRR SsR2E7D<D<E7CEE E9E7E E4 E/)E7/E44E7c ^KtjjddjdDcgc]}|s| }}|D]}ddl}|j 5|j d|jdtddd t|j}|j}tjj|t|}|rt!j"|d|tjj%|}|j&j)||j+|d{}|D]} |j-|  ycc}w#1swYxYw71#t.$r(} t!j0|d |d | Yd} ~ ad} ~ wwxYww) zGDynamically load observers from files listed in PIPECAT_OBSERVER_FILES.PIPECAT_OBSERVER_FILESrcrdrNrzObserver files (and environment variable `PIPECAT_OBSERVER_FILES`) is deprecated, use setup files instead (and `PIPECAT_SETUP_FILES`) instead.z loading observers from z' error loading external observers from r1)rfrgrNrhrrrrrrrirjrkrlrmrvrrrornrpcreate_observersrrr) rKrqobserver_filesrrsrtrurvrjrrws rMrBz!PipelineTask._load_observer_files s%'ZZ^^4Lb%Q%W%WX[%\b`a!bb WA ((* %%h/ e&  WAw("ii  ~~==k3t9ULLD6)A$!HI'^^< ? KsAci}|j|jtrd|d<|j|jj |S)z?Build and return start metadata including user-provided values.Tdeprecated_openaillmcontext)rrr1rrrm)rKrms rMrLz#PipelineTask._create_start_metadata.sI   0H I<@N8 9 dll99:rNr*processor_typecrt||r|S|jD]}|j||}|s|cSy)z?Recursively find a processor of the given type in the pipeline.N)rTrr)rKr*rr-founds rMrzPipelineTask._find_processor;sF i 0 %% A((N;E  rN)TrVrWrXrYCANCEL_TIMEOUT_SECSrr#IDLE_TIMEOUT_SECSr*r r_rurorsrrvr r rrr&r7r6r8rFpropertyrzrr(rr=rrrrrrrrrrrrrrr,r r2rrrrrrrrr!r r%r&rr.r4r=r rrJrrrrrr r_rArBr rrrLr3r?rr\r]s@rMryrysE=F,059'+%8%)%))-$%) 8HJ[7\->2626=A26'G:G:( G: %-TN G: !% G:#G:#G: "G:"#G:G:#G:G:#4;#34G:$E?G: D./!G:"!/#G:$''9:%G:&/'G:R,,1E(F,,)X.?%@))m3d5k3.>(?335%U S0@*A5511*.\.7l72tE{C7G1H245ec9I3J43tE{C7G1H355ec9I3J5 d +7;.hsm.'? 2'?T9G8Q8Q;;'5;(%3$=$=9-"669"9*8< ?x} ?' - 0+ ' 'aEa,%,<+*#5*$+t+").V%Ie%I%IN3E3n3BE(> d$S@W>@ S#X   a U]^_U` rNry)_rYrZimportlib.utilrkrfpathlibrtypingrrrrrr r r r r logururpydanticrrr6pipecat.audio.interruptions.base_interruption_strategyrpipecat.clocks.base_clockrpipecat.clocks.system_clockrpipecat.frames.framesrrrrrrrrrrrr r!r"r#pipecat.metrics.metricsr$r%pipecat.observers.base_observerr&r'(pipecat.observers.turn_tracking_observerr(+pipecat.observers.user_bot_latency_observerr)pipecat.pipeline.base_pipeliner*pipecat.pipeline.base_taskr+r,pipecat.pipeline.pipeliner-r.r/pipecat.pipeline.task_observerr0+pipecat.processors.aggregators.llm_responser1"pipecat.processors.frame_processorr2r3r4"pipecat.processors.frameworks.rtvir5r6r7"pipecat.utils.asyncio.task_managerr8r9r:pipecat.utils.tracing.setupr;%pipecat.utils.tracing.tracing_contextr<)pipecat.utils.tracing.turn_trace_observerr=rrrZrrr?rAr_ryrwrNrMrs ```11[/3"KEIN7KLL7Pbb^^^^<@G'", CL"# "#J/AY/Adg#grN