libervia-web: libervia/web/pages/calls/_browser/__init_

comparison libervia/web/pages/calls/_browser/init.py @ 1549:e47c24204449

browser (calls): update call to handle search, control buttons, and better UI/UX: - adapt to backend changes - UI and WebRTC parts are not separated - users can now be searched - add mute/fullscreen buttons - ring - cancellable dialog when a call is received - status of the call - animations - various UI/UX improvments rel 423

author	Goffi <goffi@goffi.org>
date	Wed, 09 Aug 2023 00:22:18 +0200
parents	eb00d593801d
children	83c2a6faa2ae

comparison

equal deleted inserted replaced

-:66aa6e140ebb
+:e47c24204449
 import json
-import re
 from bridge import AsyncBridge as Bridge
-from browser import aio, console as log, document, timer, window
+from browser import aio, console as log, document, window
-import errors
+from cache import cache
+import dialog
+from jid import JID
+from jid_search import JidSearch
 import loading
+from template import Template
+from webrtc import WebRTC
 log.warning = log.warn
 profile = window.profile or ""
 bridge = Bridge()
 GATHER_TIMEOUT = 10000
+ALLOWED_STATUSES = (None, "dialing")
+AUDIO = 'audio'
-class WebRTCCall:
+VIDEO = 'video'
+ALLOWED_CALL_MODES = {AUDIO, VIDEO}
+class CallUI:
 def __init__(self):
-self.reset_instance()
+self.webrtc = WebRTC()
+self.mode = "search"
-def reset_instance(self):
+self._status = None
-"""Inits or resets the instance variables to their default state."""
+self._callee = None
-self._peer_connection = None
+self.contacts_elt = document["contacts"]
-self._media_types = None
+self.search_container_elt = document["search_container"]
-self.sid = None
+self.call_container_elt = document["call_container"]
-self.local_candidates = None
+self.call_box_elt = document["call_box"]
-self.remote_stream = None
+self.call_avatar_wrapper_elt = document["call_avatar_wrapper"]
-self.candidates_buffer = {
+self.call_status_wrapper_elt = document["call_status_wrapper"]
-"audio": {"candidates": []},
+self.call_avatar_tpl = Template("call/call_avatar.html")
-"video": {"candidates": []},
+self.call_status_tpl = Template("call/call_status.html")
-}
+self.audio_player_elt = document["audio_player"]
-self.media_candidates = {}
+bridge.register_signal("action_new", self._on_action_new)
-self.candidates_gathered = aio.Future()
+bridge.register_signal("call_setup", self._on_call_setup)
+bridge.register_signal("call_ended", self._on_call_ended)
+# call/hang up buttons
+self._call_mode = VIDEO
+self.call_button_tpl = Template("call/call_button.html")
+self._update_call_button()
+document['toggle_call_mode_btn'].bind('click', self.switch_call_mode)
+document["hangup_btn"].bind(
+"click",
+lambda __: aio.run(self.hang_up())
+)
+# other buttons
+document["full_screen_btn"].bind(
+"click",
+lambda __: self.toggle_fullscreen()
+)
+document["exit_full_screen_btn"].bind(
+"click",
+lambda __: self.toggle_fullscreen()
+)
+document["mute_audio_btn"].bind(
+"click",
+self.toggle_audio_mute
+)
+document["mute_video_btn"].bind(
+"click",
+self.toggle_video_mute
+)
+# search
+self.jid_search = JidSearch(
+document["search"],
+document["contacts"],
+click_cb = self._on_entity_click
+)
+# incoming call dialog
+self.incoming_call_dialog_elt = None
 @property
-def media_types(self):
+def sid(self) -> str|None:
-if self._media_types is None:
+return self.webrtc.sid
-raise Exception("self._media_types should not be None!")
-return self._media_types
+@sid.setter
+def sid(self, new_sid) -> None:
-def get_sdp_mline_index(self, media_type):
+self.webrtc.sid = new_sid
-"""Gets the sdpMLineIndex for a given media type.
+@property
-@param media_type: The type of the media.
+def status(self):
-"""
+return self._status
-for index, m_type in self.media_types.items():
-if m_type == media_type:
+@status.setter
-return index
+def status(self, new_status):
-raise ValueError(f"Media type '{media_type}' not found")
+if new_status != self._status:
+if new_status not in ALLOWED_STATUSES:
-def extract_pwd_ufrag(self, sdp):
+raise Exception(
-"""Retrieves ICE password and user fragment for SDP offer.
+f"INTERNAL ERROR: this status is not allowed: {new_status!r}"
+)
-@param sdp: The Session Description Protocol offer string.
+tpl_data = {
-"""
+"entity": self._callee,
-ufrag_line = re.search(r"ice-ufrag:(\S+)", sdp)
+"status": new_status
-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
 }
+if self._callee is not None:
-setup_line = re.search(r"a=setup:(\S+)", sdp)
+try:
-if setup_line:
+tpl_data["name"] = cache.identities[self._callee]["nicknames"][0]
-setup = setup_line.group(1)
+except (KeyError, IndexError):
-fingerprint_data["setup"] = setup
+tpl_data["name"] = str(self._callee)
+status_elt = self.call_status_tpl.get_elt(tpl_data)
-return fingerprint_data
+self.call_status_wrapper_elt.clear()
-else:
+self.call_status_wrapper_elt <= status_elt
-raise ValueError("fingerprint should not be missing")
-def parse_ice_candidate(self, candidate_string):
+self._status = new_status
-"""Parses the ice candidate string.
+@property
-@param candidate_string: The ice candidate string to be parsed.
+def call_mode(self):
-"""
+return self._call_mode
-pattern = re.compile(
-r"candidate:(?P<foundation>\S+) (?P<component_id>\d+) (?P<transport>\S+) "
+@call_mode.setter
-r"(?P<priority>\d+) (?P<address>\S+) (?P<port>\d+) typ "
+def call_mode(self, mode):
-r"(?P<type>\S+)(?: raddr (?P<rel_addr>\S+) rport "
+if mode in ALLOWED_CALL_MODES:
-r"(?P<rel_port>\d+))?(?: generation (?P<generation>\d+))?"
+if self._call_mode == mode:
-)
-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:
+self._call_mode = mode
-media_type = self.media_types[event.candidate.sdpMLineIndex]
+self._update_call_button()
-except (TypeError, IndexError):
+with_video = mode == VIDEO
-log.error(
+for elt in self.call_box_elt.select(".is-video-only"):
-f"Can't find media type.\n{event.candidate=}\n{self._media_types=}"
+if with_video:
-)
+elt.classList.remove("is-hidden")
-return
+else:
-self.media_candidates.setdefault(media_type, []).append(parsed_candidate)
+elt.classList.add("is-hidden")
-log.debug(f"ICE candidate [{media_type}]: {event.candidate.candidate}")
+else:
-else:
+raise ValueError("Invalid call mode")
-log.debug("All ICE candidates gathered")
+def switch_call_mode(self, ev):
-def _set_media_types(self, offer):
+self.call_mode = AUDIO if self.call_mode == VIDEO else VIDEO
-"""Sets media types from offer SDP
+def _update_call_button(self):
-@param offer: RTC session description containing the offer
+new_button = self.call_button_tpl.get_elt({"call_mode": self.call_mode})
-"""
+new_button.bind(
-sdp_lines = offer.sdp.splitlines()
+"click",
-media_types = {}
+lambda __: aio.run(self.make_call(video=not self.call_mode == AUDIO))
-mline_index = 0
+)
+document['call_btn'].replaceWith(new_button)
-for line in sdp_lines:
-if line.startswith("m="):
+def _on_action_new(
-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 profile: Profile associated with the action
 """
 action_data = json.loads(action_data_s)
 if action_data.get("type") != "call":
 return
+aio.run(self.on_action_new(action_data, action_id))
+async def on_action_new(self, action_data: dict, action_id: str) -> None:
 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"]
+sid = self.sid = action_data["session_id"]
-log.debug(f"Call SID: {self.sid}")
+await cache.fill_identities([peer_jid])
+identity = cache.identities[peer_jid]
-# Answer the call
+peer_name = identity['nicknames'][0]
-offer_sdp = action_data["sdp"]
-aio.run(self.answer_call(offer_sdp, action_id))
+# we start the ring
+self.audio_player_elt.play()
-def _on_call_accepted(self, session_id: str, sdp: str, profile: str) -> None:
+# and ask user if we take the call
+try:
+self.incoming_call_dialog_elt = dialog.Confirm(
+f"{peer_name} is calling you.",
+ok_label="Answer",
+cancel_label="Reject"
+)
+accepted = await self.incoming_call_dialog_elt.ashow()
+except dialog.CancelError:
+log.info("Call has been cancelled")
+self.incoming_call_dialog_elt = None
+self.sid = None
+dialog.notification.show(
+f"{peer_name} has cancelled the call",
+level="info"
+)
+return
+self.incoming_call_dialog_elt = None
+# we stop the ring
+self.audio_player_elt.pause()
+self.audio_player_elt.currentTime = 0
+if accepted:
+log.debug(f"Call SID: {sid}")
+# Answer the call
+self.switch_mode("call")
+else:
+log.info(f"your are declining the call from {peer_jid}")
+self.sid = None
+await bridge.action_launch(
+action_id,
+json.dumps({"cancelled": not accepted})
+)
+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(json.loads(data_s)))
+def _on_call_setup(self, session_id: str, setup_data_s: 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))
+aio.run(self.on_call_setup(session_id, json.loads(setup_data_s), profile))
-def _on_call_ended(self, session_id: str, data_s: str, profile: str) -> None:
+async def on_call_setup(self, session_id: str, setup_data: dict, profile: str) -> None:
-"""Call has been terminated
+"""Call has been accepted, connection can be established
 @param session_id: Session identifier
-@param data_s: Serialised additional data on why the call has ended
+@param setup_data: Data with following keys:
-@param profile: Profile associated
+role: initiator or responser
-"""
+sdp: Session Description Protocol data
-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({
+try:
-"type": "answer",
+role = setup_data["role"]
-"sdp": sdp
+sdp = setup_data["sdp"]
-})
+except KeyError:
-await self.on_ice_candidates_new(self.candidates_buffer)
+dialog.notification.show(
-self.candidates_buffer.clear()
+f"Invalid setup data received: {setup_data}",
+level="error"
-def _on_ice_candidates_new(self, sid: str, candidates_s: str, profile: str) -> None:
+)
-"""Called when new ICE candidates are received
+return
+if role == "initiator":
-@param sid: Session identifier
+await self.webrtc.accept_call(session_id, sdp, profile)
-@param candidates_s: ICE candidates serialized
+elif role == "responder":
-@param profile: Profile associated with the action
+await self.webrtc.answer_call(session_id, sdp, profile)
-"""
+else:
-if sid != self.sid:
+dialog.notification.show(
-log.debug(
+f"Invalid role received during setup: {setup_data}",
-f"ignoring peer ice candidates for {sid=} ({self.sid=})."
+level="error"
 )
 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()
+try:
-await self._get_user_media(audio, video)
+callee_jid = JID(document["search"].value.strip())
-await self._gather_ice_candidates(True)
+if not callee_jid.is_valid:
-callee_jid = document["callee_jid"].value
+raise ValueError
+except ValueError:
-call_data = {
+dialog.notification.show(
-"sdp": self._peer_connection.localDescription.sdp
+"Invalid identifier, please use a valid callee identifier",
-}
+level="error"
-log.info(f"calling {callee_jid!r}")
+)
-self.sid = await bridge.call_start(
+return
-callee_jid,
-json.dumps(call_data)
+self._callee = callee_jid
-)
+await cache.fill_identities([callee_jid])
-log.debug(f"Call SID: {self.sid}")
+self.status = "dialing"
+call_avatar_elt = self.call_avatar_tpl.get_elt({
-async def end_call(self) -> None:
+"entity": str(callee_jid),
+"identities": cache.identities,
+})
+self.call_avatar_wrapper_elt.clear()
+self.call_avatar_wrapper_elt <= call_avatar_elt
+self.switch_mode("call")
+await self.webrtc.make_call(callee_jid, audio, video)
+async def end_call(self, data: dict) -> None:
 """Stop streaming and clean instance"""
-document["hangup_btn"].classList.add("is-hidden")
+# if there is any ringing, we stop it
-document["call_btn"].classList.remove("is-hidden")
+self.audio_player_elt.pause()
-if self._peer_connection is None:
+self.audio_player_elt.currentTime = 0
-log.debug("There is currently no call to end.")
-else:
+if self.incoming_call_dialog_elt is not None:
-self._peer_connection.removeEventListener("track", self.on_track)
+self.incoming_call_dialog_elt.cancel()
-self._peer_connection.removeEventListener("negotiationneeded", self.on_negotiation_needed)
+self.incoming_call_dialog_elt = None
-self._peer_connection.removeEventListener("icecandidate", self.on_ice_candidate)
-self._peer_connection.removeEventListener("icegatheringstatechange", self.on_ice_gathering_state_change)
+self.switch_mode("search")
-local_video = document["local_video"]
+if data.get("reason") == "busy":
-remote_video = document["remote_video"]
+assert self._callee is not None
-if local_video.srcObject:
+peer_name = cache.identities[self._callee]["nicknames"][0]
-for track in local_video.srcObject.getTracks():
+dialog.notification.show(
-track.stop()
+f"{peer_name} can't answer your call",
-if remote_video.srcObject:
+level="info",
-for track in remote_video.srcObject.getTracks():
+)
-track.stop()
+await self.webrtc.end_call()
-self._peer_connection.close()
-self.reset_instance()
+async def hang_up(self) -> None:
-async def hand_up(self) -> None:
 """Terminate the call"""
 session_id = self.sid
-await self.end_call()
+if not session_id:
+log.warning("Can't hand_up, not call in progress")
+return
+await self.end_call({"reason": "terminated"})
 await bridge.call_end(
 session_id,
 ""
 )
+def _handle_animation_end(
-webrtc_call = WebRTCCall()
+self,
+element,
-document["call_btn"].bind(
+remove = None,
-"click",
+add = None,
-lambda __: aio.run(webrtc_call.make_call())
+):
-)
+"""Return a handler that removes specified classes and the event handler.
-document["hangup_btn"].bind(
-"click",
+@param element: The element to operate on.
-lambda __: aio.run(webrtc_call.hand_up())
+@param remove: List of class names to remove from the element.
-)
+@param add: List of class names to add to the element.
+"""
-bridge.register_signal("action_new", webrtc_call.on_action_new)
+def handler(__, remove=remove, add=add):
-bridge.register_signal("call_accepted", webrtc_call._on_call_accepted)
+log.info(f"animation end OK {element=}")
-bridge.register_signal("call_ended", webrtc_call._on_call_ended)
+if add:
-bridge.register_signal("ice_candidates_new", webrtc_call._on_ice_candidates_new)
+if isinstance(add, str):
+add = [add]
+element.classList.add(*add)
+if remove:
+if isinstance(remove, str):
+remove = [remove]
+element.classList.remove(*remove)
+element.unbind('animationend', handler)
+return handler
+def switch_mode(self, mode: str) -> None:
+"""Handles the user interface changes"""
+if mode == self.mode:
+return
+if mode == "call":
+# Hide contacts with fade-out animation and bring up the call box
+self.search_container_elt.classList.add("fade-out-y")
+self.search_container_elt.bind(
+'animationend',
+self._handle_animation_end(
+self.search_container_elt,
+remove="fade-out-y",
+add="is-hidden"
+)
+)
+self.call_container_elt.classList.remove("is-hidden")
+self.call_container_elt.classList.add("slide-in")
+self.call_container_elt.bind(
+'animationend',
+self._handle_animation_end(
+self.call_container_elt,
+remove="slide-in"
+)
+)
+self.mode = mode
+elif mode == "search":
+self.toggle_fullscreen(False)
+self.search_container_elt.classList.add("fade-out-y", "animation-reverse")
+self.search_container_elt.classList.remove("is-hidden")
+self.search_container_elt.bind(
+'animationend',
+self._handle_animation_end(
+self.search_container_elt,
+remove=["fade-out-y", "animation-reverse"],
+)
+)
+self.call_container_elt.classList.add("slide-in", "animation-reverse")
+self.call_container_elt.bind(
+'animationend',
+self._handle_animation_end(
+self.call_container_elt,
+remove=["slide-in", "animation-reverse"],
+add="is-hidden"
+)
+)
+self.mode = mode
+else:
+log.error(f"Internal Error: Unknown call mode: {mode}")
+def toggle_fullscreen(self, fullscreen: bool|None = None):
+"""Toggle fullscreen mode for video elements.
+@param fullscreen: if set, determine the fullscreen state; otherwise,
+the fullscreen mode will be toggled.
+"""
+do_fullscreen = (
+document.fullscreenElement is None if fullscreen is None else fullscreen
+)
+try:
+if do_fullscreen:
+if document.fullscreenElement is None:
+self.call_box_elt.requestFullscreen()
+document["full_screen_btn"].classList.add("is-hidden")
+document["exit_full_screen_btn"].classList.remove("is-hidden")
+else:
+if document.fullscreenElement is not None:
+document.exitFullscreen()
+document["full_screen_btn"].classList.remove("is-hidden")
+document["exit_full_screen_btn"].classList.add("is-hidden")
+except Exception as e:
+dialog.notification.show(
+f"An error occurred while toggling fullscreen: {e}",
+level="error"
+)
+def toggle_audio_mute(self, evt):
+is_muted = self.webrtc.toggle_audio_mute()
+btn_elt = evt.currentTarget
+if is_muted:
+btn_elt.classList.remove("is-success")
+btn_elt.classList.add("muted", "is-warning")
+dialog.notification.show(
+f"audio is now muted",
+level="info",
+delay=2,
+)
+else:
+btn_elt.classList.remove("muted", "is-warning")
+btn_elt.classList.add("is-success")
+def toggle_video_mute(self, evt):
+is_muted = self.webrtc.toggle_video_mute()
+btn_elt = evt.currentTarget
+if is_muted:
+btn_elt.classList.remove("is-success")
+btn_elt.classList.add("muted", "is-warning")
+dialog.notification.show(
+f"video is now muted",
+level="info",
+delay=2,
+)
+else:
+btn_elt.classList.remove("muted", "is-warning")
+btn_elt.classList.add("is-success")
+def _on_entity_click(self, item: dict) -> None:
+aio.run(self.on_entity_click(item))
+async def on_entity_click(self, item: dict) -> None:
+"""Set entity JID to search bar, and start the call"""
+document["search"].value = item["entity"]
+await self.make_call()
+CallUI()
 loading.remove_loading_screen()

Mercurial > libervia-web

comparison libervia/web/pages/calls/_browser/__init__.py @ 1549:e47c24204449

comparison libervia/web/pages/calls/_browser/init.py @ 1549:e47c24204449