Mercurial > libervia-backend
view libervia/cli/cmd_call.py @ 4304:92a886f31581 default tip @
doc (components): new Email gateway documentation:
fix 449
author | Goffi <goffi@goffi.org> |
---|---|
date | Fri, 06 Sep 2024 18:07:44 +0200 |
parents | 0d7bb4df2343 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia CLI # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. from functools import partial import importlib import logging from typing import Any from libervia.backend.core.i18n import _ from libervia.backend.tools.common import data_format from libervia.cli.constants import Const as C from libervia.frontends.tools import jid from libervia.frontends.tools.webrtc_models import CallData from . import base __commands__ = ["Call"] class Common(base.CommandBase): def __init__(self, *args, **kwargs): super().__init__( *args, use_output=C.OUTPUT_CUSTOM, extra_outputs={ #: automatically select best output for current platform "default": self.auto_output, #: simple output with GStreamer ``autovideosink`` "simple": partial(self.use_output, "simple"), #: Qt GUI "gui": partial(self.use_output, "gui"), #: experimental TUI output "tui": partial(self.use_output, "tui"), }, **kwargs, ) def add_parser_options(self): self.parser.add_argument( "--no-ui", action="store_true", help=_("disable user interface") ) sources_group = self.parser.add_mutually_exclusive_group() sources_group.add_argument( "-s", "--sources", choices=["auto", "test"], default="auto", help='Well-known sources to use (default: "auto").', ) def get_call_data_kw(self) -> dict[str, Any]: """Get relevant keyword arguments for CallData""" kwargs: dict[str, Any] = {} if self.args.sources == "test": from libervia.frontends.tools.webrtc_models import SourcesTest kwargs["sources_data"] = SourcesTest() return kwargs async def start(self): root_logger = logging.getLogger() # we don't want any formatting for messages from webrtc for handler in root_logger.handlers: handler.setFormatter(None) if self.verbosity == 0: root_logger.setLevel(logging.ERROR) if self.verbosity >= 1: root_logger.setLevel(logging.WARNING) if self.verbosity >= 2: root_logger.setLevel(logging.DEBUG) async def auto_output(self, call_data: CallData) -> None: """Make a guess on the best output to use on current platform""" try: from .call_gui import AVCallUI except ImportError: # we can't import GUI, we may have missing modules await self.use_output("simple", call_data) else: if AVCallUI.can_run(): await self.use_output("gui", call_data) else: await self.use_output("simple", call_data) async def use_output(self, output_type, call_data: CallData): try: AVCall_module = importlib.import_module(f"libervia.cli.call_{output_type}") AVCallUI = AVCall_module.AVCallUI except Exception as e: self.disp(f"Error starting {output_type.upper()} UI: {e}", error=True) self.host.quit(C.EXIT_ERROR) else: try: await AVCallUI.run(self, call_data) except Exception as e: self.disp(f"Error running {output_type.upper()} UI: {e}", error=True) self.host.quit(C.EXIT_ERROR) class Make(Common): def __init__(self, host): super().__init__( host, "make", use_verbose=True, help=_("start a call"), ) def add_parser_options(self): super().add_parser_options() self.parser.add_argument( "entity", metavar="JID", help=_("JIDs of entity to call"), ) async def start(self): await super().start() await super().output( CallData(callee=jid.JID(self.args.entity), kwargs=self.get_call_data_kw()) ) class Receive(Common): def __init__(self, host): super().__init__( host, "receive", use_verbose=True, help=_("wait for a call"), ) def add_parser_options(self): super().add_parser_options() auto_accept_group = self.parser.add_mutually_exclusive_group() auto_accept_group.add_argument( "-a", "--auto-accept", action="append", metavar="JID", default=[], help=_( "automatically accept call from this jid (can be used multiple times)" ), ) auto_accept_group.add_argument( "--auto-accept-all", action="store_true", help=_("automatically accept call from anybody"), ) async def on_action_new( self, action_data_s: str, action_id: str, security_limit: int, profile: str ) -> None: if profile != self.profile: return action_data = data_format.deserialise(action_data_s) if action_data.get("type") != C.META_TYPE_CALL: return peer_jid = jid.JID(action_data["from_jid"]).bare caller = peer_jid.bare if ( not self.args.auto_accept_all and caller not in self.args.auto_accept and not await self.host.confirm( _("📞 Incoming call from {caller}, do you accept?").format(caller=caller) ) ): await self.host.bridge.action_launch( action_id, data_format.serialise({"cancelled": True}), profile ) return self.disp(_("✅ Incoming call from {caller} accepted.").format(caller=caller)) await super().output( CallData( callee=peer_jid, sid=action_data["session_id"], action_id=action_id, kwargs=self.get_call_data_kw(), ) ) async def start(self): await super().start() self.host.bridge.register_signal("action_new", self.on_action_new, "core") class Call(base.CommandBase): subcommands = (Make, Receive) def __init__(self, host): super().__init__(host, "call", use_profile=False, help=_("A/V calls and related"))