diff --git a/libervia/frontends/tools/webrtc.py b/libervia/frontends/tools/webrtc.py index d54b52b2..0d0b819e 100644 --- a/libervia/frontends/tools/webrtc.py +++ b/libervia/frontends/tools/webrtc.py @@ -17,6 +17,7 @@ # along with this program. If not, see . from collections.abc import Awaitable +import time import gi gi.require_versions({"Gst": "1.0", "GstWebRTC": "1.0"}) @@ -104,6 +105,7 @@ class WebRTC: target_size: tuple[int, int] | None = None, call_start_cb: Callable[[str, dict, str], Awaitable[str]] | None = None, dc_data_list: list[SourcesDataChannel | SinksDataChannel] | None = None, + record_fd: int | None = None ) -> None: """Initializes a new WebRTC instance. @@ -130,12 +132,14 @@ class WebRTC: used for SinksAuto only for now). @param target_size: Expected size of the final sink stream. Mainly used by composer when ``merge_pip`` is set. - None to autodetect (no real autodetection implemeted yet, default to + None to autodetect (no real autodetection implemented yet, default to (1280,720)). @param call_start_cb: Called when call is started. @param dc_data_list: Data Channels to create. If a SourcesDataChannel is used as ``sources_data``, or a SinksDataChannel is used as ``sinks_data``, they will be automatically added to this list. + @param record_fd: File descriptor for recording the incoming stream. + If None, the stream will not be recorded. """ self.main_loop = asyncio.get_event_loop() self.bridge = bridge @@ -182,6 +186,9 @@ class WebRTC: else: self.desktop_portal = None self._gap_recovery_timeout = None + self._record_fd: int | None = record_fd + self._recording_elements: list[Gst.Element] = [] + self._recording_muxer: Gst.Element | None = None self.reset_instance() @property @@ -270,6 +277,31 @@ class WebRTC: # Also send any local candidates we buffered before this point. self._send_buffered_local_candidates() + @property + def record_fd(self) -> int|None: + return self._record_fd + + @record_fd.setter + def record_fd(self, value: int|None) -> None: + """Set or unset the recording file descriptor. + + When set during an active call, starts recording. + When unset during recording, stops recording. + """ + if self._record_fd == value: + return + + if value is not None: + raise NotImplementedError( + "Changing record_fd during a call is not supported yet" + ) + + self._record_fd = value + + if value is None: + log.info("Stopping recording.") + self._remove_recording_branch() + def bind(self, **kwargs: Callable) -> None: self.bindings.clear() for key, cb in kwargs.items(): @@ -471,6 +503,8 @@ class WebRTC: def reset_instance(self): """Inits or resets the instance variables to their default state.""" + if self._recording_elements: + self._remove_recording_branch() self.role: str | None = None if self.pipeline is not None: self.set_pipeline_state(Gst.State.NULL) @@ -1019,6 +1053,50 @@ class WebRTC: else: log.error("Failed to send GstForceKeyUnit event to decodebin.") + def _remove_recording_branch(self) -> None: + """Removes the recording branch from the pipeline.""" + if not self.pipeline or not self._recording_elements: + log.warning("No recording branch to remove") + return + + log.info("Removing recording branch from the pipeline.") + + current_state = self.pipeline.get_state(0)[1] + + if current_state == Gst.State.PLAYING: + self.set_pipeline_state(Gst.State.PAUSED) + + try: + # Release muxer pads before removing elements + if self._recording_muxer: + for element in self._recording_elements: + if "depay" in element.get_name(): + depay_src_pad = element.get_static_pad("src") + if depay_src_pad: + peer_pad = depay_src_pad.get_peer() + if peer_pad and self._recording_muxer == peer_pad.get_parent(): + log.debug(f"Releasing muxer pad: {peer_pad.get_name()}") + self._recording_muxer.release_request_pad(peer_pad) + + # Remove elements in reverse order + elements_to_remove = list(self._recording_elements) + for element in reversed(elements_to_remove): + if element.get_parent() == self.pipeline: + log.debug(f"Removing recording element: {element.get_name()}") + element.set_state(Gst.State.NULL) + self.pipeline.remove(element) + self._recording_elements.remove(element) + + self._recording_muxer = None + log.info("Recording branch successfully removed") + except Exception: + log.exception("Error removing recording branch") + self._recording_elements.clear() + self._recording_muxer = None + finally: + if current_state == Gst.State.PLAYING: + self.set_pipeline_state(Gst.State.PLAYING) + def on_remote_decodebin_stream(self, decodebin, pad: Gst.Pad) -> None: """Handle the stream from the remote decodebin. @@ -1211,46 +1289,215 @@ class WebRTC: log.warning(f"Ignoring unknown pad type from decodebin: {name}") def on_pad_added(self, webrtcbin, pad: Gst.Pad) -> None: - """Handle the addition of a new pad to the element. + """Handle the addition of a new pad from webrtcbin. - 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. + Always creates a tee to allow dynamic recording branch addition. + The tee splits the stream: one branch goes to decodebin for playback, + another can go to recording if enabled. @param webrtcbin: Signal source. @param pad: The newly added pad. """ assert self.pipeline is not None - log.debug(f"Attempting to handle new source pad {pad.get_name()} from {webrtcbin.get_name()}") + log.debug(f"Handling new source pad {pad.get_name()}.") + if pad.direction != Gst.PadDirection.SRC: + log.debug(f"Ignoring non-source pad: {pad.get_name()}") return - # Create the potential downstream element. - decodebin = Gst.ElementFactory.make("decodebin") - if not decodebin: - log.error("Critical: Failed to create decodebin element") + # Determine media type from caps. + caps = pad.get_current_caps() or pad.query_caps(None) + if not caps or caps.get_size() == 0: + log.warning("Empty caps on pad, ignoring.") return - decodebin.connect("pad-added", self.on_remote_decodebin_stream) + structure = caps.get_structure(0) + structure_name = structure.get_name() + log.debug(f"Pad caps structure: {structure_name}.") - self.pipeline.add(decodebin) - decodebin.sync_state_with_parent() + # We only handle RTP stream. + if structure_name != "application/x-rtp": + log.debug(f"Ignoring non-RTP pad with caps: {structure_name}.") + return - link_ret = pad.link(decodebin.get_static_pad("sink")) + # Extract media type. + media_type = structure.get_string("media") + if media_type not in ["video", "audio"]: + log.warning(f"Ignoring RTP pad with unexpected media type: {media_type}.") + return - if link_ret == Gst.PadLinkReturn.OK: - log.debug(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.debug(f"Processing incoming {media_type} stream.") + + current_state = self.pipeline.get_state(0)[1] + needs_state_restore = (current_state == Gst.State.PLAYING) + + if needs_state_restore: + log.debug("Pausing pipeline for safe modification.") + self.set_pipeline_state(Gst.State.PAUSED) + + tee_name = f"{media_type}_tee" + decodebin_name = f"decodebin_{media_type}" + + try: + # Create tee for this media stream (always present) + tee = Gst.ElementFactory.make("tee", tee_name) + if not tee: + log.error(f"Failed to create tee for {media_type}.") + return + + 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 recording is active, add recording branch from tee + if self._record_fd is not None: + log.info(f"Recording is active, adding recording branch for {media_type}.") + recording_pad = tee.get_request_pad("src_%u") + if not self._add_rtp_recording_branch(recording_pad, media_type): + log.warning(f"Failed to add recording branch for {media_type}") + tee.release_request_pad(recording_pad) + else: + log.info(f"Recording branch added for {media_type}.") + + log.info(f"Successfully set up {media_type} pipeline with tee.") + + except Exception: + log.exception(f"Error in on_pad_added for {media_type}.") + self._cleanup_elements([tee_name, decodebin_name]) + + finally: + if needs_state_restore: + log.debug("Restoring pipeline to PLAYING state.") + self.set_pipeline_state(Gst.State.PLAYING) + log.debug("Finished handling pad addition") + + def _add_rtp_recording_branch(self, src_pad: Gst.Pad, media_type: str) -> bool: + """Add recording branch for RTP stream. + + RTP stream are muxed and written as-is, without re-encoding. + + @param src_pad: Source pad. + @param media_type: Type of the media ("video" or "audio"). + @return: True is successful. + """ + assert self.pipeline is not None + assert self._record_fd is not None + + log.debug(f"Adding RTP recording branch for {media_type}") + + queue = Gst.ElementFactory.make("queue", f"rec_queue_{media_type}") + if not queue: + log.error(f"Failed to create queue for {media_type} recording") + return False + + depayloader_type = "rtpvp8depay" if media_type == "video" else "rtpopusdepay" + depayloader_name = f"{depayloader_type}_rec_{media_type}" + depayloader = Gst.ElementFactory.make(depayloader_type, depayloader_name) + + if not depayloader: + log.error(f"Failed to create {depayloader_type}") + return False + + # Create muxer and sink if they don't exist + if self._recording_muxer is None: + self._recording_muxer = Gst.ElementFactory.make("webmmux", "recording_muxer") + fdsink = Gst.ElementFactory.make("fdsink", "recording_sink") + + if not self._recording_muxer or not fdsink: + log.error("Failed to create muxer or sink") + self._recording_muxer = None + return False + + # Configure muxer for live streaming + self._recording_muxer.set_property("streamable", True) + + fdsink.set_property("fd", self._record_fd) + fdsink.set_property("sync", False) + fdsink.set_property("async", False) + + self.pipeline.add(self._recording_muxer) + self.pipeline.add(fdsink) + + if not self._recording_muxer.link(fdsink): + log.error("Failed to link muxer to sink") + self._cleanup_elements([self._recording_muxer, fdsink], self._recording_elements) + self._recording_muxer = None + return False + + self._recording_elements.extend([self._recording_muxer, fdsink]) + self._recording_muxer.sync_state_with_parent() + fdsink.sync_state_with_parent() + + self.pipeline.add(queue) + self.pipeline.add(depayloader) + + # Link: src_pad -> queue -> depayloader + if src_pad.link(queue.get_static_pad("sink")) != Gst.PadLinkReturn.OK: + log.error(f"Failed to link tee src pad to queue for {media_type}") + self._cleanup_elements([queue, depayloader], self._recording_elements) + return False + + if not queue.link(depayloader): + log.error(f"Failed to link queue to depayloader for {media_type}") + self._cleanup_elements([queue, depayloader], self._recording_elements) + return False + + # Link depayloader to muxer + depayloader_src_pad = depayloader.get_static_pad("src") + muxer_pad_name = "video_%u" if media_type == "video" else "audio_%u" + muxer_sink_pad = self._recording_muxer.get_request_pad(muxer_pad_name) + + if not muxer_sink_pad: + log.error(f"Failed to get request pad '{muxer_pad_name}' from muxer") + self._cleanup_elements([queue, depayloader], self._recording_elements) + return False + + link_result = depayloader_src_pad.link(muxer_sink_pad) + if link_result != Gst.PadLinkReturn.OK: + if hasattr(link_result, "value_nick"): + reason = link_result.value_nick + else: + reason = link_result log.error( - f"Failed to link pad '{pad.get_name()}' for an unexpected reason: {link_ret}. " - "Cleaning up." + f"Failed to link depayloader to muxer for {media_type} (reason: {reason})" ) - decodebin.set_state(Gst.State.NULL) - self.pipeline.remove(decodebin) + self._cleanup_elements([queue, depayloader], self._recording_elements) + self._recording_muxer.release_request_pad(muxer_sink_pad) + return False + + queue.sync_state_with_parent() + depayloader.sync_state_with_parent() + + self._recording_elements.extend([queue, depayloader]) + + log.debug(f"Successfully added {media_type} RTP recording branch") + return True def _setup_stdin_source(self) -> None: """Setup stdin source with decodebin for automatic format detection.""" @@ -1859,22 +2106,22 @@ class WebRTC: # then convert it to a more standard format that videorate can handle easily. if not desktop_src.link(video_scale): log.error("Failed to link desktop_src to videoscale") - self._cleanup_desktop_elements(desktop_elements) + self._cleanup_elements(desktop_elements) return if not video_scale.link(video_convert): log.error("Failed to link videoscale to videoconvert") - self._cleanup_desktop_elements(desktop_elements) + self._cleanup_elements(desktop_elements) return if not video_convert.link(video_rate): log.error("Failed to link videoconvert to videorate") - self._cleanup_desktop_elements(desktop_elements) + self._cleanup_elements(desktop_elements) return if not video_rate.link(queue): log.error("Failed to link videorate to queue") - self._cleanup_desktop_elements(desktop_elements) + self._cleanup_elements(desktop_elements) return sink_pad_template = self.video_selector.get_pad_template("sink_%u") @@ -1883,7 +2130,7 @@ class WebRTC: ) if not self.desktop_sink_pad: log.error("Failed to request a new sink pad from video_selector.") - self._cleanup_desktop_elements(desktop_elements) + self._cleanup_elements(desktop_elements) return log.debug(f"Requested and received pad {self.desktop_sink_pad.get_name()} from video_selector.") @@ -1892,7 +2139,7 @@ class WebRTC: link_ret = queue_src_pad.link(self.desktop_sink_pad) if link_ret != Gst.PadLinkReturn.OK: log.error(f"Failed to link queue to video_selector: {link_ret}") - self._cleanup_desktop_elements(desktop_elements) + self._cleanup_elements(desktop_elements) self.video_selector.release_request_pad(self.desktop_sink_pad) self.desktop_sink_pad = None return @@ -1903,15 +2150,27 @@ class WebRTC: log.debug("Desktop source pipeline successfully created and linked.") - def _cleanup_desktop_elements(self, elements: list[Gst.Element]) -> None: - """Helper to cleanup elements on error. + def _cleanup_elements( + self, + elements: list[Gst.Element], + elements_list: list[Gst.Element]|None = None + ) -> None: + """Helper to cleanup elements. @param elements: List of elements to remove from pipeline """ + if self.pipeline is None: + return + for element in elements: - if element: - element.set_state(Gst.State.NULL) - self.pipeline.remove(element) + try: + if element: + element.set_state(Gst.State.NULL) + self.pipeline.remove(element) + if elements_list and element in elements_list: + elements_list.remove(element) + except Exception as e: + log.warning(f"Error cleaning up element {element.get_name() if element else 'Unknown'}: {e}") def _remove_desktop_source(self, desktop_src: Gst.Element) -> None: """Remove the desktop source from the pipeline. @@ -1928,10 +2187,7 @@ class WebRTC: # Unlink and remove in reverse order elements = [queue, capsfilter, video_scale, video_rate, video_convert, desktop_src] - for element in elements: - if element: - element.set_state(Gst.State.NULL) - self.pipeline.remove(element) + self._cleanup_elements(elements) # Release the pad associated with the desktop source if self.desktop_sink_pad: