From fa2144a167e82a02e9eb01fdd47f1036a69150ca Mon Sep 17 00:00:00 2001 From: Goffi Date: Wed, 19 Nov 2025 17:18:44 +0100 Subject: [PATCH] frontends (tools/webrtc): Add `SinksPipeline`: This sinks allow to use custom pipelines, or to disable playback (e.g.; if we want to record only by setting `record_fd`). --- libervia/frontends/tools/webrtc.py | 117 +++++++++++++++++----- libervia/frontends/tools/webrtc_models.py | 19 ++++ 2 files changed, 109 insertions(+), 27 deletions(-) diff --git a/libervia/frontends/tools/webrtc.py b/libervia/frontends/tools/webrtc.py index 0d0b819e..9f929847 100644 --- a/libervia/frontends/tools/webrtc.py +++ b/libervia/frontends/tools/webrtc.py @@ -48,6 +48,7 @@ from .webrtc_models import ( SinksData, SinksDataChannel, SinksNone, + SinksPipeline, SourcesAuto, SourcesData, SourcesDataChannel, @@ -626,7 +627,7 @@ class WebRTC: local_video_sink_elt = "" else: local_video_sink_elt = "autovideosink" - case SinksNone(): + case SinksNone()|SinksPipeline(): local_video_sink_elt = "" case _: raise exceptions.InternalError( @@ -1195,10 +1196,32 @@ class WebRTC: else: log.error("sinks_data.remote_video_cb is not available! Cannot display remote video.") return - elif isinstance(self.sinks_data, SinksAuto): + elif ( + isinstance(self.sinks_data, SinksAuto) + or ( + # We have custom sinks, but video is automatic. + isinstance(self.sinks_data, SinksPipeline) + and self.sinks_data.video_pipeline is None + ) + ): # if ``self.merge_pip`` is not set, we create a dedicated # ``autovideosink`` for remote stream. remote_video_sink = Gst.ElementFactory.make("autovideosink") + elif isinstance(self.sinks_data, SinksPipeline): + # We have a custom video pipeline. + if self.sinks_data.video_pipeline == "": + # Video sink is deactivated. + remote_video_sink = Gst.ElementFactory.make("fakesink") + else: + remote_video_sink = Gst.parse_bin_from_description(self.sinks_data.video_pipeline) + iterator = remote_video_sink.iterate_elements() + ret, first_element = iterator.next() + + if ret == Gst.IteratorResult.OK: + for name, value in self.sinks_data.video_properties.items(): + first_element.set_property(name, value) + else: + log.warning("No elements found in the bin.") else: raise exceptions.InternalError( f'Unhandled "sinks_data" value: {self.sinks_data!r}' @@ -1269,20 +1292,47 @@ class WebRTC: elif name.startswith('audio/'): log.debug("===> Audio OK") # Create the audio sink chain: queue -> convert -> resample -> sink + if isinstance(self.sinks_data, SinksPipeline): + # We have a custom audio pipeline. + match self.sinks_data.audio_pipeline: + case None: + # Audio sink is automatic. + remote_audio_sink = Gst.ElementFactory.make("autoaudiosink") + case "": + # Audio sink is deactivated. + remote_audio_sink = Gst.ElementFactory.make("fakesink") + self.pipeline.add(queue, remote_audio_sink) + pad.link(queue.get_static_pad("sink")) + queue.link(remote_audio_sink) + queue.sync_state_with_parent() + remote_audio_sink.sync_state_with_parent() + log.info("Audio stream won't be played live.") + return + case _: + remote_audio_sink = Gst.parse_bin_from_description(self.sinks_data.audio_pipeline) + iterator = remote_audio_sink.iterate_elements() + ret, first_element = iterator.next() + + if ret == Gst.IteratorResult.OK: + for name, value in self.sinks_data.audio_properties.items(): + first_element.set_property(name, value) + else: + log.warning("No elements found in the bin.") + else: + remote_audio_sink = Gst.ElementFactory.make("autoaudiosink") + convert = Gst.ElementFactory.make("audioconvert") resample = Gst.ElementFactory.make("audioresample") - remote_video_sink = Gst.ElementFactory.make("autoaudiosink") - - self.pipeline.add(queue, convert, resample, remote_video_sink) + self.pipeline.add(queue, convert, resample, remote_audio_sink) pad.link(queue.get_static_pad("sink")) - Gst.Element.link_many(queue, convert, resample, remote_video_sink) + Gst.Element.link_many(queue, convert, resample, remote_audio_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() + remote_audio_sink.sync_state_with_parent() log.info("Successfully linked remote audio stream.") else: @@ -1347,33 +1397,46 @@ class WebRTC: self.pipeline.add(tee) - # Create decodebin for playback - decodebin = Gst.ElementFactory.make("decodebin", decodebin_name) - if not decodebin: - log.error(f"Failed to create decodebin for {media_type}.") - self.pipeline.remove(tee) - return - - decodebin.connect("pad-added", self.on_remote_decodebin_stream) - self.pipeline.add(decodebin) - # Link: webrtcbin -> tee if pad.link(tee.get_static_pad("sink")) != Gst.PadLinkReturn.OK: log.error(f"Failed to link webrtcbin pad to tee for {media_type}.") self.pipeline.remove(tee) - self.pipeline.remove(decodebin) - return - - # Link tee -> decodebin (playback branch) - playback_pad = tee.get_request_pad("src_%u") - if playback_pad.link(decodebin.get_static_pad("sink")) != Gst.PadLinkReturn.OK: - log.error(f"Failed to link tee to decodebin for {media_type}.") - self.pipeline.remove(tee) - self.pipeline.remove(decodebin) return tee.sync_state_with_parent() - decodebin.sync_state_with_parent() + + if ( + True + # FIXME: If the playback branch is not run, only audio stream is recorded. + # Something is missing when recording branch only is used. Requesting + # keyframe doesn't solve the issue. So for now we always activate + # playback branch, until the problem is fixed. To deactivate playback + # branch when no UI is shown, just uncomment the 3 next lines. + # not isinstance(self.sinks_data, SinksPipeline) + # or self.sinks_data.audio_pipeline != "" + # or self.sinks_data.video_pipeline != "" + ): + # There is a playback branch. + + # Create decodebin for playback. + decodebin = Gst.ElementFactory.make("decodebin", decodebin_name) + if not decodebin: + log.error(f"Failed to create decodebin for {media_type}.") + self.pipeline.remove(tee) + return + + decodebin.connect("pad-added", self.on_remote_decodebin_stream) + self.pipeline.add(decodebin) + + # Link tee -> decodebin (playback branch) + playback_pad = tee.get_request_pad("src_%u") + if playback_pad.link(decodebin.get_static_pad("sink")) != Gst.PadLinkReturn.OK: + log.error(f"Failed to link tee to decodebin for {media_type}.") + self.pipeline.remove(tee) + self.pipeline.remove(decodebin) + return + + decodebin.sync_state_with_parent() # If recording is active, add recording branch from tee if self._record_fd is not None: diff --git a/libervia/frontends/tools/webrtc_models.py b/libervia/frontends/tools/webrtc_models.py index fe0880fe..dc212369 100644 --- a/libervia/frontends/tools/webrtc_models.py +++ b/libervia/frontends/tools/webrtc_models.py @@ -120,3 +120,22 @@ class SinksDataChannel(SinksData): dc_on_data_channel: ( Callable[[GstWebRTC.WebRTCDataChannel], Awaitable[None]] | None ) = None + + +class SinksPipeline(SinksData): + """Use custom pipeline description as a sink. + + @param video_pipeline: pipeline description of video sink. + None to use automatic video sink (same as SinksAuto). + Empty string to disable video playback. + @param audio_pipeline: pipeline description of audio sink. + None to use automatic audio sink (same as SinksAuto). + Empty string to disable audio playback. + @param video_properties: Elements properties to set. + @param audio_properties: Elements properties to set. + """ + + video_pipeline: str | None = None + audio_pipeline: str | None = None + video_properties: dict = Field(default_factory=lambda: {}) + audio_properties: dict = Field(default_factory=lambda: {})