qi| tdZddlZddlmZddlmZmZmZmZm Z m Z ddl m Z m Z mZmZmZddlmZmZddlmZddlmZdd lmZmZdd lmZmZeGd d eZGd deZGddeZ dddddddddede!dee e"dee e"de e de!deeedeede!de e e e e ffdZ#y)z2Testing utilities for Pipecat pipeline components.N) dataclass) AwaitableCallableListOptionalSequenceTuple)EndFrameFrameHeartbeatFrame StartFrame SystemFrame) BaseObserver FramePushed)Pipeline)PipelineRunner)PipelineParams PipelineTask)FrameDirectionFrameProcessorc eZdZUdZdZeed<y) SleepFrameaNA system frame that introduces a sleep delay in the test pipeline. This frame is used by the test framework to control timing between frame processing, allowing tests to separate system frames from data or control frames. Parameters: sleep: Duration to sleep in seconds before processing the next frame. g?sleepN)__name__ __module__ __qualname____doc__rfloat__annotations__E/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/tests/utils.pyrrsE5r!rcLeZdZdZdedeeegedfffd Zde fdZ xZ S)HeartbeatsObserverzObserver that monitors heartbeat frames from a specific processor. This observer watches for HeartbeatFrames from a target processor and invokes a callback when they are detected, useful for testing timing and lifecycle events. targetheartbeat_callbackNc @t|di|||_||_y)a.Initialize the heartbeats observer. Args: target: The frame processor to monitor for heartbeat frames. heartbeat_callback: Async callback function to invoke when heartbeats are detected. **kwargs: Additional arguments passed to the parent observer. Nr )super__init___target _callback)selfr%r&kwargs __class__s r"r)zHeartbeatsObserver.__init__2s# "6" +r!datacK|j}|j}||jk(r6t|tr%|j |j|d{yyy7w)zHandle frame push events and detect heartbeats from target processor. Args: data: The frame push event data containing source and frame information. N)sourceframer* isinstancer r+)r,r/srcr2s r" on_push_framez HeartbeatsObserver.on_push_frameDsS kk  $,, :e^#D..u5 5 5$E  5sAA#A!A#) rrrrrrr rr)rr5 __classcell__r.s@r"r$r$*sA,,%nn%EyQU%VW ,$ 6 6r!r$c^eZdZdZdddej dedeffdZde d effd Z xZ S) QueuedFrameProcessoraA processor that captures frames in a queue for testing purposes. This processor intercepts frames flowing in a specific direction and stores them in a queue for later inspection during testing, while still allowing the frames to continue through the pipeline. T) ignore_startqueuequeue_directionr:cPt|d||_||_||_y)aInitialize the queued frame processor. Args: queue: The asyncio queue to store captured frames. queue_direction: The direction of frames to capture (UPSTREAM or DOWNSTREAM). ignore_start: Whether to ignore StartFrames when capturing. T)enable_direct_modeN)r(r)_queue_queue_direction _ignore_start)r,r;r<r:r.s r"r)zQueuedFrameProcessor.__init__Ys, D1 /)r!r2 directioncKt|||d{||jk(r?t|tr |j s#|j j|d{|j||d{y7m7!7 w)zProcess frames and capture them in the queue if they match the direction. Args: frame: The frame to process. direction: The direction the frame is flowing. N) r( process_framer@r3r rAr?put push_frame)r,r2rBr.s r"rDz"QueuedFrameProcessor.process_framelswg#E9555 -- -eZ08J8Jkkooe,,,ooeY/// 6-/s4B BA B %B&B ?B B B  B ) rrrrasyncioQueuerboolr)r rDr6r7s@r"r9r9QsL" *}}*( *  *& 0 0> 0 0r!r9FT) enable_rtviexpected_down_framesexpected_up_framesr: observerspipeline_paramssend_end_frame processorrJrKrLframes_to_sendr:rMrNrOreturncK|xsg}|xs t}tj} tj} t| tj |} t| tj |} t| || g} t| d|||fd}t}tj|j|d{g}| jsL| jd{}t|trs|j!|| jsL|d}|D]}||j"j$dz } |dz }d}|D]}||j$dz }|dz }t'd |t'd |t)|t)|k(sJt+||D]\}}t||rJg}| js:| jd{}|j!|| js:|St'd |t'd |t)|t)|k(sJt+||D]\}}t||rJ||fS777w) a`Run a test pipeline with the specified processor and validate frame flow. This function creates a test pipeline with the given processor, sends the specified frames through it, and validates that the expected frames are received in both upstream and downstream directions. Args: processor: The frame processor to test. enable_rtvi: Whether RTVI should be enabled in this test. expected_down_frames: Expected frame types flowing downstream (optional). expected_up_frames: Expected frame types flowing upstream (optional). frames_to_send: Sequence of frames to send through the processor. ignore_start: Whether to ignore StartFrames in frame validation. observers: Optional list of observers to attach to the pipeline. pipeline_params: Optional pipeline parameters. send_end_frame: Whether to send an EndFrame at the end of the test. Returns: Tuple containing (downstream_frames, upstream_frames) that were received. Raises: AssertionError: If the received frames don't match the expected frame types. )r;r<r:F)cancel_on_idle_timeoutrJrMparamscPKtjdd{D]S}t|tr(tj|jd{;j |d{Ur"j t d{yy77G7/7 w)Ng{Gz?)rGrr3r queue_framer )r2rQrOtasks r" push_frameszrun_test..push_framessmmD!!!# .E%,mmEKK000&&u---  . ""8:. . .  "1- /sEB&B;B&B B&2B"3$B&B$B& B&"B&$B&N[z, ]zreceived DOWN frames =zexpected DOWN frames =zreceived UP frames =zexpected UP frames =)rrGrHr9rUPSTREAM DOWNSTREAMrrrgatherrunemptygetr3r appendr.rprintlenzip)rPrJrKrLrQr:rMrNrO received_up received_downr1sinkpipelinerYrunnerreceived_down_framesr2down_frames_printedexpected_frames_printedrealexpectedreceived_up_framesrXs ` ` @r"run_testrq{sFRI%9)9O--/KMMOM !&//!F &11! D D12H $  D / F ..D);= 999 )+!!##''))%*. ' ' .!!# '!) CE eoo&>&>%?r#B B  Cs""%) =E #%..)9'< < # =3& &(;< &(?@'(C0D,EEEE!"68LM .ND(dH- -- . ')!!oo''!!%(!% $&89 $&89%&#.@*AAAA!"46HI .ND(dH- -- . !"4 55[:*4(sPCJI=)JJ7J9BJ(J=J>%J$AJ6JJJ)$rrG dataclassesrtypingrrrrrr pipecat.frames.framesr r r r rpipecat.observers.base_observerrrpipecat.pipeline.pipelinerpipecat.pipeline.runnerrpipecat.pipeline.taskrr"pipecat.processors.frame_processorrrrr$r9rItyperqr r!r"r{s'9!GGF.2>M     $6$6N'0>'0Z5937.204w6w6w6#8D>2 w6 !$0 w6 UO w6w6\*+w6n-w6w6 8E?HUO +,w6r!