From f24c5b0aa7fdd61e56da4d95c2947911611510c3 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 12 Mar 2025 11:31:18 -0300 Subject: [PATCH] Adding support for renegotiation. --- .../typescript/src/smallWebRTCTransport.ts | 29 +++++++++------ .../aiortc/video-transform/server/server.py | 36 ++++++++++++------- .../transports/network/small_webrtc.py | 9 +++-- .../transports/network/webrtc_connection.py | 29 ++++++++++++--- 4 files changed, 71 insertions(+), 32 deletions(-) diff --git a/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts b/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts index 0c3917107..5f53f058f 100644 --- a/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts +++ b/examples/aiortc/video-transform/client/typescript/src/smallWebRTCTransport.ts @@ -27,6 +27,9 @@ export class SmallWebRTCTransport { private pc: RTCPeerConnection | null = null; private dc: RTCDataChannel | null = null; + private audioCodec: string | null = null; + private videoCodec: string | null = null; + private pc_id: string | null = null; constructor(callbacks: SmallWebRTCTransportCallbacks) { this._callbacks = callbacks @@ -78,7 +81,7 @@ export class SmallWebRTCTransport { return pc; } - private async negotiate(audioCodec: string, videoCodec: string): Promise { + private async negotiate(): Promise { if (!this.pc) { return Promise.reject('Peer connection is not initialized'); } @@ -107,22 +110,25 @@ export class SmallWebRTCTransport { let codec: string; // Filter audio codec - if (audioCodec !== 'default') { + if (this.audioCodec && this.audioCodec !== 'default') { // @ts-ignore - offerSdp.sdp = this.sdpFilterCodec('audio', audioCodec, offerSdp.sdp); + offerSdp.sdp = this.sdpFilterCodec('audio', this.audioCodec, offerSdp.sdp); } // Filter video codec - if (videoCodec !== 'default') { + if (this.videoCodec && this.videoCodec !== 'default') { // @ts-ignore - offerSdp.sdp = this.sdpFilterCodec('video', videoCodec, offerSdp.sdp); + offerSdp.sdp = this.sdpFilterCodec('video', this.videoCodec, offerSdp.sdp); } + this.log(`Will create offer for peerId: ${this.pc_id}`) + // Send offer to server const response = await fetch('/api/offer', { body: JSON.stringify({ sdp: offerSdp.sdp, type: offerSdp.type, + pc_id: this.pc_id }), headers: { 'Content-Type': 'application/json', @@ -132,6 +138,8 @@ export class SmallWebRTCTransport { const answer: RTCSessionDescriptionInit = await response.json(); // @ts-ignore + this.pc_id = answer.pc_id + // @ts-ignore this.log(`Received answer for peer connection id ${answer.pc_id}`) await this.pc!.setRemoteDescription(answer); } catch (e) { @@ -164,7 +172,9 @@ export class SmallWebRTCTransport { this.addInitialTransceivers(); this.dc = this.createDataChannel('chat', { ordered: true }); await this.addUserMedias(audioDevice, videoDevice); - await this.negotiate(audioCodec, videoCodec); + this.audioCodec = audioCodec + this.videoCodec = videoCodec + await this.negotiate(); } private async addUserMedias(audioDevice: string|undefined, videoDevice:string|undefined): Promise { @@ -195,7 +205,7 @@ export class SmallWebRTCTransport { // Check if it's a signalling message if (messageObj.type === SIGNALLING_TYPE) { - this.handleSignallingMessage(messageObj as SignallingMessageObject); // Delegate to handleSignallingMessage + void this.handleSignallingMessage(messageObj as SignallingMessageObject); // Delegate to handleSignallingMessage } else { // implement to handle the other messages in the future } @@ -205,15 +215,14 @@ export class SmallWebRTCTransport { } // Method to handle signalling messages specifically - handleSignallingMessage(messageObj: SignallingMessageObject): void { + async handleSignallingMessage(messageObj: SignallingMessageObject): Promise { // Cast the object to the correct type after verification const signallingMessage = messageObj as SignallingMessageObject; // Handle different signalling message types switch (signallingMessage.message) { case SignallingMessage.RENEGOTIATE: - this.log("Handling renegotiation..."); - // TODO: implement it + await this.negotiate() break; default: diff --git a/examples/aiortc/video-transform/server/server.py b/examples/aiortc/video-transform/server/server.py index e0e34506e..f1377ed0e 100644 --- a/examples/aiortc/video-transform/server/server.py +++ b/examples/aiortc/video-transform/server/server.py @@ -2,6 +2,7 @@ import argparse import asyncio import logging from contextlib import asynccontextmanager +from typing import Dict import uvicorn from aiortc_bot import run_bot @@ -17,33 +18,42 @@ logger = logging.getLogger("pc") app = FastAPI() - -pcs = set() +# Store connections by pc_id +pcs_map: Dict[str, SmallWebRTCConnection] = {} @app.post("/api/offer") async def offer(request: dict, background_tasks: BackgroundTasks): - pipecat_connection = SmallWebRTCConnection() - await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"]) + pc_id = request.get("pc_id") - pcs.add(pipecat_connection) + 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(): - logger.info("Discarding the peer connection.") - pcs.discard(pipecat_connection) + @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) + background_tasks.add_task(run_bot, pipecat_connection) - return pipecat_connection.get_answer() + answer = pipecat_connection.get_answer() + # Updating the peer connection inside the map + pcs_map[answer["pc_id"]] = pipecat_connection + + return answer @asynccontextmanager async def lifespan(app: FastAPI): yield # Run app - coros = [pc.close() for pc in pcs] + coros = [pc.close() for pc in pcs_map.values()] await asyncio.gather(*coros) - pcs.clear() + pcs_map.clear() if __name__ == "__main__": diff --git a/src/pipecat/transports/network/small_webrtc.py b/src/pipecat/transports/network/small_webrtc.py index 6679dbc2c..9dbbd0f5d 100644 --- a/src/pipecat/transports/network/small_webrtc.py +++ b/src/pipecat/transports/network/small_webrtc.py @@ -156,17 +156,17 @@ class SmallWebRTCClient: self._pipecat_resampler = AudioResampler("s16", "mono", 16000) @self._webrtcConnection.on("connected") - async def 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(): + async def on_disconnected(connection: SmallWebRTCConnection): logger.info("Peer connection lost.") await self._handle_client_disconnected() @self._webrtcConnection.on("closed") - async def on_closed(): + async def on_closed(connection: SmallWebRTCConnection): logger.info("Client connection closed.") await self._handle_client_closed() @@ -188,9 +188,8 @@ class SmallWebRTCClient: frame = await asyncio.wait_for(self._video_input_track.recv(), timeout=1.0) except asyncio.TimeoutError: logger.warning("Timeout: No video frame received within the specified time.") - # TODO maybe we should ask to renegotiate in this case. Need to test. - # self._webrtcConnection.renegotiate() frame = None + self._webrtcConnection.ask_to_renegotiate() if frame is None or not isinstance(frame, VideoFrame): # If no valid frame, sleep for a bit diff --git a/src/pipecat/transports/network/webrtc_connection.py b/src/pipecat/transports/network/webrtc_connection.py index ff2350b91..46d119b5b 100644 --- a/src/pipecat/transports/network/webrtc_connection.py +++ b/src/pipecat/transports/network/webrtc_connection.py @@ -1,3 +1,4 @@ +import asyncio import json import uuid from enum import Enum @@ -24,6 +25,7 @@ class SmallWebRTCConnection(EventEmitter): self._setup_listeners() self._tracks = set() self._data_channel = None + self._renegotiation_in_progress = False def _setup_listeners(self): @self.pc.on("datachannel") @@ -41,7 +43,7 @@ class SmallWebRTCConnection(EventEmitter): @self.pc.on("connectionstatechange") async def on_connectionstatechange(): logger.info(f"Connection state is {self.pc.connectionState}") - await self.emit(self.pc.connectionState) + await self.emit(self.pc.connectionState, self) if self.pc.connectionState == "failed": await self.close() @@ -57,7 +59,7 @@ class SmallWebRTCConnection(EventEmitter): self._tracks.discard(track) await self.emit("track-ended", track) - async def initialize(self, sdp: str, type: str): + async def _create_answer(self, sdp: str, type: str): offer = RTCSessionDescription(sdp=sdp, type=type) await self.pc.setRemoteDescription(offer) @@ -67,11 +69,26 @@ class SmallWebRTCConnection(EventEmitter): self.answer = await self.pc.createAnswer() - return self.pc + async def initialize(self, sdp: str, type: str): + await self._create_answer(sdp, type) async def connect(self): await self.pc.setLocalDescription(self.answer) + async def renegotiate(self, sdp: str, type: str): + logger.info(f"Renegotiating {self.pc_id}") + await self._create_answer(sdp, type) + await self.pc.setLocalDescription(self.answer) + + # TODO 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" @@ -149,7 +166,11 @@ class SmallWebRTCConnection(EventEmitter): json_message = json.dumps(message) self._data_channel.send(json_message) - def renegotiate(self): + 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} )