# # Copyright (c) 2024-2026, Daily # # SPDX-License-Identifier: BSD 2-Clause License # """LiveKit transport implementation for Pipecat. This module provides comprehensive LiveKit real-time communication integration including audio streaming, data messaging, participant management, and room event handling for conversational AI applications. """ import asyncio import json from dataclasses import dataclass from typing import Any, Awaitable, Callable, List, Optional from loguru import logger from pydantic import BaseModel from pipecat.audio.utils import create_stream_resampler from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.frames.frames import ( AudioRawFrame, BotConnectedFrame, CancelFrame, ClientConnectedFrame, EndFrame, ImageRawFrame, OutputAudioRawFrame, OutputDTMFFrame, OutputDTMFUrgentFrame, OutputTransportMessageFrame, OutputTransportMessageUrgentFrame, StartFrame, UserAudioRawFrame, UserImageRawFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.utils.asyncio.task_manager import BaseTaskManager try: from livekit import rtc from livekit.rtc._proto import video_frame_pb2 as proto_video_frame from tenacity import retry, stop_after_attempt, wait_exponential except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error("In order to use LiveKit, you need to `pip install pipecat-ai[livekit]`.") raise Exception(f"Missing module: {e}") # DTMF mapping according to RFC 4733 DTMF_CODE_MAP = { "0": 0, "1": 1, "2": 2, "3": 3, "4": 4, "5": 5, "6": 6, "7": 7, "8": 8, "9": 9, "*": 10, "#": 11, } @dataclass class LiveKitOutputTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in LiveKit rooms. Parameters: participant_id: Optional ID of the participant this message is for/from. """ participant_id: Optional[str] = None @dataclass class LiveKitOutputTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): """Frame for urgent transport messages in LiveKit rooms. Parameters: participant_id: Optional ID of the participant this message is for/from. """ participant_id: Optional[str] = None @dataclass class LiveKitTransportMessageFrame(LiveKitOutputTransportMessageFrame): """Frame for transport messages in LiveKit rooms. Parameters: participant_id: Optional ID of the participant this message is for/from. """ def __post_init__(self): super().__post_init__() import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( "LiveKitTransportMessageFrame is deprecated and will be removed in a future version. " "Instead, use LiveKitOutputTransportMessageFrame.", DeprecationWarning, stacklevel=2, ) @dataclass class LiveKitTransportMessageUrgentFrame(LiveKitOutputTransportMessageUrgentFrame): """Frame for urgent transport messages in LiveKit rooms. Parameters: participant_id: Optional ID of the participant this message is for/from. """ def __post_init__(self): super().__post_init__() import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( "LiveKitTransportMessageUrgentFrame is deprecated and will be removed in a future version. " "Instead, use LiveKitOutputTransportMessageUrgentFrame.", DeprecationWarning, stacklevel=2, ) class LiveKitParams(TransportParams): """Configuration parameters for LiveKit transport. Inherits all parameters from TransportParams without additional configuration. """ pass class LiveKitCallbacks(BaseModel): """Callback handlers for LiveKit events. Parameters: on_connected: Called when connected to the LiveKit room. on_disconnected: Called when disconnected from the LiveKit room. on_participant_connected: Called when a participant joins the room. on_participant_disconnected: Called when a participant leaves the room. on_audio_track_subscribed: Called when an audio track is subscribed. on_audio_track_unsubscribed: Called when an audio track is unsubscribed. on_data_received: Called when data is received from a participant. on_first_participant_joined: Called when the first participant joins. """ on_connected: Callable[[], Awaitable[None]] on_disconnected: Callable[[], Awaitable[None]] on_before_disconnect: Callable[[], Awaitable[None]] on_participant_connected: Callable[[str], Awaitable[None]] on_participant_disconnected: Callable[[str], Awaitable[None]] on_audio_track_subscribed: Callable[[str], Awaitable[None]] on_audio_track_unsubscribed: Callable[[str], Awaitable[None]] on_video_track_subscribed: Callable[[str], Awaitable[None]] on_video_track_unsubscribed: Callable[[str], Awaitable[None]] on_data_received: Callable[[bytes, str], Awaitable[None]] on_first_participant_joined: Callable[[str], Awaitable[None]] class LiveKitTransportClient: """Core client for interacting with LiveKit rooms. Manages the connection to LiveKit rooms and handles all low-level API interactions including room management, audio streaming, data messaging, and event handling. """ def __init__( self, url: str, token: str, room_name: str, params: LiveKitParams, callbacks: LiveKitCallbacks, transport_name: str, ): """Initialize the LiveKit transport client. Args: url: LiveKit server URL to connect to. token: Authentication token for the room. room_name: Name of the LiveKit room to join. params: Configuration parameters for the transport. callbacks: Event callback handlers. transport_name: Name identifier for the transport. """ self._url = url self._token = token self._room_name = room_name self._params = params self._callbacks = callbacks self._transport_name = transport_name self._room: Optional[rtc.Room] = None self._participant_id: str = "" self._connected = False self._disconnect_counter = 0 self._audio_source: Optional[rtc.AudioSource] = None self._audio_track: Optional[rtc.LocalAudioTrack] = None self._audio_tracks = {} self._audio_queue = asyncio.Queue() self._video_tracks = {} self._video_queue = asyncio.Queue() self._other_participant_has_joined = False self._task_manager: Optional[BaseTaskManager] = None self._async_lock = asyncio.Lock() @property def participant_id(self) -> str: """Get the participant ID for this client. Returns: The participant ID assigned by LiveKit. """ return self._participant_id @property def room(self) -> rtc.Room: """Get the LiveKit room instance. Returns: The LiveKit room object. Raises: Exception: If room object is not available. """ if not self._room: raise Exception(f"{self}: missing room object (pipeline not started?)") return self._room async def setup(self, setup: FrameProcessorSetup): """Setup the client with task manager and room initialization. Args: setup: The frame processor setup configuration. """ if self._task_manager: return self._task_manager = setup.task_manager self._room = rtc.Room(loop=self._task_manager.get_event_loop()) # Set up room event handlers self.room.on("participant_connected")(self._on_participant_connected_wrapper) self.room.on("participant_disconnected")(self._on_participant_disconnected_wrapper) self.room.on("track_subscribed")(self._on_track_subscribed_wrapper) self.room.on("track_unsubscribed")(self._on_track_unsubscribed_wrapper) self.room.on("data_received")(self._on_data_received_wrapper) self.room.on("connected")(self._on_connected_wrapper) self.room.on("disconnected")(self._on_disconnected_wrapper) async def cleanup(self): """Cleanup client resources.""" await self.disconnect() async def start(self, frame: StartFrame): """Start the client and initialize audio components. Args: frame: The start frame containing initialization parameters. """ self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) async def connect(self): """Connect to the LiveKit room with retry logic.""" async with self._async_lock: if self._connected: # Increment disconnect counter if already connected. self._disconnect_counter += 1 return logger.info(f"Connecting to {self._room_name}") try: await self.room.connect( self._url, self._token, options=rtc.RoomOptions(auto_subscribe=True), ) self._connected = True # Increment disconnect counter if we successfully connected. self._disconnect_counter += 1 self._participant_id = self.room.local_participant.sid logger.info(f"Connected to {self._room_name}") # Set up audio source and track self._audio_source = rtc.AudioSource( self._out_sample_rate, self._params.audio_out_channels ) self._audio_track = rtc.LocalAudioTrack.create_audio_track( "pipecat-audio", self._audio_source ) options = rtc.TrackPublishOptions() options.source = rtc.TrackSource.SOURCE_MICROPHONE await self.room.local_participant.publish_track(self._audio_track, options) await self._callbacks.on_connected() # Check if there are already participants in the room participants = self.get_participants() if participants and not self._other_participant_has_joined: self._other_participant_has_joined = True await self._callbacks.on_first_participant_joined(participants[0]) except Exception as e: logger.error(f"Error connecting to {self._room_name}: {e}") raise async def disconnect(self): """Disconnect from the LiveKit room.""" async with self._async_lock: # Decrement leave counter when leaving. self._disconnect_counter -= 1 if not self._connected or self._disconnect_counter > 0: return logger.info(f"Disconnecting from {self._room_name}") await self._callbacks.on_before_disconnect() await self.room.disconnect() self._connected = False logger.info(f"Disconnected from {self._room_name}") await self._callbacks.on_disconnected() async def send_data(self, data: bytes, participant_id: Optional[str] = None): """Send data to participants in the room. Args: data: The data bytes to send. participant_id: Optional specific participant to send to. """ if not self._connected: return try: if participant_id: await self.room.local_participant.publish_data( data, reliable=True, destination_identities=[participant_id] ) else: await self.room.local_participant.publish_data(data, reliable=True) except Exception as e: logger.error(f"Error sending data: {e}") async def send_dtmf(self, digit: str): r"""Send DTMF tone to the room. Args: digit: The DTMF digit to send (0-9, \*, #). """ if not self._connected: return if digit not in DTMF_CODE_MAP: logger.warning(f"Invalid DTMF digit: {digit}") return code = DTMF_CODE_MAP[digit] try: await self.room.local_participant.publish_dtmf(code=code, digit=digit) except Exception as e: logger.error(f"Error sending DTMF tone {digit}: {e}") async def publish_audio(self, audio_frame: rtc.AudioFrame) -> bool: """Publish an audio frame to the room. Args: audio_frame: The LiveKit audio frame to publish. """ if not self._connected or not self._audio_source: return False try: await self._audio_source.capture_frame(audio_frame) return True except Exception as e: logger.error(f"Error publishing audio: {e}") return False def get_participants(self) -> List[str]: """Get list of participant IDs in the room. Returns: List of participant IDs. """ return [p.sid for p in self.room.remote_participants.values()] async def get_participant_metadata(self, participant_id: str) -> dict: """Get metadata for a specific participant. Args: participant_id: ID of the participant to get metadata for. Returns: Dictionary containing participant metadata. """ participant = self.room.remote_participants.get(participant_id) if participant: return { "id": participant.sid, "name": participant.name, "metadata": participant.metadata, "is_speaking": participant.is_speaking, } return {} async def set_participant_metadata(self, metadata: str): """Set metadata for the local participant. Args: metadata: Metadata string to set. """ await self.room.local_participant.set_metadata(metadata) async def mute_participant(self, participant_id: str): """Mute a specific participant's audio tracks. Args: participant_id: ID of the participant to mute. """ participant = self.room.remote_participants.get(participant_id) if participant: for track in participant.tracks.values(): if track.kind == "audio": await track.set_enabled(False) async def unmute_participant(self, participant_id: str): """Unmute a specific participant's audio tracks. Args: participant_id: ID of the participant to unmute. """ participant = self.room.remote_participants.get(participant_id) if participant: for track in participant.tracks.values(): if track.kind == "audio": await track.set_enabled(True) # Wrapper methods for event handlers def _on_participant_connected_wrapper(self, participant: rtc.RemoteParticipant): """Wrapper for participant connected events.""" self._task_manager.create_task( self._async_on_participant_connected(participant), f"{self}::_async_on_participant_connected", ) def _on_participant_disconnected_wrapper(self, participant: rtc.RemoteParticipant): """Wrapper for participant disconnected events.""" self._task_manager.create_task( self._async_on_participant_disconnected(participant), f"{self}::_async_on_participant_disconnected", ) def _on_track_subscribed_wrapper( self, track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant, ): """Wrapper for track subscribed events.""" self._task_manager.create_task( self._async_on_track_subscribed(track, publication, participant), f"{self}::_async_on_track_subscribed", ) def _on_track_unsubscribed_wrapper( self, track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant, ): """Wrapper for track unsubscribed events.""" self._task_manager.create_task( self._async_on_track_unsubscribed(track, publication, participant), f"{self}::_async_on_track_unsubscribed", ) def _on_data_received_wrapper(self, data: rtc.DataPacket): """Wrapper for data received events.""" self._task_manager.create_task( self._async_on_data_received(data), f"{self}::_async_on_data_received", ) def _on_connected_wrapper(self): """Wrapper for connected events.""" self._task_manager.create_task(self._async_on_connected(), f"{self}::_async_on_connected") def _on_disconnected_wrapper(self): """Wrapper for disconnected events.""" self._task_manager.create_task( self._async_on_disconnected(), f"{self}::_async_on_disconnected" ) # Async methods for event handling async def _async_on_participant_connected(self, participant: rtc.RemoteParticipant): """Handle participant connected events.""" logger.info(f"Participant connected: {participant.identity}") await self._callbacks.on_participant_connected(participant.sid) if not self._other_participant_has_joined: self._other_participant_has_joined = True await self._callbacks.on_first_participant_joined(participant.sid) async def _async_on_participant_disconnected(self, participant: rtc.RemoteParticipant): """Handle participant disconnected events.""" logger.info(f"Participant disconnected: {participant.identity}") await self._callbacks.on_participant_disconnected(participant.sid) if len(self.get_participants()) == 0: self._other_participant_has_joined = False async def _async_on_track_subscribed( self, track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant, ): """Handle track subscribed events.""" if track.kind == rtc.TrackKind.KIND_AUDIO: logger.info(f"Audio track subscribed: {track.sid} from participant {participant.sid}") self._audio_tracks[participant.sid] = track audio_stream = rtc.AudioStream(track) self._task_manager.create_task( self._process_audio_stream(audio_stream, participant.sid), f"{self}::_process_audio_stream", ) await self._callbacks.on_audio_track_subscribed(participant.sid) elif track.kind == rtc.TrackKind.KIND_VIDEO: logger.info(f"Video track subscribed: {track.sid} from participant {participant.sid}") self._video_tracks[participant.sid] = track # Only process video stream if video input is enabled to prevent # unbounded queue growth when there is no consumer for video frames. if self._params.video_in_enabled: video_stream = rtc.VideoStream(track) self._task_manager.create_task( self._process_video_stream(video_stream, participant.sid), f"{self}::_process_video_stream", ) await self._callbacks.on_video_track_subscribed(participant.sid) async def _async_on_track_unsubscribed( self, track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant, ): """Handle track unsubscribed events.""" logger.info(f"Track unsubscribed: {publication.sid} from {participant.identity}") if track.kind == rtc.TrackKind.KIND_AUDIO: await self._callbacks.on_audio_track_unsubscribed(participant.sid) elif track.kind == rtc.TrackKind.KIND_VIDEO: await self._callbacks.on_video_track_unsubscribed(participant.sid) async def _async_on_data_received(self, data: rtc.DataPacket): """Handle data received events.""" await self._callbacks.on_data_received(data.data, data.participant.sid) async def _async_on_connected(self): """Handle connected events.""" await self._callbacks.on_connected() async def _async_on_disconnected(self, reason=None): """Handle disconnected events.""" self._connected = False logger.info(f"Disconnected from {self._room_name}. Reason: {reason}") await self._callbacks.on_disconnected() async def _process_audio_stream(self, audio_stream: rtc.AudioStream, participant_id: str): """Process incoming audio stream from a participant.""" logger.info(f"Started processing audio stream for participant {participant_id}") async for event in audio_stream: if isinstance(event, rtc.AudioFrameEvent): await self._audio_queue.put((event, participant_id)) else: logger.warning(f"Received unexpected event type: {type(event)}") async def get_next_audio_frame(self): """Get the next audio frame from the queue.""" while True: frame, participant_id = await self._audio_queue.get() yield frame, participant_id async def _process_video_stream(self, video_stream: rtc.VideoStream, participant_id: str): """Process incoming video stream from a participant.""" logger.info(f"Started processing video stream for participant {participant_id}") async for event in video_stream: if isinstance(event, rtc.VideoFrameEvent): await self._video_queue.put((event, participant_id)) else: logger.warning(f"Received unexpected event type: {type(event)}") async def get_next_video_frame(self): """Get the next video frame from the queue.""" while True: frame, participant_id = await self._video_queue.get() yield frame, participant_id def __str__(self): """String representation of the LiveKit transport client.""" return f"{self._transport_name}::LiveKitTransportClient" class LiveKitInputTransport(BaseInputTransport): """Handles incoming media streams and events from LiveKit rooms. Processes incoming audio streams from room participants and forwards them as Pipecat frames, including audio resampling and VAD integration. """ def __init__( self, transport: BaseTransport, client: LiveKitTransportClient, params: LiveKitParams, **kwargs, ): """Initialize the LiveKit input transport. Args: transport: The parent transport instance. client: LiveKitTransportClient instance. params: Configuration parameters. **kwargs: Additional arguments passed to parent class. """ super().__init__(params, **kwargs) self._transport = transport self._client = client self._audio_in_task = None self._video_in_task = None self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer self._resampler = create_stream_resampler() # Whether we have seen a StartFrame already. self._initialized = False @property def vad_analyzer(self) -> Optional[VADAnalyzer]: """Get the Voice Activity Detection analyzer. Returns: The VAD analyzer instance if configured. """ return self._vad_analyzer async def start(self, frame: StartFrame): """Start the input transport and connect to LiveKit room. Args: frame: The start frame containing initialization parameters. """ await super().start(frame) if self._initialized: return self._initialized = True await self._client.start(frame) await self._client.connect() if not self._audio_in_task and self._params.audio_in_enabled: self._audio_in_task = self.create_task(self._audio_in_task_handler()) if not self._video_in_task and self._params.video_in_enabled: self._video_in_task = self.create_task(self._video_in_task_handler()) await self.set_transport_ready(frame) logger.info("LiveKitInputTransport started") async def stop(self, frame: EndFrame): """Stop the input transport and disconnect from LiveKit room. Args: frame: The end frame signaling transport shutdown. """ await super().stop(frame) await self._client.disconnect() if self._audio_in_task: await self.cancel_task(self._audio_in_task) if self._video_in_task: await self.cancel_task(self._video_in_task) logger.info("LiveKitInputTransport stopped") async def cancel(self, frame: CancelFrame): """Cancel the input transport and disconnect from LiveKit room. Args: frame: The cancel frame signaling immediate cancellation. """ await super().cancel(frame) await self._client.disconnect() if self._audio_in_task and self._params.audio_in_enabled: await self.cancel_task(self._audio_in_task) if self._video_in_task and self._params.video_in_enabled: await self.cancel_task(self._video_in_task) async def setup(self, setup: FrameProcessorSetup): """Setup the input transport with shared client setup. Args: setup: The frame processor setup configuration. """ await super().setup(setup) await self._client.setup(setup) async def cleanup(self): """Cleanup input transport and shared resources.""" await super().cleanup() await self._transport.cleanup() async def push_app_message(self, message: Any, sender: str): """Push an application message as an urgent transport frame. Args: message: The message data to send. sender: ID of the message sender. """ frame = LiveKitOutputTransportMessageUrgentFrame(message=message, participant_id=sender) await self.push_frame(frame) async def _audio_in_task_handler(self): """Handle incoming audio frames from participants.""" logger.info("Audio input task started") audio_iterator = self._client.get_next_audio_frame() async for audio_data in audio_iterator: if audio_data: audio_frame_event, participant_id = audio_data pipecat_audio_frame = await self._convert_livekit_audio_to_pipecat( audio_frame_event ) # Skip frames with no audio data if len(pipecat_audio_frame.audio) == 0: continue input_audio_frame = UserAudioRawFrame( user_id=participant_id, audio=pipecat_audio_frame.audio, sample_rate=pipecat_audio_frame.sample_rate, num_channels=pipecat_audio_frame.num_channels, ) await self.push_audio_frame(input_audio_frame) async def _video_in_task_handler(self): """Handle incoming video frames from participants.""" logger.info("Video input task started") video_iterator = self._client.get_next_video_frame() async for video_data in video_iterator: if video_data: video_frame_event, participant_id = video_data pipecat_video_frame = await self._convert_livekit_video_to_pipecat( video_frame_event=video_frame_event ) # Skip frames with no video data if len(pipecat_video_frame.image) == 0: continue input_video_frame = UserImageRawFrame( user_id=participant_id, image=pipecat_video_frame.image, size=pipecat_video_frame.size, format=pipecat_video_frame.format, ) await self.push_video_frame(input_video_frame) async def _convert_livekit_audio_to_pipecat( self, audio_frame_event: rtc.AudioFrameEvent ) -> AudioRawFrame: """Convert LiveKit audio frame to Pipecat audio frame.""" audio_frame = audio_frame_event.frame audio_data = await self._resampler.resample( audio_frame.data.tobytes(), audio_frame.sample_rate, self.sample_rate ) return AudioRawFrame( audio=audio_data, sample_rate=self.sample_rate, num_channels=audio_frame.num_channels, ) async def _convert_livekit_video_to_pipecat( self, video_frame_event: rtc.VideoFrameEvent, ) -> ImageRawFrame: """Convert LiveKit video frame to Pipecat video frame.""" rgb_frame = video_frame_event.frame.convert(proto_video_frame.VideoBufferType.RGB24) image_frame = ImageRawFrame( image=rgb_frame.data, size=(rgb_frame.width, rgb_frame.height), format="RGB", ) return image_frame class LiveKitOutputTransport(BaseOutputTransport): """Handles outgoing media streams and events to LiveKit rooms. Manages sending audio frames and data messages to LiveKit room participants, including audio format conversion for LiveKit compatibility. """ def __init__( self, transport: BaseTransport, client: LiveKitTransportClient, params: LiveKitParams, **kwargs, ): """Initialize the LiveKit output transport. Args: transport: The parent transport instance. client: LiveKitTransportClient instance. params: Configuration parameters. **kwargs: Additional arguments passed to parent class. """ super().__init__(params, **kwargs) self._transport = transport self._client = client # Whether we have seen a StartFrame already. self._initialized = False async def start(self, frame: StartFrame): """Start the output transport and connect to LiveKit room. Args: frame: The start frame containing initialization parameters. """ await super().start(frame) if self._initialized: return self._initialized = True await self._client.start(frame) await self._client.connect() await self.set_transport_ready(frame) logger.info("LiveKitOutputTransport started") async def stop(self, frame: EndFrame): """Stop the output transport and disconnect from LiveKit room. Args: frame: The end frame signaling transport shutdown. """ await super().stop(frame) await self._client.disconnect() logger.info("LiveKitOutputTransport stopped") async def cancel(self, frame: CancelFrame): """Cancel the output transport and disconnect from LiveKit room. Args: frame: The cancel frame signaling immediate cancellation. """ await super().cancel(frame) await self._client.disconnect() async def setup(self, setup: FrameProcessorSetup): """Setup the output transport with shared client setup. Args: setup: The frame processor setup configuration. """ await super().setup(setup) await self._client.setup(setup) async def cleanup(self): """Cleanup output transport and shared resources.""" await super().cleanup() await self._transport.cleanup() async def send_message( self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame ): """Send a transport message to participants. Args: frame: The transport message frame to send. """ message = frame.message if isinstance(message, dict): # fix message encoding for dict-like messages, e.g. RTVI messages. message = json.dumps(message, ensure_ascii=False) if isinstance( frame, (LiveKitOutputTransportMessageFrame, LiveKitOutputTransportMessageUrgentFrame) ): await self._client.send_data(message.encode(), frame.participant_id) else: await self._client.send_data(message.encode()) async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool: """Write an audio frame to the LiveKit room. Args: frame: The audio frame to write. Returns: True if the audio frame was written successfully, False otherwise. """ livekit_audio = self._convert_pipecat_audio_to_livekit(frame.audio) return await self._client.publish_audio(livekit_audio) def _supports_native_dtmf(self) -> bool: """LiveKit supports native DTMF via telephone events. Returns: True, as LiveKit supports native DTMF transmission. """ return True async def _write_dtmf_native(self, frame: OutputDTMFFrame | OutputDTMFUrgentFrame): """Use LiveKit's native publish_dtmf method for telephone events. Args: frame: The DTMF frame to write. """ await self._client.send_dtmf(frame.button.value) def _convert_pipecat_audio_to_livekit(self, pipecat_audio: bytes) -> rtc.AudioFrame: """Convert Pipecat audio data to LiveKit audio frame.""" bytes_per_sample = 2 # Assuming 16-bit audio total_samples = len(pipecat_audio) // bytes_per_sample samples_per_channel = total_samples // self._params.audio_out_channels return rtc.AudioFrame( data=pipecat_audio, sample_rate=self.sample_rate, num_channels=self._params.audio_out_channels, samples_per_channel=samples_per_channel, ) class LiveKitTransport(BaseTransport): """Transport implementation for LiveKit real-time communication. Provides comprehensive LiveKit integration including audio streaming, data messaging, participant management, and room event handling for conversational AI applications. Event handlers available: - on_connected: Called when the bot connects to the room. - on_disconnected: Called when the bot disconnects from the room. - on_before_disconnect: [sync] Called just before the bot disconnects. - on_call_state_updated: Called when the call state changes. Args: (state: str) - on_first_participant_joined: Called when the first participant joins. Args: (participant_id: str) - on_participant_connected: Called when a participant connects. Args: (participant_id: str) - on_participant_disconnected: Called when a participant disconnects. Args: (participant_id: str) - on_participant_left: Called when a participant leaves. Args: (participant_id: str, reason: str) - on_audio_track_subscribed: Called when an audio track is subscribed. Args: (participant_id: str) - on_audio_track_unsubscribed: Called when an audio track is unsubscribed. Args: (participant_id: str) - on_video_track_subscribed: Called when a video track is subscribed. Args: (participant_id: str) - on_video_track_unsubscribed: Called when a video track is unsubscribed. Args: (participant_id: str) - on_data_received: Called when data is received from a participant. Args: (data: bytes, participant_id: str) Example:: @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant_id): await task.queue_frame(TTSSpeakFrame("Hello!")) @transport.event_handler("on_participant_disconnected") async def on_participant_disconnected(transport, participant_id): await task.queue_frame(EndFrame()) """ def __init__( self, url: str, token: str, room_name: str, params: Optional[LiveKitParams] = None, input_name: Optional[str] = None, output_name: Optional[str] = None, ): """Initialize the LiveKit transport. Args: url: LiveKit server URL to connect to. token: Authentication token for the room. room_name: Name of the LiveKit room to join. params: Configuration parameters for the transport. input_name: Optional name for the input transport. output_name: Optional name for the output transport. """ super().__init__(input_name=input_name, output_name=output_name) callbacks = LiveKitCallbacks( on_connected=self._on_connected, on_disconnected=self._on_disconnected, on_before_disconnect=self._on_before_disconnect, on_participant_connected=self._on_participant_connected, on_participant_disconnected=self._on_participant_disconnected, on_audio_track_subscribed=self._on_audio_track_subscribed, on_audio_track_unsubscribed=self._on_audio_track_unsubscribed, on_video_track_subscribed=self._on_video_track_subscribed, on_video_track_unsubscribed=self._on_video_track_unsubscribed, on_data_received=self._on_data_received, on_first_participant_joined=self._on_first_participant_joined, ) self._params = params or LiveKitParams() self._client = LiveKitTransportClient( url, token, room_name, self._params, callbacks, self.name ) self._input: Optional[LiveKitInputTransport] = None self._output: Optional[LiveKitOutputTransport] = None self._register_event_handler("on_connected") self._register_event_handler("on_disconnected") self._register_event_handler("on_participant_connected") self._register_event_handler("on_participant_disconnected") self._register_event_handler("on_audio_track_subscribed") self._register_event_handler("on_audio_track_unsubscribed") self._register_event_handler("on_video_track_subscribed") self._register_event_handler("on_video_track_unsubscribed") self._register_event_handler("on_data_received") self._register_event_handler("on_first_participant_joined") self._register_event_handler("on_participant_left") self._register_event_handler("on_call_state_updated") self._register_event_handler("on_before_disconnect", sync=True) def input(self) -> LiveKitInputTransport: """Get the input transport for receiving media and events. Returns: The LiveKit input transport instance. """ if not self._input: self._input = LiveKitInputTransport( self, self._client, self._params, name=self._input_name ) return self._input def output(self) -> LiveKitOutputTransport: """Get the output transport for sending media and events. Returns: The LiveKit output transport instance. """ if not self._output: self._output = LiveKitOutputTransport( self, self._client, self._params, name=self._output_name ) return self._output @property def participant_id(self) -> str: """Get the participant ID for this transport. Returns: The participant ID assigned by LiveKit. """ return self._client.participant_id async def send_audio(self, frame: OutputAudioRawFrame): """Send an audio frame to the LiveKit room. Args: frame: The audio frame to send. """ if self._output: await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM) def get_participants(self) -> List[str]: """Get list of participant IDs in the room. Returns: List of participant IDs. """ return self._client.get_participants() async def get_participant_metadata(self, participant_id: str) -> dict: """Get metadata for a specific participant. Args: participant_id: ID of the participant to get metadata for. Returns: Dictionary containing participant metadata. """ return await self._client.get_participant_metadata(participant_id) async def set_metadata(self, metadata: str): """Set metadata for the local participant. Args: metadata: Metadata string to set. """ await self._client.set_participant_metadata(metadata) async def mute_participant(self, participant_id: str): """Mute a specific participant's audio tracks. Args: participant_id: ID of the participant to mute. """ await self._client.mute_participant(participant_id) async def unmute_participant(self, participant_id: str): """Unmute a specific participant's audio tracks. Args: participant_id: ID of the participant to unmute. """ await self._client.unmute_participant(participant_id) async def _on_connected(self): """Handle room connected events.""" await self._call_event_handler("on_connected") if self._input: await self._input.push_frame(BotConnectedFrame()) async def _on_disconnected(self): """Handle room disconnected events.""" await self._call_event_handler("on_disconnected") async def _on_before_disconnect(self): """Handle before disconnection room events.""" await self._call_event_handler("on_before_disconnect") async def _on_participant_connected(self, participant_id: str): """Handle participant connected events.""" await self._call_event_handler("on_participant_connected", participant_id) if self._input: await self._input.push_frame(ClientConnectedFrame()) async def _on_participant_disconnected(self, participant_id: str): """Handle participant disconnected events.""" await self._call_event_handler("on_participant_disconnected", participant_id) await self._call_event_handler("on_participant_left", participant_id, "disconnected") async def _on_audio_track_subscribed(self, participant_id: str): """Handle audio track subscribed events.""" await self._call_event_handler("on_audio_track_subscribed", participant_id) participant = self._client.room.remote_participants.get(participant_id) if participant: for publication in participant.audio_tracks.values(): self._client._on_track_subscribed_wrapper( publication.track, publication, participant ) async def _on_audio_track_unsubscribed(self, participant_id: str): """Handle audio track unsubscribed events.""" await self._call_event_handler("on_audio_track_unsubscribed", participant_id) async def _on_video_track_subscribed(self, participant_id: str): """Handle video track subscribed events.""" await self._call_event_handler("on_video_track_subscribed", participant_id) participant = self._client.room.remote_participants.get(participant_id) if participant: for publication in participant.video_tracks.values(): self._client._on_track_subscribed_wrapper( publication.track, publication, participant ) async def _on_video_track_unsubscribed(self, participant_id: str): """Handle video track unsubscribed events.""" await self._call_event_handler("on_video_track_unsubscribed", participant_id) async def _on_data_received(self, data: bytes, participant_id: str): """Handle data received events.""" if self._input: await self._input.push_app_message(data.decode(), participant_id) await self._call_event_handler("on_data_received", data, participant_id) async def send_message(self, message: str, participant_id: Optional[str] = None): """Send a message to participants in the room. Args: message: The message string to send. participant_id: Optional specific participant to send to. """ if self._output: frame = LiveKitOutputTransportMessageFrame( message=message, participant_id=participant_id ) await self._output.send_message(frame) async def send_message_urgent(self, message: str, participant_id: Optional[str] = None): """Send an urgent message to participants in the room. Args: message: The urgent message string to send. participant_id: Optional specific participant to send to. """ if self._output: frame = LiveKitOutputTransportMessageUrgentFrame( message=message, participant_id=participant_id ) await self._output.send_message(frame) async def on_room_event(self, event): """Handle room events. Args: event: The room event to handle. """ # Handle room events pass async def on_participant_event(self, event): """Handle participant events. Args: event: The participant event to handle. """ # Handle participant events pass async def on_track_event(self, event): """Handle track events. Args: event: The track event to handle. """ # Handle track events pass async def _on_call_state_updated(self, state: str): """Handle call state update events.""" await self._call_event_handler("on_call_state_updated", state) async def _on_first_participant_joined(self, participant_id: str): """Handle first participant joined events.""" await self._call_event_handler("on_first_participant_joined", participant_id)