libervia-backend: libervia/frontends/tools/webrtc.py comparison

comparison libervia/frontends/tools/webrtc.py @ 4270:0d7bb4df2343

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	79c8a70e1813
children	2992f9d1e039

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 def __init__(
 self,
 bridge,
 profile: str,
-sources_data: SourcesData|None = None,
+sources_data: SourcesData | None = None,
-sinks_data: SinksData|None = None,
+sinks_data: SinksData | None = None,
 reset_cb: Callable | None = None,
 merge_pip: bool | None = None,
 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
+dc_data_list: list[SourcesDataChannel | SinksDataChannel] | None = None,
 ) -> None:
 """Initializes a new WebRTC instance.
 @param bridge: An instance of backend bridge.
 @param profile: Libervia profile.
 """Retrieves ICE password and user fragment for SDP offer.
 @param sdp: The Session Description Protocol offer string.
 """
 lines = sdp.splitlines()
-media = ''
+media = ""
 mid_media_map = {}
 bundle_media = set()
-bundle_ufrag = ''
+bundle_ufrag = ""
-bundle_pwd = ''
+bundle_pwd = ""
 in_bundle = False
 for line in lines:
-if line.startswith('m='):
+if line.startswith("m="):
-media = line.split('=')[1].split()[0]
+media = line.split("=")[1].split()[0]
-elif line.startswith('a=mid:'):
+elif line.startswith("a=mid:"):
-mid = line.split(':')[1].strip()
+mid = line.split(":")[1].strip()
 mid_media_map[mid] = media
-elif line.startswith('a=group:BUNDLE'):
+elif line.startswith("a=group:BUNDLE"):
 in_bundle = True
-bundle_media = set(line.split(':')[1].strip().split())
+bundle_media = set(line.split(":")[1].strip().split())
-elif line.startswith('a=ice-ufrag:'):
+elif line.startswith("a=ice-ufrag:"):
 if in_bundle:
-bundle_ufrag = line.split(':')[1].strip()
+bundle_ufrag = line.split(":")[1].strip()
 else:
-self.ufrag[media] = line.split(':')[1].strip()
+self.ufrag[media] = line.split(":")[1].strip()
-elif line.startswith('a=ice-pwd:'):
+elif line.startswith("a=ice-pwd:"):
 if in_bundle:
-bundle_pwd = line.split(':')[1].strip()
+bundle_pwd = line.split(":")[1].strip()
 else:
-self.pwd[media] = line.split(':')[1].strip()
+self.pwd[media] = line.split(":")[1].strip()
 else:
 in_bundle = False
 if bundle_ufrag and bundle_pwd:
 for mid in bundle_media:
 @raises AssertionError: If the role is not 'initiator' or 'responder'.
 """
 assert role in ("initiator", "responder")
 self.role = role
 if isinstance(self.sources_data, SourcesPipeline):
-if self.sources_data.video_pipeline!= "" and video_pt is None:
+if self.sources_data.video_pipeline != "" and video_pt is None:
 raise NotImplementedError(NONE_NOT_IMPLEMENTED_MSG)
-if self.sources_data.audio_pipeline!= "" and audio_pt is None:
+if self.sources_data.audio_pipeline != "" and audio_pt is None:
 raise NotImplementedError(NONE_NOT_IMPLEMENTED_MSG)
 elif isinstance(self.sources_data, SourcesNone):
 pass
 else:
 if audio_pt is None or video_pt is None:
 local_video_sink_elt = "compositor.sink_1"
 if video_source_elt:
 # Video source with an input-selector to switch between normal and video mute
 # (or desktop sharing).
-gst_pipe_elements.append(f"""
+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
 ! videoconvert
 ! vp8enc deadline=1 keyframe-max-dist=30
 ! rtpvp8pay picture-id-mode=15-bit
 ! application/x-rtp,media=video,encoding-name=VP8,payload={video_pt}
 ! sendrecv.
-""")
+"""
+)
 if local_video_sink_elt:
 # Local video feedback.
-gst_pipe_elements.append(f"""
+gst_pipe_elements.append(
+f"""
 t.
 ! queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream
 ! videoconvert
 ! {local_video_sink_elt}
-""")
+"""
+)
 if audio_source_elt:
 # Audio with a valve for muting.
-gst_pipe_elements.append(r"""
+gst_pipe_elements.append(
+r"""
 {audio_source_elt} name=audio_src
 ! valve
 ! queue max-size-buffers=10 max-size-time=0 max-size-bytes=0 leaky=downstream
 ! audioconvert
 ! audioresample
 ! opusenc audio-type=voice
 ! rtpopuspay
 ! application/x-rtp,media=audio,encoding-name=OPUS,payload={audio_pt}
 ! sendrecv.
-""")
+"""
+)
 self.gst_pipe_desc = "\n\n".join(gst_pipe_elements)
 log.debug(f"Gstreamer pipeline: {self.gst_pipe_desc}")
 )
 for dc_data in self.dc_data_list:
 self.create_data_channel(dc_data)
-def create_data_channel(self, dc_data: SourcesDataChannel|SinksDataChannel) -> None:
+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")
 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(
+raise ValueError("Only SourcesDataChannel or SinksDataChannel are allowed.")
-"Only SourcesDataChannel or SinksDataChannel are allowed."
-)
 def start_pipeline(self) -> None:
 """Starts the GStreamer pipeline."""
 log.debug("starting the pipeline")
 self.pipeline.set_state(Gst.State.PLAYING)
 assert sdp is not None
 self.extract_ufrag_pwd(sdp)
 ice_data = {
 "ufrag": self.ufrag[media_type],
 "pwd": self.pwd[media_type],
-"candidates": [parsed_candidate]
+"candidates": [parsed_candidate],
 }
 self._a_call(
 self.bridge.ice_candidates_add,
 self.sid,
 data_format.serialise({media_type: ice_data}),

Mercurial > libervia-backend

comparison libervia/frontends/tools/webrtc.py @ 4270:0d7bb4df2343