qi ZdZddlmZmZddlmZddlmZmZddl m Z m Z Gdde Z y) zGated frame aggregator for conditional frame accumulation. This module provides a gated aggregator that accumulates frames based on custom gate open/close functions, allowing for conditional frame buffering and release in frame processing pipelines. )ListTuple)logger)Frame SystemFrame)FrameDirectionFrameProcessorcTeZdZdZej fdeffd Zdedeffd ZxZ S)GatedAggregatorahAccumulate frames, with custom functions to start and stop accumulation. Yields gate-opening frame before any accumulated frames, then ensuing frames until and not including the gate-closed frame. The aggregator maintains an internal gate state that controls whether frames are passed through immediately or accumulated for later release. directioncht|||_||_||_||_g|_y)a{Initialize the gated aggregator. Args: gate_open_fn: Function that returns True when a frame should open the gate. gate_close_fn: Function that returns True when a frame should close the gate. start_open: Whether the gate should start in the open state. direction: The frame direction this aggregator operates on. N)super__init__ _gate_open_fn_gate_close_fn _gate_open _direction _accumulator)self gate_open_fn gate_close_fn start_openr __class__s V/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/processors/aggregators/gated.pyrzGatedAggregator.__init__s6 )+$#@BframecKt|||d{t|tr|j ||d{y||j k7r|j ||d{y|j }|j r|j| |_n|j||_||j k7r+|j rdnd}tjd|d||j rP|j ||d{|jD]\}}|j ||d{!g|_ y|jj||fy7T7-77a77w)zProcess incoming frames with gated accumulation logic. Args: frame: The frame to process. direction: The direction of the frame flow. Nopenclosedz Gate is now z because of ) r process_frame isinstancer push_framerrrrrdebugrappend)rrr old_statestatefdrs rr zGatedAggregator.process_frame5sSg#E9555 e[ )//%3 3 3   '//%3 3 3 OO ??"&"5"5e"<E6?-E8.E81E84E86E8) __name__ __module__ __qualname____doc__r DOWNSTREAMrrr __classcell__)rs@rr r s<%3$=$= C " C,#9#9>#9#9rr N) r,typingrrlogururpipecat.frames.framesrr"pipecat.processors.frame_processorrr r rrr4s&4MB9nB9r