From bad9977e8cb3c77b6912410b5225b5de89db230b Mon Sep 17 00:00:00 2001 From: mattie ruth backman Date: Thu, 21 Aug 2025 15:58:14 -0400 Subject: [PATCH] PR feedback and more explicit about only supporting exporting 1 video --- src/pipecat/frames/frames.py | 2 +- src/pipecat/runner/utils.py | 6 -- .../transports/network/small_webrtc.py | 59 +++++++++++-------- .../transports/network/webrtc_connection.py | 37 +++++++++--- 4 files changed, 64 insertions(+), 40 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index a8acd35bc..d1d3806d5 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -228,7 +228,7 @@ class OutputImageRawFrame(DataFrame, ImageRawFrame): def __str__(self): pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})" + return f"{self.name}(pts: {pts}, destination: {self.transport_destination}, size: {self.size}, format: {self.format})" @dataclass diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index aec9a9361..b638eb11e 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -233,15 +233,12 @@ async def maybe_capture_participant_camera( framerate: Video capture framerate. Defaults to 0 (auto). """ try: - from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.services.daily import DailyTransport if isinstance(transport, DailyTransport): await transport.capture_participant_video( client["id"], framerate=framerate, video_source="camera" ) - elif isinstance(transport, SmallWebRTCTransport): - await transport.capture_participant_video(video_source="camera") except ImportError: pass @@ -257,15 +254,12 @@ async def maybe_capture_participant_screen( framerate: Video capture framerate. Defaults to 0 (auto). """ try: - from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.services.daily import DailyTransport if isinstance(transport, DailyTransport): await transport.capture_participant_video( client["id"], framerate=framerate, video_source="screenVideo" ) - elif isinstance(transport, SmallWebRTCTransport): - await transport.capture_participant_video(video_source="screenVideo") except ImportError: pass diff --git a/src/pipecat/transports/network/small_webrtc.py b/src/pipecat/transports/network/small_webrtc.py index 539368630..080491319 100644 --- a/src/pipecat/transports/network/small_webrtc.py +++ b/src/pipecat/transports/network/small_webrtc.py @@ -51,6 +51,10 @@ except ModuleNotFoundError as e: logger.error("In order to use the SmallWebRTC, you need to `pip install pipecat-ai[webrtc]`.") raise Exception(f"Missing module: {e}") +CAM_VIDEO_SOURCE = "camera" +SCREEN_VIDEO_SOURCE = "screenVideo" +MIC_AUDIO_SOURCE = "microphone" + class SmallWebRTCCallbacks(BaseModel): """Callback handlers for SmallWebRTC events. @@ -288,7 +292,9 @@ class SmallWebRTCClient: """ while True: video_track = ( - self._video_input_track if video_source == "camera" else self._screen_video_track + self._video_input_track + if video_source == CAM_VIDEO_SOURCE + else self._screen_video_track ) if video_track is None: await asyncio.sleep(0.01) @@ -562,7 +568,7 @@ class SmallWebRTCInputTransport(BaseInputTransport): if not self._receive_audio_task and self._params.audio_in_enabled: self._receive_audio_task = self.create_task(self._receive_audio()) if not self._receive_video_task and self._params.video_in_enabled: - self._receive_video_task = self.create_task(self._receive_video("camera")) + self._receive_video_task = self.create_task(self._receive_video(CAM_VIDEO_SOURCE)) async def _stop_tasks(self): """Stop all background tasks.""" @@ -665,23 +671,27 @@ class SmallWebRTCInputTransport(BaseInputTransport): # If we're not already receiving video, try to get a frame now if ( - frame.video_source == "camera" + frame.video_source == CAM_VIDEO_SOURCE and not self._receive_video_task and self._params.video_in_enabled ): # Start video reception if it's not already running - self._receive_video_task = self.create_task(self._receive_video("camera")) + self._receive_video_task = self.create_task(self._receive_video(CAM_VIDEO_SOURCE)) elif ( - frame.video_source == "screenVideo" + frame.video_source == SCREEN_VIDEO_SOURCE and not self._receive_screen_video_task and self._params.video_in_enabled ): + print(f"Starting screen video task in request_participant_image") + # Start screen video reception if it's not already running - self._receive_screen_video_task = self.create_task(self._receive_video("screenVideo")) + self._receive_screen_video_task = self.create_task( + self._receive_video(SCREEN_VIDEO_SOURCE) + ) async def capture_participant_media( self, - source: str = "camera", + source: str = CAM_VIDEO_SOURCE, ): """Capture media from a specific participant. @@ -690,22 +700,29 @@ class SmallWebRTCInputTransport(BaseInputTransport): """ # If we're not already receiving video, try to get a frame now if ( - source == "microphone" + source == MIC_AUDIO_SOURCE and not self._receive_audio_task and self._params.audio_in_enabled ): # Start audio reception if it's not already running self._receive_audio_task = self.create_task(self._receive_audio()) - elif source == "camera" and not self._receive_video_task and self._params.video_in_enabled: - # Start video reception if it's not already running - self._receive_video_task = self.create_task(self._receive_video("camera")) elif ( - source == "screenVideo" + source == CAM_VIDEO_SOURCE + and not self._receive_video_task + and self._params.video_in_enabled + ): + # Start video reception if it's not already running + self._receive_video_task = self.create_task(self._receive_video(CAM_VIDEO_SOURCE)) + elif ( + source == SCREEN_VIDEO_SOURCE and not self._receive_screen_video_task and self._params.video_in_enabled ): # Start screen video reception if it's not already running - self._receive_screen_video_task = self.create_task(self._receive_video("screenVideo")) + print(f"Starting screen video task in capture_participant_media") + self._receive_screen_video_task = self.create_task( + self._receive_video(SCREEN_VIDEO_SOURCE) + ) class SmallWebRTCOutputTransport(BaseOutputTransport): @@ -895,34 +912,24 @@ class SmallWebRTCTransport(BaseTransport): async def capture_participant_video( self, - participant_id: str = None, - framerate: int = 30, - video_source: str = "camera", - color_format: str = "RGB", + video_source: str = CAM_VIDEO_SOURCE, ): """Capture video from a specific participant. Args: - participant_id: Unused parameter, kept for compatibility. - framerate: Unused parameter, kept for compatibility. - video_source: Video source to capture from. - color_format: Unused parameter, kept for compatibility. + video_source: Video source to capture from ("camera" or "screenVideo"). """ if self._input: await self._input.capture_participant_media(source=video_source) async def capture_participant_audio( self, - participant_id: str = None, - audio_source: str = "microphone", - sample_rate: int = 16000, + audio_source: str = MIC_AUDIO_SOURCE, ): """Capture audio from a specific participant. Args: - participant_id: Unused parameter, kept for compatibility. audio_source: Audio source to capture from. (currently, "microphone" is the only supported option) - sample_rate: Unused parameter, kept for compatibility. """ if self._input: await self._input.capture_participant_media(source=audio_source) diff --git a/src/pipecat/transports/network/webrtc_connection.py b/src/pipecat/transports/network/webrtc_connection.py index 890aaef68..d84216d68 100644 --- a/src/pipecat/transports/network/webrtc_connection.py +++ b/src/pipecat/transports/network/webrtc_connection.py @@ -95,7 +95,7 @@ class SmallWebRTCTrack: enable/disable control and frame discarding for audio and video streams. """ - def __init__(self, track: MediaStreamTrack, index: int): + def __init__(self, track: MediaStreamTrack): """Initialize the WebRTC track wrapper. Args: @@ -104,7 +104,6 @@ class SmallWebRTCTrack: """ self._track = track self._enabled = True - self.source_index = index def set_enabled(self, enabled: bool) -> None: """Enable or disable the track. @@ -350,7 +349,11 @@ class SmallWebRTCConnection(BaseObject): screen_video_input_track = self.screen_video_input_track() if screen_video_input_track: await self.screen_video_input_track().discard_old_frames() - self.ask_to_renegotiate() + if video_input_track or screen_video_input_track: + # This prevents an issue where sometimes the WebRTC connection can be established + # before the bot is ready to receive video. When that happens, we can lose a couple + # of seconds of video before we received a key frame to finally start displaying it. + self.ask_to_renegotiate() async def renegotiate(self, sdp: str, type: str, restart_pc: bool = False): """Renegotiate the WebRTC connection with new parameters. @@ -385,7 +388,11 @@ class SmallWebRTCConnection(BaseObject): def force_transceivers_to_send_recv(self): """Force all transceivers to bidirectional send/receive mode.""" for transceiver in self._pc.getTransceivers(): - transceiver.direction = "sendrecv" + # For now, we only support sendrecv for camera audio and video (the first two transceivers) + if transceiver.mid == "0" or transceiver.mid == "1": + transceiver.direction = "sendrecv" + else: + transceiver.direction = "recvonly" # logger.debug( # f"Transceiver: {transceiver}, Mid: {transceiver.mid}, Direction: {transceiver.direction}" # ) @@ -423,6 +430,22 @@ class SmallWebRTCConnection(BaseObject): else: logger.warning("Video transceiver not found. Cannot replace video track.") + def replace_screen_video_track(self, track): + """Replace the screen video track in the second transceiver. + + Args: + track: The new screen video track to use for sending. + """ + logger.debug(f"Replacing screen 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) > 2 and transceivers[2].sender: + transceivers[2].sender.replaceTrack(track) + else: + logger.warning("Screen video transceiver not found. Cannot replace screen video track.") + async def disconnect(self): """Disconnect from the WebRTC peer connection.""" self.send_app_message({"type": SIGNALLING_TYPE, "message": PeerLeftMessage().model_dump()}) @@ -503,7 +526,7 @@ class SmallWebRTCConnection(BaseObject): return None track = transceivers[AUDIO_TRANSCEIVER_INDEX].receiver.track - audio_track = SmallWebRTCTrack(track, AUDIO_TRANSCEIVER_INDEX) if track else None + audio_track = SmallWebRTCTrack(track) if track else None self._track_map[AUDIO_TRANSCEIVER_INDEX] = audio_track return audio_track @@ -525,7 +548,7 @@ class SmallWebRTCConnection(BaseObject): return None track = transceivers[VIDEO_TRANSCEIVER_INDEX].receiver.track - video_track = SmallWebRTCTrack(track, VIDEO_TRANSCEIVER_INDEX) if track else None + video_track = SmallWebRTCTrack(track) if track else None self._track_map[VIDEO_TRANSCEIVER_INDEX] = video_track return video_track @@ -547,7 +570,7 @@ class SmallWebRTCConnection(BaseObject): return None track = transceivers[SCREEN_VIDEO_TRANSCEIVER_INDEX].receiver.track - video_track = SmallWebRTCTrack(track, SCREEN_VIDEO_TRANSCEIVER_INDEX) if track else None + video_track = SmallWebRTCTrack(track) if track else None self._track_map[SCREEN_VIDEO_TRANSCEIVER_INDEX] = video_track return video_track