o iJ @s~dZddlZddlZddlZddlZddlmZmZmZddl m Z ddl m Z ddl mZmZmZmZmZmZmZmZmZmZmZddlmZddlmZdd lmZdd lm Z dd l!m"Z"m#Z#zddl$Z$dd l%m&Z'dd l(m)Z)Wn e*yZ+ze ,de+e ,de-de+dZ+[+wwGddde#Z.Gddde Z/GdddeZ0Gddde Z1Gddde"Z2dS)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: c@s>eZdZUdZdZeed<dZee ed<dZ ee ed<dS)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__rrrrintr%r%W/home/ubuntu/.local/lib/python3.10/site-packages/pipecat/transports/websocket/server.pyr4s  rc@sneZdZUdZeejgedfed<eejgedfed<eejgedfed<egedfed<dS)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) rrr r!r websocketsWebSocketServerProtocolrr#r%r%r%r&r'Bs  r'c seZdZdZdededededef fdd Z d e ffd d Z d e ffd d Z d effdd ZfddZddZdejfddZdejdefddZZS)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 sXtj|fi|||_||_||_||_||_d|_d|_d|_ t |_ d|_ dS)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)selfr/r0r1r2r3kwargs __class__r%r&r5Ys  z&WebsocketServerInputTransport.__init__framecsjt|IdH|jrdSd|_|jjr |jj|IdH|js+|||_| |IdHdS)zStart the WebSocket server and initialize components. Args: frame: The start frame containing initialization parameters. NT) r4startrAr9rsetupr< create_task_server_task_handlerset_transport_readyrBrFrDr%r&rGsz#WebsocketServerInputTransport.startcs\t|IdH|j|jr||jIdHd|_|jr,|jIdHd|_dSdS)zStop the WebSocket server and cleanup resources. Args: frame: The end frame signaling transport shutdown. N)r4stopr@setr= cancel_taskr<rLrDr%r&rMs   z"WebsocketServerInputTransport.stopcsXt|IdH|jr||jIdHd|_|jr*||jIdHd|_dSdS)zCancel the WebSocket server and stop all processing. Args: frame: The cancel frame signaling immediate cancellation. N)r4cancelr=rOr<rLrDr%r&rPs z$WebsocketServerInputTransport.cancelc&tIdH|jIdHdSz'Cleanup resources and parent transport.Nr4cleanupr6rBrDr%r&rTz%WebsocketServerInputTransport.cleanupc std|jd|jt|j|j|j4IdH}|jIdH|j IdHWdIdHdS1IdHs=wYdS)z7Handle WebSocket server startup and client connections.zStarting websocket server on :N) rinfor7r8websocket_serve_client_handlerr:r+r@wait)rBserverr%r%r&rJs .z2WebsocketServerInputTransport._server_task_handler websocketc std|j|jr|jIdHtd||_|j|IdH|js9|j j r9| | ||j j |_zE|2z?3dHW}|j j sGq<|j j |IdH}|sTqsleepstaterCLOSEDr:r*CancelledErrorrrXra)rBr]rr%r%r&rds z0WebsocketServerInputTransport._monitor_websocket)rrr r!rstrr$rr'r5rrGr rMrrPrTrJr,r-rZrd __classcell__r%r%rDr&r.Rs0' .r.cseZdZdZdedeffdd Zdeej fddZ d e ffd d Z d e ffd d Zd effdd ZfddZd edeffdd Zd eeBfddZd edefddZd efddZddZZS)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. r/r2c s<tj|fi|||_||_d|_d|_d|_d|_dS)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)r4r5r6r9r;_send_interval_next_send_timerA)rBr/r2rCrDr%r&r5s z'WebsocketServerOutputTransport.__init__r]cs,|jr|jIdHtd||_dS)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)r;rbrrcrBr]r%r%r&set_client_connections   z4WebsocketServerOutputTransport.set_client_connectionrFcsft|IdH|jrdSd|_|jjr |jj|IdH|j|jd|_| |IdHdS)zStart the output transport and initialize components. Args: frame: The start frame containing initialization parameters. NT) r4rGrAr9rrHaudio_chunk_size sample_ratertrKrLrDr%r&rG#sz$WebsocketServerOutputTransport.startc(t|IdH||IdHdS)zStop the output transport and send final frame. Args: frame: The end frame signaling transport shutdown. N)r4rM _write_framerLrDr%r&rM5z#WebsocketServerOutputTransport.stopcr{)zCancel the output transport and send cancellation frame. Args: frame: The cancel frame signaling immediate cancellation. N)r4rPr|rLrDr%r&rP>r}z%WebsocketServerOutputTransport.cancelcrQrRrSrUrDr%r&rTGrVz&WebsocketServerOutputTransport.cleanup directioncs>t||IdHt|tr||IdHd|_dSdS)zProcess frames and handle interruption timing. Args: frame: The frame to process. direction: The direction of frame flow in the pipeline. Nr)r4 process_framerfr r|ru)rBrFr~rDr%r&rLs   z,WebsocketServerOutputTransport.process_framecs||IdHdS)z}Send a transport message frame to the client. Args: frame: The transport message frame to send. N)r|rLr%r%r& send_messageYsz+WebsocketServerOutputTransport.send_messagereturnc s|jsdSt|j|j|jjd}|jjrctB}t |d}| d| |j ||j||jWdn1sBwYt||j|j d}|}Wdn1s^wY||IdH|IdHdS)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)audiorz num_channelswbrxN)rzrT)r;rrrzr9audio_out_channelsrioBytesIOwaveopen setsampwidth setnchannelsr setframerate writeframesgetvaluer|_write_audio_sleep)rBrFbufferwf wav_framer%r%r&write_audio_framecs4      z0WebsocketServerOutputTransport.write_audio_framec s|jjsdSz!|jj|IdH}|r#|jr&|j|IdHWdSWdSWdStyK}zt|d|jj d|dWYd}~dSd}~ww)z3Serialize and send a frame to the WebSocket client.Nz exception sending data: r_r`) r9r serializer;sendrjrrkrEr)rBrFpayloadrlr%r%r&r|s  .z+WebsocketServerOutputTransport._write_framecsZt}td|j|}t|IdH|dkr#t|j|_dS|j|j7_dS)z>Simulate audio device timing by sleeping between audio chunks.rN)time monotonicmaxrur>rmrt)rB current_timesleep_durationr%r%r&rsz1WebsocketServerOutputTransport._write_audio_sleep)rrr r!rrr5rr,r-rwrrGr rMrrPrTr rrrrrrr"rr|rrrr%r%rDr&rss     ' rsc seZdZdZ    ddedededeed eef fd d Zd e fd dZ d e fddZ ddZ ddZddZddZZS)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): ... localhost="Nr2r0r1 input_name output_namecsztj||d||_||_||_t|j|j|j|j d|_ d|_ d|_ d|_ |d|d|d|ddS)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+)r4r5r7r8r9r'_on_client_connected_on_client_disconnected_on_session_timeout_on_websocket_readyr:_input_outputr;_register_event_handler)rBr2r0r1rrrDr%r&r5s"   z!WebsocketServerTransport.__init__rcCs,|jst||j|j|j|j|jd|_|jS)zGet the input transport for receiving client data. Returns: The WebSocket server input transport instance. name)rr.r7r8r9r: _input_namerUr%r%r&inputs zWebsocketServerTransport.inputcCs |js t||j|jd|_|jS)zGet the output transport for sending data to clients. Returns: The WebSocket server output transport instance. r)rrsr9 _output_namerUr%r%r&outputs  zWebsocketServerTransport.outputcs\|jr'|j|IdH|d|IdH|jr%|jtIdHdSdStddS)z Handle client connection events.Nr(|jr|jdIdH|d|IdHdStddS)z#Handle client disconnection events.Nr)r)rrwrrrkrvr%r%r&rs z0WebsocketServerTransport._on_client_disconnectedcs|d|IdHdS)z%Handle client session timeout events.r*Nrrvr%r%r&rsz,WebsocketServerTransport._on_session_timeoutcs|dIdHdS)z%Handle WebSocket server ready events.r+NrrUr%r%r&r sz,WebsocketServerTransport._on_websocket_ready)rrNN)rrr r!rrqr$rr5r.rrsrrrrrrrr%r%rDr&rs.'   r)3r!r>rrrtypingrrrlogururpydanticrpipecat.frames.framesrrr r r r r rrrr"pipecat.processors.frame_processorr#pipecat.serializers.base_serializerrpipecat.transports.base_inputrpipecat.transports.base_outputr!pipecat.transports.base_transportrrr,websockets.asyncio.serverrrYwebsockets.protocolrModuleNotFoundErrorrlrkrjrr'r.rsrr%r%r%r&s>  4     &,