view libervia/web/pages/calls/_browser/__init__.py @ 1553:83c2a6faa2ae

browser (calls): screen sharing implementation: - the new screen sharing button toggle screen sharing state - the button reflect the screen sharing state (green crossed when not sharing, red uncrossed otherwise) - the screen sharing stream replaces the camera one, and vice versa. No re-negociation is needed. - stopping the sharing through browser's dialog is supported - the screen sharing button is only visibile if supported by the platform rel 432
author Goffi <goffi@goffi.org>
date Mon, 14 Aug 2023 16:49:02 +0200
parents e47c24204449
children 855729ef75f2
line wrap: on
line source

import json

from bridge import AsyncBridge as Bridge
from browser import aio, console as log, document, window
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"
VIDEO = "video"
ALLOWED_CALL_MODES = {AUDIO, VIDEO}
INACTIVE_CLASS = "inactive"
MUTED_CLASS = "muted"
SCREEN_OFF_CLASS = "screen-off"


class CallUI:
    def __init__(self):
        self.webrtc = WebRTC()
        self.webrtc.screen_sharing_cb = self.on_sharing_screen
        self.mode = "search"
        self._status = None
        self._callee = None
        self.contacts_elt = document["contacts"]
        self.search_container_elt = document["search_container"]
        self.call_container_elt = document["call_container"]
        self.call_box_elt = document["call_box"]
        self.call_avatar_wrapper_elt = document["call_avatar_wrapper"]
        self.call_status_wrapper_elt = document["call_status_wrapper"]
        self.call_avatar_tpl = Template("call/call_avatar.html")
        self.call_status_tpl = Template("call/call_status.html")
        self.audio_player_elt = document["audio_player"]
        bridge.register_signal("action_new", self._on_action_new)
        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)
        self.share_desktop_btn_elt = document["share_desktop_btn"]
        if hasattr(window.navigator.mediaDevices, "getDisplayMedia"):
            self.share_desktop_btn_elt.classList.remove("is-hidden-touch")
            # screen sharing is supported
            self.share_desktop_btn_elt.bind("click", self.toggle_screen_sharing)
        else:
            self.share_desktop_btn_elt.classList.add("is-hidden")

        # 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 sid(self) -> str | None:
        return self.webrtc.sid

    @sid.setter
    def sid(self, new_sid) -> None:
        self.webrtc.sid = new_sid

    @property
    def status(self):
        return self._status

    @status.setter
    def status(self, new_status):
        if new_status != self._status:
            if new_status not in ALLOWED_STATUSES:
                raise Exception(
                    f"INTERNAL ERROR: this status is not allowed: {new_status!r}"
                )
            tpl_data = {"entity": self._callee, "status": new_status}
            if self._callee is not None:
                try:
                    tpl_data["name"] = cache.identities[self._callee]["nicknames"][0]
                except (KeyError, IndexError):
                    tpl_data["name"] = str(self._callee)
            status_elt = self.call_status_tpl.get_elt(tpl_data)
            self.call_status_wrapper_elt.clear()
            self.call_status_wrapper_elt <= status_elt

        self._status = new_status

    @property
    def call_mode(self):
        return self._call_mode

    @call_mode.setter
    def call_mode(self, mode):
        if mode in ALLOWED_CALL_MODES:
            if self._call_mode == mode:
                return
            self._call_mode = mode
            self._update_call_button()
            with_video = mode == VIDEO
            for elt in self.call_box_elt.select(".is-video-only"):
                if with_video:
                    elt.classList.remove("is-hidden")
                else:
                    elt.classList.add("is-hidden")
        else:
            raise ValueError("Invalid call mode")

    def switch_call_mode(self, ev):
        self.call_mode = AUDIO if self.call_mode == VIDEO else VIDEO

    def _update_call_button(self):
        new_button = self.call_button_tpl.get_elt({"call_mode": self.call_mode})
        new_button.bind(
            "click", lambda __: aio.run(self.make_call(video=not self.call_mode == AUDIO))
        )
        document["call_btn"].replaceWith(new_button)

    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
        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
        sid = self.sid = action_data["session_id"]
        await cache.fill_identities([peer_jid])
        identity = cache.identities[peer_jid]
        peer_name = identity["nicknames"][0]

        # we start the ring
        self.audio_player_elt.play()

        # 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_setup(session_id, json.loads(setup_data_s), profile))

    async def on_call_setup(
        self, session_id: str, setup_data: dict, profile: str
    ) -> None:
        """Call has been accepted, connection can be established

        @param session_id: Session identifier
        @param setup_data: Data with following keys:
            role: initiator or responser
            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
        try:
            role = setup_data["role"]
            sdp = setup_data["sdp"]
        except KeyError:
            dialog.notification.show(
                f"Invalid setup data received: {setup_data}", level="error"
            )
            return
        if role == "initiator":
            await self.webrtc.accept_call(session_id, sdp, profile)
        elif role == "responder":
            await self.webrtc.answer_call(session_id, sdp, profile)
        else:
            dialog.notification.show(
                f"Invalid role received during setup: {setup_data}", level="error"
            )
            return

    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
        """
        try:
            callee_jid = JID(document["search"].value.strip())
            if not callee_jid.is_valid:
                raise ValueError
        except ValueError:
            dialog.notification.show(
                "Invalid identifier, please use a valid callee identifier", level="error"
            )
            return

        self._callee = callee_jid
        await cache.fill_identities([callee_jid])
        self.status = "dialing"
        call_avatar_elt = self.call_avatar_tpl.get_elt(
            {
                "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"""
        # if there is any ringing, we stop it
        self.audio_player_elt.pause()
        self.audio_player_elt.currentTime = 0

        if self.incoming_call_dialog_elt is not None:
            self.incoming_call_dialog_elt.cancel()
            self.incoming_call_dialog_elt = None

        self.switch_mode("search")

        if data.get("reason") == "busy":
            assert self._callee is not None
            peer_name = cache.identities[self._callee]["nicknames"][0]
            dialog.notification.show(
                f"{peer_name} can't answer your call",
                level="info",
            )

        await self.webrtc.end_call()

    async def hang_up(self) -> None:
        """Terminate the call"""
        session_id = self.sid
        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(
        self,
        element,
        remove=None,
        add=None,
    ):
        """Return a handler that removes specified classes and the event handler.

        @param element: The element to operate on.
        @param remove: List of class names to remove from the element.
        @param add: List of class names to add to the element.
        """

        def handler(__, remove=remove, add=add):
            log.info(f"animation end OK {element=}")
            if add:
                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(INACTIVE_CLASS, MUTED_CLASS, "is-warning")
            dialog.notification.show(
                f"audio is now muted",
                level="info",
                delay=2,
            )
        else:
            btn_elt.classList.remove(INACTIVE_CLASS, MUTED_CLASS, "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(INACTIVE_CLASS, MUTED_CLASS, "is-warning")
            dialog.notification.show(
                f"video is now muted",
                level="info",
                delay=2,
            )
        else:
            btn_elt.classList.remove(INACTIVE_CLASS, MUTED_CLASS, "is-warning")
            btn_elt.classList.add("is-success")

    def toggle_screen_sharing(self, evt):
        aio.run(self.webrtc.toggle_screen_sharing())

    def on_sharing_screen(self, sharing: bool) -> None:
        """Called when screen sharing state changes"""
        share_desktop_btn_elt = self.share_desktop_btn_elt
        if sharing:
            share_desktop_btn_elt.classList.add("is-danger")
            share_desktop_btn_elt.classList.remove(INACTIVE_CLASS, SCREEN_OFF_CLASS)
        else:
            share_desktop_btn_elt.classList.remove("is-danger")
            share_desktop_btn_elt.classList.add(INACTIVE_CLASS, SCREEN_OFF_CLASS)

    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()