From 8f2dadf5a06a83ac954fd2e8cdcfa17a0ea833ba Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 13 Mar 2025 17:07:32 -0300 Subject: [PATCH] Improving the reconnection logic to be able to recreate the peer connection in some cases. --- .../typescript/src/smallWebRTCTransport.ts | 58 ++++++++++++++----- .../aiortc/video-transform/server/server.py | 4 +- .../transports/network/small_webrtc.py | 44 +++++++++----- .../transports/network/webrtc_connection.py | 24 +++++++- 4 files changed, 96 insertions(+), 34 deletions(-) diff --git a/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts b/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts index 2c8186977..a79467177 100644 --- a/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts +++ b/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts @@ -35,9 +35,14 @@ export class SmallWebRTCTransport { private maxReconnectionAttempts = 3; private isReconnecting = false; private keepAliveInterval: number | null = null; + private audioDevice: string | undefined; + private videoDevice: string | undefined; constructor(callbacks: SmallWebRTCTransportCallbacks) { this._callbacks = callbacks + // for testing reconnections + // @ts-ignore + window.attemptReconnection = this.attemptReconnection.bind(this) } private log(message: string): void { @@ -65,6 +70,9 @@ export class SmallWebRTCTransport { pc.addEventListener('signalingstatechange', () => { this.log(`signalingState: ${this.pc!.signalingState}`) + if (this.pc!.signalingState == 'stable') { + this.handleReconnectionCompleted() + } }); this.log(`signalingState: ${pc.signalingState}`) @@ -94,20 +102,24 @@ export class SmallWebRTCTransport { } } + private handleReconnectionCompleted() { + this.reconnectionAttempts = 0; + this.isReconnecting = false; + } + private handleConnectionStateChange(): void { if (!this.pc) return; this.log(`Connection State: ${this.pc.connectionState}`); if (this.pc.connectionState === "connected") { - this.reconnectionAttempts = 0; - this.isReconnecting = false; + this.handleReconnectionCompleted() this._callbacks.onConnected(); } else if (this.pc.connectionState === "failed") { void this.attemptReconnection(true); } } - private async attemptReconnection(iceRestart: boolean = false): Promise { + private async attemptReconnection(recreatePeerConnection: boolean = false): Promise { if (this.isReconnecting) { this.log("Reconnection already in progress, skipping."); return; @@ -120,17 +132,24 @@ export class SmallWebRTCTransport { this.isReconnecting = true; this.reconnectionAttempts++; this.log(`Reconnection attempt ${this.reconnectionAttempts}...`); - await this.negotiate(iceRestart); + // aiortc it is not working fine when just trying to restart the ice + // so in this case we are creating a new peer connection on both sides + if (recreatePeerConnection) { + //this.pc?.close() + await this.startNewPeerConnection(recreatePeerConnection) + } else { + await this.negotiate(); + } } - private async negotiate(iceRestart: boolean = false): Promise { + private async negotiate(recreatePeerConnection: boolean = false): Promise { if (!this.pc) { return Promise.reject('Peer connection is not initialized'); } try { // Create offer - const offer = await this.pc.createOffer({iceRestart}); + const offer = await this.pc.createOffer(); await this.pc.setLocalDescription(offer); // Wait for ICE gathering to complete @@ -170,7 +189,8 @@ export class SmallWebRTCTransport { body: JSON.stringify({ sdp: offerSdp.sdp, type: offerSdp.type, - pc_id: this.pc_id + pc_id: this.pc_id, + restart_pc: recreatePeerConnection }), headers: { 'Content-Type': 'application/json', @@ -212,18 +232,24 @@ export class SmallWebRTCTransport { } async start(audioDevice: string | undefined, audioCodec: string, videoCodec: string, videoDevice: string | undefined): Promise { + this.audioDevice = audioDevice + this.videoDevice = videoDevice + this.audioCodec = audioCodec + this.videoCodec = videoCodec + await this.startNewPeerConnection() + } + + private async startNewPeerConnection(recreatePeerConnection: boolean = false) { this.pc = this.createPeerConnection(); this.addInitialTransceivers(); this.dc = this.createDataChannel('chat', { ordered: true }); - await this.addUserMedias(audioDevice, videoDevice); - this.audioCodec = audioCodec - this.videoCodec = videoCodec - await this.negotiate(); + await this.addUserMedias(); + await this.negotiate(recreatePeerConnection); } - private async addUserMedias(audioDevice: string|undefined, videoDevice:string|undefined): Promise { + private async addUserMedias(): Promise { this.log("Will send the audio and video"); - const constraints = this.createMediaConstraints(audioDevice, videoDevice); + const constraints = this.createMediaConstraints(); if (constraints.audio || constraints.video) { try { @@ -314,16 +340,16 @@ export class SmallWebRTCTransport { return dc; } - private createMediaConstraints(audioDevice: string|undefined, videoDevice:string|undefined): MediaStreamConstraints { + private createMediaConstraints(): MediaStreamConstraints { const constraints: MediaStreamConstraints = { audio: false, video: false }; const audioConstraints: MediaTrackConstraints = {}; - if (audioDevice) audioConstraints.deviceId = { exact: audioDevice }; + if (this.audioDevice) audioConstraints.deviceId = { exact: this.audioDevice }; constraints.audio = Object.keys(audioConstraints).length ? audioConstraints : true; const videoConstraints: MediaTrackConstraints = {}; - if (videoDevice) videoConstraints.deviceId = { exact: videoDevice }; + if (this.videoDevice) videoConstraints.deviceId = { exact: this.videoDevice }; constraints.video = Object.keys(videoConstraints).length ? videoConstraints : true; diff --git a/examples/aiortc/video-transform/server/server.py b/examples/aiortc/video-transform/server/server.py index f1377ed0e..5a0680650 100644 --- a/examples/aiortc/video-transform/server/server.py +++ b/examples/aiortc/video-transform/server/server.py @@ -29,7 +29,9 @@ async def offer(request: dict, background_tasks: BackgroundTasks): 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"]) + await pipecat_connection.renegotiate( + sdp=request["sdp"], type=request["type"], restart_pc=request.get("restart_pc", False) + ) else: pipecat_connection = SmallWebRTCConnection() await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"]) diff --git a/src/pipecat/transports/network/small_webrtc.py b/src/pipecat/transports/network/small_webrtc.py index 3fe244969..411c32838 100644 --- a/src/pipecat/transports/network/small_webrtc.py +++ b/src/pipecat/transports/network/small_webrtc.py @@ -14,7 +14,7 @@ from typing import Any, Awaitable, Callable, Optional import cv2 import numpy as np from aiortc import VideoStreamTrack -from aiortc.mediastreams import AudioStreamTrack, VideoFrame +from aiortc.mediastreams import AudioStreamTrack, MediaStreamError, VideoFrame from av import AudioFrame, AudioResampler from loguru import logger from pydantic import BaseModel @@ -191,6 +191,9 @@ class SmallWebRTCClient: 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 @@ -237,6 +240,9 @@ class SmallWebRTCClient: 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). @@ -265,11 +271,11 @@ class SmallWebRTCClient: yield audio_frame async def write_raw_audio_frames(self, data: bytes): - if self._can_send(): + 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(): + if self._can_send() and self._video_output_track: self._video_output_track.add_video_frame(frame) async def setup(self, _params: TransportParams, frame): @@ -279,23 +285,13 @@ class SmallWebRTCClient: self._params = _params async def connect(self): - if self._audio_output_track or self._video_output_track: + if self._webrtcConnection.is_connected() or self._webrtcConnection.is_connecting(): # already initialized return + logger.info(f"Connecting to Small WebRTC") await self._webrtcConnection.connect() - logger.info(f"Connecting to Small WebRTC") - - 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) async def disconnect(self): if self.is_connected and not self.is_closing: @@ -311,12 +307,30 @@ class SmallWebRTCClient: async def _handle_client_connected(self): 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): diff --git a/src/pipecat/transports/network/webrtc_connection.py b/src/pipecat/transports/network/webrtc_connection.py index 2a8b30689..2b7ba0d3c 100644 --- a/src/pipecat/transports/network/webrtc_connection.py +++ b/src/pipecat/transports/network/webrtc_connection.py @@ -1,7 +1,7 @@ import asyncio import json -import uuid import time +import uuid from enum import Enum from typing import Any, Optional @@ -20,6 +20,11 @@ class SignallingMessage(Enum): class SmallWebRTCConnection(EventEmitter): def __init__(self): super().__init__() + self._is_connecting = False + self._initialize() + + def _initialize(self): + logger.info("Initializing new peer connection") self.answer: Optional[RTCSessionDescription] = None self.pc = RTCPeerConnection() self.pc_id = "PeerConnection(%s)" % uuid.uuid4() @@ -87,10 +92,21 @@ class SmallWebRTCConnection(EventEmitter): await self._create_answer(sdp, type) async def connect(self): + self._is_connecting = True await self.pc.setLocalDescription(self.answer) - async def renegotiate(self, sdp: str, type: str): + 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) await self.pc.setLocalDescription(self.answer) @@ -136,6 +152,7 @@ class SmallWebRTCConnection(EventEmitter): async def close(self): if self.pc: await self.pc.close() + self._is_connecting = False def get_answer(self): if not self.answer: @@ -166,6 +183,9 @@ class SmallWebRTCConnection(EventEmitter): # Checks if the last received ping was within the last 3 seconds. return (time.time() - self._last_received_time) < 3 + def is_connecting(self): + return self._is_connecting + 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,