Mercurial > libervia-web
changeset 1517:b8ed9726525b
browser: "calls" implementation, first draft:
Basic page to make a call. Call are automatically accepted for the moment.
fix 422
author | Goffi <goffi@goffi.org> |
---|---|
date | Thu, 01 Jun 2023 21:42:02 +0200 |
parents | a3ca1bab6eb1 |
children | eb00d593801d |
files | libervia/pages/calls/_browser/__init__.py libervia/pages/calls/page_meta.py libervia/server/constants.py |
diffstat | 3 files changed, 533 insertions(+), 0 deletions(-) [+] |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/libervia/pages/calls/_browser/__init__.py Thu Jun 01 21:42:02 2023 +0200 @@ -0,0 +1,513 @@ +import json +import re + +from bridge import AsyncBridge as Bridge +from browser import aio, console as log, document, timer, window +import errors +import loading + +log.warning = log.warn +profile = window.profile or "" +bridge = Bridge() +GATHER_TIMEOUT = 10000 + + +class WebRTCCall: + + def __init__(self): + self.reset_instance() + + def reset_instance(self): + """Inits or resets the instance variables to their default state.""" + self._peer_connection = None + self._media_types = None + self.sid = None + self.local_candidates = None + self.remote_stream = None + self.candidates_buffer = { + "audio": {"candidates": []}, + "video": {"candidates": []}, + } + self.media_candidates = {} + self.candidates_gathered = aio.Future() + + @property + def media_types(self): + if self._media_types is None: + raise Exception("self._media_types should not be None!") + return self._media_types + + def get_sdp_mline_index(self, media_type): + """Gets the sdpMLineIndex for a given media type. + + @param media_type: The type of the media. + """ + for index, m_type in self.media_types.items(): + if m_type == media_type: + return index + raise ValueError(f"Media type '{media_type}' not found") + + def extract_pwd_ufrag(self, sdp): + """Retrieves ICE password and user fragment for SDP offer. + + @param sdp: The Session Description Protocol offer string. + """ + ufrag_line = re.search(r"ice-ufrag:(\S+)", sdp) + pwd_line = re.search(r"ice-pwd:(\S+)", sdp) + + if ufrag_line and pwd_line: + return ufrag_line.group(1), pwd_line.group(1) + else: + log.error(f"SDP with missing ice-ufrag or ice-pwd:\n{sdp}") + raise ValueError("Can't extract ice-ufrag and ice-pwd from SDP") + + def extract_fingerprint_data(self, sdp): + """Retrieves fingerprint data from an SDP offer. + + @param sdp: The Session Description Protocol offer string. + @return: A dictionary containing the fingerprint data. + """ + fingerprint_line = re.search(r"a=fingerprint:(\S+)\s+(\S+)", sdp) + if fingerprint_line: + algorithm, fingerprint = fingerprint_line.groups() + fingerprint_data = { + "hash": algorithm, + "fingerprint": fingerprint + } + + setup_line = re.search(r"a=setup:(\S+)", sdp) + if setup_line: + setup = setup_line.group(1) + fingerprint_data["setup"] = setup + + return fingerprint_data + else: + raise ValueError("fingerprint should not be missing") + + def parse_ice_candidate(self, candidate_string): + """Parses the ice candidate string. + + @param candidate_string: The ice candidate string to be parsed. + """ + pattern = re.compile( + r"candidate:(?P<foundation>\S+) (?P<component_id>\d+) (?P<transport>\S+) " + r"(?P<priority>\d+) (?P<address>\S+) (?P<port>\d+) typ " + r"(?P<type>\S+)(?: raddr (?P<rel_addr>\S+) rport " + r"(?P<rel_port>\d+))?(?: generation (?P<generation>\d+))?" + ) + match = pattern.match(candidate_string) + if match: + candidate_dict = match.groupdict() + + # Apply the correct types to the dictionary values + candidate_dict["component_id"] = int(candidate_dict["component_id"]) + candidate_dict["priority"] = int(candidate_dict["priority"]) + candidate_dict["port"] = int(candidate_dict["port"]) + + if candidate_dict["rel_port"]: + candidate_dict["rel_port"] = int(candidate_dict["rel_port"]) + + if candidate_dict["generation"]: + candidate_dict["generation"] = candidate_dict["generation"] + + # Remove None values + return {k: v for k, v in candidate_dict.items() if v is not None} + else: + log.warning(f"can't parse candidate: {candidate_string!r}") + return None + + def build_ice_candidate(self, parsed_candidate): + """Builds ICE candidate + + @param parsed_candidate: Dictionary containing parsed ICE candidate + """ + base_format = ( + "candidate:{foundation} {component_id} {transport} {priority} " + "{address} {port} typ {type}" + ) + + if ((parsed_candidate.get('rel_addr') + and parsed_candidate.get('rel_port'))): + base_format += " raddr {rel_addr} rport {rel_port}" + + if parsed_candidate.get('generation'): + base_format += " generation {generation}" + + return base_format.format(**parsed_candidate) + + def on_ice_candidate(self, event): + """Handles ICE candidate event + + @param event: Event containing the ICE candidate + """ + log.debug(f"on ice candidate {event.candidate=}") + if event.candidate and event.candidate.candidate: + window.last_event = event + parsed_candidate = self.parse_ice_candidate(event.candidate.candidate) + if parsed_candidate is None: + return + try: + media_type = self.media_types[event.candidate.sdpMLineIndex] + except (TypeError, IndexError): + log.error( + f"Can't find media type.\n{event.candidate=}\n{self._media_types=}" + ) + return + self.media_candidates.setdefault(media_type, []).append(parsed_candidate) + log.debug(f"ICE candidate [{media_type}]: {event.candidate.candidate}") + else: + log.debug("All ICE candidates gathered") + + def _set_media_types(self, offer): + """Sets media types from offer SDP + + @param offer: RTC session description containing the offer + """ + sdp_lines = offer.sdp.splitlines() + media_types = {} + mline_index = 0 + + for line in sdp_lines: + if line.startswith("m="): + media_types[mline_index] = line[2:line.find(" ")] + mline_index += 1 + + self._media_types = media_types + + def on_ice_gathering_state_change(self, event): + """Handles ICE gathering state change + + @param event: Event containing the ICE gathering state change + """ + connection = event.target + log.debug(f"on_ice_gathering_state_change {connection.iceGatheringState=}") + if connection.iceGatheringState == "complete": + log.info("ICE candidates gathering done") + self.candidates_gathered.set_result(None) + + async def _create_peer_connection( + self, + ): + """Creates peer connection""" + if self._peer_connection is not None: + raise Exception("create_peer_connection can't be called twice!") + + external_disco = json.loads(await bridge.external_disco_get("")) + ice_servers = [] + + for server in external_disco: + ice_server = {} + if server["type"] == "stun": + ice_server["urls"] = f"stun:{server['host']}:{server['port']}" + elif server["type"] == "turn": + ice_server["urls"] = ( + f"turn:{server['host']}:{server['port']}?transport={server['transport']}" + ) + ice_server["username"] = server["username"] + ice_server["credential"] = server["password"] + ice_servers.append(ice_server) + + rtc_configuration = {"iceServers": ice_servers} + + peer_connection = window.RTCPeerConnection.new(rtc_configuration) + peer_connection.addEventListener("track", self.on_track) + peer_connection.addEventListener("negotiationneeded", self.on_negotiation_needed) + peer_connection.addEventListener("icecandidate", self.on_ice_candidate) + peer_connection.addEventListener("icegatheringstatechange", self.on_ice_gathering_state_change) + + self._peer_connection = peer_connection + window.pc = self._peer_connection + + async def _get_user_media( + self, + audio: bool = True, + video: bool = True + ): + """Gets user media + + @param audio: True if an audio flux is required + @param video: True if a video flux is required + """ + media_constraints = {'audio': audio, 'video': video} + local_stream = await window.navigator.mediaDevices.getUserMedia(media_constraints) + document["local_video"].srcObject = local_stream + + for track in local_stream.getTracks(): + self._peer_connection.addTrack(track) + + async def _gather_ice_candidates(self, is_initiator: bool, remote_candidates=None): + """Get ICE candidates and wait to have them all before returning them + + @param is_initiator: Boolean indicating if the user is the initiator of the connection + @param remote_candidates: Remote ICE candidates, if any + """ + if self._peer_connection is None: + raise Exception("The peer connection must be created before gathering ICE candidates!") + + self.media_candidates.clear() + gather_timeout = timer.set_timeout( + lambda: self.candidates_gathered.set_exception( + errors.TimeoutError("ICE gathering time out") + ), + GATHER_TIMEOUT + ) + + if is_initiator: + offer = await self._peer_connection.createOffer() + self._set_media_types(offer) + await self._peer_connection.setLocalDescription(offer) + else: + answer = await self._peer_connection.createAnswer() + self._set_media_types(answer) + await self._peer_connection.setLocalDescription(answer) + + if not is_initiator: + log.debug(self._peer_connection.localDescription.sdp) + await self.candidates_gathered + log.debug(self._peer_connection.localDescription.sdp) + timer.clear_timeout(gather_timeout) + ufrag, pwd = self.extract_pwd_ufrag(self._peer_connection.localDescription.sdp) + return { + "ufrag": ufrag, + "pwd": pwd, + "candidates": self.media_candidates, + } + + def on_action_new( + self, action_data_s: str, action_id: str, security_limit: int, profile: str + ) -> None: + """Called when a call is received + + @param action_data_s: Action data serialized + @param action_id: Unique identifier for the action + @param security_limit: Security limit for the action + @param profile: Profile associated with the action + """ + action_data = json.loads(action_data_s) + if action_data.get("type") != "call": + return + peer_jid = action_data["from_jid"] + log.info( + f"{peer_jid} wants to start a call ({action_data['sub_type']})" + ) + if self.sid is not None: + log.warning( + f"already in a call ({self.sid}), can't receive a new call from " + f"{peer_jid}" + ) + return + self.sid = action_data["session_id"] + log.debug(f"Call SID: {self.sid}") + + # Answer the call + offer_sdp = action_data["sdp"] + aio.run(self.answer_call(offer_sdp, action_id)) + + def _on_call_accepted(self, session_id: str, sdp: str, profile: str) -> None: + """Called when we have received answer SDP from responder + + @param session_id: Session identifier + @param sdp: Session Description Protocol data + @param profile: Profile associated with the action + """ + aio.run(self.on_call_accepted(session_id, sdp, profile)) + + def _on_call_ended(self, session_id: str, data_s: str, profile: str) -> None: + """Call has been terminated + + @param session_id: Session identifier + @param data_s: Serialised additional data on why the call has ended + @param profile: Profile associated + """ + if self.sid is None: + log.debug("there are no calls in progress") + return + if session_id != self.sid: + log.debug( + f"ignoring call_ended not linked to our call ({self.sid}): {session_id}" + ) + return + aio.run(self.end_call()) + + async def on_call_accepted(self, session_id: str, sdp: str, profile: str) -> None: + """Call has been accepted, connection can be established + + @param session_id: Session identifier + @param sdp: Session Description Protocol data + @param profile: Profile associated + """ + if self.sid != session_id: + log.debug( + f"Call ignored due to different session ID ({self.sid=} {session_id=})" + ) + return + await self._peer_connection.setRemoteDescription({ + "type": "answer", + "sdp": sdp + }) + await self.on_ice_candidates_new(self.candidates_buffer) + self.candidates_buffer.clear() + + def _on_ice_candidates_new(self, sid: str, candidates_s: str, profile: str) -> None: + """Called when new ICE candidates are received + + @param sid: Session identifier + @param candidates_s: ICE candidates serialized + @param profile: Profile associated with the action + """ + if sid != self.sid: + log.debug( + f"ignoring peer ice candidates for {sid=} ({self.sid=})." + ) + return + candidates = json.loads(candidates_s) + aio.run(self.on_ice_candidates_new(candidates)) + + async def on_ice_candidates_new(self, candidates: dict) -> None: + """Called when new ICE canidates are received from peer + + @param candidates: Dictionary containing new ICE candidates + """ + log.debug(f"new peer candidates received: {candidates}") + if ( + self._peer_connection is None + or self._peer_connection.remoteDescription is None + ): + for media_type in ("audio", "video"): + media_candidates = candidates.get(media_type) + if media_candidates: + buffer = self.candidates_buffer[media_type] + buffer["candidates"].extend(media_candidates["candidates"]) + return + for media_type, ice_data in candidates.items(): + for candidate in ice_data["candidates"]: + candidate_sdp = self.build_ice_candidate(candidate) + try: + sdp_mline_index = self.get_sdp_mline_index(media_type) + except Exception as e: + log.warning(e) + continue + ice_candidate = window.RTCIceCandidate.new({ + "candidate": candidate_sdp, + "sdpMLineIndex": sdp_mline_index + } + ) + await self._peer_connection.addIceCandidate(ice_candidate) + + def on_track(self, event): + """New track has been received from peer + + @param event: Event associated with the new track + """ + if event.streams and event.streams[0]: + remote_stream = event.streams[0] + document["remote_video"].srcObject = remote_stream + else: + if self.remote_stream is None: + self.remote_stream = window.MediaStream.new() + document["remote_video"].srcObject = self.remote_stream + self.remote_stream.addTrack(event.track) + + document["call_btn"].classList.add("is-hidden") + document["hangup_btn"].classList.remove("is-hidden") + + def on_negotiation_needed(self, event) -> None: + log.debug(f"on_negotiation_needed {event=}") + # TODO + + async def answer_call(self, offer_sdp: str, action_id: str): + """We respond to the call""" + log.debug("answering call") + await self._create_peer_connection() + + await self._peer_connection.setRemoteDescription({ + "type": "offer", + "sdp": offer_sdp + }) + await self.on_ice_candidates_new(self.candidates_buffer) + self.candidates_buffer.clear() + await self._get_user_media() + + # Gather local ICE candidates + local_ice_data = await self._gather_ice_candidates(False) + self.local_candidates = local_ice_data["candidates"] + + await bridge.action_launch( + action_id, + json.dumps({ + "sdp": self._peer_connection.localDescription.sdp, + }) + ) + + async def make_call(self, audio: bool = True, video: bool = True) -> None: + """Start a WebRTC call + + @param audio: True if an audio flux is required + @param video: True if a video flux is required + """ + await self._create_peer_connection() + await self._get_user_media(audio, video) + await self._gather_ice_candidates(True) + callee_jid = document["callee_jid"].value + + call_data = { + "sdp": self._peer_connection.localDescription.sdp + } + log.info(f"calling {callee_jid!r}") + self.sid = await bridge.call_start( + callee_jid, + json.dumps(call_data) + ) + log.debug(f"Call SID: {self.sid}") + + async def end_call(self) -> None: + """Stop streaming and clean instance""" + document["hangup_btn"].classList.add("is-hidden") + document["call_btn"].classList.remove("is-hidden") + if self._peer_connection is None: + log.debug("There is currently no call to end.") + else: + self._peer_connection.removeEventListener("track", self.on_track) + self._peer_connection.removeEventListener("negotiationneeded", self.on_negotiation_needed) + self._peer_connection.removeEventListener("icecandidate", self.on_ice_candidate) + self._peer_connection.removeEventListener("icegatheringstatechange", self.on_ice_gathering_state_change) + + local_video = document["local_video"] + remote_video = document["remote_video"] + if local_video.srcObject: + for track in local_video.srcObject.getTracks(): + track.stop() + if remote_video.srcObject: + for track in remote_video.srcObject.getTracks(): + track.stop() + + self._peer_connection.close() + self.reset_instance() + + async def hand_up(self) -> None: + """Terminate the call""" + session_id = self.sid + await self.end_call() + await bridge.call_end( + session_id, + "" + ) + + +webrtc_call = WebRTCCall() + +document["call_btn"].bind( + "click", + lambda __: aio.run(webrtc_call.make_call()) +) +document["hangup_btn"].bind( + "click", + lambda __: aio.run(webrtc_call.hand_up()) +) + +bridge.register_signal("action_new", webrtc_call.on_action_new) +bridge.register_signal("call_accepted", webrtc_call._on_call_accepted) +bridge.register_signal("call_ended", webrtc_call._on_call_ended) +bridge.register_signal("ice_candidates_new", webrtc_call._on_ice_candidates_new) + +loading.remove_loading_screen()
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/libervia/pages/calls/page_meta.py Thu Jun 01 21:42:02 2023 +0200 @@ -0,0 +1,19 @@ +#!/usr/bin/env python3 + + +from sat.core.i18n import _ +from sat.core.log import getLogger +from sat.tools.common import data_format +from twisted.internet import defer +import datetime +import time +from dateutil import tz + +from libervia.server.constants import Const as C + +log = getLogger(__name__) + + +name = "calls" +access = C.PAGES_ACCESS_PROFILE +template = "call/call.html"
--- a/libervia/server/constants.py Thu Jun 01 20:44:57 2023 +0200 +++ b/libervia/server/constants.py Thu Jun 01 21:42:02 2023 +0200 @@ -95,6 +95,7 @@ "events", "lists", "merge-requests", + "calls" # XXX: app is not available anymore since removal of pyjamas code with Python 3 # port. It should come back at a later point with an alternative (Brython # probably).