o iN @sdZddlZddlZddlZddlmZmZmZmZddl m Z ddl m Z ddl mZmZmZmZddlmZded efd d Zd e fd dZdeded efddZ d.dededefddZ d.dededefddZdeded efddZded efddZdedeed efd d!Zd"ed#eeefd efd$d%Z   d/d e d&eed'ed(ed ef d)d*Z d+ed#eeefd efd,d-Z!dS)0aTransport utility functions and FastAPI route setup helpers. This module provides common functionality for setting up transport-specific FastAPI routes and handling WebRTC/WebSocket connections. It includes SDP manipulation utilities for WebRTC compatibility and transport detection helpers. Key features: - WebRTC route setup with connection management - WebSocket route setup for telephony providers - SDP munging for ESP32 and other WebRTC compatibility - Transport client ID detection across different transport types - Video capture utilities for Daily transports The utilities are designed to be transport-agnostic where possible, with specific handlers for each transport type's unique requirements. Example:: from pipecat.runner.utils import parse_telephony_websocket async def telephony_websocket_handler(websocket: WebSocket): transport_type, call_data = await parse_telephony_websocket(websocket) N)AnyCallableDictOptional) WebSocket)logger)DailyRunnerArgumentsLiveKitRunnerArgumentsSmallWebRTCRunnerArgumentsWebSocketRunnerArguments) BaseTransport message_datareturncCstd|ddkr'd|vr'd|divr'd|divr'tddSd|vr>d|vr>d |divr>td d Sd|vrYd |divrYd |divrYtddS|ddkrd|vrd|divrd|divrd|divrtddStddS)zGAttempt to auto-detect transport type from WebSocket message structure.z=== Auto-Detection Analysis ===eventstart streamSidcallSidzAuto-detected: TWILIOtwilio stream_idcall_control_idzAuto-detected: TELNYXtelnyxstreamIdcallIdzAuto-detected: PLIVOplivo stream_sidcall_sid account_sidzAuto-detected: EXOTELexotelz&Auto-detection failed - unknown formatunknown)rtraceget)r r!H/home/ubuntu/.local/lib/python3.10/site-packages/pipecat/runner/utils.py#_detect_transport_type_from_message1s4      r# websocketc s|}i}i}z|IdH}td||r t|ni}Wntjy-Yn ty7tdwz|IdH}td||rOt|ni}Wntjy\Yntyit dYnwzt |}t |}|dkr|}|} t d|dn|dkr|}|} t d|d n d}|} t d |d kr| d i} | d i} | d| d| d} nu|dkr| d| d i d| d i di d| d i dd| d i ddd} nC|dkr| d i} | d| dd} n,|dkr0| d i} | d| d | d!| dd| dd| d"dd#} ni} t d$|d%| || fWSt yW} z td&| d} ~ ww)'aParse telephony WebSocket messages and return transport type and call data. Args: websocket: FastAPI WebSocket connection from telephony provider. Returns: tuple: (transport_type: str, call_data: dict) call_data contains provider-specific fields: - Twilio:: { "stream_id": str, "call_id": str, "body": dict } - Telnyx:: { "stream_id": str, "call_control_id": str, "outbound_encoding": str, "from": str, "to": str, } - Plivo:: { "stream_id": str, "call_id": str, } - Exotel:: { "stream_id": str, "call_id": str, "account_sid": str, "from": str, "to": str, } Raises: ValueError: If WebSocket closes before sending any messages. Example usage:: transport_type, call_data = await parse_telephony_websocket(websocket) if transport_type == "twilio": user_id = call_data["body"]["user_id"] NzFirst message: z>WebSocket closed before receiving telephony handshake messageszSecond message: z1Only received one WebSocket message, expected tworzDetected transport: z (from first message)z (from second message)z$Could not auto-detect transport typerrcustomParametersrr)rcall_idbodyrrr media_formatencodingfromto)rroutbound_encodingr*r,rrr)rr&rrrrcustom_parameters)rr&rr*r,r.zParsed - Type: z, Data: z#Error parsing telephony WebSocket: ) iter_text __anext__rrjsonloadsJSONDecodeErrorStopAsyncIteration ValueErrorwarningr#debugr Exceptionerror)r$message_stream first_messagesecond_messagefirst_message_rawsecond_message_rawdetected_type_firstdetected_type_secondtransport_type call_data_raw start_data body_data call_dataer!r!r"parse_telephony_websocket`s8              rG transportclientcCszddlm}t||r|jWSWn tyYnwzddlm}t||r,|dWSWn ty6Ynwtdt |dS)zGet client identifier from transport-specific client object. Args: transport: The transport instance. client: Transport-specific client object. Returns: Client identifier string, empty if transport not supported. rSmallWebRTCTransportDailyTransportidz3Unable to get client id from unsupported transport r+) (pipecat.transports.smallwebrtc.transportrK isinstancepc_id ImportError"pipecat.transports.daily.transportrMrr6type)rHrIrKrMr!r!r"get_transport_client_ids$       rU framerateczddlm}t||r|j|d|ddIdHWn ty$Ynwzddlm}t||r=|jddIdHWdSWdStyIYdSw) zCapture participant camera video if transport supports it. Args: transport: The transport instance. client: Transport-specific client object. framerate: Video capture framerate. Defaults to 0 (auto). rrLrNcamerarV video_sourceNrJrZrSrMrPcapture_participant_videorRrOrKrHrIrVrMrKr!r!r" maybe_capture_participant_cameras&        r_crW) zCapture participant screen video if transport supports it. Args: transport: The transport instance. client: Transport-specific client object. framerate: Video capture framerate. Defaults to 0 (auto). rrLrN screenVideorYNrJr[r\r^r!r!r" maybe_capture_participant_screen2s&        ratextpatterncCshtdg}|}|D]}td|r't||r&td|s&||q ||q d|dS)zClean up ICE candidates in SDP text for SmallWebRTC. Args: text: SDP text to clean up. pattern: Pattern to match for candidate filtering. Returns: Cleaned SDP text with filtered ICE candidates. z,Removing unsupported ICE candidates from SDPz a=candidateraddr rr7 splitlinesresearchappendjoin)rbrcresultlinesliner!r!r"'_smallwebrtc_sdp_cleanup_ice_candidatesPs    rocCsPtdg}|}|D]}td|s td|s ||q d|dS)zRemove unsupported fingerprint algorithms from SDP text. Args: text: SDP text to clean up. Returns: SDP text with sha-384 and sha-512 fingerprints removed. z*Removing unsupported fingerprints from SDPzsha-384zsha-512rerf)rbrlrmrnr!r!r"%_smallwebrtc_sdp_cleanup_fingerprintsfs  rpsdphostcCst|}|r t||}|S)zApply SDP modifications for SmallWebRTC compatibility. Args: sdp: Original SDP string. host: Host address for ICE candidate filtering. Returns: Modified SDP string with fingerprint and ICE candidate cleanup. )rpro)rqrrr!r!r"smallwebrtc_sdp_mungingxs  rs transport_keytransport_paramscCs<||vrtd|d|d||}td||S)aaGet transport parameters from factory function. Args: transport_key: The transport key to look up transport_params: Dict mapping transport names to parameter factory functions Returns: Transport parameters from the factory function Raises: ValueError: If transport key is missing from transport_params zMissing transport params for 'z'. Please add 'z$' key to your transport_params dict.zUsing transport params for )r5rr7)rtruparamsr!r!r"_get_transport_paramss  rwrvrArEc s:ddlm}|durtdd|_td||dkr:ddlm}||d |d t d d t d d d|_ n]|dkrYddl m }||d |d|ddt dd d|_ n>|dkryddl m}||d |d t dd t dd d|_ n|dkrddlm}||d |d d|_ ntd|d|||d S)!aCreate a telephony transport with pre-parsed WebSocket data. Args: websocket: FastAPI WebSocket connection from telephony provider params: FastAPIWebsocketParams (required) transport_type: Pre-detected provider type ("twilio", "telnyx", "plivo") call_data: Pre-parsed call data dict with provider-specific fields Returns: Configured FastAPIWebsocketTransport ready for telephony use. r)FastAPIWebsocketTransportNzeFastAPIWebsocketParams must be provided. The serializer and add_wav_header will be set automatically.Fz'Using pre-detected telephony provider: r)TwilioFrameSerializerrr&TWILIO_ACCOUNT_SIDr+TWILIO_AUTH_TOKEN)rrr auth_tokenr)TelnyxFrameSerializerrr-PCMUTELNYX_API_KEY)rrr-inbound_encodingapi_keyr)PlivoFrameSerializer PLIVO_AUTH_IDPLIVO_AUTH_TOKEN)rr&auth_idr|r)ExotelFrameSerializer)rrz Unsupported telephony provider: z4. Supported providers: twilio, telnyx, plivo, exotel)r$rv)$pipecat.transports.websocket.fastapirxr5add_wav_headerrinfopipecat.serializers.twilioryosgetenv serializerpipecat.serializers.telnyxr}pipecat.serializers.plivorpipecat.serializers.exotelr) r$rvrArErxryr}rrr!r!r"_create_telephony_transportsV                r runner_argscst|trtd|}ddlm}||j|jd|dSt|tr2td|}ddlm }|||j dSt|t rQt |j Id H\}}t||}t|j |||Id HSt|trltd |}dd lm}||j|j|j|dStd t|) aH Create a transport from runner arguments using factory functions. This function uses the clean transport_params factory pattern where users define a dictionary mapping transport names to parameter factory functions. Args: runner_args: Arguments from the runner. transport_params: Dict mapping transport names to parameter factory functions. Keys should be: "daily", "webrtc", "twilio", "telnyx", "plivo", "exotel" Values should be functions that return transport parameters when called. Returns: Configured transport instance. Raises: ValueError: If transport key is missing from transport_params or runner_args type is unsupported. ImportError: If required dependencies are not installed. Example:: transport_params = { "daily": lambda: DailyParams( audio_in_enabled=True, audio_out_enabled=True, vad_analyzer=SileroVADAnalyzer(), ), "webrtc": lambda: TransportParams( audio_in_enabled=True, audio_out_enabled=True, vad_analyzer=SileroVADAnalyzer(), ), "twilio": lambda: FastAPIWebsocketParams( audio_in_enabled=True, audio_out_enabled=True, vad_analyzer=SileroVADAnalyzer(), # add_wav_header and serializer will be set automatically ), "telnyx": lambda: FastAPIWebsocketParams( audio_in_enabled=True, audio_out_enabled=True, vad_analyzer=SileroVADAnalyzer(), # add_wav_header and serializer will be set automatically ), "plivo": lambda: FastAPIWebsocketParams( audio_in_enabled=True, audio_out_enabled=True, vad_analyzer=SileroVADAnalyzer(), # add_wav_header and serializer will be set automatically ), "exotel": lambda: FastAPIWebsocketParams( audio_in_enabled=True, audio_out_enabled=True, vad_analyzer=SileroVADAnalyzer(), # add_wav_header and serializer will be set automatically ), } transport = await create_transport(runner_args, transport_params) dailyrrLz Pipecat Bot)rvwebrtcrJ)rvwebrtc_connectionNlivekit)LiveKitTransportz#Unsupported runner arguments type: )rPrrwrSrMroom_urltokenr rOrKrr rGr$rr $pipecat.transports.livekit.transportrurl room_namer5rT)rrurvrMrKrArErr!r!r"create_transportsB ?           r)r)NNN)"__doc__r1rrhtypingrrrrfastapirlogururpipecat.runner.typesrr r r !pipecat.transports.base_transportr dictstrr#rGrUintr_rarorprsrwrrr!r!r!r"sl   /    J