qi`VdZddlZddlZddlZddlZddlZddlmZmZmZddl m Z ddl m Z ddl mZmZmZmZmZmZmZmZmZmZmZddlmZddlmZdd lmZdd lm Z dd l!m"Z"m#Z# dd l$m%Z%dd l&m'Z'Gdde#Z,Gdde Z-GddZ.GddZ/GddeZ0Gdde Z1Gdde"Z2y#e($r7Z)e jTde)e jTde+de)dZ)[)wwxYw)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\eZdZUdZdZeed<dZee ed<dZ ee ed<dZ ee ed<y)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__rrrrintr V/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/transports/websocket/fastapi.pyrr6s?!ND ,0J)0%)OXc])-1Xc]1r)rcfeZdZUdZeegedfed<eegedfed<eegedfed<y)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)r!r"r#r$rrrr&r(r)r*r,r,GsK"9+y">??$i[)D/%ABB )io!=>>r)r,c4eZdZdZdefdZdZdeezfdZ y)_WebSocketMessageIteratorzGAsync iterator for WebSocket messages that yields both binary and text. websocketc||_yN) _websocketselfr2s r*__init__z"_WebSocketMessageIterator.__init__Xs #r)c|Sr4r(r7s r* __aiter__z#_WebSocketMessageIterator.__aiter__[s r)returncK|jjd{}|ddk(rtd|vr |d|dSd|vr |d|dSt74w)Ntypezwebsocket.disconnectbytestext)r5receiveStopAsyncIteration)r7messages r* __anext__z#_WebSocketMessageIterator.__anext__^sq//11 6?4 4$ $ g ''"2">7# # W !<6? "  2sAA5AN) r!r"r#r$rr8r;r?strrDr(r)r*r1r1Us&Q$)$!!r)r1ceZdZdZdedefdZdefdZde je e zfdZ d e e zfd Zd Zd Zd ZdZdZedefdZedefdZy)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. r2 callbacksc<||_d|_||_d|_y)zInitialize the WebSocket client. Args: websocket: The FastAPI WebSocket connection. callbacks: Event callback functions. FrN)r5_closing _callbacks_leave_counter)r7r2rHs r*r8zFastAPIWebsocketClient.__init__ps!$ #r)_c6K|xjdz c_yw)z]Set up the WebSocket client. Args: _: The start frame (unused). N)rL)r7rMs r*setupzFastAPIWebsocketClient.setup|s q sr<c,t|jS)zGet an async iterator for receiving WebSocket messages. Returns: An async iterator yielding bytes or strings. )r1r5r:s r*rAzFastAPIWebsocketClient.receives )99r)datac TK |jrXt|tr$|jj |d{y|jj |d{yy7*7#t $r}tj|d|jjd|d|jj|jjtjk(r2|js!tjdd|_Yd}~yYd}~yYd}~yd}~wwxYww)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 isinstancer?r5 send_bytes send_text Exceptionrerror __class__r!application_stater DISCONNECTED is_closingwarningrJ)r7rRes r*sendzFastAPIWebsocketClient.sends  %~~dE*//44T::://33D999  :9 % LL&1!++2F2F1Gr!LbcgcrcrdEdEcFG  11^5P5PPHI $ (Q %s_D(>A1A-A1D(A1&A/'A1+D(-A1/A11 D%:BD D( D%%D(cBK|xjdzc_|jdkDry|jr8|js+d|_ |jj d{yyy7#t $r$}tj|d|Yd}~yd}~wwxYww)z Disconnect the WebSocket client.rOrNTz( exception while closing the websocket: ) rL is_connectedr_rJr5closerZrr[)r7ras r* disconnectz!FastAPIWebsocketClient.disconnects q     "    T__ DM Soo++---&5 . S v%MaSQRR SsBABA/%A-&A/*B-A// B8BBBBcjK|jj|jd{y7w)z)Trigger the client disconnected callback.N)rKr.r5r:s r*trigger_client_disconnectedz2FastAPIWebsocketClient.trigger_client_disconnecteds"oo44T__EEE )313cjK|jj|jd{y7w)z&Trigger the client connected callback.N)rKr-r5r:s r*trigger_client_connectedz/FastAPIWebsocketClient.trigger_client_connecteds"oo11$//BBBricjK|jj|jd{y7w)z$Trigger the client timeout callback.N)rKr/r5r:s r*trigger_client_timeoutz-FastAPIWebsocketClient.trigger_client_timeouts"oo00AAAric8|jxr |j S)z0Check if data can be sent through the WebSocket.)rdr_r:s r*rVz FastAPIWebsocketClient._can_sends  8%88r)cP|jjtjk(S)zCheck if the WebSocket is currently connected. Returns: True if the WebSocket is in connected state. )r5 client_stater CONNECTEDr:s r*rdz#FastAPIWebsocketClient.is_connecteds++~/G/GGGr)c|jS)zCheck if the WebSocket is currently closing. Returns: True if the WebSocket is in the process of closing. )rJr:s r*r_z!FastAPIWebsocketClient.is_closings}}r)N)r!r"r#r$rr,r8rrPtyping AsyncIteratorr?rErArbrfrhrkrmrVpropertyr%rdr_r(r)r*rGrGis  )  8Q  !Z!:--eck::%sU{%2 SFCB9HdHHDr)rGc|eZdZdZdededeffd Zdeffd Z dZ de ffd Z de ffd Zfd Zd Zd ZxZS)FastAPIWebsocketInputTransportzInput transport for FastAPI WebSocket connections. Handles incoming WebSocket messages, deserializes frames, and manages connection monitoring with optional session timeouts. transportclientparamsc zt||fi|||_||_||_d|_d|_d|_y)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)superr8 _transport_client_params _receive_task_monitor_websocket_task _initializedr7rxryrzkwargsr\s r*r8z'FastAPIWebsocketInputTransport.__init__sF *6*#  !'+$"r)framecKt||d{|jryd|_|jj |d{|j j r-|j j j |d{|js:|j jr$|j|j|_|jjd{|jtd{|js$|j|j|_|j!|d{y7R777w7X7w)zStart the input transport and begin message processing. Args: frame: The start frame containing initialization parameters. NT)r|startrr~rPrrrr create_task_monitor_websocketrk push_framer r_receive_messagesset_transport_readyr7rr\s r*rz$FastAPIWebsocketInputTransport.starts' gmE"""     ll  ''' << " ",,))//6 6 6++ 0L0L+/+;+;DAB0B1BBcBK|j|d{y7w)zoSend a transport message frame. Args: frame: The transport message frame to send. N)r)r7rs r* send_messagez,FastAPIWebsocketOutputTransport.send_messages&&&s r<c8K|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 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)r~r_rdrrrraudio_out_channelsrioBytesIOwaveopen setsampwidth setnchannelsr setframerate writeframesgetvaluer_write_audio_sleep)r7rbufferwf wav_frames r*write_audio_framez1FastAPIWebsocketOutputTransport.write_audio_framesF << " "$,,*C*C#++((88  << & & "YYvt,0OOA&OOE$6$67OOE$5$56NN5;;/ 0 0OO% % 1 1!&!3!3 " "&&&%%'''#00 " " ' (sUBFF 'A#E> :F F F!F8F9F>F F  FFFc tK|jjs|jjsy|jjsy |jjj |d{}|r|jj }|rt|ttfr|jjt|t|j|k\rct|jd|}|jd|=|jj|d{t|j|k\rcy|jj|d{yy77E7 #t$r<}tj |d|j"j$d|dYd}~yd}~wwxYww)z1Serialize and send a frame through the WebSocket.NrTrUr)r~r_rdrr serializer rWr?rrextendlenrbrZrr[r\r!)r7rpayload packet_byteschunkras r*rz,FastAPIWebsocketOutputTransport._write_framest << " "$,,*C*C ||&&  Y LL33==eDDG $||CC Jw 8J$K++225>Bd556,F %d&=&=m|&L M 33M\MB"ll//666d556,Fll''000!E71 Y LLD6!:1;;;O;O:PPRSTRUUVW X X YssAF8(E00E*1B5E0&E,'E0F8E0#E.$E0(F8*E0,E0.E00 F592F0+F80F55F8c:Ktj}td|j|z }t j |d{|dk(r'tj|j z|_y|xj|j z c_y7Pw)z7Simulate audio playback timing with appropriate delays.rN)time monotonicmaxrrrr)r7 current_timesleep_durations r*rz2FastAPIWebsocketOutputTransport._write_audio_sleep s{~~' Q 4 4| CDmmN+++ Q #'>>#3d6I6I#ID  D$7$7 7 ,sABB AB)r!r"r#r$rrGrr8rrr rrrrr rrrrrrr%rrrrrs@r*rrOs &" &"'&"' &"P..&(((+(( %%>%"'03TT'%-@%T%NYY< 8r)rc peZdZdZ d dededeedeeffd Zde fdZ de fd Z d Z d Zd ZxZS)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): ... r2rz input_name output_namect|||||_t|j|j |j |_t||j|_ t||j|j|j|_ t||j|j|j|_|j!d|j!d|j!dy)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)r|r8rr,_on_client_connected_on_client_disconnected_on_session_timeoutrKrGr~rw _input_name_inputr _output_name_output_register_event_handler)r7r2rzrrr\s r*r8z"FastAPIWebsocketTransport.__init__)s JKH 3 $ 9 9#'#?#?#77  .iI 4 $,, 43C3C  7 $,, 43D3D  $$%:; $$%=> $$%9:r)r<c|jS)zqGet the input transport processor. Returns: The WebSocket input transport instance. )rr:s r*inputzFastAPIWebsocketTransport.inputQs {{r)c|jS)zsGet the output transport processor. Returns: The WebSocket output transport instance. )rr:s r*outputz FastAPIWebsocketTransport.outputYs ||r)cDK|jd|d{y7w)zHandle client connected event.r-N_call_event_handlerr6s r*rz.FastAPIWebsocketTransport._on_client_connectedas&&'rs 00    >?<>L,!32_2" ? ?!!(iiXw4%7w4tD8&9D8NUH UHQ,FLL;qc"#FLL^ &qc* ++ ,s( CC<2C77C<