diff --git a/libervia/frontends/tools/webrtc.py b/libervia/frontends/tools/webrtc.py index 0097ff80..d54b52b2 100644 --- a/libervia/frontends/tools/webrtc.py +++ b/libervia/frontends/tools/webrtc.py @@ -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(