o i @sldZddlZddlmZmZmZddlmZmZm Z m Z ddl m Z m Z ddlmZmZGddde ZdS) zFConsumer processor for consuming frames from ProducerProcessor queues.N) AwaitableCallableOptional) CancelFrameEndFrameFrame StartFrame)FrameDirectionFrameProcessor)ProducerProcessoridentity_transformercseZdZdZeejddedee ge e fdeffddZ de deffd d Z d e fd d Zd efddZd efddZddZZS)ConsumerProcessora=Frame processor that consumes frames from a ProducerProcessor's queue. This processor passes through frames normally while also consuming frames from a ProducerProcessor's queue. When frames are received from the producer queue, they are optionally transformed and pushed in the specified direction. ) transformer directionproducerrrc s.tjdi|||_||_||_d|_dS)a|Initialize the consumer processor. Args: producer: The producer processor to consume frames from. transformer: Function to transform frames before pushing. Defaults to identity_transformer. direction: Direction to push consumed frames. Defaults to DOWNSTREAM. **kwargs: Additional arguments passed to parent class. N)super__init__ _transformer _direction _producer_consumer_task)selfrrrkwargs __class__rY/home/ubuntu/.local/lib/python3.10/site-packages/pipecat/processors/consumer_processor.pyrs  zConsumerProcessor.__init__framecs~t||IdHt|tr||IdHnt|tr'||IdHn t|tr4||IdH| ||IdHdS)zProcess incoming frames and handle lifecycle events. Args: frame: The frame to process. direction: The direction the frame is traveling. N) r process_frame isinstancer_startr_stopr_cancel push_frame)rrrrrrr/s   zConsumerProcessor.process_frame_cs,|js|j|_|||_dSdS)z7Start the consumer task and register with the producer.N)rr add_consumer_queue create_task_consumer_task_handlerrr$rrrr As  zConsumerProcessor._startc"|jr||jIdHdSdS)zStop the consumer task.Nr cancel_taskr)rrrr!GzConsumerProcessor._stopcr*)zCancel the consumer task.Nr+r)rrrr"Lr-zConsumerProcessor._cancelcs: |jIdH}||IdH}|||jIdHq)z0Handle consuming frames from the producer queue.TN)r&getr queue_framer)rr new_framerrrr(Qs z(ConsumerProcessor._consumer_task_handler)__name__ __module__ __qualname____doc__r r DOWNSTREAMr rrrrrrr rr!rr"r( __classcell__rrrrr s  r )r4asynciotypingrrrpipecat.frames.framesrrrr"pipecat.processors.frame_processorr r %pipecat.processors.producer_processorr r r rrrrs