erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

cli (call): tui output implementation:

Browse source 
- 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
This commit is contained in:
Goffi 2024-02-16 18:46:06 +01:00
commit 430f2d9354
6 changed files with 560 additions and 227 deletions
Download patch file Download diff file Expand all files Collapse all files

27
libervia/cli/call_gui.py
View file

@ -47,10 +47,17 @@ from PyQt6.QtWidgets import (
QVBoxLayout,
QWidget,
)
from gi.repository import Gst
import gi
from libervia.backend.core.i18n import _
from libervia.cli.call_webrtc import WebRTCCall
from libervia.frontends.tools import aio, display_servers, webrtc
gi.require_versions({
"Gst": "1.0",
"GstWebRTC": "1.0"
})
from gi.repository import Gst
ICON_SIZE = QSize(45, 45)
@ -181,7 +188,7 @@ class X11DesktopScreenDialog(QDialog):
return await self.__a_result
class AVCallGUI(QMainWindow):
class AVCallUI(QMainWindow):
def __init__(self, host, icons_path: Path):
super().__init__()
self.host = host
@ -196,13 +203,16 @@ class AVCallGUI(QMainWindow):
await asyncio.sleep(0.1)
@classmethod
async def run(cls, parent, call_data, icons_path: Path):
async def run(cls, parent, call_data):
"""Run PyQt loop and show the app"""
print("Starting GUI...")
media_dir = Path(await parent.host.bridge.config_get("", "media_dir"))
icons_path = media_dir / "fonts/fontello/svg"
app = QApplication([])
av_call_gui = cls(parent.host, icons_path)
av_call_gui.show()
webrtc_call = await parent.make_webrtc_call(
webrtc_call = await WebRTCCall.make_webrtc_call(
parent.host,
parent.profile,
call_data,
sinks=webrtc.SINKS_APP,
appsink_data=webrtc.AppSinkData(
@ -372,7 +382,7 @@ class AVCallGUI(QMainWindow):
self.fullscreen_btn.setIcon(self.fullscreen_icon_normal)
self.showNormal()
def closeEvent(self, a0: QCloseEvent) -> None:
def closeEvent(self, a0: QCloseEvent|None) -> None:
super().closeEvent(a0)
global running
running = False
@ -400,6 +410,11 @@ class AVCallGUI(QMainWindow):
def hang_up(self):
self.close()
@staticmethod
def can_run():
# if a known display server is detected, we should be able to run
return display_servers.detect() is not None
async def show_X11_screen_dialog(self):
assert self.webrtc_call is not None
windows_data = display_servers.x11_list_windows()

168
libervia/cli/call_simple.py Normal file
View file

@ -0,0 +1,168 @@
#!/usr/bin/env python3
# Libervia CLI
# Copyright (C) 2009-2024 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/>.
import asyncio
from functools import partial
import os
from prompt_toolkit.input import create_input
from prompt_toolkit.keys import Keys
from rich.align import Align
from rich.columns import Columns
from rich.console import group
from rich.live import Live
from rich.panel import Panel
from rich.text import Text
from .call_webrtc import CallData, WebRTCCall
class BaseAVTUI:
def __init__(self, host, webrtc=None, align: str = "left"):
self.host = host
self.webrtc = webrtc
self.align = align
self.input = create_input()
self.done = asyncio.Event()
self.target_size: tuple[int, int] | None = None
@staticmethod
def parse_output_opts(parent) -> dict:
"""Parse output options.
This method should be called in a loop checking all output options.
It will set the relevant attributes.
@return: keyword argument to use to instanciate WebRTCCall
"""
kwargs = {}
for oo in parent.args.output_opts:
if oo.startswith("size="):
try:
width_s, height_s = oo[5:].lower().strip().split("x", 1)
width, height = int(width_s), int(height_s)
except ValueError:
parent.parser.error(
"Invalid size; it must be in the form widthxheight "
"(e.g., 640x360)."
)
else:
kwargs["target_size"] = (width, height)
return kwargs
def styled_shortcut_key(self, word: str, key: str | None = None) -> Text:
"""Return a word with the specified key or the first letter underlined."""
if key is None:
key = word[0]
index = word.find(key)
before, keyword, after = word[:index], word[index], word[index + 1 :]
return Text(before) + Text(keyword, style="shortcut") + Text(after)
def get_micro_display(self):
assert self.webrtc is not None
if self.webrtc.audio_muted:
return Panel(Text("🔇 ") + self.styled_shortcut_key("Muted"), expand=False)
else:
return Panel(
Text("🎤 ") + self.styled_shortcut_key("Unmuted", "m"), expand=False
)
def get_video_display(self):
assert self.webrtc is not None
if self.webrtc.video_muted:
return Panel(Text("❌ ") + self.styled_shortcut_key("Video Off"), expand=False)
else:
return Panel(Text("🎥 ") + self.styled_shortcut_key("Video On"), expand=False)
def get_phone_display(self):
return Panel(Text("📞 ") + self.styled_shortcut_key("Hang up"), expand=False)
@group()
def generate_control_bar(self):
"""Return the full interface display."""
assert self.webrtc is not None
yield Align(
Columns(
[
Text(),
self.get_micro_display(),
self.get_video_display(),
self.get_phone_display(),
],
expand=False,
title="Calling [bold center]{}[/]".format(self.webrtc.callee),
),
align=self.align,
)
def keys_ready(self, live=None):
assert self.webrtc is not None
for key_press in self.input.read_keys():
char = key_press.key.lower()
if char == "m":
# audio mute
self.webrtc.audio_muted = not self.webrtc.audio_muted
if live is not None:
live.update(self.generate_control_bar(), refresh=True)
elif char == "v":
# video mute
self.webrtc.video_muted = not self.webrtc.video_muted
if live is not None:
live.update(self.generate_control_bar(), refresh=True)
elif char == "h" or key_press.key == Keys.ControlC:
# Hang up
self.done.set()
elif char == "d":
# generate dot file for debugging. Only available if
# ``GST_DEBUG_DUMP_DOT_DIR`` is set. Filename is "pipeline.dot" with a
# timestamp.
if os.getenv("GST_DEBUG_DUMP_DOT_DIR"):
self.webrtc.generate_dot_file()
self.host.disp("Dot file generated.")
class AVCallUI(BaseAVTUI):
def __init__(self, host, webrtc):
super().__init__(host, webrtc)
async def start(self):
assert self.webrtc is not None
with Live(
self.generate_control_bar(), console=self.host.console, auto_refresh=False
) as live:
with self.input.raw_mode():
with self.input.attach(partial(self.keys_ready, live)):
await self.done.wait()
await self.webrtc.end_call()
await self.host.a_quit()
@classmethod
async def run(cls, parent, call_data: CallData) -> None:
kwargs = cls.parse_output_opts(parent)
merge_pip = False if "split" in parent.args.output_opts else None
webrtc_call = await WebRTCCall.make_webrtc_call(
parent.host,
parent.profile,
call_data,
merge_pip=merge_pip,
**kwargs,
)
if not parent.args.no_ui:
ui = cls(parent.host, webrtc_call.webrtc)
await ui.start()

194
libervia/cli/call_tui.py Normal file
View file

@ -0,0 +1,194 @@
#!/usr/bin/env python3
# Libervia CLI
# Copyright (C) 2009-2024 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/>.
import asyncio
from functools import partial
import sys
from PIL import Image
import gi
from rich.padding import Padding
from term_image import image as t_image
from libervia.cli.constants import Const as C
from libervia.frontends.tools import webrtc
from .call_simple import BaseAVTUI
from .call_webrtc import CallData, WebRTCCall
gi.require_versions({"Gst": "1.0", "GstWebRTC": "1.0"})
from gi.repository import Gst
class AVCallUI(BaseAVTUI):
def __init__(self, parent):
super().__init__(parent.host, align="center")
self.parent = parent
self.image_rows = 0
self.buffer = None
self._processing = False
self.render_class: t_image.ImageMeta | None = None
self.target_size
self.fps = 25
for oo in parent.args.output_opts:
if oo.startswith("renderer="):
renderer = oo[9:].lower().strip()
match renderer:
case "auto":
# we let None as it will auto-detect best option later
pass
case "block":
self.render_class = t_image.BlockImage
case "iterm2":
self.render_class = t_image.ITerm2Image
case "kitty":
self.render_class = t_image.KittyImage
case _:
parent.parser.error(f"Invalid renderer: {renderer!r}")
elif oo.startswith("fps="):
try:
self.fps = int(oo[4:])
except ValueError:
parent.parser.error(f"Invalid FPS: {oo[4:]!r}")
async def init_call(self, call_data):
kwargs = self.parse_output_opts(self.parent)
if "target_size" not in kwargs:
# we use low res by default for performance reason
kwargs["target_size"] = (640, 380)
webrtc_call = await WebRTCCall.make_webrtc_call(
self.parent.host,
self.parent.profile,
call_data,
sinks=webrtc.SINKS_APP,
appsink_data=webrtc.AppSinkData(
local_video_cb=partial(self.on_new_sample, video_stream="local"),
remote_video_cb=None,
),
merge_pip=True,
**kwargs,
)
self.webrtc = webrtc_call.webrtc
async def start(self, call_data):
term_rows = self.host.console.size.height
if term_rows < 20:
self.host.disp(
"Your terminal must have a height of a least 20 rows.", error=True
)
self.host.a_quit(C.EXIT_ERROR)
self.image_rows = term_rows - 6
self.image_cols = self.host.console.size.width
await self.init_call(call_data)
assert self.webrtc is not None
idx = 0
self.buffer = ""
# we detect render
if self.render_class is None:
self.render_class = t_image.auto_image_class()
loop_sleep = 1/self.fps
with self.input.raw_mode():
# for whatever reason, using self.input.attach is breaking KittyImage and uses
# a BlockImage style rendering instead. So we don't use it and we call
# ``self.keys_ready()`` ourself in the loop, below.
# cursor is not restored despite the ``screen`` context if peer is hanging up,
# so we reactivate cursor here
self.host.add_on_quit_callback(self.host.console.show_cursor, True)
with self.host.console.screen():
while True:
idx += 1
if self.buffer is not None:
sys.stdout.write(self.buffer)
sys.stdout.write("\n")
self.parent.console.print(
# the padding avoid artifact when toggling buttons
Padding(self.generate_control_bar(), (0, 2, 0, 0))
)
sys.stdout.flush()
rendered = True
else:
rendered = False
await asyncio.sleep(loop_sleep)
self.keys_ready()
if self.done.is_set():
break
if rendered:
# we put cursor back at the top of image to print the next frame
# FIXME: we use +4 for the controls because we know the height of the
# renderable, but it would be better to measure it dynamically.
sys.stdout.write(f"\033[{self.image_rows+4}A")
await self.webrtc.end_call()
await self.host.a_quit()
@classmethod
async def run(cls, parent, call_data: CallData) -> None:
ui = cls(parent)
await ui.start(call_data)
def on_new_sample(self, video_sink, video_stream: str) -> bool:
if self._processing:
# image encoding for terminal is slow, if one is already processing, we don't
# bother going further
return False
sample = video_sink.emit("pull-sample")
if sample is None:
return False
video_pad = video_sink.get_static_pad("sink")
assert video_pad is not None
s = video_pad.get_current_caps().get_structure(0)
stream_size = (s.get_value("width"), s.get_value("height"))
buf = sample.get_buffer()
result, mapinfo = buf.map(Gst.MapFlags.READ)
if result and self.render_class is not None:
self._processing = True
image_data = mapinfo.data
image = Image.frombuffer("RGB", stream_size, image_data, "raw", "RGB", 0, 1)
img_renderer = self.render_class(image, height=self.image_rows)
img_fmt = f"<{self.image_cols}.^1"
if self.render_class == t_image.KittyImage:
# we don't do compression to speed up things
img_fmt += "+Wc0"
self.host.loop.loop.call_soon_threadsafe(
self.update_sample,
sample,
stream_size,
video_stream,
format(img_renderer, img_fmt),
)
self._processing = False
buf.unmap(mapinfo)
return False
def update_sample(self, sample, stream_size, video_stream: str, buffer) -> None:
if sample is None:
return
if video_stream == "remote":
return
self.buffer = buffer

135
libervia/cli/call_webrtc.py Normal file
View file

@ -0,0 +1,135 @@
#!/usr/bin/env python3
# Libervia CLI
# Copyright (C) 2009-2024 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 dataclasses import dataclass
from libervia.backend.tools.common import data_format
from libervia.frontends.tools import aio, jid
@dataclass
class CallData:
callee: jid.JID
sid: str | None = None
action_id: str | None = None
class WebRTCCall:
def __init__(self, host, profile: str, callee: jid.JID, **kwargs):
"""Create and setup a webRTC instance
@param profile: profile making or receiving the call
@param callee: peer jid
@param kwargs: extra kw args to use when instantiating WebRTC
"""
from libervia.frontends.tools import webrtc
aio.install_glib_asyncio_iteration()
self.host = host
self.profile = profile
self.webrtc = webrtc.WebRTC(host.bridge, profile, **kwargs)
self.webrtc.callee = callee
host.bridge.register_signal(
"ice_candidates_new", self.on_ice_candidates_new, "plugin"
)
host.bridge.register_signal("call_setup", self.on_call_setup, "plugin")
host.bridge.register_signal("call_ended", self.on_call_ended, "plugin")
@classmethod
async def make_webrtc_call(
cls,
host,
profile: str,
call_data: CallData,
**kwargs
) -> "WebRTCCall":
"""Create the webrtc_call instance
@param call_data: Call data of the command
@param kwargs: extra args used to instanciate WebRTCCall
"""
webrtc_call = cls(host, profile, call_data.callee, **kwargs)
if call_data.sid is None:
# we are making the call
await webrtc_call.start()
else:
# we are receiving the call
webrtc_call.sid = call_data.sid
if call_data.action_id is not None:
await host.bridge.action_launch(
call_data.action_id,
data_format.serialise({"cancelled": False}),
profile
)
return webrtc_call
@property
def sid(self) -> str | None:
return self.webrtc.sid
@sid.setter
def sid(self, new_sid: str | None) -> None:
self.webrtc.sid = new_sid
async def on_ice_candidates_new(
self, sid: str, candidates_s: str, profile: str
) -> None:
if sid != self.webrtc.sid or profile != self.profile:
return
self.webrtc.on_ice_candidates_new(
data_format.deserialise(candidates_s),
)
async def on_call_setup(self, sid: str, setup_data_s: str, profile: str) -> None:
if sid != self.webrtc.sid or profile != self.profile:
return
setup_data = data_format.deserialise(setup_data_s)
try:
role = setup_data["role"]
sdp = setup_data["sdp"]
except KeyError:
self.host.disp(f"Invalid setup data received: {setup_data}", error=True)
return
if role == "initiator":
self.webrtc.on_accepted_call(sdp, profile)
elif role == "responder":
await self.webrtc.answer_call(sdp, profile)
else:
self.host.disp(
f"Invalid role received during setup: {setup_data}", error=True
)
# we want to be sure that call is ended if user presses `Ctrl + c` or anything
# else stops the session.
self.host.add_on_quit_callback(
lambda: self.host.bridge.call_end(sid, "", profile)
)
async def on_call_ended(self, sid: str, data_s: str, profile: str) -> None:
if sid != self.webrtc.sid or profile != self.profile:
return
await self.webrtc.end_call()
await self.host.a_quit()
async def start(self):
"""Start a call.
To be used only if we are initiator
"""
await self.webrtc.setup_call("initiator")
self.webrtc.start_pipeline()

254
libervia/cli/cmd_call.py
View file

@ -18,199 +18,22 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from argparse import ArgumentParser
import asyncio
from dataclasses import dataclass
from functools import partial
import importlib
import logging
import os
from pathlib import Path
from typing import Callable
from prompt_toolkit.input import create_input
from prompt_toolkit.keys import Keys
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 aio, jid
from rich.columns import Columns
from rich.console import group
from rich.live import Live
from rich.panel import Panel
from rich.text import Text
from libervia.frontends.tools import jid
from . import base
from .call_webrtc import CallData, WebRTCCall
__commands__ = ["Call"]
@dataclass
class CallData:
callee: jid.JID
sid: str|None = None
action_id: str|None = None
class WebRTCCall:
def __init__(self, host, profile: str, callee: jid.JID, **kwargs):
"""Create and setup a webRTC instance
@param profile: profile making or receiving the call
@param callee: peer jid
@param kwargs: extra kw args to use when instantiating WebRTC
"""
from libervia.frontends.tools import webrtc
aio.install_glib_asyncio_iteration()
self.host = host
self.profile = profile
self.webrtc = webrtc.WebRTC(host.bridge, profile, **kwargs)
self.webrtc.callee = callee
host.bridge.register_signal(
"ice_candidates_new", self.on_ice_candidates_new, "plugin"
)
host.bridge.register_signal("call_setup", self.on_call_setup, "plugin")
host.bridge.register_signal("call_ended", self.on_call_ended, "plugin")
@property
def sid(self) -> str | None:
return self.webrtc.sid
@sid.setter
def sid(self, new_sid: str | None) -> None:
self.webrtc.sid = new_sid
async def on_ice_candidates_new(
self, sid: str, candidates_s: str, profile: str
) -> None:
if sid != self.webrtc.sid or profile != self.profile:
return
self.webrtc.on_ice_candidates_new(
data_format.deserialise(candidates_s),
)
async def on_call_setup(self, sid: str, setup_data_s: str, profile: str) -> None:
if sid != self.webrtc.sid or profile != self.profile:
return
setup_data = data_format.deserialise(setup_data_s)
try:
role = setup_data["role"]
sdp = setup_data["sdp"]
except KeyError:
self.host.disp(f"Invalid setup data received: {setup_data}", error=True)
return
if role == "initiator":
self.webrtc.on_accepted_call(sdp, profile)
elif role == "responder":
await self.webrtc.answer_call(sdp, profile)
else:
self.host.disp(
f"Invalid role received during setup: {setup_data}", error=True
)
# we want to be sure that call is ended if user presses `Ctrl + c` or anything
# else stops the session.
self.host.add_on_quit_callback(
lambda: self.host.bridge.call_end(sid, "", profile)
)
async def on_call_ended(self, sid: str, data_s: str, profile: str) -> None:
if sid != self.webrtc.sid or profile != self.profile:
return
await self.webrtc.end_call()
await self.host.a_quit()
async def start(self):
"""Start a call.
To be used only if we are initiator
"""
await self.webrtc.setup_call("initiator")
self.webrtc.start_pipeline()
class UI:
def __init__(self, host, webrtc):
self.host = host
self.webrtc = webrtc
def styled_shortcut_key(self, word: str, key: str | None = None) -> Text:
"""Return a word with the specified key or the first letter underlined."""
if key is None:
key = word[0]
index = word.find(key)
before, keyword, after = word[:index], word[index], word[index + 1 :]
return Text(before) + Text(keyword, style="shortcut") + Text(after)
def get_micro_display(self):
if self.webrtc.audio_muted:
return Panel(Text("🔇 ") + self.styled_shortcut_key("Muted"), expand=False)
else:
return Panel(
Text("🎤 ") + self.styled_shortcut_key("Unmuted", "m"), expand=False
)
def get_video_display(self):
if self.webrtc.video_muted:
return Panel(Text("❌ ") + self.styled_shortcut_key("Video Off"), expand=False)
else:
return Panel(Text("🎥 ") + self.styled_shortcut_key("Video On"), expand=False)
def get_phone_display(self):
return Panel(Text("📞 ") + self.styled_shortcut_key("Hang up"), expand=False)
@group()
def generate_control_bar(self):
"""Return the full interface display."""
yield Columns(
[
self.get_micro_display(),
self.get_video_display(),
self.get_phone_display(),
],
expand=False,
title="Calling [bold center]{}[/]".format(self.webrtc.callee),
)
async def start(self):
done = asyncio.Event()
input = create_input()
def keys_ready(live):
for key_press in input.read_keys():
char = key_press.key.lower()
if char == "m":
# audio mute
self.webrtc.audio_muted = not self.webrtc.audio_muted
live.update(self.generate_control_bar(), refresh=True)
elif char == "v":
# video mute
self.webrtc.video_muted = not self.webrtc.video_muted
live.update(self.generate_control_bar(), refresh=True)
elif char == "h" or key_press.key == Keys.ControlC:
# Hang up
done.set()
elif char == "d":
# generate dot file for debugging. Only available if
# ``GST_DEBUG_DUMP_DOT_DIR`` is set. Filename is "pipeline.dot" with a
# timestamp.
if os.getenv("GST_DEBUG_DUMP_DOT_DIR"):
self.webrtc.generate_dot_file()
self.host.disp("Dot file generated.")
with Live(
self.generate_control_bar(),
console=self.host.console,
auto_refresh=False
) as live:
with input.raw_mode():
with input.attach(partial(keys_ready, live)):
await done.wait()
await self.webrtc.end_call()
await self.host.a_quit()
class Common(base.CommandBase):
@ -219,9 +42,14 @@ class Common(base.CommandBase):
*args,
use_output=C.OUTPUT_CUSTOM,
extra_outputs={
#: automatically select best output for current platform
"default": self.auto_output,
"simple": self.simple_output,
"gui": self.gui_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
)
@ -243,50 +71,32 @@ class Common(base.CommandBase):
if self.verbosity >= 2:
root_logger.setLevel(logging.DEBUG)
async def make_webrtc_call(self, call_data: CallData, **kwargs) -> WebRTCCall:
"""Create the webrtc_call instance
@param call_data: Call data of the command
@param kwargs: extra args used to instanciate WebRTCCall
"""
webrtc_call = WebRTCCall(self.host, self.profile, call_data.callee, **kwargs)
if call_data.sid is None:
# we are making the call
await webrtc_call.start()
else:
# we are receiving the call
webrtc_call.sid = call_data.sid
if call_data.action_id is not None:
await self.host.bridge.action_launch(
call_data.action_id,
data_format.serialise({"cancelled": False}),
self.profile
)
return webrtc_call
async def auto_output(self, call_data: CallData):
async def auto_output(self, call_data: CallData) -> None:
"""Make a guess on the best output to use on current platform"""
# For now we just use simple output
await self.simple_output(call_data)
async def simple_output(self, call_data: CallData):
"""Run simple output, with GStreamer ``autovideosink``"""
webrtc_call = await self.make_webrtc_call(call_data)
if not self.args.no_ui:
ui = UI(self.host, webrtc_call.webrtc)
await ui.start()
async def gui_output(self, call_data: CallData):
"""Run GUI output"""
media_dir = Path(await self.host.bridge.config_get("", "media_dir"))
icons_path = media_dir / "fonts/fontello/svg"
try:
from .call_gui import AVCallGUI
await AVCallGUI.run(self, call_data, icons_path)
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 GUI: {e}", error=True)
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):

11
pyproject.toml
View file

@ -80,6 +80,17 @@ keywords = [
SVG = [
"CairoSVG",
]
GUI = [
"PyQt6"
]
TUI = [
"term-image ~= 0.7.1"
]
all = [
"libervia-backed[SVG]",
"libervia-backed[GUI]",
"libervia-backed[TUI]",
]
[project.scripts]
jp = "libervia.cli.base:LiberviaCli.run"