wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

PR feedback and more explicit about only supporting exporting 1 video

Browse Source
This commit is contained in:
mattie ruth backman
2025-08-21 15:58:14 -04:00
committed by Mattie Ruth
parent b987579d54
commit bad9977e8c
4 changed files with 64 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/pipecat/frames/frames.py
View File

@@ -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

6
src/pipecat/runner/utils.py
View File

@@ -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

59
src/pipecat/transports/network/small_webrtc.py
View File

@@ -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)

37
src/pipecat/transports/network/webrtc_connection.py
View File

@@ -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