Mercurial > libervia-backend

comparison libervia/cli/cmd_call.py @ 4210:9218d4331bb2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
cli (call): `tui` output implementation: - Moved original UI to a separated class, and use if with the `simple` output - By default, best output is automatically selected. For now `gui` is selected if possible, and `simple` is used as fallback. - The new `tui` output can be used to have the videos directly embedded in the terminal, either with real videos for compatible terminal emulators, or with Unicode blocks. - Text contrôls are used for both `simple` and `tui` outputs - several options can be used with `--oo` (will be documented in next commit). rel 428
author Goffi <goffi@goffi.org>
date Fri, 16 Feb 2024 18:46:06 +0100
parents 0f8ea0768a3b
children d01b8d002619
comparison
equal deleted inserted replaced
4209:fe29fbdabce6 4210:9218d4331bb2
16 16
17 # You should have received a copy of the GNU Affero General Public License 17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>. 18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19 19
20 20
21 from argparse import ArgumentParser
22 import asyncio
23 from dataclasses import dataclass
24 from functools import partial 21 from functools import partial
22 import importlib
25 import logging 23 import logging
26 import os
27 from pathlib import Path
28 from typing import Callable
29 24
30 from prompt_toolkit.input import create_input
31 from prompt_toolkit.keys import Keys
32 25
33 from libervia.backend.core.i18n import _ 26 from libervia.backend.core.i18n import _
34 from libervia.backend.tools.common import data_format 27 from libervia.backend.tools.common import data_format
35 from libervia.cli.constants import Const as C 28 from libervia.cli.constants import Const as C
36 from libervia.frontends.tools import aio, jid 29 from libervia.frontends.tools import jid
37 from rich.columns import Columns
38 from rich.console import group
39 from rich.live import Live
40 from rich.panel import Panel
41 from rich.text import Text
42 30
43 from . import base 31 from . import base
32 from .call_webrtc import CallData, WebRTCCall
44 33
45 __commands__ = ["Call"] 34 __commands__ = ["Call"]
46
47
48 @dataclass
49 class CallData:
50 callee: jid.JID
51 sid: str|None = None
52 action_id: str|None = None
53
54
55 class WebRTCCall:
56 def __init__(self, host, profile: str, callee: jid.JID, **kwargs):
57 """Create and setup a webRTC instance
58
59 @param profile: profile making or receiving the call
60 @param callee: peer jid
61 @param kwargs: extra kw args to use when instantiating WebRTC
62 """
63 from libervia.frontends.tools import webrtc
64
65 aio.install_glib_asyncio_iteration()
66 self.host = host
67 self.profile = profile
68 self.webrtc = webrtc.WebRTC(host.bridge, profile, **kwargs)
69 self.webrtc.callee = callee
70 host.bridge.register_signal(
71 "ice_candidates_new", self.on_ice_candidates_new, "plugin"
72 )
73 host.bridge.register_signal("call_setup", self.on_call_setup, "plugin")
74 host.bridge.register_signal("call_ended", self.on_call_ended, "plugin")
75
76 @property
77 def sid(self) -> str | None:
78 return self.webrtc.sid
79
80 @sid.setter
81 def sid(self, new_sid: str | None) -> None:
82 self.webrtc.sid = new_sid
83
84 async def on_ice_candidates_new(
85 self, sid: str, candidates_s: str, profile: str
86 ) -> None:
87 if sid != self.webrtc.sid or profile != self.profile:
88 return
89 self.webrtc.on_ice_candidates_new(
90 data_format.deserialise(candidates_s),
91 )
92
93 async def on_call_setup(self, sid: str, setup_data_s: str, profile: str) -> None:
94 if sid != self.webrtc.sid or profile != self.profile:
95 return
96 setup_data = data_format.deserialise(setup_data_s)
97 try:
98 role = setup_data["role"]
99 sdp = setup_data["sdp"]
100 except KeyError:
101 self.host.disp(f"Invalid setup data received: {setup_data}", error=True)
102 return
103 if role == "initiator":
104 self.webrtc.on_accepted_call(sdp, profile)
105 elif role == "responder":
106 await self.webrtc.answer_call(sdp, profile)
107 else:
108 self.host.disp(
109 f"Invalid role received during setup: {setup_data}", error=True
110 )
111 # we want to be sure that call is ended if user presses `Ctrl + c` or anything
112 # else stops the session.
113 self.host.add_on_quit_callback(
114 lambda: self.host.bridge.call_end(sid, "", profile)
115 )
116
117 async def on_call_ended(self, sid: str, data_s: str, profile: str) -> None:
118 if sid != self.webrtc.sid or profile != self.profile:
119 return
120 await self.webrtc.end_call()
121 await self.host.a_quit()
122
123 async def start(self):
124 """Start a call.
125
126 To be used only if we are initiator
127 """
128 await self.webrtc.setup_call("initiator")
129 self.webrtc.start_pipeline()
130
131
132 class UI:
133 def __init__(self, host, webrtc):
134 self.host = host
135 self.webrtc = webrtc
136
137 def styled_shortcut_key(self, word: str, key: str | None = None) -> Text:
138 """Return a word with the specified key or the first letter underlined."""
139 if key is None:
140 key = word[0]
141 index = word.find(key)
142 before, keyword, after = word[:index], word[index], word[index + 1 :]
143 return Text(before) + Text(keyword, style="shortcut") + Text(after)
144
145 def get_micro_display(self):
146 if self.webrtc.audio_muted:
147 return Panel(Text("🔇 ") + self.styled_shortcut_key("Muted"), expand=False)
148 else:
149 return Panel(
150 Text("🎤 ") + self.styled_shortcut_key("Unmuted", "m"), expand=False
151 )
152
153 def get_video_display(self):
154 if self.webrtc.video_muted:
155 return Panel(Text("❌ ") + self.styled_shortcut_key("Video Off"), expand=False)
156 else:
157 return Panel(Text("🎥 ") + self.styled_shortcut_key("Video On"), expand=False)
158
159 def get_phone_display(self):
160 return Panel(Text("📞 ") + self.styled_shortcut_key("Hang up"), expand=False)
161
162 @group()
163 def generate_control_bar(self):
164 """Return the full interface display."""
165 yield Columns(
166 [
167 self.get_micro_display(),
168 self.get_video_display(),
169 self.get_phone_display(),
170 ],
171 expand=False,
172 title="Calling [bold center]{}[/]".format(self.webrtc.callee),
173 )
174
175 async def start(self):
176 done = asyncio.Event()
177 input = create_input()
178
179 def keys_ready(live):
180 for key_press in input.read_keys():
181 char = key_press.key.lower()
182 if char == "m":
183 # audio mute
184 self.webrtc.audio_muted = not self.webrtc.audio_muted
185 live.update(self.generate_control_bar(), refresh=True)
186 elif char == "v":
187 # video mute
188 self.webrtc.video_muted = not self.webrtc.video_muted
189 live.update(self.generate_control_bar(), refresh=True)
190 elif char == "h" or key_press.key == Keys.ControlC:
191 # Hang up
192 done.set()
193 elif char == "d":
194 # generate dot file for debugging. Only available if
195 # ``GST_DEBUG_DUMP_DOT_DIR`` is set. Filename is "pipeline.dot" with a
196 # timestamp.
197 if os.getenv("GST_DEBUG_DUMP_DOT_DIR"):
198 self.webrtc.generate_dot_file()
199 self.host.disp("Dot file generated.")
200
201 with Live(
202 self.generate_control_bar(),
203 console=self.host.console,
204 auto_refresh=False
205 ) as live:
206 with input.raw_mode():
207 with input.attach(partial(keys_ready, live)):
208 await done.wait()
209
210 await self.webrtc.end_call()
211 await self.host.a_quit()
212 35
213 36
214 class Common(base.CommandBase): 37 class Common(base.CommandBase):
215 38
216 39
217 def __init__(self, *args, **kwargs): 40 def __init__(self, *args, **kwargs):
218 super().__init__( 41 super().__init__(
219 *args, 42 *args,
220 use_output=C.OUTPUT_CUSTOM, 43 use_output=C.OUTPUT_CUSTOM,
221 extra_outputs={ 44 extra_outputs={
45 #: automatically select best output for current platform
222 "default": self.auto_output, 46 "default": self.auto_output,
223 "simple": self.simple_output, 47 #: simple output with GStreamer ``autovideosink``
224 "gui": self.gui_output, 48 "simple": partial(self.use_output, "simple"),
49 #: Qt GUI
50 "gui": partial(self.use_output, "gui"),
51 #: experimental TUI output
52 "tui": partial(self.use_output, "tui"),
225 }, 53 },
226 **kwargs 54 **kwargs
227 ) 55 )
228 56
229 def add_parser_options(self): 57 def add_parser_options(self):
241 if self.verbosity >= 1: 69 if self.verbosity >= 1:
242 root_logger.setLevel(logging.WARNING) 70 root_logger.setLevel(logging.WARNING)
243 if self.verbosity >= 2: 71 if self.verbosity >= 2:
244 root_logger.setLevel(logging.DEBUG) 72 root_logger.setLevel(logging.DEBUG)
245 73
246 async def make_webrtc_call(self, call_data: CallData, **kwargs) -> WebRTCCall: 74 async def auto_output(self, call_data: CallData) -> None:
247 """Create the webrtc_call instance 75 """Make a guess on the best output to use on current platform"""
76 try:
77 from .call_gui import AVCallUI
78 except ImportError:
79 # we can't import GUI, we may have missing modules
80 await self.use_output("simple", call_data)
81 else:
82 if AVCallUI.can_run():
83 await self.use_output("gui", call_data)
84 else:
85 await self.use_output("simple", call_data)
248 86
249 @param call_data: Call data of the command 87 async def use_output(self, output_type, call_data: CallData):
250 @param kwargs: extra args used to instanciate WebRTCCall 88 try:
251 89 AVCall_module = importlib.import_module(f"libervia.cli.call_{output_type}")
252 """ 90 AVCallUI = AVCall_module.AVCallUI
253 webrtc_call = WebRTCCall(self.host, self.profile, call_data.callee, **kwargs) 91 except Exception as e:
254 if call_data.sid is None: 92 self.disp(f"Error starting {output_type.upper()} UI: {e}", error=True)
255 # we are making the call 93 self.host.quit(C.EXIT_ERROR)
256 await webrtc_call.start()
257 else: 94 else:
258 # we are receiving the call 95 try:
259 webrtc_call.sid = call_data.sid 96 await AVCallUI.run(self, call_data)
260 if call_data.action_id is not None: 97 except Exception as e:
261 await self.host.bridge.action_launch( 98 self.disp(f"Error running {output_type.upper()} UI: {e}", error=True)
262 call_data.action_id, 99 self.host.quit(C.EXIT_ERROR)
263 data_format.serialise({"cancelled": False}),
264 self.profile
265 )
266 return webrtc_call
267
268 async def auto_output(self, call_data: CallData):
269 """Make a guess on the best output to use on current platform"""
270 # For now we just use simple output
271 await self.simple_output(call_data)
272
273 async def simple_output(self, call_data: CallData):
274 """Run simple output, with GStreamer ``autovideosink``"""
275 webrtc_call = await self.make_webrtc_call(call_data)
276 if not self.args.no_ui:
277 ui = UI(self.host, webrtc_call.webrtc)
278 await ui.start()
279
280 async def gui_output(self, call_data: CallData):
281 """Run GUI output"""
282 media_dir = Path(await self.host.bridge.config_get("", "media_dir"))
283 icons_path = media_dir / "fonts/fontello/svg"
284 try:
285 from .call_gui import AVCallGUI
286 await AVCallGUI.run(self, call_data, icons_path)
287 except Exception as e:
288 self.disp(f"Error starting GUI: {e}", error=True)
289 self.host.quit(C.EXIT_ERROR)
290 100
291 101
292 class Make(Common): 102 class Make(Common):
293 def __init__(self, host): 103 def __init__(self, host):
294 super().__init__( 104 super().__init__(