+
+
+
+
+
+
+
+
+
+ Disconnected
+ + + + + + diff --git a/examples/p2p-webrtc/voice-agent/requirements.txt b/examples/p2p-webrtc/voice-agent/requirements.txt new file mode 100644 index 000000000..d8ffd53ef --- /dev/null +++ b/examples/p2p-webrtc/voice-agent/requirements.txt @@ -0,0 +1,5 @@ +python-dotenv +fastapi[all] +uvicorn +aiortc +pipecat-ai[google,silero] \ No newline at end of file diff --git a/examples/p2p-webrtc/voice-agent/server.py b/examples/p2p-webrtc/voice-agent/server.py new file mode 100644 index 000000000..3e5f8dcfb --- /dev/null +++ b/examples/p2p-webrtc/voice-agent/server.py @@ -0,0 +1,81 @@ +import argparse +import asyncio +import logging +from contextlib import asynccontextmanager +from typing import Dict + +import uvicorn +from bot import run_bot +from dotenv import load_dotenv +from fastapi import BackgroundTasks, FastAPI +from fastapi.responses import FileResponse + +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +# Load environment variables +load_dotenv(override=True) + +logger = logging.getLogger("pc") + +app = FastAPI() + +# Store connections by pc_id +pcs_map: Dict[str, SmallWebRTCConnection] = {} + + +@app.post("/api/offer") +async def offer(request: dict, background_tasks: BackgroundTasks): + pc_id = request.get("pc_id") + + if pc_id and pc_id in pcs_map: + pipecat_connection = pcs_map[pc_id] + logger.info(f"Reusing existing connection for pc_id: {pc_id}") + await pipecat_connection.renegotiate(sdp=request["sdp"], type=request["type"]) + else: + pipecat_connection = SmallWebRTCConnection() + await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"]) + + @pipecat_connection.on("closed") + async def handle_disconnected(webrtc_connection: SmallWebRTCConnection): + logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}") + pcs_map.pop(webrtc_connection.pc_id, None) + + background_tasks.add_task(run_bot, pipecat_connection) + + answer = pipecat_connection.get_answer() + # Updating the peer connection inside the map + pcs_map[answer["pc_id"]] = pipecat_connection + + return answer + + +@app.get("/") +async def serve_index(): + return FileResponse("index.html") + + +@asynccontextmanager +async def lifespan(app: FastAPI): + yield # Run app + coros = [pc.close() for pc in pcs_map.values()] + await asyncio.gather(*coros) + pcs_map.clear() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="WebRTC demo") + parser.add_argument( + "--host", default="localhost", help="Host for HTTP server (default: localhost)" + ) + parser.add_argument( + "--port", type=int, default=7860, help="Port for HTTP server (default: 7860)" + ) + parser.add_argument("--verbose", "-v", action="count") + args = parser.parse_args() + + if args.verbose: + logging.basicConfig(level=logging.DEBUG) + else: + logging.basicConfig(level=logging.INFO) + + uvicorn.run(app, host=args.host, port=args.port) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 801af46b4..965648ada 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -7,6 +7,7 @@ import asyncio import base64 import json +import time from dataclasses import dataclass from enum import Enum from typing import Any, Dict, List, Mapping, Optional, Union @@ -177,6 +178,7 @@ class GeminiMultimodalLiveLLMService(LLMService): **kwargs, ): super().__init__(base_url=base_url, **kwargs) + self._last_sent_time = 0 self.api_key = api_key self.base_url = base_url self.set_model_name(model) @@ -548,7 +550,13 @@ class GeminiMultimodalLiveLLMService(LLMService): async def _send_user_video(self, frame): if self._video_input_paused: return - # logger.debug(f"Sending video frame to Gemini: {frame}") + + now = time.time() + if now - self._last_sent_time < 1: + return # Ignore if less than 1 second has passed + + self._last_sent_time = now # Update last sent time + logger.debug(f"Sending video frame to Gemini: {frame}") evt = events.VideoInputMessage.from_image_frame(frame) await self.send_client_event(evt) diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 06f6cb920..30e126ae3 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -19,6 +19,7 @@ from pipecat.utils.base_object import BaseObject class TransportParams(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) + camera_in_enabled: bool = False camera_out_enabled: bool = False camera_out_is_live: bool = False camera_out_width: int = 1024 diff --git a/src/pipecat/transports/network/small_webrtc.py b/src/pipecat/transports/network/small_webrtc.py new file mode 100644 index 000000000..bb3044a75 --- /dev/null +++ b/src/pipecat/transports/network/small_webrtc.py @@ -0,0 +1,512 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import fractions +import logging +import time +from collections import deque +from typing import Any, Awaitable, Callable, Optional + +import cv2 +import numpy as np +from aiortc import VideoStreamTrack +from aiortc.mediastreams import AudioStreamTrack, MediaStreamError, VideoFrame +from av import AudioFrame, AudioResampler +from loguru import logger +from pydantic import BaseModel + +# Get the logger for aiortc +# aiortc_logger = logging.getLogger("aiortc") +# aiortc_logger.setLevel(logging.DEBUG) +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + InputAudioRawFrame, + InputImageRawFrame, + OutputImageRawFrame, + StartFrame, + TransportMessageFrame, + TransportMessageUrgentFrame, +) +from pipecat.transports.base_input import BaseInputTransport +from pipecat.transports.base_output import BaseOutputTransport +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + + +class SmallWebRTCCallbacks(BaseModel): + on_app_message: Callable[[Any], Awaitable[None]] + on_client_connected: Callable[[SmallWebRTCConnection], Awaitable[None]] + on_client_disconnected: Callable[[SmallWebRTCConnection], Awaitable[None]] + on_client_closed: Callable[[SmallWebRTCConnection], Awaitable[None]] + + +class RawAudioTrack(AudioStreamTrack): + def __init__(self, sample_rate): + super().__init__() + self._sample_rate = sample_rate + self._samples_per_frame = self._sample_rate // 50 # 20ms per frame + self._timestamp = 0 + self._audio_buffer = deque() + self._start = time.time() + + def add_audio_bytes(self, audio_bytes: bytes): + """ + Adds bytes to the audio buffer and returns a Future that completes when the data is processed. + """ + if len(audio_bytes) % 2 != 0: + raise ValueError("Audio bytes length must be even (16-bit samples).") + future = asyncio.get_running_loop().create_future() + self._audio_buffer.append((audio_bytes, future)) + return future + + async def recv(self): + """ + Returns the next audio frame, generating silence if needed. + """ + # Compute required wait time for synchronization + if self._timestamp > 0: + wait = self._start + (self._timestamp / self._sample_rate) - time.time() + if wait > 0: + await asyncio.sleep(wait) + + # Check if we have enough data + needed_bytes = self._samples_per_frame * 2 # 16-bit (2 bytes per sample) + available_bytes = sum(len(audio_bytes) for audio_bytes, _ in self._audio_buffer) + consumed_futures = [] # Track futures for processed data + if available_bytes >= needed_bytes: + # Extract data from deque + chunk = bytearray() + while len(chunk) < needed_bytes: + audio_bytes, future = self._audio_buffer.popleft() + chunk.extend(audio_bytes) + consumed_futures.append(future) # Track the future + chunk = bytes(chunk[:needed_bytes]) # Trim excess bytes + else: + chunk = bytes(needed_bytes) # Generate silent frame + + # Convert the byte data to an ndarray of int16 samples + samples = np.frombuffer(chunk, dtype=np.int16) + + # Create AudioFrame + frame = AudioFrame.from_ndarray(samples[None, :], layout="mono") + self._timestamp += self._samples_per_frame + frame.pts = self._timestamp + frame.sample_rate = self._sample_rate + frame.time_base = fractions.Fraction(1, self._sample_rate) + + # Resolve all futures corresponding to consumed data + for future in consumed_futures: + if not future.done(): + future.set_result(True) + + return frame + + +class RawVideoTrack(VideoStreamTrack): + def __init__(self, width, height): + super().__init__() + self._width = width + self._height = height + self._video_buffer = asyncio.Queue() + + def add_video_frame(self, frame): + """Adds a raw video frame to the buffer.""" + self._video_buffer.put_nowait(frame) + + async def recv(self): + """Returns the next video frame, waiting if the buffer is empty.""" + raw_frame = await self._video_buffer.get() + + # Convert bytes to NumPy array + frame_data = np.frombuffer(raw_frame.image, dtype=np.uint8).reshape( + (self._height, self._width, 3) + ) + + frame = VideoFrame.from_ndarray(frame_data, format="rgb24") + + # Assign timestamp + frame.pts, frame.time_base = await self.next_timestamp() + + return frame + + +class SmallWebRTCClient: + def __init__(self, webrtc_connection: SmallWebRTCConnection, callbacks: SmallWebRTCCallbacks): + self._webrtcConnection = webrtc_connection + self._closing = False + self._callbacks = callbacks + + self._audio_output_track = None + self._video_output_track = None + self._audio_input_track: Optional[AudioStreamTrack] = None + self._video_input_track: Optional[VideoStreamTrack] = None + + self._params = None + self._audio_in_channels = None + self._in_sample_rate = None + self._out_sample_rate = None + + # We are always resampling it for 16000 if the sample_rate that we receive is bigger than that. + # otherwise we face issues with Silero VAD + self._pipecat_resampler = AudioResampler("s16", "mono", 16000) + + @self._webrtcConnection.on("connected") + async def on_connected(connection: SmallWebRTCConnection): + logger.info("Peer connection established.") + await self._handle_client_connected() + + @self._webrtcConnection.on("disconnected") + async def on_disconnected(connection: SmallWebRTCConnection): + logger.info("Peer connection lost.") + await self._handle_client_disconnected() + + @self._webrtcConnection.on("closed") + async def on_closed(connection: SmallWebRTCConnection): + logger.info("Client connection closed.") + await self._handle_client_closed() + + @self._webrtcConnection.on("appMessage") + async def on_app_message(message: Any): + await self._handle_app_message(message) + + async def read_video_frame(self): + """ + Reads a video frame from the given MediaStreamTrack, converts it to RGB, + and creates an InputImageRawFrame. + """ + while True: + if self._video_input_track is None: + await asyncio.sleep(0.01) + continue + + try: + frame = await asyncio.wait_for(self._video_input_track.recv(), timeout=2.0) + except asyncio.TimeoutError: + if self._webrtcConnection.is_connected(): + logger.warning("Timeout: No video frame received within the specified time.") + # self._webrtcConnection.ask_to_renegotiate() + frame = None + except MediaStreamError: + logger.warning("Received an unexpected media stream error while reading the audio.") + frame = None + + if frame is None or not isinstance(frame, VideoFrame): + # If no valid frame, sleep for a bit + await asyncio.sleep(0.01) + continue + + format_name = frame.format.name + + # Convert frame to NumPy array in its native format + frame_array = frame.to_ndarray(format=format_name) + + # Handle different formats dynamically + if format_name == "yuv420p": + frame_rgb = cv2.cvtColor(frame_array, cv2.COLOR_YUV2RGB_I420) + elif format_name == "nv12": + frame_rgb = cv2.cvtColor(frame_array, cv2.COLOR_YUV2RGB_NV12) + elif format_name == "gray": + frame_rgb = cv2.cvtColor(frame_array, cv2.COLOR_GRAY2RGB) + elif format_name.startswith("rgb"): # Already RGB, no conversion needed + frame_rgb = frame_array + else: + raise ValueError(f"Unsupported format: {format_name}") + + image_frame = InputImageRawFrame( + image=frame_rgb.tobytes(), + size=(frame.width, frame.height), + format="RGB", + ) + + yield image_frame + + async def read_audio_frame(self): + """ + Reads 20ms of audio from the given MediaStreamTrack and creates an InputAudioRawFrame. + """ + while True: + if self._audio_input_track is None: + await asyncio.sleep(0.01) + continue + + try: + frame = await asyncio.wait_for(self._audio_input_track.recv(), timeout=2.0) + except asyncio.TimeoutError: + if self._webrtcConnection.is_connected(): + logger.warning("Timeout: No audio frame received within the specified time.") + frame = None + except MediaStreamError: + logger.warning("Received an unexpected media stream error while reading the audio.") + frame = None + + if frame is None or not isinstance(frame, AudioFrame): + # If we don't read any audio let's sleep for a little bit (i.e. busy wait). + await asyncio.sleep(0.01) + continue + + if frame.sample_rate > self._in_sample_rate: + resampled_frames = self._pipecat_resampler.resample(frame) + for resampled_frame in resampled_frames: + # 16-bit PCM bytes + pcm_bytes = resampled_frame.to_ndarray().astype(np.int16).tobytes() + audio_frame = InputAudioRawFrame( + audio=pcm_bytes, + sample_rate=resampled_frame.sample_rate, + num_channels=self._audio_in_channels, + ) + yield audio_frame + else: + # 16-bit PCM bytes + pcm_bytes = frame.to_ndarray().astype(np.int16).tobytes() + audio_frame = InputAudioRawFrame( + audio=pcm_bytes, + sample_rate=frame.sample_rate, + num_channels=self._audio_in_channels, + ) + yield audio_frame + + async def write_raw_audio_frames(self, data: bytes): + if self._can_send() and self._audio_output_track: + await self._audio_output_track.add_audio_bytes(data) + + async def write_frame_to_camera(self, frame: OutputImageRawFrame): + if self._can_send() and self._video_output_track: + self._video_output_track.add_video_frame(frame) + + async def setup(self, _params: TransportParams, frame): + self._audio_in_channels = _params.audio_in_channels + self._in_sample_rate = _params.audio_in_sample_rate or frame.audio_in_sample_rate + self._out_sample_rate = _params.audio_out_sample_rate or frame.audio_out_sample_rate + self._params = _params + + async def connect(self): + if self._webrtcConnection.is_connected(): + # already initialized + return + + logger.info(f"Connecting to Small WebRTC") + await self._webrtcConnection.connect() + + async def disconnect(self): + if self.is_connected and not self.is_closing: + logger.info(f"Disconnecting to Small WebRTC") + self._closing = True + await self._webrtcConnection.close() + await self._handle_client_disconnected() + + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + if self._can_send(): + self._webrtcConnection.send_app_message(frame.message) + + async def _handle_client_connected(self): + # There is nothing to do here yet, the pipeline is still not ready + if not self._params: + return + + self._audio_input_track = self._webrtcConnection.audio_input_track() + self._video_input_track = self._webrtcConnection.video_input_track() + if self._params.audio_out_enabled: + self._audio_output_track = RawAudioTrack(sample_rate=self._out_sample_rate) + self._webrtcConnection.replace_audio_track(self._audio_output_track) + + if self._params.camera_out_enabled: + self._video_output_track = RawVideoTrack( + width=self._params.camera_out_width, height=self._params.camera_out_height + ) + self._webrtcConnection.replace_video_track(self._video_output_track) + + await self._callbacks.on_client_connected(self._webrtcConnection) + + async def _handle_client_disconnected(self): + self._audio_input_track = None + self._video_input_track = None + self._audio_output_track = None + self._video_output_track = None + await self._callbacks.on_client_disconnected(self._webrtcConnection) + + async def _handle_client_closed(self): + self._audio_input_track = None + self._video_input_track = None + self._audio_output_track = None + self._video_output_track = None + await self._callbacks.on_client_closed(self._webrtcConnection) + + async def _handle_app_message(self, message: Any): + await self._callbacks.on_app_message(message) + + def _can_send(self): + return self.is_connected and not self.is_closing + + @property + def is_connected(self) -> bool: + return self._webrtcConnection.is_connected() + + @property + def is_closing(self) -> bool: + return self._closing + + +class SmallWebRTCInputTransport(BaseInputTransport): + def __init__( + self, + client: SmallWebRTCClient, + params: TransportParams, + **kwargs, + ): + super().__init__(params, **kwargs) + self._client = client + self._params = params + self._receive_audio_task = None + self._receive_video_task = None + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._client.setup(self._params, frame) + await self._client.connect() + if not self._receive_audio_task and ( + self._params.audio_in_enabled or self._params.vad_enabled + ): + self._receive_audio_task = self.create_task(self._receive_audio()) + if not self._receive_video_task and self._params.camera_in_enabled: + self._receive_video_task = self.create_task(self._receive_video()) + + async def _stop_tasks(self): + if self._receive_audio_task: + await self.cancel_task(self._receive_audio_task) + self._receive_audio_task = None + if self._receive_video_task: + await self.cancel_task(self._receive_video_task) + self._receive_video_task = None + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._stop_tasks() + await self._client.disconnect() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._stop_tasks() + await self._client.disconnect() + + async def _receive_audio(self): + try: + async for audio_frame in self._client.read_audio_frame(): + if audio_frame: + await self.push_audio_frame(audio_frame) + + except Exception as e: + logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})") + + async def _receive_video(self): + try: + async for video_frame in self._client.read_video_frame(): + if video_frame: + await self.push_frame(video_frame) + + except Exception as e: + logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})") + + async def push_app_message(self, message: Any): + logger.info(f"Received app message inside SmallWebRTCInputTransport {message}") + frame = TransportMessageUrgentFrame(message=message) + await self.push_frame(frame) + + +class SmallWebRTCOutputTransport(BaseOutputTransport): + def __init__( + self, + client: SmallWebRTCClient, + params: TransportParams, + **kwargs, + ): + super().__init__(params, **kwargs) + self._client = client + self._params = params + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._client.setup(self._params, frame) + await self._client.connect() + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._client.disconnect() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._client.disconnect() + + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + await self._client.send_message(frame) + + async def write_raw_audio_frames(self, frames: bytes): + await self._client.write_raw_audio_frames(frames) + + async def write_frame_to_camera(self, frame: OutputImageRawFrame): + await self._client.write_frame_to_camera(frame) + + +class SmallWebRTCTransport(BaseTransport): + def __init__( + self, + webrtc_connection: SmallWebRTCConnection, + params: TransportParams, + input_name: Optional[str] = None, + output_name: Optional[str] = None, + ): + super().__init__(input_name=input_name, output_name=output_name) + self._params = params + + self._callbacks = SmallWebRTCCallbacks( + on_app_message=self._on_app_message, + on_client_connected=self._on_client_connected, + on_client_disconnected=self._on_client_disconnected, + on_client_closed=self._on_client_closed, + ) + + self._client = SmallWebRTCClient(webrtc_connection, self._callbacks) + + self._input = SmallWebRTCInputTransport(self._client, self._params, name=self._input_name) + self._output = SmallWebRTCOutputTransport( + self._client, self._params, name=self._output_name + ) + + # Register supported handlers. The user will only be able to register + # these handlers. + self._register_event_handler("on_app_message") + self._register_event_handler("on_client_connected") + self._register_event_handler("on_client_disconnected") + self._register_event_handler("on_client_closed") + + def input(self) -> SmallWebRTCInputTransport: + if not self._input: + self._input = SmallWebRTCInputTransport( + self._client, self._params, name=self._input_name + ) + return self._input + + def output(self) -> SmallWebRTCOutputTransport: + if not self._output: + self._output = SmallWebRTCOutputTransport( + self._client, self._params, name=self._input_name + ) + return self._output + + async def _on_app_message(self, message: Any): + if self._input: + await self._input.push_app_message(message) + await self._call_event_handler("on_app_message", message) + + async def _on_client_connected(self, webrtc_connection): + await self._call_event_handler("on_client_connected", webrtc_connection) + + async def _on_client_disconnected(self, webrtc_connection): + await self._call_event_handler("on_client_disconnected", webrtc_connection) + + async def _on_client_closed(self, webrtc_connection): + await self._call_event_handler("on_client_closed", webrtc_connection) diff --git a/src/pipecat/transports/network/webrtc_connection.py b/src/pipecat/transports/network/webrtc_connection.py new file mode 100644 index 000000000..e18879add --- /dev/null +++ b/src/pipecat/transports/network/webrtc_connection.py @@ -0,0 +1,246 @@ +import asyncio +import json +import time +import uuid +from enum import Enum +from typing import Any, Optional + +from aiortc import RTCConfiguration, RTCIceServer, RTCPeerConnection, RTCSessionDescription +from loguru import logger + +from pipecat.utils.event_emitter import EventEmitter + +SIGNALLING_TYPE = "signalling" + + +class SignallingMessage(Enum): + RENEGOTIATE = "renegotiate" + + +class SmallWebRTCConnection(EventEmitter): + def __init__(self, ice_servers=None): + super().__init__() + if ice_servers: + self.ice_servers = [RTCIceServer(urls=server) for server in ice_servers] + else: + self.ice_servers = [] + self._connect_invoked = False + self._initialize() + + def _initialize(self): + logger.info("Initializing new peer connection") + rtc_config = RTCConfiguration(iceServers=self.ice_servers) + + self.answer: Optional[RTCSessionDescription] = None + self.pc = RTCPeerConnection(rtc_config) + self.pc_id = "PeerConnection(%s)" % uuid.uuid4() + self._setup_listeners() + self._tracks = set() + self._data_channel = None + self._renegotiation_in_progress = False + self._last_received_time = None + + def _setup_listeners(self): + @self.pc.on("datachannel") + def on_datachannel(channel): + self._data_channel = channel + + @channel.on("message") + async def on_message(message): + try: + # aiortc does not provide any way so we can be aware when we are disconnected, + # so we are using this keep alive message as a way to implement that + if isinstance(message, str) and message.startswith("ping"): + self._last_received_time = time.time() + else: + json_message = json.loads(message) + await self.emit("appMessage", json_message) + except Exception as e: + logger.exception(f"Error parsing JSON message {message}, {e}") + + # Despite the fact that aiortc provides this listener, they don't have a status for "disconnected" + # So, in case we loose connection, this event will not be triggered + @self.pc.on("connectionstatechange") + async def on_connectionstatechange(): + await self._handle_new_connection_state() + + # Despite the fact that aiortc provides this listener, they don't have a status for "disconnected" + # So, in case we loose connection, this event will not be triggered + @self.pc.on("iceconnectionstatechange") + async def on_iceconnectionstatechange(): + logger.info( + f"Ice connection state is {self.pc.iceConnectionState}, connection is {self.pc.connectionState}" + ) + + @self.pc.on("icegatheringstatechange") + async def on_icegatheringstatechange(): + logger.info(f"Ice gathering state is {self.pc.iceGatheringState}") + + @self.pc.on("track") + async def on_track(track): + logger.info(f"Track {track.kind} received") + self._tracks.add(track) + await self.emit("track-started", track) + + @track.on("ended") + async def on_ended(): + logger.info(f"Track {track.kind} ended") + self._tracks.discard(track) + await self.emit("track-ended", track) + + async def _create_answer(self, sdp: str, type: str): + offer = RTCSessionDescription(sdp=sdp, type=type) + await self.pc.setRemoteDescription(offer) + + # For some reason, aiortc is not respecting the SDP for the transceivers to be sendrcv + # so we are basically forcing it to act this way + self.force_transceivers_to_send_recv() + + # this answer does not contain the ice candidates, which will be gathered later, after the setLocalDescription + logger.info(f"Creating answer") + local_answer = await self.pc.createAnswer() + await self.pc.setLocalDescription(local_answer) + logger.info(f"Setting the answer after the local description is created") + self.answer = self.pc.localDescription + + async def initialize(self, sdp: str, type: str): + await self._create_answer(sdp, type) + + async def connect(self): + self._connect_invoked = True + # If we already connected, trigger again the connected event + if self.is_connected(): + await self.emit("connected", self) + # We are renegotiating here, because likely we have loose the first video frames + # and aiortc does not handle that pretty well. + self.ask_to_renegotiate() + + async def renegotiate(self, sdp: str, type: str, restart_pc: bool = False): + logger.info(f"Renegotiating {self.pc_id}") + + if restart_pc: + await self.emit("disconnected", self) + logger.info("Closing old peer connection") + # removing the listeners to prevent the bot from closing + self.pc.remove_all_listeners() + await self.close() + # we are initializing a new peer connection in this case. + self._initialize() + + await self._create_answer(sdp, type) + + # Maybe we should refactor to receive a message from the client side when the renegotiation is completed. + # or look at the peer connection listeners + # but this is good enough for now for testing. + async def delayed_task(): + await asyncio.sleep(2) + self._renegotiation_in_progress = False + + asyncio.create_task(delayed_task()) + + def force_transceivers_to_send_recv(self): + for transceiver in self.pc.getTransceivers(): + transceiver.direction = "sendrecv" + # logger.info( + # f"Transceiver: {transceiver}, Mid: {transceiver.mid}, Direction: {transceiver.direction}" + # ) + # logger.info(f"Sender track: {transceiver.sender.track}") + + def replace_audio_track(self, track): + logger.info(f"Replacing audio track {track.kind}") + # Transceivers always appear in creation-order for both peers + # For now we are only considering that we are going to have 02 transceivers, + # one for audio and one for video + transceivers = self.pc.getTransceivers() + if len(transceivers) > 0 and transceivers[0].sender: + transceivers[0].sender.replaceTrack(track) + else: + logger.warning("Audio transceiver not found. Cannot replace audio track.") + + def replace_video_track(self, track): + logger.info(f"Replacing video track {track.kind}") + # Transceivers always appear in creation-order for both peers + # For now we are only considering that we are going to have 02 transceivers, + # one for audio and one for video + transceivers = self.pc.getTransceivers() + if len(transceivers) > 1 and transceivers[1].sender: + transceivers[1].sender.replaceTrack(track) + else: + logger.warning("Video transceiver not found. Cannot replace video track.") + + async def close(self): + if self.pc: + await self.pc.close() + + def get_answer(self): + if not self.answer: + return None + + return { + "sdp": self.answer.sdp, + "type": self.answer.type, + "pc_id": self.pc_id, + } + + async def _handle_new_connection_state(self): + state = self.pc.connectionState + logger.info(f"Connection state changed to: {state}") + await self.emit(state, self) + if state == "failed": + logger.warning("Connection failed, closing peer connection.") + await self.close() + + # Despite the fact that aiortc provides this listener, they don't have a status for "disconnected" + # So, there is no advantage in looking at self.pc.connectionState + # That is why we are trying to keep our own state + def is_connected(self): + # If the small webrtc transport has never invoked to connect + # we are acting like if we are not connected + if not self._connect_invoked: + return False + + if self._last_received_time is None: + # if we have never received a message, it is probably because the client has not created a data channel + # so we are going to trust aiortc in this case + return self.pc.connectionState == "connected" + # Checks if the last received ping was within the last 3 seconds. + return (time.time() - self._last_received_time) < 3 + + def audio_input_track(self): + # Transceivers always appear in creation-order for both peers + # For now we are only considering that we are going to have 02 transceivers, + # one for audio and one for video + transceivers = self.pc.getTransceivers() + if len(transceivers) == 0 or not transceivers[0].receiver: + logger.warning("No audio transceiver is available") + return None + + return transceivers[0].receiver.track + + def video_input_track(self): + # Transceivers always appear in creation-order for both peers + # For now we are only considering that we are going to have 02 transceivers, + # one for audio and one for video + transceivers = self.pc.getTransceivers() + if len(transceivers) <= 1 or not transceivers[1].receiver: + logger.warning("No video transceiver is available") + return None + + return transceivers[1].receiver.track + + def tracks(self): + return self._tracks + + def send_app_message(self, message: Any): + if self._data_channel: + json_message = json.dumps(message) + self._data_channel.send(json_message) + + def ask_to_renegotiate(self): + if self._renegotiation_in_progress: + return + + self._renegotiation_in_progress = True + self.send_app_message( + {"type": SIGNALLING_TYPE, "message": SignallingMessage.RENEGOTIATE.value} + ) diff --git a/src/pipecat/utils/event_emitter.py b/src/pipecat/utils/event_emitter.py new file mode 100644 index 000000000..dcffd2fd1 --- /dev/null +++ b/src/pipecat/utils/event_emitter.py @@ -0,0 +1,20 @@ +class EventEmitter: + def __init__(self): + self._events = {} + + def on(self, event_name): + """Decorator to register an event handler.""" + + def decorator(func): + if event_name not in self._events: + self._events[event_name] = [] + self._events[event_name].append(func) + return func + + return decorator + + async def emit(self, event_name, *args, **kwargs): + """Trigger all handlers for a given event.""" + if event_name in self._events: + for handler in self._events[event_name]: + await handler(*args, **kwargs)