view libervia/web/pages/calls/_browser/__init__.py @ 1565:d282dbdd5ffd

browser (calls): restore UI state on reset: `on_reset` is now called when webRTC reset happens (notably when a call ends). The UI state is reset, in particular mute buttons so they don't appear in the wrong state during the next call.
author Goffi <goffi@goffi.org>
date Mon, 21 Aug 2023 15:23:22 +0200
parents bd3c880f4a47
children 0a4433a343a3
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",
    "ringing",
    "in-call",
    "on-hold",
    "connecting",
    "connection-lost",
    "reconnecting",
)
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(
            screen_sharing_cb=self.on_sharing_screen,
            on_connection_established_cb=self.on_connection_established,
            on_reconnect_cb=self.on_reconnect,
            on_connection_lost_cb=self.on_connection_lost,
            on_video_devices=self.on_video_devices,
            on_reset_cb=self.on_reset_cb,
        )
        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_info", self._on_call_info)
        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
        document["video_call_btn"].bind("click", lambda __: aio.run(self.make_call()))
        document["audio_call_btn"].bind(
            "click", lambda __: aio.run(self.make_call(video=False))
        )
        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_col_elt = document["share_desktop_column"]
        if hasattr(window.navigator.mediaDevices, "getDisplayMedia"):
            self.share_desktop_col_elt.classList.remove("is-hidden-touch")
            # screen sharing is supported
            document["share_desktop_btn"].bind("click", self.toggle_screen_sharing)
        else:
            self.share_desktop_col_elt.classList.add("is-hidden")
        document["switch_camera_btn"].bind("click", self.on_switch_camera)

        # search
        self.search_elt = document["search"]
        self.jid_search = JidSearch(
            self.search_elt,
            document["contacts"],
            click_cb=self._on_entity_click,
            template="call/search_item.html",
            options={
                "no_group": True,
                "extra_cb": {
                    ".dropdown-trigger": lambda evt, item: aio.run(
                        self.on_entity_action(evt, "menu", item)
                    ),
                    ".click-to-video": lambda evt, item: aio.run(
                        self.on_entity_action(evt, VIDEO, item)
                    ),
                    ".click-to-audio": lambda evt, item: aio.run(
                        self.on_entity_action(evt, AUDIO, item)
                    ),
                },
            },
        )
        document["clear_search_btn"].bind("click", self.on_clear_search)

        # 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
            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 set_avatar(self, entity_jid: JID | str) -> None:
        """Set the avatar element from entity_jid

        @param entity_jid: bare jid of the entity
        """
        call_avatar_elt = self.call_avatar_tpl.get_elt(
            {
                "entity": str(entity_jid),
                "identities": cache.identities,
            }
        )
        self.call_avatar_wrapper_elt.clear()
        self.call_avatar_wrapper_elt <= call_avatar_elt

    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 = JID(action_data["from_jid"]).bare
        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]
        self._callee = 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.set_avatar(peer_jid)
            self.status = "connecting"
            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_info(self, session_id: str, info_type, info_data_s: str, profile: str):
        if self.sid != session_id:
            return
        if info_type == "ringing":
            self.status = "ringing"

    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

    def on_connection_established(self):
        self.status = "in-call"

    def on_reconnect(self):
        self.status = "reconnecting"

    def on_connection_lost(self):
        self.status = "connection-lost"

    def on_video_devices(self, has_multiple_cameras: bool) -> None:
        switch_camera_col_elt = document["switch_camera_column"]
        if has_multiple_cameras:
            switch_camera_col_elt.classList.remove("is-hidden", "is-hidden-desktop")
        else:
            switch_camera_col_elt.classList.add("is-hidden")

    def on_reset_cb(self) -> None:
        """Call when webRTC connection is reset, we reset buttons statuses"""
        document["full_screen_btn"].classList.remove("is-hidden")
        document["exit_full_screen_btn"].classList.add("is-hidden")
        for btn_elt in document["mute_audio_btn"], document["mute_video_btn"]:
            btn_elt.classList.remove(INACTIVE_CLASS, MUTED_CLASS, "is-warning")
            btn_elt.classList.add("is-success")

    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
        """
        self.call_mode = VIDEO if video else AUDIO
        try:
            callee_jid = JID(self.search_elt.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"
        self.set_avatar(callee_jid)

        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 on_clear_search(self, ev) -> None:
        """Clear the search input and trigger its 'input' event.

        @param ev: the event object from the button click.
        """
        if not self.search_elt.value:
            return
        # clear the search field
        self.search_elt.value = ""
        # and dispatch the input event so items are updated
        self.search_elt.dispatchEvent(window.Event.new("input"))

    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 = document["share_desktop_btn"]
        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_switch_camera(self, __) -> None:
        aio.run(self.webrtc.switch_camera())

    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"""
        self.search_elt.value = item["entity"]

        await self.make_call()

    async def on_entity_action(self, evt, action: str, item: dict) -> None:
        """Handle extra actions on search items"""
        evt.stopPropagation()
        if action == "menu":
            evt.currentTarget.parent.classList.toggle("is-active")
        elif action in (VIDEO, AUDIO):
            self.search_elt.value = item["entity"]
            # we want the dropdown to be inactive
            evt.currentTarget.closest(".dropdown").classList.remove("is-active")
            await self.make_call(video=action == VIDEO)


CallUI()
loading.remove_loading_screen()