erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tools (webrtc): improved pipeline:

Browse source 
- Adjust parameters and elements to better handle the buffering and jitter.
- Add logging, better error handling.
This commit is contained in:
Goffi 2025-11-06 22:10:13 +01:00
commit d55315e75f
1 changed files with 99 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

160
libervia/frontends/tools/webrtc.py
View file

@ -537,7 +537,7 @@ class WebRTC:
case SinksApp():
local_video_sink_elt = (
"appsink name=local_video_sink emit-signals=true drop=true "
"max-buffers=1 sync=True"
"max-buffers=2 sync=false"
)
case SinksAuto():
if isinstance(self.sources_data, SourcesNone):
@ -552,7 +552,7 @@ class WebRTC:
)
gst_pipe_elements = [
"webrtcbin latency=30 name=sendrecv bundle-policy=max-bundle"
"webrtcbin latency=100 name=sendrecv bundle-policy=max-bundle"
]
if self.merge_pip and local_video_sink_elt:
@ -572,12 +572,11 @@ class WebRTC:
gst_pipe_elements.append(
f"""
input-selector name=video_selector
! videorate drop-only=1 max-rate=30
! video/x-raw,framerate=30/1
! tee name=t
{video_source_elt} name=video_src
! queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream
! queue max-size-buffers=3 leaky=downstream
! video_selector.
videotestsrc name=muted_src is-live=true pattern=black
@ -585,10 +584,10 @@ class WebRTC:
! video_selector.
t.
! queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream
! queue max-size-buffers=3 leaky=downstream
! videoscale
! videoconvert
! vp8enc deadline=1 keyframe-max-dist=30
! vp8enc deadline=1 keyframe-max-dist=30 cpu-used=5 target-bitrate=1000000
! rtpvp8pay picture-id-mode=15-bit
! application/x-rtp,media=video,encoding-name=VP8,payload={video_pt}
! sendrecv.
@ -600,7 +599,7 @@ class WebRTC:
gst_pipe_elements.append(
f"""
t.
! queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream
! queue max-size-buffers=3 leaky=downstream
! videoconvert
! {local_video_sink_elt}
"""
@ -766,11 +765,16 @@ class WebRTC:
if ret == Gst.StateChangeReturn.FAILURE:
log.error(f"Failed to change pipeline state to {state}")
return False
elif ret == Gst.StateChangeReturn.ASYNC:
log.debug(f"Waiting for async state change to {state.value_nick.upper()}...")
ret, _, _ = self.pipeline.get_state(timeout=5 * Gst.SECOND)
if ret != Gst.StateChangeReturn.SUCCESS:
log.warning(
f"Async state change to {state.value_nick.upper()} timed out or failed!"
)
return False
__, pending_state = self.pipeline.get_state(Gst.CLOCK_TIME_NONE)[:2]
if pending_state != Gst.State.VOID_PENDING:
log.warning(f"State change to {state} still pending")
log.debug(f"Pipeline state successfully changed to {state.value_nick.upper()}")
return True
def monitor_connection_state(self):
@ -852,7 +856,9 @@ class WebRTC:
)
return False
elif isinstance(dest, Gst.Element):
return source.link(dest)
if not source.link(dest):
log.error(f"Failed to link {source.get_name()} to {dest.get_name()}")
return False
else:
log.error(f"Unexpected type for dest: {type(dest)}")
return False
@ -879,29 +885,27 @@ class WebRTC:
new_height = int(max_width / aspect_ratio)
return max_width, new_height
def on_remote_decodebin_stream(self, _, pad: Gst.Pad) -> None:
def on_remote_decodebin_stream(self, decodebin, pad: Gst.Pad) -> None:
"""Handle the stream from the remote decodebin.
This method processes the incoming stream from the remote decodebin, determining
whether it's video or audio. It then sets up the appropriate GStreamer elements
for video/audio processing and adds them to the pipeline.
@param decodebin: signal source.
@param pad: The Gst.Pad from the remote decodebin producing the stream.
"""
assert self.pipeline is not None
if not pad.has_current_caps():
log.error(f"{pad} has no caps, ignoring")
return
caps = pad.get_current_caps()
assert len(caps)
s = caps[0]
name = s.get_name()
log.debug(f"====> NAME START: {name}")
if not caps:
caps = pad.query_caps(None)
q = Gst.ElementFactory.make("queue")
name = caps.get_structure(0).get_name()
log.debug(f"Handling decoded pad of type {name}")
if name.startswith("video"):
queue = Gst.ElementFactory.make("queue")
if name.startswith('video/'):
log.debug("===> VIDEO OK")
self._remote_video_pad = pad
@ -921,7 +925,9 @@ class WebRTC:
height = (height + 3) // 4 * 4
log.info(f"Adjusted video size: {width}x{height}")
conv = Gst.ElementFactory.make("videoconvert")
convert = Gst.ElementFactory.make("videoconvert")
if self.merge_pip:
# with ``merge_pip`` set, we plug the remote stream to the composer
compositor = self.pipeline.get_by_name("compositor")
@ -949,31 +955,32 @@ class WebRTC:
remote_video_sink.set_property("width", width)
remote_video_sink.set_property("height", height)
remote_video_sink.set_property("sizing-policy", 1)
elif isinstance(self.sinks_data, SinksApp):
# ``app`` sink without ``self.merge_pip`` set, be create the sink and
# connect it to the ``remote_video_cb``.
remote_video_sink = Gst.ElementFactory.make("appsink")
remote_video_caps = Gst.Caps.from_string("video/x-raw,format=RGB")
remote_video_sink.set_property("caps", remote_video_caps)
remote_video_sink = Gst.ElementFactory.make("appsink", "remote_video_sink")
remote_video_sink.set_property("caps", Gst.Caps.from_string("video/x-raw,format=RGB"))
remote_video_sink.set_property("emit-signals", True)
remote_video_sink.set_property("drop", True)
remote_video_sink.set_property("max-buffers", 1)
remote_video_sink.set_property("sync", True)
remote_video_sink.connect("new-sample", self.sinks_data.remote_video_cb)
self.pipeline.add(remote_video_sink)
remote_video_sink.set_property("max-buffers", 2)
remote_video_sink.set_property("sync", False)
if self.sinks_data.remote_video_cb:
remote_video_sink.connect("new-sample", self.sinks_data.remote_video_cb)
else:
log.error("sinks_data.remote_video_cb is not available! Cannot display remote video.")
return
elif isinstance(self.sinks_data, SinksAuto):
# if ``self.merge_pip`` is not set, we create a dedicated
# ``autovideosink`` for remote stream.
remote_video_sink = Gst.ElementFactory.make("autovideosink")
self.pipeline.add(remote_video_sink)
else:
raise exceptions.InternalError(
f'Unhandled "sinks_data" value: {self.sinks_data!r}'
)
if adjust_resolution:
log.debug("Applying adjust resolution pipeline.")
videoscale = Gst.ElementFactory.make("videoscale")
adjusted_caps = Gst.Caps.from_string(
f"video/x-raw,width={width},height={height}"
@ -981,20 +988,20 @@ class WebRTC:
capsfilter = Gst.ElementFactory.make("capsfilter")
capsfilter.set_property("caps", adjusted_caps)
self.pipeline.add(q, conv, videoscale, capsfilter)
self.pipeline.add(queue, convert, videoscale, capsfilter)
self.pipeline.sync_children_states()
ret = pad.link(q.get_static_pad("sink"))
ret = pad.link(queue.get_static_pad("sink"))
if ret != Gst.PadLinkReturn.OK:
log.error(f"Error linking pad to queue: {ret}")
return
if not q.link(conv):
if not queue.link(convert):
log.error("Failed to link queue to videoconvert")
return
if not conv.link(videoscale):
if not convert.link(videoscale):
log.error("Failed to link videoconvert to videoscale")
return
@ -1007,50 +1014,81 @@ class WebRTC:
return
else:
self.pipeline.add(q, conv)
self.pipeline.add(queue, convert, remote_video_sink)
self.pipeline.sync_children_states()
ret = pad.link(q.get_static_pad("sink"))
if ret != Gst.PadLinkReturn.OK:
log.error(f"Error linking pad: {ret}")
q.link(conv)
self.link_element_or_pad(conv, remote_video_sink)
pad.link(queue.get_static_pad("sink"))
Gst.Element.link_many(queue, convert, remote_video_sink)
elif name.startswith("audio"):
queue.sync_state_with_parent()
convert.sync_state_with_parent()
remote_video_sink.sync_state_with_parent()
log.info("Successfully linked remote video stream.")
elif name.startswith('audio/'):
log.debug("===> Audio OK")
conv = Gst.ElementFactory.make("audioconvert")
# Create the audio sink chain: queue -> convert -> resample -> sink
convert = Gst.ElementFactory.make("audioconvert")
resample = Gst.ElementFactory.make("audioresample")
remote_audio_sink = Gst.ElementFactory.make("autoaudiosink")
self.pipeline.add(q, conv, resample, remote_audio_sink)
self.pipeline.sync_children_states()
ret = pad.link(q.get_static_pad("sink"))
if ret != Gst.PadLinkReturn.OK:
log.error(f"Error linking pad: {ret}")
q.link(conv)
conv.link(resample)
resample.link(remote_audio_sink)
remote_video_sink = Gst.ElementFactory.make("autoaudiosink")
self.pipeline.add(queue, convert, resample, remote_video_sink)
pad.link(queue.get_static_pad("sink"))
Gst.Element.link_many(queue, convert, resample, remote_video_sink)
# Bring all new elements to the PLAYING state
queue.sync_state_with_parent()
convert.sync_state_with_parent()
resample.sync_state_with_parent()
remote_video_sink.sync_state_with_parent()
log.info("Successfully linked remote audio stream.")
else:
log.warning(f"unmanaged name: {name!r}")
log.warning(f"Ignoring unknown pad type from decodebin: {name}")
def on_pad_added(self, __, pad: Gst.Pad) -> None:
def on_pad_added(self, webrtcbin, pad: Gst.Pad) -> None:
"""Handle the addition of a new pad to the element.
When a new source pad is added to the element, this method creates a decodebin,
connects it to handle the stream, and links the pad to the decodebin.
@param __: Placeholder for the signal source. Not used in this method.
@param webrtcbin: Signal source.
@param pad: The newly added pad.
"""
log.debug("on_pad_added")
assert self.pipeline is not None
log.debug(f"Attempting to handle new source pad {pad.get_name()} from {webrtcbin.get_name()}")
if pad.direction != Gst.PadDirection.SRC:
return
# Create the potential downstream element.
decodebin = Gst.ElementFactory.make("decodebin")
if not decodebin:
log.error("Critical: Failed to create decodebin element")
return
decodebin.connect("pad-added", self.on_remote_decodebin_stream)
self.pipeline.add(decodebin)
decodebin.sync_state_with_parent()
pad.link(decodebin.get_static_pad("sink"))
link_ret = pad.link(decodebin.get_static_pad("sink"))
if link_ret == Gst.PadLinkReturn.OK:
log.info(f"Successfully linked pad '{pad.get_name()}' to a new decodebin.")
elif link_ret == Gst.PadLinkReturn.WAS_LINKED:
log.debug(f"Pad '{pad.get_name()}' was already linked. Cleaning up redundant decodebin.")
# The pad is already handled. We must clean up the decodebin we just created.
decodebin.set_state(Gst.State.NULL)
self.pipeline.remove(decodebin)
else:
log.error(
f"Failed to link pad '{pad.get_name()}' for an unexpected reason: {link_ret}. "
"Cleaning up."
)
decodebin.set_state(Gst.State.NULL)
self.pipeline.remove(decodebin)
async def _call_start(self, callee: jid.JID, call_data: dict, profile: str) -> str:
return await self.bridge.call_start(