Mercurial > libervia-backend

comparison libervia/cli/cmd_call.py @ 4270:0d7bb4df2343

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Reformatted code base using black.
author Goffi <goffi@goffi.org>
date Wed, 19 Jun 2024 18:44:57 +0200
parents 79c8a70e1813
children
comparison
equal deleted inserted replaced
4269:64a85ce8be70 4270:0d7bb4df2343
34 34
35 __commands__ = ["Call"] 35 __commands__ = ["Call"]
36 36
37 37
38 class Common(base.CommandBase): 38 class Common(base.CommandBase):
39
40 39
41 def __init__(self, *args, **kwargs): 40 def __init__(self, *args, **kwargs):
42 super().__init__( 41 super().__init__(
43 *args, 42 *args,
44 use_output=C.OUTPUT_CUSTOM, 43 use_output=C.OUTPUT_CUSTOM,
50 #: Qt GUI 49 #: Qt GUI
51 "gui": partial(self.use_output, "gui"), 50 "gui": partial(self.use_output, "gui"),
52 #: experimental TUI output 51 #: experimental TUI output
53 "tui": partial(self.use_output, "tui"), 52 "tui": partial(self.use_output, "tui"),
54 }, 53 },
55 **kwargs 54 **kwargs,
56 ) 55 )
57 56
58 def add_parser_options(self): 57 def add_parser_options(self):
59 self.parser.add_argument( 58 self.parser.add_argument(
60 "--no-ui", action="store_true", help=_("disable user interface") 59 "--no-ui", action="store_true", help=_("disable user interface")
61 ) 60 )
62 sources_group = self.parser.add_mutually_exclusive_group() 61 sources_group = self.parser.add_mutually_exclusive_group()
63 sources_group.add_argument( 62 sources_group.add_argument(
64 "-s", "--sources", choices=['auto', 'test'], default='auto', 63 "-s",
65 help='Well-known sources to use (default: "auto").' 64 "--sources",
65 choices=["auto", "test"],
66 default="auto",
67 help='Well-known sources to use (default: "auto").',
66 ) 68 )
67 69
68 def get_call_data_kw(self) -> dict[str, Any]: 70 def get_call_data_kw(self) -> dict[str, Any]:
69 """Get relevant keyword arguments for CallData""" 71 """Get relevant keyword arguments for CallData"""
70 kwargs: dict[str, Any] = {} 72 kwargs: dict[str, Any] = {}
71 if self.args.sources == "test": 73 if self.args.sources == "test":
72 from libervia.frontends.tools.webrtc_models import SourcesTest 74 from libervia.frontends.tools.webrtc_models import SourcesTest
75
73 kwargs["sources_data"] = SourcesTest() 76 kwargs["sources_data"] = SourcesTest()
74 return kwargs 77 return kwargs
75
76 78
77 async def start(self): 79 async def start(self):
78 root_logger = logging.getLogger() 80 root_logger = logging.getLogger()
79 # we don't want any formatting for messages from webrtc 81 # we don't want any formatting for messages from webrtc
80 for handler in root_logger.handlers: 82 for handler in root_logger.handlers:
131 help=_("JIDs of entity to call"), 133 help=_("JIDs of entity to call"),
132 ) 134 )
133 135
134 async def start(self): 136 async def start(self):
135 await super().start() 137 await super().start()
136 await super().output(CallData( 138 await super().output(
137 callee=jid.JID(self.args.entity), 139 CallData(callee=jid.JID(self.args.entity), kwargs=self.get_call_data_kw())
138 kwargs=self.get_call_data_kw() 140 )
139 ))
140 141
141 142
142 class Receive(Common): 143 class Receive(Common):
143 def __init__(self, host): 144 def __init__(self, host):
144 super().__init__( 145 super().__init__(
155 "-a", 156 "-a",
156 "--auto-accept", 157 "--auto-accept",
157 action="append", 158 action="append",
158 metavar="JID", 159 metavar="JID",
159 default=[], 160 default=[],
160 help=_("automatically accept call from this jid (can be used multiple times)") 161 help=_(
162 "automatically accept call from this jid (can be used multiple times)"
163 ),
161 ) 164 )
162 auto_accept_group.add_argument( 165 auto_accept_group.add_argument(
163 "--auto-accept-all", 166 "--auto-accept-all",
164 action="store_true", 167 action="store_true",
165 help=_("automatically accept call from anybody") 168 help=_("automatically accept call from anybody"),
166 ) 169 )
167 170
168 async def on_action_new( 171 async def on_action_new(
169 self, action_data_s: str, action_id: str, security_limit: int, profile: str 172 self, action_data_s: str, action_id: str, security_limit: int, profile: str
170 ) -> None: 173 ) -> None:
177 caller = peer_jid.bare 180 caller = peer_jid.bare
178 if ( 181 if (
179 not self.args.auto_accept_all 182 not self.args.auto_accept_all
180 and caller not in self.args.auto_accept 183 and caller not in self.args.auto_accept
181 and not await self.host.confirm( 184 and not await self.host.confirm(
182 _("📞 Incoming call from {caller}, do you accept?").format( 185 _("📞 Incoming call from {caller}, do you accept?").format(caller=caller)
183 caller=caller
184 )
185 ) 186 )
186 ): 187 ):
187 await self.host.bridge.action_launch( 188 await self.host.bridge.action_launch(
188 action_id, data_format.serialise({"cancelled": True}), profile 189 action_id, data_format.serialise({"cancelled": True}), profile
189 ) 190 )
190 return 191 return
191 192
192 self.disp(_("✅ Incoming call from {caller} accepted.").format(caller=caller)) 193 self.disp(_("✅ Incoming call from {caller} accepted.").format(caller=caller))
193 194
194 await super().output(CallData( 195 await super().output(
195 callee=peer_jid, 196 CallData(
196 sid=action_data["session_id"], 197 callee=peer_jid,
197 action_id=action_id, 198 sid=action_data["session_id"],
198 kwargs=self.get_call_data_kw() 199 action_id=action_id,
199 )) 200 kwargs=self.get_call_data_kw(),
201 )
202 )
200 203
201 async def start(self): 204 async def start(self):
202 await super().start() 205 await super().start()
203 self.host.bridge.register_signal("action_new", self.on_action_new, "core") 206 self.host.bridge.register_signal("action_new", self.on_action_new, "core")
204 207