qiJdZddlZddlZddlZddlZddlmZmZmZddl m Z ddl m Z ddl mZmZmZmZmZmZmZmZmZmZmZddlmZddlmZdd lmZdd lm Z dd l!m"Z"m#Z# ddl$Z$dd l%m&Z'dd l(m)Z)Gdde#Z.Gdde Z/GddeZ0Gdde Z1Gdde"Z2y#e*$r7Z+e jXde+e jXde-de+dZ+[+wwxYw)zWebSocket server transport implementation for Pipecat. This module provides WebSocket server transport functionality for real-time audio and data streaming, including client connection management, session handling, and frame serialization. N) AwaitableCallableOptional)logger) BaseModel) CancelFrameClientConnectedFrameEndFrameFrameInputAudioRawFrameInputTransportMessageFrameInterruptionFrameOutputAudioRawFrameOutputTransportMessageFrame!OutputTransportMessageUrgentFrame StartFrame)FrameDirection)FrameSerializer)BaseInputTransport)BaseOutputTransport) BaseTransportTransportParams)serve)Statez Exception: zLIn order to use websockets, you need to `pip install pipecat-ai[websocket]`.zMissing module: cHeZdZUdZdZeed<dZee ed<dZ ee ed<y)WebsocketServerParamsaConfiguration parameters for WebSocket server transport. Parameters: add_wav_header: Whether to add WAV headers to audio frames. serializer: Frame serializer for message encoding/decoding. session_timeout: Timeout in seconds for client sessions. Fadd_wav_headerN serializersession_timeout) __name__ __module__ __qualname____doc__rbool__annotations__rrrrintU/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/transports/websocket/server.pyrr4s0!ND ,0J)0%)OXc])r(rceZdZUdZeej gedfed<eej gedfed<eej gedfed<egedfed<y)WebsocketServerCallbacksaCallback functions for WebSocket server events. Parameters: on_client_connected: Called when a client connects to the server. on_client_disconnected: Called when a client disconnects from the server. on_session_timeout: Called when a client session times out. on_websocket_ready: Called when the WebSocket server is ready to accept connections. Non_client_connectedon_client_disconnectedon_session_timeouton_websocket_ready) r r!r"r#r websocketsWebSocketServerProtocolrr%r'r(r)r+r+Bsw":#E#E"F RV"WXX$j&H&H%I9UY?%Z[[ *"D"D!EyQU!VWW Yt_!455r(r+c eZdZdZdededededef fd Z de ffd Z de ffd Z deffd Zfd Zd Zdej&fdZdej&defdZxZS)WebsocketServerInputTransportzWebSocket server input transport for receiving client data. Handles incoming WebSocket connections, message processing, and client session management including timeout monitoring and connection lifecycle. transporthostportparams callbacksc t||fi|||_||_||_||_||_d|_d|_d|_ tj|_ d|_ y)aInitialize the WebSocket server input transport. Args: transport: The parent transport instance. host: Host address to bind the WebSocket server to. port: Port number to bind the WebSocket server to. params: WebSocket server configuration parameters. callbacks: Callback functions for WebSocket events. **kwargs: Additional arguments passed to parent class. NF)super__init__ _transport_host_port_params _callbacks _websocket _server_task _monitor_taskasyncioEvent_stop_server_event _initialized)selfr4r5r6r7r8kwargs __class__s r)r;z&WebsocketServerInputTransport.__init__Ysk& *6*#   #HL "")--/"r(framecKt||d{|jryd|_|jjr-|jjj |d{|j s$|j|j|_|j|d{y77P7 w)zStart the WebSocket server and initialize components. Args: frame: The start frame containing initialization parameters. NT) r:startrGr?rsetuprB create_task_server_task_handlerset_transport_readyrHrKrJs r)rMz#WebsocketServerInputTransport.starts gmE"""      << " ",,))//6 6 6  $ 0 01J1J1L MD &&u--- # 7 .s5CB<AC-B>.AC6C7C>CCc8Kt||d{|jj|jr*|j |jd{d|_|j r|j d{d|_yy7}767w)zStop the WebSocket server and cleanup resources. Args: frame: The end frame signaling transport shutdown. N)r:stoprFsetrC cancel_taskrBrRs r)rTz"WebsocketServerInputTransport.stops gl5!!! ##%   ""4#5#56 6 6!%D    ## # # $D   " 7 $s4BBABB &BBBBBc"Kt||d{|jr*|j|jd{d|_|jr+|j|jd{d|_yy7r7E7w)zCancel the WebSocket server and stop all processing. Args: frame: The cancel frame signaling immediate cancellation. N)r:cancelrCrVrBrRs r)rXz$WebsocketServerInputTransport.cancels gnU###   ""4#5#56 6 6!%D    ""4#4#45 5 5 $D   $ 6 6s3BB .BB 5B;B <B B BcKt|d{|jjd{y7'7wz'Cleanup resources and parent transport.Nr:cleanupr<rHrJs r)r\z%WebsocketServerInputTransport.cleanup6gooo%%''' 'A=!A?AAcKtjd|jd|jt |j |j|j4d{}|j jd{|jjd{dddd{y7Y7977 #1d{7swYyxYww)z7Handle WebSocket server startup and client connections.zStarting websocket server on :N) rinfor=r>websocket_serve_client_handlerr@r/rFwait)rHservers r)rPz2WebsocketServerInputTransport._server_task_handlers 3DJJJ  GA J /G1G5G6G9AG9G:+G%G &9GG  G9G :G?J J GG G G GGJ  H2H J H!J 7H:8"J I0J rcDK tj|d{|jtjur$|j j |d{yy7E7#tj$r$tjd|jwxYww)z1Monitor WebSocket connection for session timeout.NzMonitoring task cancelled for: ) rDsleepstaterCLOSEDr@r.CancelledErrorrrbrl)rHrgrs r)roz0WebsocketServerInputTransport._monitor_websockets --0 0 0ell2oo88CCC3 1C%%  KK9):R:R9ST U  s>B A&A">A&A$A& B "A&$A&&7BB )r r!r"r#rstrr&rr+r;rrMr rTrrXr\rPr0r1rdro __classcell__rJs@r)r3r3Rs %" %"%" %" & %" , %"N..& % % %+ %( 1,Gz/Q/Q,G\ #;; NQ r(r3ceZdZdZdedeffd ZdeejfdZ de ffd Z de ffd Zdeffd Zfd Zded effd ZdeezfdZdedefdZdefdZdZxZS)WebsocketServerOutputTransportzWebSocket server output transport for sending data to clients. Handles outgoing frame serialization, audio streaming with timing control, and client connection management for WebSocket communication. r4r7c zt||fi|||_||_d|_d|_d|_d|_y)aInitialize the WebSocket server output transport. Args: transport: The parent transport instance. params: WebSocket server configuration parameters. **kwargs: Additional arguments passed to parent class. NrF)r:r;r<r?rA_send_interval_next_send_timerG)rHr4r7rIrJs r)r;z'WebsocketServerOutputTransport.__init__sH *6*# HL  "r(rgcK|jr7|jjd{tjd||_y7!w)zSet the active client WebSocket connection. Args: websocket: The WebSocket connection to set as active, or None to clear. Nz-Only one client allowed, using new connection)rArmrrnrHrgs r)set_client_connectionz4WebsocketServerOutputTransport.set_client_connections> ??//'') ) ) NNJ K# *s*AA"ArKcjKt||d{|jryd|_|jjr-|jjj |d{|j |jz dz |_|j|d{y77A7 w)zStart the output transport and initialize components. Args: frame: The start frame containing initialization parameters. NT) r:rMrGr?rrNaudio_chunk_size sample_raterrQrRs r)rMz$WebsocketServerOutputTransport.start#s gmE"""      << " ",,))//6 6 6#44t7G7GG1L&&u--- # 7-s4B3B-AB3-B/.9B3'B1(B3/B31B3cvKt||d{|j|d{y77w)zStop the output transport and send final frame. Args: frame: The end frame signaling transport shutdown. N)r:rT _write_framerRs r)rTz#WebsocketServerOutputTransport.stop5s8 gl5!!!&&& "&959799cvKt||d{|j|d{y77w)zCancel the output transport and send cancellation frame. Args: frame: The cancel frame signaling immediate cancellation. N)r:rXrrRs r)rXz%WebsocketServerOutputTransport.cancel>s8 gnU###&&& $&rcKt|d{|jjd{y7'7wrZr[r]s r)r\z&WebsocketServerOutputTransport.cleanupGr^r_ directioncKt|||d{t|tr!|j |d{d|_yy767w)zProcess frames and handle interruption timing. Args: frame: The frame to process. direction: The direction of frame flow in the pipeline. Nr)r: process_framerqrrr)rHrKrrJs r)rz,WebsocketServerOutputTransport.process_frameLsUg#E9555 e. /##E* * *#$D  0 6 +s!AA(AAAAcBK|j|d{y7w)z}Send a transport message frame to the client. Args: frame: The transport message frame to send. N)r)rHrKs r) send_messagez+WebsocketServerOutputTransport.send_messageYs&&& returncK|jsyt|j|j|jj }|jj rtj5}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 client with timing control. Args: frame: The output audio frame to write. Returns: True if the audio frame was written successfully, False otherwise. F)audior num_channelswbrN)rrT)rArrrr?audio_out_channelsrioBytesIOwaveopen setsampwidth setnchannelsr setframerate writeframesgetvaluer_write_audio_sleep)rHrKbufferwf wav_frames r)write_audio_framez0WebsocketServerOutputTransport.write_audio_framecs4#++((88  << & & "YYvt,0OOA&OOE$6$67OOE$5$56NN5;;/ 0 0OO% % 1 1!&!3!3 " "&&&%%'''#00 " " ' (sUA.E:0E*A#E*:E*$E:E6E:E8E:E' #E**E3/E:8E:c K|jjsy |jjj|d{}|r1|jr$|jj |d{yyy787 #t $r<}t j|d|jjd|dYd}~yd}~wwxYww)z3Serialize and send a frame to the WebSocket client.Nz exception sending data: rjrk) r?r serializerAsendrurrvrJr )rHrKpayloadrws r)rz+WebsocketServerOutputTransport._write_frames||&&  Y LL33==eDDG4??oo**7333+wE3 Y LLD6!:1;;;O;O:PPRSTRUUVW X X YsRC(BA<0B4A>5B9C<B>B C 2C;CCCc:Ktj}td|j|z }t j |d{|dk(r'tj|j z|_y|xj|j z c_y7Pw)z>Simulate audio device timing by sleeping between audio chunks.rN)time monotonicmaxrrDryr)rH current_timesleep_durations r)rz1WebsocketServerOutputTransport._write_audio_sleeps{~~' Q 4 4| CDmmN+++ Q #'>>#3d6I6I#ID  D$7$7 7 ,sABB AB)r r!r"r#rrr;rr0r1rrrMr rTrrXr\r rrrrrrr$rrrr~rs@r)rrs "-"9N"4 $Xj>`>`5a $..$'''+'( % %> %'03TT'%-@%T%N Y Y 8r(rc ~eZdZdZ ddedededeedeef fd Zde fd Z de fd Z d Z d Zd ZdZxZS)WebsocketServerTransportaWebSocket server transport for bidirectional real-time communication. Provides a complete WebSocket server implementation with separate input and output transports, client connection management, and event handling for real-time audio and data streaming applications. Event handlers available: - on_client_connected(transport, websocket): Client WebSocket connected - on_client_disconnected(transport, websocket): Client WebSocket disconnected - on_session_timeout(transport, websocket): Session timed out - on_websocket_ready(transport): WebSocket server is ready to accept connections Example:: @transport.event_handler("on_client_connected") async def on_client_connected(transport, websocket): ... r7r5r6 input_name output_namec|t|||||_||_||_t |j |j|j|j|_ d|_ d|_ d|_ |jd|jd|jd|jdy)aInitialize the WebSocket server transport. Args: params: WebSocket server configuration parameters. host: Host address to bind the server to. Defaults to "localhost". port: Port number to bind the server to. Defaults to 8765. input_name: Optional name for the input processor. output_name: Optional name for the output processor. )rr)r,r-r.r/Nr,r-r.r/)r:r;r=r>r?r+_on_client_connected_on_client_disconnected_on_session_timeout_on_websocket_readyr@_input_outputrA_register_event_handler)rHr7r5r6rrrJs r)r;z!WebsocketServerTransport.__init__s" JKH   2 $ 9 9#'#?#?#77#77   @D AE HL $$%:; $$%=> $$%9: $$%9:r(rc|jsHt||j|j|j|j |j |_|jS)zGet the input transport for receiving client data. Returns: The WebSocket server input transport instance. name)rr3r=r>r?r@ _input_namerHs r)inputzWebsocketServerTransport.inputsG {{7djj$**dllDOORVRbRbDK{{r(c|js't||j|j|_|jS)zGet the output transport for sending data to clients. Returns: The WebSocket server output transport instance. r)rrr? _output_namers r)outputzWebsocketServerTransport.outputs4 ||9dll):):DL||r(cFK|jrv|jj|d{|jd|d{|jr,|jj t d{yyt jdy7m7U7 w)z Handle client connection events.Nr,B?B!B!B!cK|jr>|jjdd{|jd|d{ytjdy757w)z#Handle client disconnection events.Nr-r)rrrrrvrs r)rz0WebsocketServerTransport._on_client_disconnectedsR <<,,44T: : :**+CYO O O LLW X ; Os!+A'A#A'A%A'%A'cDK|jd|d{y7w)z%Handle client session timeout events.r.Nrrs r)rz,WebsocketServerTransport._on_session_timeouts&&';YGGGs   cBK|jdd{y7w)z%Handle WebSocket server ready events.r/Nrrs r)rz,WebsocketServerTransport._on_websocket_ready s&&';<<rs 00    >?<>L,B) *O * 6y 6 b$6bJh8%8h8Vl=}l=i ,FLL;qc"#FLL_` &qc* ++,s$B,,C(12C##C(