qiVdZddlZddlZddlmZmZmZmZddlm Z ddl m Z m Z m Z ddlmZe GddZGd d ZGd d e Zy) zTask observer for managing pipeline frame observers. This module provides a proxy observer system that manages multiple observers for pipeline frame events, ensuring that observer processing doesn't block the main pipeline execution. N)AnyDictListOptional) dataclass) BaseObserverFrameProcessed FramePushed)BaseTaskManagercXeZdZUdZej ed<ejed<eed<y)Proxya{Proxy data for managing observer tasks and queues. This represents is the data received from the main observer that is queued for later processing. Parameters: queue: Queue for frame data awaiting observer processing. task: Asyncio task running the observer's frame processing loop. observer: The actual observer instance being proxied. queuetaskobserverN) __name__ __module__ __qualname____doc__asyncioQueue__annotations__TaskrP/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/pipeline/task_observer.pyr r s#  == ,,rr ceZdZdZy)_PipelineStartedSignalzDInternal sentinel queued to observers when the pipeline has started.N)rrrrrrrrr*sNrrceZdZdZdddeeedeffdZdefdZ defd Z d Z d Z fd Z d ZdefdZdefdZdedefdZdeedeeeffdZdefdZdej4defdZxZS) TaskObserveraProxy observer that manages multiple observers without blocking the pipeline. This is a pipeline frame observer that is meant to be used as a proxy to the user provided observers. That is, this is the observer that should be passed to the frame processors. Then, every time a frame is pushed this observer will call all the observers registered to the pipeline task. This observer makes sure that passing frames to observers doesn't block the pipeline by creating a queue and a task for each user observer. When a frame is received, it will be put in a queue for efficiency and later processed by each task. N) observersr task_managerc Vt|di||xsg|_||_d|_y)aInitialize the TaskObserver. Args: observers: List of observers to manage. Defaults to empty list. task_manager: Task manager for creating and managing observer tasks. **kwargs: Additional arguments passed to the base observer. Nr)super__init__ _observers _task_manager_proxies)selfr r!kwargs __class__s rr$zTaskObserver.__init__>s2 "6"#/r)  rrc|jj||jr!|j|}||j|<yy)zjAdd a new observer to the managed list. Args: observer: The observer to add. N)r%appendr' _create_proxyr(rproxys r add_observerzTaskObserver.add_observerSsB x( ==&&x0E&+DMM( # rc*K|jrW||jvrI|j|}|j|=|jj|jd{||jvr|jj |yy7/w)ztRemove an observer and clean up its resources. Args: observer: The observer to remove. N)r'r& cancel_taskrr%remover.s rremove_observerzTaskObserver.remove_observerbs| ==X6MM(+E h'$$00< < < t & OO " "8 , ' =sAB!B"0BcLK|j|j|_yw)zStart all proxy observer tasks.N)_create_proxiesr%r'r(s rstartzTaskObserver.startts,,T__= s"$cK|jsy|jjD]/}|jj|jd{1y7w)zStop all proxy observer tasks.N)r'valuesr&r2r)r(r/s rstopzTaskObserver.stopxsM}} ]]))+ =E$$00< < < = >SWSaSa#00666D.1//555 OO   %575ssAF(E:8FF'F>F ?A!F F !F:F ;)F$F%F:F? F F F FF)rrrrrrrr r$r0r4r8r;r=r@r rFr rJr r-rr6rr?rrrN __classcell__)r*s@rrr0s  37 D./ &  * ,\ ,-l-$>="<(>(( (lul);\SXEX@Y((w}} rr)rrrVtypingrrrrattrrpipecat.observers.base_observerrr r "pipecat.utils.asyncio.task_managerr r rrrrrrmsU,,UU>  "  c<cr