o i`V @sdZddlZddlZddlZddlZddlZddlmZmZmZddl m Z ddl m Z ddl mZmZmZmZmZmZmZmZmZmZmZddlmZddlmZdd lmZdd lm Z dd l!m"Z"m#Z#zdd l$m%Z%dd l&m'Z'Wn e(yZ)ze *de)e *de+de)dZ)[)wwGddde#Z,Gddde Z-GdddZ.GdddZ/GdddeZ0Gddde Z1Gddde"Z2dS)aFastAPI WebSocket transport implementation for Pipecat. This module provides WebSocket-based transport for real-time audio/video streaming using FastAPI and WebSocket connections. Supports binary and text serialization with configurable session timeouts and WAV header generation. N) AwaitableCallableOptional)logger) BaseModel) CancelFrameClientConnectedFrameEndFrameFrameInputAudioRawFrameInputTransportMessageFrameInterruptionFrameOutputAudioRawFrameOutputTransportMessageFrame!OutputTransportMessageUrgentFrame StartFrame)FrameDirection)FrameSerializer)BaseInputTransport)BaseOutputTransport) BaseTransportTransportParams) WebSocket)WebSocketStatez Exception: zTIn order to use FastAPI websockets, you need to `pip install pipecat-ai[websocket]`.zMissing module: c@sNeZdZUdZdZeed<dZee ed<dZ ee ed<dZ ee ed<dS)FastAPIWebsocketParamsaConfiguration parameters for FastAPI WebSocket transport. Parameters: add_wav_header: Whether to add WAV headers to audio frames. serializer: Frame serializer for encoding/decoding messages. session_timeout: Session timeout in seconds, None for no timeout. fixed_audio_packet_size: Optional fixed-size packetization for raw PCM audio payloads. Useful when the remote WebSocket media endpoint requires strict audio framing. Fadd_wav_headerN serializersession_timeoutfixed_audio_packet_size) __name__ __module__ __qualname____doc__rbool__annotations__rrrrintrr&r&X/home/ubuntu/.local/lib/python3.10/site-packages/pipecat/transports/websocket/fastapi.pyr6s  rc@sTeZdZUdZeegedfed<eegedfed<eegedfed<dS)FastAPIWebsocketCallbacksa"Callback functions for WebSocket events. Parameters: on_client_connected: Called when a client connects to the WebSocket. on_client_disconnected: Called when a client disconnects from the WebSocket. on_session_timeout: Called when a session timeout occurs. Non_client_connectedon_client_disconnectedon_session_timeout)rr r!r"rrrr$r&r&r&r'r(Gs r(c@s8eZdZdZdefddZddZdeeBfdd Z d S) _WebSocketMessageIteratorzGAsync iterator for WebSocket messages that yields both binary and text. websocketcCs ||_dSN) _websocketselfr-r&r&r'__init__Xs z"_WebSocketMessageIterator.__init__cCs|Sr.r&r1r&r&r' __aiter__[sz#_WebSocketMessageIterator.__aiter__returncs^|jIdH}|ddkrtd|vr|ddur|dSd|vr-|ddur-|dSt)Ntypezwebsocket.disconnectbytestext)r/receiveStopAsyncIteration)r1messager&r&r' __anext__^s z#_WebSocketMessageIterator.__anext__N) rr r!r"rr2r4r7strr<r&r&r&r'r,Us r,c@seZdZdZdedefddZdefddZd e j e e Bfd d Z d e e Bfd dZddZddZddZddZddZed efddZed efddZdS)FastAPIWebsocketClientzWebSocket client wrapper for handling connections and message passing. Manages WebSocket state, message sending/receiving, and connection lifecycle with support for both binary and text message types. r- callbackscCs||_d|_||_d|_dS)zInitialize the WebSocket client. Args: websocket: The FastAPI WebSocket connection. callbacks: Event callback functions. FrN)r/_closing _callbacks_leave_counter)r1r-r?r&r&r'r2ps zFastAPIWebsocketClient.__init___cs|jd7_dS)z]Set up the WebSocket client. Args: _: The start frame (unused). N)rB)r1rCr&r&r'setup|szFastAPIWebsocketClient.setupr5cCs t|jS)zGet an async iterator for receiving WebSocket messages. Returns: An async iterator yielding bytes or strings. )r,r/r3r&r&r'r9s zFastAPIWebsocketClient.receivedatac sz$|r#t|tr|j|IdHWdS|j|IdHWdSWdStyl}z;t|d|j j d|d|jj |jj t j krZ|jsatdd|_WYd}~dSWYd}~dSWYd}~dSd}~ww)zySend data through the WebSocket connection. Args: data: The data to send (string or bytes). N exception sending data:  (z), application_state: z'Closing already disconnected websocket!T) _can_send isinstancer7r/ send_bytes send_text Exceptionrerror __class__rapplication_stater DISCONNECTED is_closingwarningr@)r1rFer&r&r'sends&   zFastAPIWebsocketClient.sendc s|jd8_|jdkrdS|jrA|jsCd|_z |jIdHWdSty@}zt|d|WYd}~dSd}~wwdSdS)z Disconnect the WebSocket client.rDrNTz( exception while closing the websocket: ) rB is_connectedrRr@r/closerMrrN)r1rTr&r&r' disconnects  "z!FastAPIWebsocketClient.disconnectc|j|jIdHdS)z)Trigger the client disconnected callback.N)rAr*r/r3r&r&r'trigger_client_disconnectedz2FastAPIWebsocketClient.trigger_client_disconnectedcrY)z&Trigger the client connected callback.N)rAr)r/r3r&r&r'trigger_client_connectedr[z/FastAPIWebsocketClient.trigger_client_connectedcrY)z$Trigger the client timeout callback.N)rAr+r/r3r&r&r'trigger_client_timeoutr[z-FastAPIWebsocketClient.trigger_client_timeoutcCs|jo|j S)z0Check if data can be sent through the WebSocket.)rVrRr3r&r&r'rIsz FastAPIWebsocketClient._can_sendcCs|jjtjkS)zCheck if the WebSocket is currently connected. Returns: True if the WebSocket is in connected state. )r/ client_stater CONNECTEDr3r&r&r'rVsz#FastAPIWebsocketClient.is_connectedcC|jS)zCheck if the WebSocket is currently closing. Returns: True if the WebSocket is in the process of closing. )r@r3r&r&r'rRsz!FastAPIWebsocketClient.is_closingN)rr r!r"rr(r2rrEtyping AsyncIteratorr7r=r9rUrXrZr\r]rIpropertyr#rVrRr&r&r&r'r>is  r>cseZdZdZdededeffdd Zdeffdd Z d d Z de ffd d Z de ffdd ZfddZddZddZZS)FastAPIWebsocketInputTransportzInput transport for FastAPI WebSocket connections. Handles incoming WebSocket messages, deserializes frames, and manages connection monitoring with optional session timeouts. transportclientparamsc s<tj|fi|||_||_||_d|_d|_d|_dS)a$Initialize the WebSocket input transport. Args: transport: The parent transport instance. client: The WebSocket client wrapper. params: Transport configuration parameters. **kwargs: Additional arguments passed to parent class. NF)superr2 _transport_client_params _receive_task_monitor_websocket_task _initializedr1rerfrgkwargsrOr&r'r2s z'FastAPIWebsocketInputTransport.__init__framecst|IdH|jrdSd|_|j|IdH|jjr)|jj|IdH|js8|jjr8| | |_|j IdH| t IdH|jsT| ||_||IdHdS)zStart the input transport and begin message processing. Args: frame: The start frame containing initialization parameters. NT)rhstartrnrjrErkrrmr create_task_monitor_websocketr\ push_framerrl_receive_messagesset_transport_readyr1rrrqr&r'rssz$FastAPIWebsocketInputTransport.startcsF|jr||jIdHd|_|jr!||jIdHd|_dSdS)zStop all running tasks.N)rm cancel_taskrlr3r&r&r' _stop_tasks s z*FastAPIWebsocketInputTransport._stop_tasksc6t|IdH|IdH|jIdHdS)zStop the input transport and cleanup resources. Args: frame: The end frame signaling transport shutdown. N)rhstopr{rjrXryrqr&r'r}z#FastAPIWebsocketInputTransport.stopcr|)zCancel the input transport and stop all processing. Args: frame: The cancel frame signaling immediate cancellation. N)rhcancelr{rjrXryrqr&r'r r~z%FastAPIWebsocketInputTransport.cancelc&tIdH|jIdHdSzClean up transport resources.Nrhcleanuprir3rqr&r'r*z&FastAPIWebsocketInputTransport.cleanupc szH|j2z?3dHW}|jjsq|jj|IdH}|sqt|tr-||IdHqt|tr>|j t|j dIdHq| |IdHq6Wn"t yk}zt |d|jjd|dWYd}~nd}~ww|jjsz|jIdHdSdS)z3Main message receiving loop for WebSocket messages.N)r;z exception receiving data: rH))rjr9rkr deserializerJr push_audio_framer broadcast_framer;rvrMrrNrOrrRrZ)r1r;rrrTr&r&r'rw/s*  ,z0FastAPIWebsocketInputTransport._receive_messagescs*t|jjIdH|jIdHdS)zeWait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event.N)asynciosleeprkrrjr]r3r&r&r'ruIsz1FastAPIWebsocketInputTransport._monitor_websocket)rr r!r"rr>rr2rrsr{r r}rrrrwru __classcell__r&r&rqr'rds   rdcseZdZdZdededeffdd Zdeffdd Z de ffd d Z de ffd d Z fddZdedeffdd ZdeeBfddZdedefddZdefddZddZZS)FastAPIWebsocketOutputTransportzOutput transport for FastAPI WebSocket connections. Handles outgoing frame serialization, audio streaming with timing simulation, and WebSocket message transmission with optional WAV header generation. rerfrgc sDtj|fi|||_||_||_d|_d|_t|_d|_ dS)a%Initialize the WebSocket output transport. Args: transport: The parent transport instance. client: The WebSocket client wrapper. params: Transport configuration parameters. **kwargs: Additional arguments passed to parent class. rFN) rhr2rirjrk_send_interval_next_send_time bytearray_audio_send_bufferrnrorqr&r'r2Vs z(FastAPIWebsocketOutputTransport.__init__rrcsxt|IdH|jrdSd|_|j|IdH|jjr)|jj|IdH|j|jd|_ | |IdHdS)zStart the output transport and initialize timing. Args: frame: The start frame containing initialization parameters. NT) rhrsrnrjrErkraudio_chunk_size sample_raterrxryrqr&r'rs~sz%FastAPIWebsocketOutputTransport.startc8t|IdH||IdH|jIdHdS)zStop the output transport and cleanup resources. Args: frame: The end frame signaling transport shutdown. N)rhr} _write_framerjrXryrqr&r'r}z$FastAPIWebsocketOutputTransport.stopcr)zCancel the output transport and stop all processing. Args: frame: The cancel frame signaling immediate cancellation. N)rhrrrjrXryrqr&r'rrz&FastAPIWebsocketOutputTransport.cancelcrrrr3rqr&r'rrz'FastAPIWebsocketOutputTransport.cleanup directioncsPt||IdHt|tr&|jjr|j||IdHd|_ dSdS)zProcess outgoing frames with special handling for interruptions. Args: frame: The frame to process. direction: The direction of frame flow in the pipeline. Nr) rh process_framerJr rkrrclearrr)r1rrrrqr&r'rs   z-FastAPIWebsocketOutputTransport.process_framecs||IdHdS)zoSend a transport message frame. Args: frame: The transport message frame to send. N)rryr&r&r' send_messagesz,FastAPIWebsocketOutputTransport.send_messager5c 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 timing simulation. Args: frame: The output audio frame to write. Returns: True if the audio frame was written successfully, False otherwise. F)audior num_channelswbrN)rrT)rjrRrVrrrrkaudio_out_channelsrioBytesIOwaveopen setsampwidth setnchannelsr setframerate writeframesgetvaluer_write_audio_sleep)r1rrbufferwf wav_framer&r&r'write_audio_frames4      z1FastAPIWebsocketOutputTransport.write_audio_framec s|jjs |jjs dS|jjsdSzY|jj|IdH}|rh|jj}|r\t|tt fr\|j t|t |j |krYt|j d|}|j d|=|j |IdHt |j |ks:WdS|j |IdHWdSWdSty}zt|d|jjd|dWYd}~dSd}~ww)z1Serialize and send a frame through the WebSocket.NrGrHr)rjrRrVrkr serializerrJr7rrextendlenrUrMrrNrOr)r1rrpayload packet_byteschunkrTr&r&r'rs. .z,FastAPIWebsocketOutputTransport._write_framecsZt}td|j|}t|IdH|dkr#t|j|_dS|j|j7_dS)z7Simulate audio playback timing with appropriate delays.rN)time monotonicmaxrrrr)r1 current_timesleep_durationr&r&r'r sz2FastAPIWebsocketOutputTransport._write_audio_sleep)rr r!r"rr>rr2rrsr r}rrrr rrrrrrr#rrrrr&r&rqr'rOs((   'rc steZdZdZ  ddededeedeeffdd Zd e fd d Z d e fd d Z ddZ ddZddZZS)FastAPIWebsocketTransportaxFastAPI WebSocket transport for real-time audio/video streaming. Provides bidirectional WebSocket communication with frame serialization, session management, and event handling for client connections and timeouts. 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 Example:: @transport.event_handler("on_client_connected") async def on_client_connected(transport, websocket): ... Nr-rg input_name output_namecstj||d||_t|j|j|jd|_t||j|_ t ||j |j|j d|_ t ||j |j|jd|_|d|d|ddS)a4Initialize the FastAPI WebSocket transport. Args: websocket: The FastAPI WebSocket connection. params: Transport configuration parameters. input_name: Optional name for the input processor. output_name: Optional name for the output processor. )rr)r)r*r+)namer)r*r+N)rhr2rkr(_on_client_connected_on_client_disconnected_on_session_timeoutrAr>rjrd _input_name_inputr _output_name_output_register_event_handler)r1r-rgrrrqr&r'r2)s"  z"FastAPIWebsocketTransport.__init__r5cCr`)zqGet the input transport processor. Returns: The WebSocket input transport instance. )rr3r&r&r'inputQzFastAPIWebsocketTransport.inputcCr`)zsGet the output transport processor. Returns: The WebSocket output transport instance. )rr3r&r&r'outputYrz FastAPIWebsocketTransport.outputc|d|IdHdS)zHandle client connected event.r)N_call_event_handlerr0r&r&r'raz.FastAPIWebsocketTransport._on_client_connectedcr)z!Handle client disconnected event.r*Nrr0r&r&r'rerz1FastAPIWebsocketTransport._on_client_disconnectedcr)zHandle session timeout event.r+Nrr0r&r&r'rirz-FastAPIWebsocketTransport._on_session_timeout)NN)rr r!r"rrrr=r2rdrrrrrrrr&r&rqr'rs$(r)3r"rrrrarrrrlogururpydanticrpipecat.frames.framesrrr r r r r rrrr"pipecat.processors.frame_processorr#pipecat.serializers.base_serializerrpipecat.transports.base_inputrpipecat.transports.base_outputr!pipecat.transports.base_transportrrfastapirstarlette.websocketsrModuleNotFoundErrorrTrNrMrr(r,r>rdrrr&r&r&r'sD  4    lzH