frontends (tools/webrtc): Handle SourcesStdin:
Update the pipeline workflow to handle `SourcesStdin`. A `decodebin` is used to autodetect the stream type, then audio and/or video elements are created and plugged accordingly.
This commit is contained in:
Goffi
parent
59cf0b846d
commit
ed1a652029
1 changed files with 257 additions and 41 deletions
|
|
@ -52,6 +52,7 @@ from .webrtc_models import (
|
|||
SourcesDataChannel,
|
||||
SourcesNone,
|
||||
SourcesPipeline,
|
||||
SourcesStdin,
|
||||
SourcesTest,
|
||||
)
|
||||
|
||||
|
|
@ -545,6 +546,10 @@ class WebRTC:
|
|||
raise NotImplementedError(NONE_NOT_IMPLEMENTED_MSG)
|
||||
elif isinstance(self.sources_data, SourcesNone):
|
||||
pass
|
||||
elif isinstance(self.sources_data, SourcesStdin):
|
||||
# SourcesStdin uses decodebin which will dynamically provide streams.
|
||||
if audio_pt is None or video_pt is None:
|
||||
raise NotImplementedError(NONE_NOT_IMPLEMENTED_MSG)
|
||||
else:
|
||||
if audio_pt is None or video_pt is None:
|
||||
raise NotImplementedError(NONE_NOT_IMPLEMENTED_MSG)
|
||||
|
|
@ -568,6 +573,9 @@ class WebRTC:
|
|||
case SourcesTest():
|
||||
video_source_elt = "videotestsrc is-live=true pattern=ball"
|
||||
audio_source_elt = "audiotestsrc"
|
||||
case SourcesStdin():
|
||||
video_source_elt = ""
|
||||
audio_source_elt = ""
|
||||
case _:
|
||||
raise exceptions.InternalError(
|
||||
f'Unexpected "sources_data" value: {self.sources_data!r}'
|
||||
|
|
@ -606,35 +614,42 @@ class WebRTC:
|
|||
)
|
||||
local_video_sink_elt = "compositor.sink_1"
|
||||
|
||||
if video_source_elt:
|
||||
if video_source_elt or isinstance(self.sources_data, SourcesStdin):
|
||||
# Video source with an input-selector to switch between normal and video mute
|
||||
# (or desktop sharing).
|
||||
gst_pipe_elements.append(
|
||||
f"""
|
||||
input-selector name=video_selector
|
||||
! tee name=t
|
||||
video_elements = [
|
||||
"input-selector name=video_selector",
|
||||
"! tee name=t",
|
||||
"",
|
||||
]
|
||||
|
||||
{video_source_elt} name=video_src
|
||||
! queue max-size-buffers=3 leaky=downstream
|
||||
! video_selector.
|
||||
if video_source_elt:
|
||||
video_elements.extend([
|
||||
f"{video_source_elt} name=video_src",
|
||||
"! queue max-size-buffers=3 leaky=downstream",
|
||||
"! video_selector.",
|
||||
"",
|
||||
])
|
||||
|
||||
videotestsrc name=muted_src is-live=true pattern=black
|
||||
! queue leaky=downstream
|
||||
! video_selector.
|
||||
video_elements.extend([
|
||||
"videotestsrc name=muted_src is-live=true pattern=black",
|
||||
"! queue leaky=downstream",
|
||||
"! video_selector.",
|
||||
"",
|
||||
"t.",
|
||||
"! queue max-size-buffers=3 leaky=downstream",
|
||||
"! videorate",
|
||||
f"! capsfilter name=main_rate_caps caps=video/x-raw,framerate={WEBCAM.fps}/1",
|
||||
"! videoconvert",
|
||||
f"! vp8enc name=vp8enc deadline=1 keyframe-max-dist={WEBCAM.keyframe_max} cpu-used=5 target-bitrate={WEBCAM.bitrate} error-resilient=1",
|
||||
"! rtpvp8pay picture-id-mode=15-bit mtu=1200",
|
||||
f"! application/x-rtp,media=video,encoding-name=VP8,payload={video_pt}",
|
||||
"! sendrecv.",
|
||||
])
|
||||
|
||||
t.
|
||||
! queue max-size-buffers=3 leaky=downstream
|
||||
! videorate
|
||||
! capsfilter name=main_rate_caps caps=video/x-raw,framerate={WEBCAM.fps}/1
|
||||
! videoconvert
|
||||
! vp8enc name=vp8enc deadline=1 keyframe-max-dist={WEBCAM.keyframe_max} cpu-used=5 target-bitrate={WEBCAM.bitrate} error-resilient=1
|
||||
! rtpvp8pay picture-id-mode=15-bit mtu=1200
|
||||
! application/x-rtp,media=video,encoding-name=VP8,payload={video_pt}
|
||||
! sendrecv.
|
||||
"""
|
||||
)
|
||||
gst_pipe_elements.append("\n".join(video_elements))
|
||||
|
||||
if local_video_sink_elt:
|
||||
if local_video_sink_elt and (video_source_elt or isinstance(self.sources_data, SourcesStdin)):
|
||||
# Local video feedback.
|
||||
gst_pipe_elements.append(
|
||||
f"""
|
||||
|
|
@ -645,21 +660,30 @@ class WebRTC:
|
|||
"""
|
||||
)
|
||||
|
||||
if audio_source_elt:
|
||||
if audio_source_elt or isinstance(self.sources_data, SourcesStdin):
|
||||
# Audio with a valve for muting.
|
||||
gst_pipe_elements.append(
|
||||
f"""
|
||||
{audio_source_elt} name=audio_src
|
||||
! valve name=audio_valve
|
||||
! queue max-size-buffers=10 max-size-time=0 max-size-bytes=0 leaky=downstream
|
||||
! audioconvert
|
||||
! audioresample
|
||||
! opusenc audio-type=voice bitrate=32000
|
||||
! rtpopuspay
|
||||
! application/x-rtp,media=audio,encoding-name=OPUS,payload={audio_pt}
|
||||
! sendrecv.
|
||||
"""
|
||||
)
|
||||
audio_elements = []
|
||||
|
||||
if audio_source_elt:
|
||||
audio_elements.extend([
|
||||
f"{audio_source_elt} name=audio_src",
|
||||
"! valve name=audio_valve",
|
||||
])
|
||||
else:
|
||||
# For stdin, just create the valve for dynamic linking
|
||||
audio_elements.append("valve name=audio_valve")
|
||||
|
||||
audio_elements.extend([
|
||||
"! queue max-size-buffers=10 max-size-time=0 max-size-bytes=0 leaky=downstream",
|
||||
"! audioconvert",
|
||||
"! audioresample",
|
||||
"! opusenc audio-type=voice bitrate=32000",
|
||||
"! rtpopuspay",
|
||||
f"! application/x-rtp,media=audio,encoding-name=OPUS,payload={audio_pt}",
|
||||
"! sendrecv.",
|
||||
])
|
||||
|
||||
gst_pipe_elements.append("\n".join(audio_elements))
|
||||
|
||||
self.gst_pipe_desc = "\n\n".join(gst_pipe_elements)
|
||||
|
||||
|
|
@ -743,6 +767,9 @@ class WebRTC:
|
|||
local_video_sink_caps = Gst.Caps.from_string(f"video/x-raw,format=RGB")
|
||||
local_video_sink.set_property("caps", local_video_sink_caps)
|
||||
|
||||
if isinstance(self.sources_data, SourcesStdin):
|
||||
self._setup_stdin_source()
|
||||
|
||||
# Create bus and associate signal watchers
|
||||
self.bus = self.pipeline.get_bus()
|
||||
if not self.bus:
|
||||
|
|
@ -1225,6 +1252,195 @@ class WebRTC:
|
|||
decodebin.set_state(Gst.State.NULL)
|
||||
self.pipeline.remove(decodebin)
|
||||
|
||||
def _setup_stdin_source(self) -> None:
|
||||
"""Setup stdin source with decodebin for automatic format detection."""
|
||||
assert self.pipeline is not None
|
||||
|
||||
fdsrc = Gst.ElementFactory.make("fdsrc")
|
||||
if not fdsrc:
|
||||
log.error("Critical: Failed to create fdsrc element")
|
||||
return
|
||||
|
||||
fdsrc.set_property("fd", 0)
|
||||
# Make it a live source to run at real-time speed.
|
||||
fdsrc.set_property("is-live", True)
|
||||
fdsrc.set_property("do-timestamp", True)
|
||||
|
||||
# ``decodebin`` is used for automatic format detection.
|
||||
decodebin = Gst.ElementFactory.make("decodebin")
|
||||
if not decodebin:
|
||||
log.error("Critical: Failed to create decodebin element")
|
||||
return
|
||||
|
||||
self.pipeline.add(fdsrc)
|
||||
self.pipeline.add(decodebin)
|
||||
|
||||
if not fdsrc.link(decodebin):
|
||||
log.error("Failed to link fdsrc to decodebin")
|
||||
return
|
||||
|
||||
decodebin.connect("pad-added", self._on_stdin_decodebin_stream)
|
||||
|
||||
fdsrc.sync_state_with_parent()
|
||||
decodebin.sync_state_with_parent()
|
||||
|
||||
log.info("Successfully setup stdin source.")
|
||||
|
||||
def _on_stdin_decodebin_stream(self, decodebin, pad: Gst.Pad) -> None:
|
||||
"""Handle streams from stdin decodebin."""
|
||||
assert self.pipeline is not None
|
||||
|
||||
caps = pad.get_current_caps()
|
||||
if not caps:
|
||||
caps = pad.query_caps(None)
|
||||
|
||||
name = caps.get_structure(0).get_name()
|
||||
log.info(f"Handling stdin decoded pad of type {name}")
|
||||
log.info(f"Pad caps: {caps.to_string()}")
|
||||
|
||||
if name.startswith('video/'):
|
||||
log.info("===> VIDEO (stdin)")
|
||||
|
||||
video_selector = self.pipeline.get_by_name("video_selector")
|
||||
if not video_selector:
|
||||
log.error("video_selector not found in pipeline")
|
||||
return
|
||||
|
||||
log.debug("video_selector pads before:")
|
||||
for p in video_selector.pads:
|
||||
log.debug(f" {p.get_name()}: direction={p.direction}")
|
||||
|
||||
try:
|
||||
active_pad = video_selector.get_property("active-pad")
|
||||
current_active_pad = active_pad.get_name() if active_pad else "none"
|
||||
log.debug(f"Current active pad: {current_active_pad}.")
|
||||
except:
|
||||
log.warning("Could not get active-pad property.")
|
||||
|
||||
# Synchronize buffers to pipeline clock to avoid video playing too fast.
|
||||
clocksync = Gst.ElementFactory.make("clocksync")
|
||||
if not clocksync:
|
||||
log.warning("clocksync element not available, video may play too fast.")
|
||||
else:
|
||||
self.pipeline.add(clocksync)
|
||||
log.debug("Created clocksync element for real-time playback")
|
||||
|
||||
queue = Gst.ElementFactory.make("queue")
|
||||
if not queue:
|
||||
log.error("Failed to create queue for stdin video")
|
||||
return
|
||||
|
||||
queue.set_property("max-size-buffers", 30)
|
||||
queue.set_property("max-size-bytes", 0)
|
||||
queue.set_property("max-size-time", 2000000000)
|
||||
self.pipeline.add(queue)
|
||||
|
||||
if clocksync:
|
||||
sink_pad = clocksync.get_static_pad("sink")
|
||||
link_result = pad.link(sink_pad)
|
||||
if link_result != Gst.PadLinkReturn.OK:
|
||||
log.error(
|
||||
f"Failed to link decodebin pad to clocksync sink: {link_result}."
|
||||
)
|
||||
return
|
||||
|
||||
if not clocksync.link(queue):
|
||||
log.error("Failed to link clocksync to queue")
|
||||
return
|
||||
else:
|
||||
sink_pad = queue.get_static_pad("sink")
|
||||
link_result = pad.link(sink_pad)
|
||||
if link_result != Gst.PadLinkReturn.OK:
|
||||
log.error(
|
||||
f"Failed to link decodebin pad to queue sink: {link_result}."
|
||||
)
|
||||
return
|
||||
|
||||
src_pad = queue.get_static_pad("src")
|
||||
sink_pad = video_selector.get_request_pad("sink_%u")
|
||||
if not src_pad or not sink_pad:
|
||||
log.error("Failed to get pads for linking queue to video_selector")
|
||||
return
|
||||
|
||||
link_result = src_pad.link(sink_pad)
|
||||
if link_result != Gst.PadLinkReturn.OK:
|
||||
log.error(f"Failed to link queue to video_selector: {link_result}")
|
||||
return
|
||||
|
||||
log.info("video_selector pads after linking:")
|
||||
for p in video_selector.pads:
|
||||
log.debug(f" {p.get_name()}: direction={p.direction}")
|
||||
|
||||
video_selector.set_property("active-pad", sink_pad)
|
||||
log.debug("Set video_selector active pad")
|
||||
|
||||
try:
|
||||
active_pad = video_selector.get_property("active-pad")
|
||||
current_active_pad = active_pad.get_name() if active_pad else "none"
|
||||
log.debug(f"New active pad: {current_active_pad}.")
|
||||
except:
|
||||
log.warning("Could not verify active-pad property")
|
||||
|
||||
queue.sync_state_with_parent()
|
||||
|
||||
state, _, _ = self.pipeline.get_state(0)
|
||||
log.debug(f"Pipeline state: {state}.")
|
||||
|
||||
elif name.startswith('audio/'):
|
||||
log.info("===> AUDIO (stdin)")
|
||||
|
||||
audio_valve = self.pipeline.get_by_name("audio_valve")
|
||||
if not audio_valve:
|
||||
log.error("audio_valve not found in pipeline")
|
||||
return
|
||||
|
||||
# Synchronize buffers to pipeline clock to avoid audio playing too fast.
|
||||
clocksync = Gst.ElementFactory.make("clocksync")
|
||||
if not clocksync:
|
||||
log.warning("clocksync element not available, audio may play too fast.")
|
||||
else:
|
||||
self.pipeline.add(clocksync)
|
||||
log.debug("Created clocksync element for real-time audio playback")
|
||||
|
||||
queue = Gst.ElementFactory.make("queue")
|
||||
if not queue:
|
||||
log.error("Failed to create queue for stdin audio")
|
||||
return
|
||||
|
||||
self.pipeline.add(queue)
|
||||
|
||||
if clocksync:
|
||||
sink_pad = clocksync.get_static_pad("sink")
|
||||
link_result = pad.link(sink_pad)
|
||||
if link_result != Gst.PadLinkReturn.OK:
|
||||
log.error(f"Failed to link decodebin pad to clocksync sink: {link_result}")
|
||||
return
|
||||
|
||||
if not clocksync.link(queue):
|
||||
log.error("Failed to link clocksync to queue")
|
||||
return
|
||||
else:
|
||||
sink_pad = queue.get_static_pad("sink")
|
||||
link_result = pad.link(sink_pad)
|
||||
if link_result != Gst.PadLinkReturn.OK:
|
||||
log.error(f"Failed to link decodebin pad to queue sink: {link_result}")
|
||||
return
|
||||
|
||||
link_result = queue.link(audio_valve)
|
||||
if not link_result:
|
||||
log.error("Failed to link queue to audio_valve")
|
||||
return
|
||||
|
||||
if clocksync:
|
||||
clocksync.sync_state_with_parent()
|
||||
queue.sync_state_with_parent()
|
||||
log.debug("Audio elements synced to pipeline state")
|
||||
|
||||
state, _, _ = self.pipeline.get_state(0)
|
||||
log.debug(f"Pipeline state: {state}.")
|
||||
|
||||
else:
|
||||
log.warning(f"Ignoring unknown pad type from stdin decodebin: {name!r}.")
|
||||
|
||||
async def _call_start(self, callee: jid.JID, call_data: dict, profile: str) -> str:
|
||||
return await self.bridge.call_start(
|
||||
|
|
|
|||
Loading…
Reference in a new issue