From 2fce8ba7d6c04bf9c69b1a9e07ed84ce159e6dcf Mon Sep 17 00:00:00 2001 From: Goffi Date: Wed, 29 Oct 2025 13:20:45 +0100 Subject: [PATCH] frontends (tools/webrtc): Better error handling and logging --- libervia/frontends/tools/webrtc.py | 172 ++++++++++++++++++++++------- 1 file changed, 133 insertions(+), 39 deletions(-) diff --git a/libervia/frontends/tools/webrtc.py b/libervia/frontends/tools/webrtc.py index e3cbf0d1..8140ba60 100644 --- a/libervia/frontends/tools/webrtc.py +++ b/libervia/frontends/tools/webrtc.py @@ -296,6 +296,7 @@ class WebRTC: mline_index += 1 self.media_types = media_types + log.debug(f"Media types mapping: {media_types}") def _a_call(self, method, *args, **kwargs): """Call an async method in main thread""" @@ -335,6 +336,10 @@ class WebRTC: @param candidate_string: The ice candidate string to be parsed. """ + # Skip empty candidates + if not candidate_string or candidate_string.strip() == "": + return None + pattern = re.compile( r"candidate:(?P\S+) (?P\d+) (?P\S+) " r"(?P\d+) (?P
\S+) (?P\d+) typ " @@ -417,15 +422,21 @@ class WebRTC: if bundle_ufrag and bundle_pwd: for mid in bundle_media: - media = mid_media_map[mid] - self.ufrag[media] = bundle_ufrag - self.pwd[media] = bundle_pwd + media = mid_media_map.get(mid) + if media: + self.ufrag[media] = bundle_ufrag + self.pwd[media] = bundle_pwd + log.debug( + f"Set bundle ICE credentials for {media}: ufrag={bundle_ufrag[:4]}..." + ) + else: + log.warning(f"No media type found for mid {mid}") def reset_instance(self): """Inits or resets the instance variables to their default state.""" self.role: str | None = None if self.pipeline is not None: - self.pipeline.set_state(Gst.State.NULL) + self.set_pipeline_state(Gst.State.NULL) self.pipeline = None self._remote_video_pad = None self.sid: str | None = None @@ -635,6 +646,8 @@ class WebRTC: if self.webrtcbin is None: raise exceptions.InternalError("Can't get the pipeline.") + self.monitor_connection_state() + # If video or audio sources are not created, ``get_by_name`` will return None. self.video_src = self.pipeline.get_by_name("video_src") self.muted_src = self.pipeline.get_by_name("muted_src") @@ -716,30 +729,68 @@ class WebRTC: def create_data_channel(self, dc_data: SourcesDataChannel | SinksDataChannel) -> None: """Create a Data Channel and connect relevant callbacks.""" assert self.pipeline is not None - if isinstance(dc_data, SourcesDataChannel): - # Data channel configuration for compatibility with browser defaults - data_channel_options = Gst.Structure.new_empty("data-channel-options") - data_channel_options.set_value("ordered", True) - data_channel_options.set_value("protocol", "") + match dc_data: + case SourcesDataChannel(): + # Data channel configuration for compatibility with browser defaults + data_channel_options = Gst.Structure.new_empty("data-channel-options") + data_channel_options.set_value("ordered", True) + data_channel_options.set_value("protocol", "") - # Create the data channel - self.pipeline.set_state(Gst.State.READY) - self.data_channels[dc_data.name] = data_channel = self.webrtcbin.emit( - "create-data-channel", dc_data.name, data_channel_options - ) - if data_channel is None: - log.error("Failed to create data channel") - return - data_channel.connect("on-open", dc_data.dc_open_cb) - elif isinstance(dc_data, SinksDataChannel): - self.webrtcbin.connect("on-data-channel", dc_data.dc_on_data_channel) - else: - raise ValueError("Only SourcesDataChannel or SinksDataChannel are allowed.") + # Create the data channel + self.pipeline.set_state(Gst.State.READY) + self.data_channels[dc_data.name] = data_channel = self.webrtcbin.emit( + "create-data-channel", dc_data.name, data_channel_options + ) + if data_channel is None: + log.error("Failed to create data channel") + return + data_channel.connect("on-open", dc_data.dc_open_cb) + case SinksDataChannel(): + self.webrtcbin.connect("on-data-channel", dc_data.dc_on_data_channel) + case _: + raise ValueError( + "Only SourcesDataChannel or SinksDataChannel are allowed." + ) + + def set_pipeline_state(self, state: Gst.State) -> bool: + """Safely set pipeline state with verification. + + @param state: The target state + @return: True if state change was successful + """ + if not self.pipeline: + log.error("Pipeline not initialized") + return False + + ret = self.pipeline.set_state(state) + if ret == Gst.StateChangeReturn.FAILURE: + log.error(f"Failed to change pipeline state to {state}") + 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") + + return True + + def monitor_connection_state(self): + """Monitor WebRTC connection state""" + + def on_notify_connection_state(pspect, param): + state = self.webrtcbin.get_property("connection-state") + log.info(f"WebRTC connection state: {state}") + + if state == GstWebRTC.WebRTCPeerConnectionState.FAILED: + log.error("WebRTC connection failed") + elif state == GstWebRTC.WebRTCPeerConnectionState.CONNECTED: + log.info("WebRTC connection established") + + self.webrtcbin.connect("notify::connection-state", on_notify_connection_state) def start_pipeline(self) -> None: """Starts the GStreamer pipeline.""" log.debug("starting the pipeline") - self.pipeline.set_state(Gst.State.PLAYING) + self.set_pipeline_state(Gst.State.PLAYING) def on_negotiation_needed(self, webrtc): """Initiate SDP offer when negotiation is needed.""" @@ -933,13 +984,27 @@ class WebRTC: self.pipeline.add(q, conv, videoscale, capsfilter) 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(videoscale) - videoscale.link(capsfilter) - self.link_element_or_pad(capsfilter.link, remote_video_sink) + log.error(f"Error linking pad to queue: {ret}") + return + + if not q.link(conv): + log.error("Failed to link queue to videoconvert") + return + + if not conv.link(videoscale): + log.error("Failed to link videoconvert to videoscale") + return + + if not videoscale.link(capsfilter): + log.error("Failed to link videoscale to capsfilter") + return + + if not self.link_element_or_pad(capsfilter.link, remote_video_sink): + log.error("Failed to link capsfilter to sink") + return else: self.pipeline.add(q, conv) @@ -1010,16 +1075,29 @@ class WebRTC: if not self.pwd: sdp = self.webrtcbin.props.local_description.sdp.as_text() self.extract_ufrag_pwd(sdp) + ice_data = {} for media_type, candidates in self.local_candidates_buffer.items(): + if media_type not in self.ufrag or media_type not in self.pwd: + log.error(f"Missing ICE credentials for {media_type}") + log.debug(f"Available ufrag keys: {list(self.ufrag.keys())}") + log.debug(f"Available pwd keys: {list(self.pwd.keys())}") + continue + ice_data[media_type] = { "ufrag": self.ufrag[media_type], "pwd": self.pwd[media_type], "candidates": candidates, } - await self.bridge.ice_candidates_add( - self.sid, data_format.serialise(ice_data), self.profile - ) + + if ice_data: + log.debug(f"Sending ICE data for media types: {list(ice_data.keys())}") + await self.bridge.ice_candidates_add( + self.sid, data_format.serialise(ice_data), self.profile + ) + else: + log.error("No ICE data to send - missing credentials!") + self.local_candidates_buffer.clear() def _remote_sdp_set(self, promise) -> None: @@ -1079,6 +1157,11 @@ class WebRTC: @param mlineindex: The mline index. @param candidate: The ICE candidate. """ + # Empty candidate signals end of candidates for this mline + if not candidate_sdp or candidate_sdp.strip() == "": + log.debug(f"ICE gathering complete for mline {mline_index}") + return + log.debug( f"Local ICE candidate. MLine Index: {mline_index}, Candidate: {candidate_sdp}" ) @@ -1086,16 +1169,20 @@ class WebRTC: if parsed_candidate is None: log.warning(f"Can't parse candidate: {candidate_sdp}") return + try: media_type = self.media_types[mline_index] except KeyError: - raise exceptions.InternalError("can't find media type") + log.error(f"Can't find media type for mline {mline_index}") + log.debug(f"{self.media_types=}") + return if self.sid is None: - log.debug("buffering local ICE candidate") + log.debug(f"buffering local ICE candidate for {media_type}") self.local_candidates_buffer.setdefault(media_type, []).append( parsed_candidate ) + else: sdp = self.webrtcbin.props.local_description.sdp.as_text() assert sdp is not None @@ -1119,6 +1206,8 @@ class WebRTC: keys and corresponding ICE data as values. @raise exceptions.InternalError: Raised when sdp mline index is not found. + + """ if not self.sdp_set: log.debug("buffering remote ICE candidate") @@ -1136,9 +1225,9 @@ class WebRTC: except Exception as e: raise exceptions.InternalError(f"Can't find sdp mline index: {e}") self.webrtcbin.emit("add-ice-candidate", mline_index, candidate_sdp) - log.warning( + log.debug( f"Remote ICE candidate added. MLine Index: {mline_index}, " - f"{candidate_sdp}" + f"Candidate: {candidate_sdp}" ) def on_ice_gathering_state_change(self, pspec, __): @@ -1147,9 +1236,14 @@ class WebRTC: def on_ice_connection_state(self, pspec, __): state = self.webrtcbin.props.ice_connection_state + log.info(f"ICE connection state changed to: {state}") + if state == GstWebRTC.WebRTCICEConnectionState.FAILED: - log.error("ICE connection failed") - log.info(f"ICE connection state changed to {state}") + log.error("ICE connection failed - check firewall/STUN/TURN configuration") + elif state == GstWebRTC.WebRTCICEConnectionState.CONNECTED: + log.info("ICE connection established successfully") + elif state == GstWebRTC.WebRTCICEConnectionState.COMPLETED: + log.info("ICE connectivity checks completed") def on_bus_error(self, bus: Gst.Bus, message: Gst.Message) -> None: """Handles the GStreamer bus error messages. @@ -1238,7 +1332,7 @@ class WebRTC: f"Invalid source: {source!r}, use one of {'video', 'muted', 'desktop'}" ) - self.pipeline.set_state(Gst.State.PAUSED) + self.set_pipeline_state(Gst.State.PAUSED) # Create a new desktop source if necessary if source == "desktop": @@ -1271,7 +1365,7 @@ class WebRTC: if pad is not None: self.video_selector.props.active_pad = pad - self.pipeline.set_state(Gst.State.PLAYING) + self.set_pipeline_state(Gst.State.PLAYING) def _setup_desktop_source(self, properties: dict[str, object] | None) -> None: """Set up a new desktop source.