o i*H@sJdZddlZddlZddlZddlZddlmZmZmZddl Z ddl m Z ddl m Z ddlmZddlmZmZmZmZmZmZmZmZmZddlmZdd lmZdd lm Z dd l!m"Z"dd l#m$Z$dd l%m&Z&m'Z'ddl(m)Z)Gddde'Z*Gddde Z+GdddZ,Gddde"Z-Gddde$Z.Gddde&Z/dS)aWebSocket client transport implementation for Pipecat. This module provides a WebSocket client transport that enables bidirectional communication over WebSocket connections, with support for audio streaming, frame serialization, and connection management. N) AwaitableCallableOptional)logger) BaseModel)connect) CancelFrameEndFrameFrameInputAudioRawFrameInputTransportMessageFrameOutputAudioRawFrameOutputTransportMessageFrame!OutputTransportMessageUrgentFrame StartFrame)FrameProcessorSetup)FrameSerializer)ProtobufFrameSerializer)BaseInputTransport)BaseOutputTransport) BaseTransportTransportParams)BaseTaskManagerc@sFeZdZUdZdZeed<dZee e e fed<dZ ee ed<dS)WebsocketClientParamszConfiguration parameters for WebSocket client transport. Parameters: add_wav_header: Whether to add WAV headers to audio frames. serializer: Frame serializer for encoding/decoding messages. Tadd_wav_headerNadditional_headers serializer) __name__ __module__ __qualname____doc__rbool__annotations__rrdictstrrrr%r%W/home/ubuntu/.local/lib/python3.10/site-packages/pipecat/transports/websocket/client.pyr-s  rc@s^eZdZUdZeejgedfed<eejgedfed<eejej gedfed<dS)WebsocketClientCallbacksaCallback functions for WebSocket client events. Parameters: on_connected: Called when WebSocket connection is established. on_disconnected: Called when WebSocket connection is closed. on_message: Called when a message is received from the WebSocket. N on_connectedon_disconnected on_message) rrrr r websocketsWebSocketClientProtocolrr"Datar%r%r%r&r':s  r'c@seZdZdZdedededefddZede fd d Z d e fd d Z ddZ ddZ dejdefddZedefddZedefddZddZddZdS)WebsocketClientSessionzManages a WebSocket client connection session. Handles connection lifecycle, message sending/receiving, and provides callback mechanisms for connection events. uriparams callbackstransport_namecCs.||_||_||_||_d|_d|_d|_dS)a4Initialize the WebSocket client session. Args: uri: The WebSocket URI to connect to. params: Configuration parameters for the session. callbacks: Callback functions for session events. transport_name: Name of the parent transport for logging. rN)_uri_params _callbacks_transport_name_leave_counter _task_manager _websocket)selfr/r0r1r2r%r%r&__init__Os zWebsocketClientSession.__init__returncCs|js t|jd|jS)zGet the task manager for this session. Returns: The task manager instance. Raises: Exception: If task manager is not initialized. zM::WebsocketClientSession: TaskManager not initialized (pipeline not started?))r8 Exceptionr6r:r%r%r& task_managergs   z#WebsocketClientSession.task_managerr?cs$|jd7_|js||_dSdS)zSet up the session with a task manager. Args: task_manager: The task manager to use for session tasks. N)r7r8)r:r?r%r%r&setupws  zWebsocketClientSession.setupcs|jrdSz)t|jd|jjdIdH|_|j||jd|_ |j |jIdHWdSt yBt d|jYdSw)z Connect to the WebSocket server.N )r/ open_timeoutrz.::WebsocketClientSession::_client_task_handlerzTimeout connecting to )r9websocket_connectr3r4rr? create_task_client_task_handlerr6 _client_taskr5r( TimeoutErrorrerrorr>r%r%r&rs"  zWebsocketClientSession.connectcsR|jd8_|jr|jdkrdS|j|jIdH|jIdHd|_dS)z%Disconnect from the WebSocket server.r@rN)r7r9r? cancel_taskrGcloser>r%r%r& disconnects z!WebsocketClientSession.disconnectmessagec sd}z;z|jr|j|IdHd}WW|SWW|Sty>}zt|d|jjd|dWYd}~W|Sd}~ww|YS)zwSend a message through the WebSocket connection. Args: message: The message data to send. FNTz exception sending data:  ())r9sendr=rrI __class__r)r:rMresulter%r%r&rPs,zWebsocketClientSession.sendcC|jr |jjtjjkSdS)zCheck if the WebSocket is currently connected. Returns: True if the WebSocket is in connected state. F)r9stater+StateOPENr>r%r%r& is_connectedz#WebsocketClientSession.is_connectedcCrT)zCheck if the WebSocket is currently closing. Returns: True if the WebSocket is in the process of closing. F)r9rUr+rVCLOSINGr>r%r%r& is_closingrYz!WebsocketClientSession.is_closingc sz|j2z3dHW}|j|j|IdHq6Wn"ty;}zt|d|jjd|dWYd}~nd}~ww|j|jIdHdS)z7Handle incoming messages from the WebSocket connection.Nz exception receiving data: rNrO) r9r5r*r=rrIrQrr))r:rMrSr%r%r&rFs,z+WebsocketClientSession._client_task_handlercCs |jdS)z6String representation of the WebSocket client session.z::WebsocketClientSession)r6r>r%r%r&__str__s zWebsocketClientSession.__str__N)rrrr r$rr'r;propertyrr?rArrLr+r-r!rPrXr[rFr\r%r%r%r&r.Hs.    r.cseZdZdZdededeffdd Zdeffdd Z d e ffd d Z d e ffd d Z d effdd ZfddZddZZS)WebsocketClientInputTransportzWebSocket client input transport for receiving frames. Handles incoming WebSocket messages, deserializes them to frames, and pushes them downstream in the processing pipeline. transportsessionr0cs(t|||_||_||_d|_dS)aInitialize the WebSocket client input transport. Args: transport: The parent transport instance. session: The WebSocket session to use for communication. params: Configuration parameters for the transport. FN)superr; _transport_sessionr4 _initializedr:r_r`r0rQr%r&r;s  z&WebsocketClientInputTransport.__init__rAc,t|IdH|j|jIdHdS)zSet up the input transport with the frame processor setup. Args: setup: The frame processor setup configuration. NrarArcr?r:rArfr%r&rAz#WebsocketClientInputTransport.setupframecsdt|IdH|jrdSd|_|jjr |jj|IdH|jIdH||IdHdS)zStart the input transport and initialize the WebSocket connection. Args: frame: The start frame containing initialization parameters. NT) rastartrdr4rrArcrset_transport_readyr:rkrfr%r&rlsz#WebsocketClientInputTransport.startc(t|IdH|jIdHdS)zStop the input transport and disconnect from WebSocket. Args: frame: The end frame signaling transport shutdown. NrastoprcrLrnrfr%r&rq z"WebsocketClientInputTransport.stopcro)zCancel the input transport and disconnect from WebSocket. Args: frame: The cancel frame signaling immediate cancellation. NracancelrcrLrnrfr%r&rtrrz$WebsocketClientInputTransport.cancelc&tIdH|jIdHdS)z'Clean up the input transport resources.Nracleanuprbr>rfr%r&rwz%WebsocketClientInputTransport.cleanupcs|jjsdS|jj|IdH}|sdSt|tr(|jjr(||IdHdSt|tr:|jt|j dIdHdS| |IdHdS)zHandle incoming WebSocket messages. Args: websocket: The WebSocket connection that received the message. message: The received message data. N)rM) r4r deserialize isinstancer audio_in_enabledpush_audio_framer broadcast_framerM push_frame)r: websocketrMrkr%r%r&r* s z(WebsocketClientInputTransport.on_message)rrrr rr.rr;rrArrlr rqrrtrwr* __classcell__r%r%rfr&r^s  r^cseZdZdZdededeffdd Zdeffdd Z d e ffd d Z d e ffd d Z d effdd ZfddZd eeBfddZd edefddZd efddZddZZS)WebsocketClientOutputTransportzWebSocket client output transport for sending frames. Handles outgoing frames, serializes them for WebSocket transmission, and manages audio streaming with proper timing simulation. r_r`r0cs4t|||_||_||_d|_d|_d|_dS)aInitialize the WebSocket client output transport. Args: transport: The parent transport instance. session: The WebSocket session to use for communication. params: Configuration parameters for the transport. rFN)rar;rbrcr4_send_interval_next_send_timerdrerfr%r&r;;s  z'WebsocketClientOutputTransport.__init__rAcrg)zSet up the output transport with the frame processor setup. Args: setup: The frame processor setup configuration. Nrhrirfr%r&rAYrjz$WebsocketClientOutputTransport.setuprkcsvt|IdH|jrdSd|_|j|jd|_|jjr)|jj|IdH|j IdH| |IdHdS)zStart the output transport and initialize the WebSocket connection. Args: frame: The start frame containing initialization parameters. NT) rarlrdaudio_chunk_size sample_raterr4rrArcrrmrnrfr%r&rlbsz$WebsocketClientOutputTransport.startcro)zStop the output transport and disconnect from WebSocket. Args: frame: The end frame signaling transport shutdown. Nrprnrfr%r&rqurrz#WebsocketClientOutputTransport.stopcro)zCancel the output transport and disconnect from WebSocket. Args: frame: The cancel frame signaling immediate cancellation. Nrsrnrfr%r&rt~rrz%WebsocketClientOutputTransport.cancelcru)z(Clean up the output transport resources.Nrvr>rfr%r&rwrxz&WebsocketClientOutputTransport.cleanupcs||IdHdS)zSend a transport message through the WebSocket. Args: frame: The transport message frame to send. N) _write_framernr%r%r& send_messagesz+WebsocketClientOutputTransport.send_messager<c s|jjs |jjs dSt|j|j|jjd}|jjrht B}t |d}| d||j||j||jWdn1sGwYt||j|jd}|}Wdn1scwY||IdH|IdHdS)zWrite an audio frame to the WebSocket with optional WAV header. Args: frame: The output audio frame to write. Returns: True if the audio frame was written successfully, False otherwise. F)audior num_channelswbrN)rrT)rcr[rXr rrr4audio_out_channelsrioBytesIOwaveopen setsampwidth setnchannelsr setframerate writeframesgetvaluer_write_audio_sleep)r:rkbufferwf wav_framer%r%r&write_audio_frames4      z0WebsocketClientOutputTransport.write_audio_framecsT|jjs |jjs dS|jjsdS|jj|IdH}|r(|j|IdHdSdS)z3Write a frame to the WebSocket after serialization.N)rcr[rXr4r serializerP)r:rkpayloadr%r%r&rsz+WebsocketClientOutputTransport._write_framecsZt}td|j|}t|IdH|dkr#t|j|_dS|j|j7_dS)z1Simulate audio playback timing with sleep delays.rN)time monotonicmaxrasynciosleepr)r: current_timesleep_durationr%r%r&rsz1WebsocketClientOutputTransport._write_audio_sleep)rrrr rr.rr;rrArrlr rqrrtrwrrrr r!rr rrrr%r%rfr&r4s(    ' rcsfeZdZdZ ddedeeffdd Zdefdd Z de fd d Z d d Z ddZ ddZZS)WebsocketClientTransportaWebSocket client transport for bidirectional communication. Provides a complete WebSocket client transport implementation with input and output capabilities, connection management, and event handling. Event handlers available: - on_connected(transport): Connected to WebSocket server - on_disconnected(transport): Disconnected from WebSocket server Example:: @transport.event_handler("on_connected") async def on_connected(transport): ... Nr/r0cstt|p t|_|jjpt|j_t|j|j|j d}t ||j||j |_ d|_ d|_|d|ddS)zInitialize the WebSocket client transport. Args: uri: The WebSocket URI to connect to. params: Optional configuration parameters for the transport. )r(r)r*Nr(r))rar;rr4rrr' _on_connected_on_disconnected _on_messager.namerc_input_output_register_event_handler)r:r/r0r1rfr%r&r;s  z!WebsocketClientTransport.__init__r<cC|js t||j|j|_|jS)zGet the input transport for receiving frames. Returns: The WebSocket client input transport instance. )rr^rcr4r>r%r%r&inputzWebsocketClientTransport.inputcCr)zGet the output transport for sending frames. Returns: The WebSocket client output transport instance. )rrrcr4r>r%r%r&outputrzWebsocketClientTransport.outputc|d|IdHdS)z.Handle WebSocket connection established event.r(N_call_event_handlerr:rr%r%r&rz&WebsocketClientTransport._on_connectedcr)z)Handle WebSocket connection closed event.r)Nrrr%r%r&rrz)WebsocketClientTransport._on_disconnectedcs$|jr|j||IdHdSdS)z"Handle incoming WebSocket message.N)rr*)r:rrMr%r%r&r"sz$WebsocketClientTransport._on_message)N)rrrr r$rrr;r^rrrrrrrr%r%rfr&rs  r)0r rrrrtypingrrrr+logurur pydantic.mainrwebsockets.asyncio.clientrrDpipecat.frames.framesrr r r r r rrr"pipecat.processors.frame_processorr#pipecat.serializers.base_serializerrpipecat.serializers.protobufrpipecat.transports.base_inputrpipecat.transports.base_outputr!pipecat.transports.base_transportrr"pipecat.utils.asyncio.task_managerrrr'r.r^rrr%r%r%r&s4   ,       c"