o i @sdZddlZddlZddlZddlmZddlmZmZm Z m Z m Z ddl Z ddlmZddlmZddlmZmZmZmZmZmZmZmZmZmZmZmZmZm Z ddl!m"Z"dd l#m$Z$dd l%m&Z&dd l'm(Z(m)Z)dd l*m+Z+zddl,Z,dd l-m.Z.ddl/m0Z0m1Z1ddl2m3Z3m4Z4m5Z5Wn e6yZ7ze8de7e8de9de7dZ7[7wwdZ:dZ;dZGddde.Z?GdddZ@Gddde$ZAGd d!d!e&ZBGd"d#d#e(ZCdS)$a Small WebRTC transport implementation for Pipecat. This module provides a WebRTC transport implementation using aiortc for real-time audio and video communication. It supports bidirectional media streaming, application messaging, and client connection management. N)deque)Any AwaitableCallableListOptional)logger) BaseModel) CancelFrameClientConnectedFrameEndFrameFrameInputAudioRawFrameInputTransportMessageFrameOutputAudioRawFrameOutputImageRawFrameOutputTransportMessageFrame!OutputTransportMessageUrgentFrame SpriteFrame StartFrameUserImageRawFrameUserImageRequestFrame)FrameDirection)BaseInputTransport)BaseOutputTransport) BaseTransportTransportParams)SmallWebRTCConnection)VideoStreamTrack)AudioStreamTrackMediaStreamError) AudioFrameAudioResampler VideoFramez Exception: zNIn order to use the SmallWebRTC, you need to `pip install pipecat-ai[webrtc]`.zMissing module: camera screenVideo microphonec@sVeZdZUdZeeegedfed<ee gedfed<ee gedfed<dS)SmallWebRTCCallbacksaCallback handlers for SmallWebRTC events. Parameters: on_app_message: Called when an application message is received. on_client_connected: Called when a client establishes connection. on_client_disconnected: Called when a client disconnects. Non_app_messageon_client_connectedon_client_disconnected) __name__ __module__ __qualname____doc__rrstrr__annotations__rr1r1\/home/ubuntu/.local/lib/python3.10/site-packages/pipecat/transports/smallwebrtc/transport.pyr'=s r'cs6eZdZdZfddZdefddZddZZS) RawAudioTrackzCustom audio stream track for WebRTC output. Handles audio frame generation and timing for WebRTC transmission, supporting queued audio data with proper synchronization. csFt||_|dd|_|jd|_d|_t|_t|_ dS)zmInitialize the raw audio track. Args: sample_rate: The audio sample rate in Hz. irN) super__init__ _sample_rate_samples_per_10ms_bytes_per_10ms _timestamptime_startr _chunk_queue)self sample_rate __class__r1r2r7Rs    zRawAudioTrack.__init__ audio_bytescCs~t||jdkr tdt}tdt||jD] }||||j}||jt|kr2|nd}|j||fq|S)a5Add audio bytes to the buffer for transmission. Args: audio_bytes: Raw audio data to queue for transmission. Returns: A Future that completes when the data is processed. Raises: ValueError: If audio bytes are not a multiple of 10ms size. rz,Audio bytes must be a multiple of 10ms size.N) lenr: ValueErrorasyncioget_running_loop create_futureranger>append)r?rCfutureichunkfutr1r1r2add_audio_bytesas  zRawAudioTrack.add_audio_bytescs|jdkr|j|j|jt}|dkrt|IdH|jr5|j\}}|r4|s4| dnt |j }t j |t jd}tj|dddfdd}|j|_|j|_td|j|_|j|j7_|S)zReturn the next audio frame for WebRTC transmission. Returns: An AudioFrame containing the next audio data or silence. rNTdtypemono)layout)r;r=r8r<rFsleepr>popleftdone set_resultbytesr:np frombufferint16r! from_ndarrayr@pts fractionsFraction time_baser9)r?waitrMrKsamplesframer1r1r2recvzs$    zRawAudioTrack.recv) r+r,r-r.r7rYrOre __classcell__r1r1rAr2r3Ks  r3cs0eZdZdZfddZddZddZZS) RawVideoTrackzzCustom video stream track for WebRTC output. Handles video frame queuing and conversion for WebRTC transmission. cs$t||_||_t|_dS)zInitialize the raw video track. Args: width: Video frame width in pixels. height: Video frame height in pixels. N)r6r7_width_heightrFQueue _video_buffer)r?widthheightrAr1r2r7s zRawVideoTrack.__init__cCs|j|dS)zAdd a video frame to the transmission buffer. Args: frame: The video frame to queue for transmission. N)rk put_nowaitr?rdr1r1r2add_video_frameszRawVideoTrack.add_video_framecs\|jIdH}tj|jtjd|j|jdf}t j |dd}| IdH\|_ |_ |S)zReturn the next video frame for WebRTC transmission. Returns: A VideoFrame ready for WebRTC transmission. NrPrgb24format)rkgetrZr[imageuint8reshaperirhr#r]next_timestampr^ra)r? raw_frame frame_datardr1r1r2res zRawVideoTrack.recv)r+r,r-r.r7rprerfr1r1rAr2rgs   rgc@seZdZdZejejejejdZde de fddZ de j ded e j fd d Zd efd dZddZded efddZded efddZdefddZddZddZdeeBfddZdd Zd!d"Zd#d$Z d%e!d&efd'd(Z"d)d*Z#e$d efd+d,Z%e$d efd-d.Z&d/S)0SmallWebRTCClientzWebRTC client implementation for handling connections and media streams. Manages WebRTC peer connections, audio/video streaming, and application messaging through the SmallWebRTCConnection interface. )yuv420pyuvj420pnv12graywebrtc_connection callbackscs|_d_|_d_d_d_d_d_d_d_ d_ d_ d_ d_ jddtffdd }jddtffd d }jd dtffd d }jddtdtffdd }dS)zInitialize the WebRTC client. Args: webrtc_connection: The underlying WebRTC connection handler. callbacks: Event callbacks for connection and message handling. FNr connected connectionctdIdHdS)NzPeer connection established.)rdebug_handle_client_connectedrr?r1r2 on_connected z0SmallWebRTCClient.__init__..on_connected disconnectedcr)NzPeer connection lost.)rr_handle_peer_disconnectedrrr1r2on_disconnectedrz3SmallWebRTCClient.__init__..on_disconnectedclosedcr)NzClient connection closed.)rr_handle_client_closedrrr1r2 on_closedrz-SmallWebRTCClient.__init__..on_closedz app-messagemessagecs||jIdHdS)N)_handle_app_messagepc_id)rrrr1r2r(z2SmallWebRTCClient.__init__..on_app_message)_webrtc_connection_closing _callbacks_audio_output_track_video_output_track_audio_input_track_video_input_track_screen_video_track_params_audio_in_channels_in_sample_rate_out_sample_rate_leave_counter_audio_in_resampler event_handlerrr)r?rrrrrr(r1rr2r7s,    zSmallWebRTCClient.__init__ frame_array format_namereturncCs<|dr|Stj|}|durtd|t||S)aXConvert a video frame to RGB format based on the input format. Args: frame_array: The input frame as a NumPy array. format_name: The format of the input frame. Returns: The converted RGB frame as a NumPy array. Raises: ValueError: If the format is unsupported. rgbNzUnsupported format: ) startswithr|FORMAT_CONVERSIONSrurEcv2cvtColor)r?rrconversion_coder1r1r2_convert_frames   z SmallWebRTCClient._convert_frame video_sourcec Cs, |tkr |jn|j}|durtdIdHqztj|ddIdH}Wn+tjyC|j r?|r?| r?t dd}Ynt yRt dd}Ynw|dus\t|tsetdIdHq|jj}|j|d}|||}~|}~t|jj||j|jfd d }||_|j|_~~|Vq) amRead video frames from the WebRTC connection. Reads a video frame from the given MediaStreamTrack, converts it to RGB, and creates an InputImageRawFrame. Args: video_source: Video source to capture ("camera" or "screenVideo"). Yields: UserImageRawFrame objects containing video data from the peer. TN{Gz?@timeoutz;Timeout: No video frame received within the specified time.zBReceived an unexpected media stream error while reading the video.rsRGB)user_idrvsizert)CAM_VIDEO_SOURCErrrFrUwait_forre TimeoutErrorr is_connected is_enabledrwarningr isinstancer#rtname to_ndarrayrtobytesrrrlrmtransport_sourcer^) r?r video_trackrdrr frame_rgb image_bytes image_framer1r1r2read_video_framesZ       z"SmallWebRTCClient.read_video_framecCs& |jdurtdIdHqztj|jddIdH}Wn-tjy=|jr9|jr9|jr9t dd}Ynt yLt dd}Ynw|dusVt |t s_tdIdHq|j|jkrk|j|n|g}|D] }|tj}|}~t||j|jd}|j|_~|Vqp~q) zRead audio frames from the WebRTC connection. Reads 20ms of audio from the given MediaStreamTrack and creates an InputAudioRawFrame. Yields: InputAudioRawFrame objects containing audio data from the peer. TNrrrz;Timeout: No audio frame received within the specified time.zBReceived an unexpected media stream error while reading the audio.)audior@ num_channels)rrFrUrrerrrrrrr rr!r@rrresamplerastyperZr\rrrr^)r?rdframes_to_processprocessed_frame pcm_array pcm_bytes audio_framer1r1r2read_audio_frameZsR      z"SmallWebRTCClient.read_audio_framerdcs,|r|jr|j|jIdHdSdS)zWrite an audio frame to the WebRTC connection. Args: frame: The audio frame to transmit. Returns: True if the audio frame was written successfully, False otherwise. NTF) _can_sendrrOrror1r1r2write_audio_frames  z#SmallWebRTCClient.write_audio_framecs$|r|jr|j|dSdS)zWrite a video frame to the WebRTC connection. Args: frame: The video frame to transmit. Returns: True if the video frame was written successfully, False otherwise. TF)rrrpror1r1r2write_video_frames  z#SmallWebRTCClient.write_video_framercsN|j|_|jp |j|_|jp|j|_||_|jd7_tdd|j|_ dS)zSet up the client with transport parameters. Args: _params: Transport configuration parameters. frame: The initialization frame containing setup data. rTs16rRN) audio_in_channelsraudio_in_sample_rateraudio_out_sample_raterrrr"r)r?rrdr1r1r2setupszSmallWebRTCClient.setupcs.|jrdStd|jIdHdS)z Establish the WebRTC connection.NzConnecting to Small WebRTC)rrrinfoconnectrr1r1r2rs   zSmallWebRTCClient.connectcsd|jd8_|jdkrdS|jr.|js0tdd|_|jIdH|IdHdSdSdS)z Disconnect from the WebRTC peer.rTrNzDisconnecting to Small WebRTCT) rr is_closingrrrr disconnectrrr1r1r2rs   zSmallWebRTCClient.disconnectcs |r|j|jdSdS)zSend an application message through the WebRTC connection. Args: frame: The message frame to send. N)rrsend_app_messagerror1r1r2 send_messageszSmallWebRTCClient.send_messagecs|jsdS|j|_|j|_|j|_|jjr*t |j d|_ |j |j |jj r@t|jj|jjd|_|j|j|j|jIdHdS)z'Handle client connection establishment.N)r@)rlrm)rraudio_input_trackrvideo_input_trackrscreen_video_input_trackraudio_out_enabledr3rrreplace_audio_trackvideo_out_enabledrgvideo_out_widthvideo_out_heightrreplace_video_trackrr)rr1r1r2rs    z*SmallWebRTCClient._handle_client_connectedcs$d|_d|_d|_d|_d|_dS)z"Handle peer disconnection cleanup.N)rrrrrrr1r1r2rs  z+SmallWebRTCClient._handle_peer_disconnectedcsBd|_d|_d|_d|_d|_|js|j|jIdHdSdS)z!Handle client connection closure.N) rrrrrrrr*rrr1r1r2rsz'SmallWebRTCClient._handle_client_closedrsendercs|j||IdHdS)%Handle incoming application messages.N)rr(r?rrr1r1r2rrz%SmallWebRTCClient._handle_app_messagecCs|jo|j S)z2Check if the connection is ready for sending data.)rrrr1r1r2r szSmallWebRTCClient._can_sendcCs |jS)ztCheck if the WebRTC connection is established. Returns: True if connected to the peer. )rrrr1r1r2rs zSmallWebRTCClient.is_connectedcCs|jS)zCheck if the connection is in the process of closing. Returns: True if the connection is closing. )rrr1r1r2rszSmallWebRTCClient.is_closingN)'r+r,r-r.rCOLOR_YUV2RGB_I420COLOR_YUV2RGB_NV12COLOR_GRAY2RGBrrr'r7rZndarrayr/rrrrboolrrrrrrrrrrrrrrrrpropertyrrr1r1r1r2r|s8-@9     r|cseZdZdZdedeffdd Zdedeffdd Z de ffd d Z d d Z de ffdd Zdeffdd ZddZdefddZdefddZdefddZefdefddZZS)SmallWebRTCInputTransportzInput transport implementation for SmallWebRTC. Handles incoming audio and video streams from WebRTC peers, including user image requests and application message handling. clientparamsc sBtj|fi|||_||_d|_d|_d|_g|_d|_dS)zInitialize the WebRTC input transport. Args: client: The WebRTC client instance. params: Transport configuration parameters. **kwargs: Additional arguments passed to parent class. NF) r6r7_clientr_receive_audio_task_receive_video_task_receive_screen_video_task_image_requests _initializedr?rrkwargsrAr1r2r7)s  z"SmallWebRTCInputTransport.__init__rd directioncs8t||IdHt|tr||IdHdSdS)zProcess incoming frames including user image requests. Args: frame: The frame to process. direction: The direction of frame flow in the pipeline. N)r6 process_framerrrequest_participant_image)r?rdrrAr1r2rAs  z'SmallWebRTCInputTransport.process_framecst|IdH|jrdSd|_|j|j|IdH|jIdH||IdH|js<|jj r<| | |_|j sN|jj rP| |t|_ dSdSdS)zStart the input transport and establish WebRTC connection. Args: frame: The start frame containing initialization parameters. NT)r6startrrrrrset_transport_readyraudio_in_enabled create_task_receive_audiorvideo_in_enabled_receive_videorrorAr1r2rMszSmallWebRTCInputTransport.startcsF|jr||jIdHd|_|jr!||jIdHd|_dSdS)zStop all background tasks.N)r cancel_taskrrr1r1r2 _stop_tasksbs z%SmallWebRTCInputTransport._stop_tasksc6t|IdH|IdH|jIdHdS)zStop the input transport and disconnect from WebRTC. Args: frame: The end frame signaling transport shutdown. N)r6stoprrrrorAr1r2r kzSmallWebRTCInputTransport.stopcr)zCancel the input transport and disconnect immediately. Args: frame: The cancel frame signaling immediate cancellation. N)r6cancelrrrrorAr1r2r ur z SmallWebRTCInputTransport.cancelc sz|j}|2z3dHW}|r||IdHq 6WdSty@}zt|d|jjd|dWYd}~dSd}~ww)z7Background task for receiving audio frames from WebRTC.N exception receiving data:  ())rrpush_audio_frame ExceptionrerrorrBr+)r?audio_iteratorrer1r1r2rs .z(SmallWebRTCInputTransport._receive_audiorc szZ|j|}|2zM3dHW}|rW||IdH|jddD]5}|r(|jnd}|r/|jnd}|j|krVt|j|j |j |j |||d}||_ ||IdH|j |q!q 6WdSty~}zt|d|jjd|dWYd}~dSd}~ww)zBackground task for receiving video frames from WebRTC. Args: video_source: Video source to capture ("camera" or "screenVideo"). N)rrvrrttextappend_to_contextrequestr r r)rrpush_video_framerrrrrrrvrrtrremoverrrrBr+) r?rvideo_iterator video_frame request_frame request_textadd_to_contextrrr1r1r2rs8    .z(SmallWebRTCInputTransport._receive_videorcs*td||jt|dIdHdS)zPush an application message into the pipeline. Args: message: The application message to process. z7Received app message inside SmallWebRTCInputTransport )rN)rrbroadcast_framer)r?rr1r1r2push_app_messagesz*SmallWebRTCInputTransport.push_app_messagecstd|j|j||jdurt|_|jtkr/|js/|jj r/| | t|_dS|jt krF|j sH|jj rJ| | t |_ dSdSdSdS)a.Request an image frame from the participant's video stream. When a UserImageRequestFrame is received, this method will store the request and the next video frame received will be converted to a UserImageRawFrame. Args: frame: The user image request frame. z#Requesting image from participant: N)rrrrrJrrrrrrrSCREEN_VIDEO_SOURCError1r1r2rs(     z3SmallWebRTCInputTransport.request_participant_imagesourcecs|tkr|js|jjr|||_dS|tkr,|js,|jjr,|| t|_dS|t krB|j sD|jjrF|| t |_ dSdSdSdS)zCapture media from a specific participant. Args: source: Media source to capture from. ("camera", "microphone", or "screenVideo") N) MIC_AUDIO_SOURCErrrrrrrrrr r)r?r!r1r1r2capture_participant_medias,  z3SmallWebRTCInputTransport.capture_participant_media)r+r,r-r.r|rr7r rrrrrr r r r rr/rrrrrrr#rfr1r1rAr2r"s(     & %rcseZdZdZdedeffdd Zdeffdd Zde ffd d Z de ffd d Z de eBfd dZdedefddZdedefddZZS)SmallWebRTCOutputTransportzOutput transport implementation for SmallWebRTC. Handles outgoing audio and video streams to WebRTC peers, including transport message sending. rrc s*tj|fi|||_||_d|_dS)zInitialize the WebRTC output transport. Args: client: The WebRTC client instance. params: Transport configuration parameters. **kwargs: Additional arguments passed to parent class. FN)r6r7rrrrrAr1r2r7s  z#SmallWebRTCOutputTransport.__init__rdcs^t|IdH|jrdSd|_|j|j|IdH|jIdH||IdHdS)zStart the output transport and establish WebRTC connection. Args: frame: The start frame containing initialization parameters. NT)r6rrrrrrrrorAr1r2rsz SmallWebRTCOutputTransport.startc(t|IdH|jIdHdS)zStop the output transport and disconnect from WebRTC. Args: frame: The end frame signaling transport shutdown. N)r6r rrrorAr1r2r ,zSmallWebRTCOutputTransport.stopcr%)zCancel the output transport and disconnect immediately. Args: frame: The cancel frame signaling immediate cancellation. N)r6r rrrorAr1r2r 5r&z!SmallWebRTCOutputTransport.cancelcs|j|IdHdS)zSend a transport message through the WebRTC connection. Args: frame: The transport message frame to send. N)rrror1r1r2r>sz'SmallWebRTCOutputTransport.send_messagerc|j|IdHS)zWrite an audio frame to the WebRTC connection. Args: frame: The output audio frame to transmit. Returns: True if the audio frame was written successfully, False otherwise. N)rrror1r1r2rH z,SmallWebRTCOutputTransport.write_audio_framecr')zWrite a video frame to the WebRTC connection. Args: frame: The output video frame to transmit. Returns: True if the video frame was written successfully, False otherwise. N)rrror1r1r2rSr(z,SmallWebRTCOutputTransport.write_video_frame)r+r,r-r.r|rr7rrr r r r rrrrrrrrrfr1r1rAr2r$s     r$c seZdZdZ  d!dededeedeeffdd Zd e fd d Z d e fd d Z de eBfddZdefddZdedefddZddZddZefdefddZefdefdd ZZS)"SmallWebRTCTransportaWebRTC transport implementation for real-time communication. Provides bidirectional audio and video streaming over WebRTC connections with support for application messaging and connection event handling. Event handlers available: - on_client_connected(transport, client): Client connected to WebRTC session - on_client_disconnected(transport, client): Client disconnected from WebRTC session - on_client_message(transport, message, client): Received a data channel message Example:: @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): ... Nrr input_name output_namecshtj||d||_t|j|j|jd|_t||j|_ d|_ d|_ | d| d| ddS)a9Initialize the WebRTC transport. Args: webrtc_connection: The underlying WebRTC connection handler. params: Transport configuration parameters. input_name: Optional name for the input processor. output_name: Optional name for the output processor. )r*r+)r(r)r*Nr(r)r*) r6r7rr'_on_app_message_on_client_connected_on_client_disconnectedrr|r_input_output_register_event_handler)r?rrr*r+rAr1r2r7rs  zSmallWebRTCTransport.__init__rcC"|jst|j|j|jd|_|jS)zGet the input transport processor. Returns: The input transport for handling incoming media streams. r)r/rrr _input_namerr1r1r2input  zSmallWebRTCTransport.inputcCr2)zGet the output transport processor. Returns: The output transport for handling outgoing media streams. r3)r0r$rrr4rr1r1r2outputr6zSmallWebRTCTransport.outputrdc&|jr|j|tjIdHdSdS)znSend an image frame through the transport. Args: frame: The image frame to send. Nr0 queue_framer DOWNSTREAMror1r1r2 send_imagezSmallWebRTCTransport.send_imagecr8)znSend an audio frame through the transport. Args: frame: The audio frame to send. Nr9ror1r1r2 send_audior=zSmallWebRTCTransport.send_audiorrcs2|jr |j|IdH|d||IdHdS)rNr()r/r_call_event_handlerrr1r1r2r,sz$SmallWebRTCTransport._on_app_messagecs6|d|IdH|jr|jtIdHdSdS)z Handle client connection events.r)N)r?r/ push_framer r?rr1r1r2r-s z)SmallWebRTCTransport._on_client_connectedcs|d|IdHdS)z#Handle client disconnection events.r*N)r?rAr1r1r2r.sz,SmallWebRTCTransport._on_client_disconnectedrc$|jr|jj|dIdHdSdS)zCapture video from a specific participant. Args: video_source: Video source to capture from ("camera" or "screenVideo"). r!Nr/r#)r?rr1r1r2capture_participant_video z.SmallWebRTCTransport.capture_participant_video audio_sourcecrB)zCapture audio from a specific participant. Args: audio_source: Audio source to capture from. (currently, "microphone" is the only supported option) rCNrD)r?rGr1r1r2capture_participant_audiorFz.SmallWebRTCTransport.capture_participant_audio)NN)r+r,r-r.rrrr/r7rr5r$r7rrr<rr>rr,r-r.rrEr"rHrfr1r1rAr2r)_s8#     r))Dr.rFr_r< collectionsrtypingrrrrrnumpyrZlogururpydanticr pipecat.frames.framesr r r r rrrrrrrrrr"pipecat.processors.frame_processorrpipecat.transports.base_inputrpipecat.transports.base_outputr!pipecat.transports.base_transportrr)pipecat.transports.smallwebrtc.connectionrraiortcraiortc.mediastreamsrr avr!r"r#ModuleNotFoundErrorrrrrr r"r'r3rgr|rr$r)r1r1r1r2sN   @      N/\__