qi bdZddlZddlmZmZmZddlmZddlm Z m Z defdZ Gdd e Z y) z8Producer processor for frame filtering and distribution.N) AwaitableCallableList)Frame)FrameDirectionFrameProcessorframecK|Sw)zDefault transformer that returns the frame unchanged. Args: frame: The frame to transform. Returns: The same frame without modifications. )r s W/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/processors/producer_processor.pyidentity_transformerr s  LsceZdZdZedddeegeefdeegeefdeffdZ dZ d ed e ffd Z d efd Z xZS) ProducerProcessora|A processor that filters frames and distributes them to multiple consumers. This processor receives frames, applies a filter to determine which frames should be sent to consumers (ConsumerProcessor), optionally transforms those frames, and distributes them to registered consumer queues. It can also pass frames through to the next processor in the pipeline. T) transformer passthroughfilterrrcZt|||_||_||_g|_y)a2Initialize the producer processor. Args: filter: Async function that determines if a frame should be produced. Must return True for frames to be sent to consumers. transformer: Async function to transform frames before sending to consumers. Defaults to identity_transformer which returns frames unchanged. passthrough: Whether to pass frames through to the next processor. If True, all frames continue downstream regardless of filter result. N)super__init___filter _transformer _passthrough _consumers)selfrrr __class__s r rzProducerProcessor.__init__%s."  ''/1cdtj}|jj||S)zAdd a new consumer and return its associated queue. Returns: asyncio.Queue: The queue for the newly added consumer. )asyncioQueuerappend)rqueues r add_consumerzProducerProcessor.add_consumer<s&   u% rr directionc:Kt|||d{|j|d{rA|j|d{|jr|j ||d{yy|j ||d{y7y7b7K7'7 w)a}Process an incoming frame and determine whether to produce it. If the frame meets the filter criteria, it will be transformed and added to all consumer queues. If passthrough is enabled, the original frame will also be sent downstream. Args: frame: The frame to process. direction: The direction of the frame flow. N)r process_framer_producer push_frame)rr r#rs r r%zProducerProcessor.process_frameFsg#E9555e$ $ $--& & &  ooeY777!//%3 3 3 6 $ &7 3sUBBBBB B %B/B0B B BBBBBcK|jD]4}|j|d{}|j|d{6y7 7 w)z!Produce a frame to all consumers.N)rrput)rr consumer new_frames r r&zProducerProcessor._produceZsE *H"//66I,,y) ) ) *6 )s $A AA A A  A )__name__ __module__ __qualname____doc__r rrrboolrr"rr%r& __classcell__)rs@r rrsy4(*E*rr) r/rtypingrrrpipecat.frames.framesr"pipecat.processors.frame_processorrrr rr rr r5s3?,,'M e B*B*r