o io @sdZddlZddlZddlZddlZddlZddlmZddlm Z m Z m Z m Z ddl Z ddlmZddlmZddlmZmZmZddlmZdd lmZmZdd lmZmZdd lmZdd l m!Z!dd l"m#Z#zddl$m%Z%ddl&m'Z'ddl(m)Z*ddl+m,Z,Wn e-yZ.ze/de.e/de0de.dZ.[.wwdZ1GdddeZ2GdddeZ3GdddZ4dS)a HeyGen implementation for Pipecat. This module provides integration with the HeyGen platform for creating conversational AI applications with avatars. It manages conversation sessions and provides real-time audio/video streaming capabilities through the HeyGen API. N)Enum) AwaitableCallableOptionalUnion)logger) BaseModel) AudioRawFrame ImageRawFrame StartFrame)FrameProcessorSetup) HeyGenApiNewSessionRequest) LiveAvatarApiLiveAvatarNewSessionRequest)StandardSessionResponse)TransportParams)BaseTaskManager)rtc)VideoBufferType)connect)ConnectionClosedOKz Exception: zEIn order to use HeyGen, you need to `pip install pipecat-ai[heygen]`.zMissing module: i]c@seZdZdZdZdZdS) ServiceTypezEnum for HeyGen service types.INTERACTIVE_AVATAR LIVE_AVATARN)__name__ __module__ __qualname____doc__rrrrR/home/ubuntu/.local/lib/python3.10/site-packages/pipecat/services/heygen/client.pyr6src@sReZdZUdZegedfed<eegedfed<eegedfed<dS)HeyGenCallbacksaCallback handlers for HeyGen events. Parameters: on_connected: Called when the bot connects to the LiveKit room. on_participant_connected: Called when a participant connects. on_participant_disconnected: Called when a participant disconnects. N on_connectedon_participant_connectedon_participant_disconnected)rrrrrr__annotations__strrrrr r!=s r!c@seZdZdZdddddedejdedee e e fd ee d e d ed dfd dZddZded dfddZdQddZdeded dfddZdQddZddZddZd ed dfd!d"ZdQd#d$Zd%ed dfd&d'Zd(ed dfd)d*ZdQd+d,ZdQd-d.Z dQd/d0Z!e"d efd1d2Z#e"d efd3d4Z$d5e%d(ed dfd6d7Z&d8d9Z'd:d;Z(d(ed dfded dfd?d@Z*d>ed dfdAdBZ+dCe,j-fdDdEZ.dCe,j/fdFdGZ0dHdIZ1dJdKZ2dLdMZ3dNe4j5fdOdPZ6dS)R HeyGenClientaA client for interacting with HeyGen's Interactive Avatar Realtime API. This client manages both WebSocket and LiveKit connections for real-time avatar streaming, handling bi-directional audio/video communication and avatar control. It implements the API defined in https://docs.heygen.com/docs/interactive-avatar-realtime-api The client manages the following connections: 1. WebSocket connection for avatar control and audio streaming 2. LiveKit connection for receiving avatar video and audio Attributes: HEY_GEN_SAMPLE_RATE (int): The required sample rate for HeyGen's audio processing (24000 Hz) NF)session_request service_typeconnect_as_userapi_keysessionparamsr(r) callbacksr*returncCs<|dur|ntj|_|dur5|dur5|tjkr$t|ts$tdd}n|tjkr5t|ts5tdd}|durK|jtjkrFtddd}ntdd}|jtjkrYt ||d |_ nt ||d |_ d|_ d|_ d|_||_d |_d |_d |_||_||_d|_d|_d|_d|_d|_d|_d |_d |_d |_d |_||_dS) aInitialize the HeyGen client. Args: api_key: HeyGen API key for authentication session: HTTP client session for API requests params: Transport configuration parameters session_request: Configuration for the HeyGen session (optional) service_type: Type of service to use callbacks: Callback handlers for HeyGen events connect_as_user: Whether to connect using the user token or not (default: False) NzeService type is LIVE_AVATAR but session_request is not SessionTokenRequest. Ignoring session_request.zjService type is INTERACTIVE_AVATAR but session_request is not NewSessionRequest. Ignoring session_request.Shawn_Therapist_publicv2) avatar_idversionz$1c690fe7-23e0-49f9-bfba-14344450285b)r2)r,rF) rr _service_typer isinstancerrwarningrr _apir_heyGen_session _websocket _task_manager_params_in_sample_rate_out_sample_rate _connected_session_request _callbacks _event_queue _event_task _video_task _audio_task_video_frame_callback_audio_frame_callback_send_interval_next_send_time_audio_seconds_sent_transport_ready_connect_as_user)selfr+r,r-r(r)r.r*rrr __init__Zsb     zHeyGenClient.__init__cs|j|jIdH|_td|jjtd|jjtd|jjtd|jj t d|jjd|jj t ddS)NzHeyGen sessionId: zHeyGen realtime_endpoint: zHeyGen livekit URL: zHeyGen livekit token: z5Full Link: https://meet.livekit.io/custom?liveKitUrl=z&token=zHeyGen session started) r8 new_sessionr@r9rdebug session_idws_url livekit_url access_tokeninforMrrr _initializeszHeyGenClient._initializesetupc s|jdur tddS|j|_z|IdHt|_|j | |j|d|_ WdSt yQ}zt d||IdHWYd}~dSd}~ww)zSetup the client and initialize the conversation. Establishes a new session with HeyGen's API if one doesn't exist. Args: setup: The frame processor setup configuration. Nz"heygen_session already initializedz::event_callback_taskzFailed to setup HeyGenClient: )r9rrP task_managerr;rWasyncioQueuerB create_task_callback_task_handlerrC Exceptionerrorcleanup)rMrXerrr rXs"     zHeyGenClient.setupc sz2|jdur|j|jjIdHd|_d|_|jr.|jr1|j|jIdHd|_WdSWdSWdStyN}zt d|WYd}~dSd}~ww)zgCleanup client resources. Closes the active HeyGen session and resets internal state. NFzException during cleanup: ) r9r8 close_sessionrQr?rCr; cancel_taskr^rr_rMrarrr r`s    zHeyGenClient.cleanupframeaudio_chunk_sizecs|jr tddStd|jjp|j|_|jjp|j|_||jd|_td|j| IdH| IdH| |j j dS)a|Start the client and establish all necessary connections. Initializes WebSocket and LiveKit connections using the provided configuration. Sets up audio processing with the specified sample rates. Args: frame: Initial configuration frame containing audio parameters audio_chunk_size: Audio chunk size for output processing zheygen client already startedNzHeyGenClient startingzHeyGenClient send_interval: )r:rrPr<audio_in_sample_rater=audio_out_sample_rater>rH _ws_connect_livekit_connect_call_event_callbackrAr")rMrerfrrr starts  zHeyGenClient.startcs:td|IdH|IdH|IdHdS)zStop the client and terminate all connections. Disconnects from WebSocket and LiveKit endpoints, and performs cleanup. zHeyGenVideoService stoppingN)rrP_ws_disconnect_livekit_disconnectr`rVrrr stops  zHeyGenClient.stopc sz,|jr tdWdStdt|jjdIdH|_d|_|jj| dd|_ WdSt yM}zt |d|d|_WYd}~dSd}~ww) z%Connect to HeyGen websocket endpoint.z"HeyGenClient ws already connected!NzHeyGenClient ws connecting)uriTHeyGenClient_Websocketnamez initialization error: ) r:rrPwebsocket_connectr9rRr?r;r\_ws_receive_task_handler _receive_taskr^r_rdrrr rj s$  zHeyGenClient._ws_connectc s|jrEz|jIdH}t|}||IdHWn$ty%YdSty?}zt d|WYd}~dSd}~ww|jsdSdS)z#Handle incoming WebSocket messages.Nz$Error processing WebSocket message: ) r?r:recvjsonloads_handle_ws_server_eventrr^rr_)rMmessageparsed_messagerarrr rvs  z%HeyGenClient._ws_receive_task_handlereventcs<|d}|dkrtd|dStd|dS)z&Handle an event from HeyGen websocket.typez agent.statez'HeyGenClient ws received agent status: z(HeyGenClient ws received unknown event: N)getrrPtrace)rMr~ event_typerrr r{,s  z$HeyGenClient._handle_ws_server_eventc s~z9zd|_|jr|jIdHWnty.}zt|d|WYd}~n d}~wwWd|_dSWd|_dSd|_w)z*Disconnect from HeyGen websocket endpoint.FNz disconnect error: )r?r:closer^rr_rdrrr rn4s  zHeyGenClient._ws_disconnectr|c st|jst|ddSz|jr!|jt|IdHWdSWdSty9}z td||d}~ww)z#Send a message to HeyGen websocket.z$ websocket is not connected anymore.Nz+Error sending message to HeyGen websocket: ) r?rrPr:sendrydumpsr^r_)rMr|rarrr _ws_send?szHeyGenClient._ws_sendevent_idcs.td||d|dIdHdS)zInterrupt the avatar's current action. Stops the current animation/speech and returns the avatar to idle state. Useful for handling user interruptions during avatar speech. zHeyGenClient interruptzagent.interruptrrN)rrP_reset_audio_timingrrMrrrr interruptKs zHeyGenClient.interruptcs.td|dttdIdHdS)zStart the avatar's listening animation. Triggers visual cues indicating the avatar is listening to user input. z"HeyGenClient start_agent_listeningzagent.start_listeningrN)rrPrr&uuiduuid4rVrrr start_agent_listeningZs  z"HeyGenClient.start_agent_listeningcs$|dttdIdHdS)zoStop the avatar's listening animation. Returns the avatar to idle state from listening state. zagent.stop_listeningrN)rr&rrrVrrr stop_agent_listeninggs  z!HeyGenClient.stop_agent_listeningcCs d|_dS)zHIndicates that the output transport is ready and able to receive frames.TN)rKrVrrr transport_readyss zHeyGenClient.transport_readycC|jS)z`Get the output sample rate. Returns: The output sample rate in Hz. )r>rVrrr out_sample_ratewzHeyGenClient.out_sample_ratecCr)z^Get the input sample rate. Returns: The input sample rate in Hz. )r=rVrrr in_sample_raterzHeyGenClient.in_sample_rateaudiocs<t|d}|d||dIdH|IdHdS)zSend audio data to the agent speak. Args: audio: Audio data in base64 encoded format event_id: Unique identifier for the event zutf-8z agent.speak)rrrN)base64 b64encodedecoder_write_audio_sleep)rMrr audio_base64rrr agent_speaks zHeyGenClient.agent_speakcCsd|_d|_dS)z%Reset audio timing control variables.r4rN)rJrIrVrrr rs z HeyGenClient._reset_audio_timingcs|jdkr|j|j7_t|j|_dSt}td|j|}|dkr:t|IdH|j|j7_dSt|j|_dS)z7Simulate audio playback timing with appropriate delays.g@Nr)rJrHtime monotonicrImaxrZsleep)rM current_timesleep_durationrrr rs zHeyGenClient._write_audio_sleepcs$||d|dIdHdS)zSend signaling that the agent has finished speaking. Args: event_id: Unique identifier for the event zagent.speak_endrN)rrrrrr agent_speak_endszHeyGenClient.agent_speak_endparticipant_idcstd|d|j||_|jdurtddS|jr`||jjvrb|jj|}|j D]3}|j t j j kr_|jdur_td|jt j|j|jd}|jj||dd|_dSq0dSdSdS) zCapture audio frames from the HeyGen avatar. Args: participant_id: Identifier of the participant to capture audio from callback: Async function to handle received audio frames zcapture_participant_audio: z, sample_rate: NLTrying to capture more than one audio stream. It is currently not supported.z+Starting audio capture for existing track: track sample_rateHeyGenClient_Receive_Audiors)rrPr=rGrEr7 _livekit_roomremote_participantstrack_publicationsvalueskindr TrackKind KIND_AUDIOrsid AudioStreamr;r\_process_audio_frames)rMrcallback participant track_pub audio_streamrrr capture_participant_audios2   z&HeyGenClient.capture_participant_audiocstd|||_|jdurtddS|jrY||jjvr[|jj|}|jD]0}|j t j j krX|j durXtd|jt |j }|jj||dd|_dSq,dSdSdS)zCapture video frames from the HeyGen avatar. Args: participant_id: Identifier of the participant to capture video from callback: Async function to handle received video frames zcapture_participant_video: Nrz+Starting video capture for existing track: HeyGenClient_Receive_Videors)rrPrFrDr7rrrrrrr KIND_VIDEOrr VideoStreamr;r\_process_video_frames)rMrrrr video_streamrrr capture_participant_videos*    z&HeyGenClient.capture_participant_videostreamc szvzKtd|2z@3dHW}z |j}t|j}t||jdd}|jr/|jr/||IdHWq t yJ}zt d|WYd}~q d}~ww6Wnt yg}zt d|WYd}~n d}~wwWtddSWtddStdw)z)Process audio frames from LiveKit stream.z"Starting audio frame processing...N)rr num_channelszError processing audio frame: zError processing audio frames: zAudio frame processing ended.) rrPrebytesdatar rrKrGr^r_)rMr frame_event audio_frame audio_datararrr rs8   z"HeyGenClient._process_audio_framesc s"zz^td|2zS3dHW}z3|j}|jtjkr |tj}tt|j |j |j fdd}|j d|_ |jrB|jrB||IdHWq ty]}ztd|WYd}~q d}~ww6Wntyz}ztd|WYd}~n d}~wwWtddSWtddStdw) z)Process video frames from LiveKit stream.z"Starting video frame processing...NRGB)imagesizeformatiz)Error processing individual video frame: zError processing video frames: zVideo frame processing ended.)rrPrerrRGB24convertr rrwidthheight timestamp_usptsrKrFr^r_)rMrr video_frame image_framerarrr rs<      z"HeyGenClient._process_video_framesc sztdjjt_jddtjffdd }jddtj dtj dtjffd d }jd dtj dtj dtjfd d }jddtjffdd }j sbjj njj }jjj|IdHtdjjtdjjjtdtjjjjD]1}td|jd|jjj|j|jD]}td|jd|jd|jqqWdSty}ztd|d_WYd}~dSd}~ww)zConnect to LiveKit room.z-HeyGenClient livekit connecting to room URL: participant_connectedrc sbtd|jd|j|jD]}td|jd|jd|jqj j |jdS)NzParticipant connected - SID: , Identity: zAvailable track - SID: , Kind: , Name: ) rrPridentityrrrrtrlrAr#)rrrVrr r#9s z?HeyGenClient._livekit_connect..on_participant_connectedtrack_subscribedr publicationcs|jtjjkr-jdur-jdur-td|jt |}j j |dd_dS|jtjj kr]jdur_jduratd|jtj|jd}j j |dd_dSdSdSdS)Nz+Creating video stream processor for track: rrsz+Creating audio stream processor for track: rr)rrrrrFrDrrPrrr;r\rrrGrErr=r)rrrrrrVrr on_track_subscribedFs&         z:HeyGenClient._livekit_connect..on_track_subscribedtrack_unsubscribedcSstd|jd|jdS)NzTrack unsubscribed - SID: r)rrPrr)rrrrrr on_track_unsubscribedasz.on_track_unsubscribedparticipant_disconnectedcs0td|jd|jjj|jdS)Nz Participant disconnected - SID: r)rrPrrrlrAr$)rrVrr r$is  zBHeyGenClient._livekit_connect..on_participant_disconnectedNz(Successfully connected to LiveKit room: zLocal participant SID: zNumber of remote participants: zExisting participant - SID: rzExisting track - SID: rrzLiveKit initialization error: )rrPr9rSrRoomronRemoteParticipantTrackRemoteTrackPublicationrLlivekit_agent_tokenrTrrtlocal_participantrlenrrrrlrAr#rrr^r_) rMr#rrr$rTrrrarrVr rk1sl         zHeyGenClient._livekit_connectc szCtd|jr|j|jIdHd|_|jr'|j|jIdHd|_|jrBtd|jIdHd|_tdWdSWdSty_}zt d|WYd}~dSd}~ww)zDisconnect from LiveKit room.zStarting LiveKit disconnect...NzDisconnecting from LiveKit roomz+Successfully disconnected from LiveKit roomzLiveKit disconnect error: ) rrPrDr;rcrEr disconnectr^r_rdrrr ros&  z HeyGenClient._livekit_disconnectcGs|j|g|RdS)z,Queue an event callback for async execution.N)rB put_nowait)rMrargsrrr rlsz!HeyGenClient._call_event_callbackqueuecs. |IdH^}}||IdH|q)z1Handle queued callbacks from the specified queue.TN)r task_done)rMrrrrrr r]s z#HeyGenClient._callback_task_handler)r/N)7rrrrr&aiohttp ClientSessionrrrrrrr!boolrNrWr rXr`r intrmrprjrvdictr{rnrrrrrpropertyrrrrrrrrrrrrrrrkrorlrZr[r]rrrr r'Ksh  ]       _r')5rrZrryrrenumrtypingrrrrrlogururpydanticrpipecat.frames.framesr r r "pipecat.processors.frame_processorr .pipecat.services.heygen.api_interactive_avatarr r&pipecat.services.heygen.api_liveavatarrr pipecat.services.heygen.base_apir!pipecat.transports.base_transportr"pipecat.utils.asyncio.task_managerrlivekitr"livekit.rtc._proto.video_frame_pb2rwebsockets.asyncio.clientrruwebsockets.exceptionsrModuleNotFoundErrorrar_r^HEY_GEN_SAMPLE_RATErr!r'rrrr sB