qi*HndZddlZddlZddlZddlZddlmZmZmZddl 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)Gdde'Z*Gdde Z+GddZ,Gdde"Z-Gdde$Z.Gdde&Z/y)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)BaseTaskManagercReZdZUdZdZeed<dZee e e fed<dZ ee ed<y)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__rrdictstrrrU/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/transports/websocket/client.pyrr-s: ND37c3h07,0J)0r(rceZdZUdZeej gedfed<eej gedfed<eej ejgedfed<y)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) rr r!r"r websocketsWebSocketClientProtocolrr$Datar'r(r)r+r+:siJ>>?4PQQzAABIdOSTT*< Exceptionr<r@s r) task_managerz#WebsocketClientSession.task_managergs:!!''((uv !!!r(rFc^K|xjdz c_|js||_yyw)zSet up the session with a task manager. Args: task_manager: The task manager to use for session tasks. N)r=r>)r@rFs r)setupzWebsocketClientSession.setupws/ q !!!-D "s+-cK|jry t|jd|jjd{|_|j j |j|jd|_ |jj|jd{y7s7#t$r%tjd|jYywxYww)z Connect to the WebSocket server.N )r4 open_timeoutrz.::WebsocketClientSession::_client_task_handlerzTimeout connecting to )r?websocket_connectr9r:rrF create_task_client_task_handlerr< _client_taskr;r, TimeoutErrorrerrorrEs r)rzWebsocketClientSession.connects ??  ?$5II#'<<#B#B%DO !% 1 1 = =))+''((VW!D //..t? ? ? @ ? LL1$))= > ?sMC)/B8B4A-B8.B6/B83C)4B86B88+C&#C)%C&&C)c"K|xjdzc_|jr|jdkDry|jj|jd{|jj d{d|_y7.7w)z%Disconnect from the WebSocket server.rHrN)r=r?rF cancel_taskrPcloserEs r) disconnectz!WebsocketClientSession.disconnectst q $"5"5"9 ++D,=,=>>>oo##%%% ?%s$ABB !B>B ? B Bmessagec Kd} |jr%|jj|d{d}|S7#t$r=}tj|d|j j d|dYd}~|Sd}~wwxYw#|ccYSxYww)zwSend a message through the WebSocket connection. Args: message: The message data to send. FNTz exception sending data:  ())r?sendrDrrR __class__r)r@rWresultes r)r[zWebsocketClientSession.sends  oo**7333M 4 Y LLD6!:1;;;O;O:PPRSTRUUVW X XM YMsIB*;9;B; B2A<6B:B<BBB  Bc|jr1|jjtjjk(SdS)zCheck if the WebSocket is currently connected. Returns: True if the WebSocket is in connected state. F)r?stater/StateOPENrEs r) is_connectedz#WebsocketClientSession.is_connecteds3BFt$$ (8(8(=(==[V[[r(c|jr1|jjtjjk(SdS)zCheck if the WebSocket is currently closing. Returns: True if the WebSocket is in the process of closing. F)r?r`r/raCLOSINGrEs r) is_closingz!WebsocketClientSession.is_closings3EIOOt$$ (8(8(@(@@^Y^^r(c K |j23d{}|jj|j|d{87376nH#t$r<}t j |d|j jd|dYd}~nd}~wwxYw|jj|jd{7yw)z7Handle incoming messages from the WebSocket connection.Nz exception receiving data: rYrZ) r?r;r.rDrrRr\rr-)r@rWr^s r)rOz+WebsocketClientSession._client_task_handlers [!% K Kgoo00'JJJ KJ"1 [ LLD6!>>slC AA AA *AA AA  A A C B2B CB+C>C?Cc |jdS)z6String representation of the WebSocket client session.z::WebsocketClientSession)r<rEs r)__str__zWebsocketClientSession.__str__s&&''?@@r(N)rr r!r"r&rr+rApropertyrrFrIrrVr/r1r#r[rcrfrOrir'r(r)r3r3Hs M M&M, M  M0 "o " "..?& *//d \d\\_D__ ?Ar(r3ceZdZdZdededeffd Zdeffd Z de ffd Z de ffd Z deffd Zfd Zd ZxZS)WebsocketClientInputTransportzWebSocket client input transport for receiving frames. Handles incoming WebSocket messages, deserializes them to frames, and pushes them downstream in the processing pipeline. transportsessionr5c\t||||_||_||_d|_y)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)superrA _transport_sessionr: _initializedr@rmrnr5r\s r)rAz&WebsocketClientInputTransport.__init__s1  #  "r(rIcKt||d{|jj|jd{y727w)zSet up the input transport with the frame processor setup. Args: setup: The frame processor setup configuration. NrprIrrrFr@rIr\s r)rIz#WebsocketClientInputTransport.setupB gmE"""mm!!%"4"4555 #5!A A ,A A A  A framecpKt||d{|jryd|_|jjr-|jjj |d{|j jd{|j|d{y77B7"7 w)zStart the input transport and initialize the WebSocket connection. Args: frame: The start frame containing initialization parameters. NT) rpstartrsr:rrIrrrset_transport_readyr@rzr\s r)r|z#WebsocketClientInputTransport.starts gmE"""      << " ",,))//6 6 6mm##%%%&&u--- # 7%-sFB6B.AB6-B0.!B6B2B6(B4)B60B62B64B6cKt||d{|jjd{y7'7w)zStop the input transport and disconnect from WebSocket. Args: frame: The end frame signaling transport shutdown. NrpstoprrrVr~s r)rz"WebsocketClientInputTransport.stop : gl5!!!mm&&((( "(A>!AAAAcKt||d{|jjd{y7'7w)zCancel the input transport and disconnect from WebSocket. Args: frame: The cancel frame signaling immediate cancellation. NrpcancelrrrVr~s r)rz$WebsocketClientInputTransport.cancel: gnU###mm&&((( $(rcKt|d{|jjd{y7'7w)z'Clean up the input transport resources.Nrpcleanuprqr@r\s r)rz%WebsocketClientInputTransport.cleanup6gooo%%''' 'A=!A?AAcK|jjsy|jjj|d{}|syt|tr0|jj r|j |d{yt|tr*|jt|jd{y|j|d{y77[7#7 w)zHandle incoming WebSocket messages. Args: websocket: The WebSocket connection that received the message. message: The received message data. N)rW) r:r deserialize isinstancer audio_in_enabledpush_audio_framer broadcast_framerW push_frame)r@ websocketrWrzs r)r.z(WebsocketClientInputTransport.on_message s||&& ll--99'BB  e/ 0T\\5R5R''. . . 9 :&&'A5==&Y Y Y//%( ( (C / Y (sIAC&CAC&C 9C&>C"?C&C$C& C&"C&$C&)rr r!r"rr3rrArrIrr|r rr rrr. __classcell__r\s@r)rlrlsd " "("& ",6!46..$)))+)( )r(rlceZdZdZdededeffd Zdeffd Z de ffd Z de ffd Z deffd Zfd Zdeezfd ZdedefdZdefdZdZxZS)WebsocketClientOutputTransportzWebSocket client output transport for sending frames. Handles outgoing frames, serializes them for WebSocket transmission, and manages audio streaming with proper timing simulation. rmrnr5cxt||||_||_||_d|_d|_d|_y)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)rprArqrrr:_send_interval_next_send_timersrts r)rAz'WebsocketClientOutputTransport.__init__;sC  #    "r(rIcKt||d{|jj|jd{y727w)zSet up the output transport with the frame processor setup. Args: setup: The frame processor setup configuration. Nrvrws r)rIz$WebsocketClientOutputTransport.setupYrxryrzcKt||d{|jryd|_|j|jz dz |_|j jr-|j jj|d{|jjd{|j|d{y77B7"7 w)zStart the output transport and initialize the WebSocket connection. Args: frame: The start frame containing initialization parameters. NT) rpr|rsaudio_chunk_size sample_raterr:rrIrrrr}r~s r)r|z$WebsocketClientOutputTransport.startbs gmE"""     #44t7G7GG1L << " ",,))//6 6 6mm##%%%&&u--- # 7%-sFCCA7CC!C0C1C C CCCCcKt||d{|jjd{y7'7w)zStop the output transport and disconnect from WebSocket. Args: frame: The end frame signaling transport shutdown. Nrr~s r)rz#WebsocketClientOutputTransport.stopurrcKt||d{|jjd{y7'7w)zCancel the output transport and disconnect from WebSocket. Args: frame: The cancel frame signaling immediate cancellation. Nrr~s r)rz%WebsocketClientOutputTransport.cancel~rrcKt|d{|jjd{y7'7w)z(Clean up the output transport resources.Nrrs r)rz&WebsocketClientOutputTransport.cleanuprrcBK|j|d{y7w)zSend a transport message through the WebSocket. Args: frame: The transport message frame to send. N) _write_frame)r@rzs r) send_messagez+WebsocketClientOutputTransport.send_messages&&&s rBc8K|jjs|jjsyt|j|j |j j}|j jrtj5}tj|d5}|jd|j|j|j!|j |j#|jdddt|j%|j |j}|}ddd|j'|d{|j)d{y#1swYuxYw#1swYGxYw757w)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)rrrfrcrrrr:audio_out_channelsrioBytesIOwaveopen setsampwidth setnchannelsr setframerate writeframesgetvaluer_write_audio_sleep)r@rzbufferwf wav_frames r)write_audio_framez0WebsocketClientOutputTransport.write_audio_framesF == # #4==+E+E#++((88  << & & "YYvt,0OOA&OOE$6$67OOE$5$56NN5;;/ 0 0OO% % 1 1!&!3!3 " "&&&%%'''#00 " " ' (sUBFF 'A#E> :F F F!F8F9F>F F  FFFcBK|jjs|jjsy|jjsy|jjj |d{}|r$|jj |d{yy7+7w)z3Write a frame to the WebSocket after serialization.N)rrrfrcr:r serializer[)r@rzpayloads r)rz+WebsocketClientOutputTransport._write_framesy == # #4==+E+E ||&&  //99%@@ --$$W- - - A -s$A-B/B0$BBBBc:Ktj}td|j|z }t j |d{|dk(r'tj|j z|_y|xj|j z c_y7Pw)z1Simulate audio playback timing with sleep delays.rN)time monotonicmaxrasynciosleepr)r@ current_timesleep_durations r)rz1WebsocketClientOutputTransport._write_audio_sleeps{~~' Q 4 4| CDmmN+++ Q #'>>#3d6I6I#ID  D$7$7 7 ,sABB AB)rr r!r"rr3rrArrIrr|r rr rrrrrrr#rr rrrrs@r)rr4s " "("& "<6!46..&)))+)( '03TT'%-@%T%N . . 8r(rc`eZdZdZ d dedeeffd ZdefdZ de fdZ dZ d Z d ZxZS) 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): ... r4r5ct||xs t|_|jjxs t |j_t |j|j|j}t||j||j|_ d|_ d|_|jd|jdy)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-)rprArr:rrr+ _on_connected_on_disconnected _on_messager3namerr_input_output_register_event_handler)r@r4r5r6r\s r)rAz!WebsocketClientTransport.__init__s 8!6!8 "&,,"9"9"V=T=V ,++ 11'' /sDLL)TYYW ?C AE  $$^4 $$%67r(rBc~|js&t||j|j|_|jS)zGet the input transport for receiving frames. Returns: The WebSocket client input transport instance. )rrlrrr:rEs r)inputzWebsocketClientTransport.inputs- {{7dmmT\\ZDK{{r(c~|js&t||j|j|_|jS)zGet the output transport for sending frames. Returns: The WebSocket client output transport instance. )rrrrr:rEs r)outputzWebsocketClientTransport.outputs- ||9$ t||\DL||r(cDK|jd|d{y7w)z.Handle WebSocket connection established event.r,N_call_event_handlerr@rs r)rz&WebsocketClientTransport._on_connecteds&&~yAAA   cDK|jd|d{y7w)z)Handle WebSocket connection closed event.r-Nrrs r)rz)WebsocketClientTransport._on_disconnecteds&&'8)DDDrcrK|jr%|jj||d{yy7w)z"Handle incoming WebSocket message.N)rr.)r@rrWs r)rz$WebsocketClientTransport._on_message"s/ ;;++((G< < <  46BE=r(r)0r"rrrrtypingrrrr/logurur pydantic.mainrwebsockets.asyncio.clientrrMpipecat.frames.framesr r r r r rrrr"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+r3rlrrr'r(r)rs 00#B   C?@<>L> 1O 1 ay aFAFAR`)$6`)F^8%8^8BP=}P=r(