diff --git a/libervia/cli/call_gui.py b/libervia/cli/call_gui.py
index a4060af8..4c7117aa 100644
--- a/libervia/cli/call_gui.py
+++ b/libervia/cli/call_gui.py
@@ -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()
diff --git a/libervia/cli/call_simple.py b/libervia/cli/call_simple.py
new file mode 100644
index 00000000..3ddf6303
--- /dev/null
+++ b/libervia/cli/call_simple.py
@@ -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 .
+
+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()
diff --git a/libervia/cli/call_tui.py b/libervia/cli/call_tui.py
new file mode 100644
index 00000000..0ea7e596
--- /dev/null
+++ b/libervia/cli/call_tui.py
@@ -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 .
+
+
+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
diff --git a/libervia/cli/call_webrtc.py b/libervia/cli/call_webrtc.py
new file mode 100644
index 00000000..7e9ac3ed
--- /dev/null
+++ b/libervia/cli/call_webrtc.py
@@ -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 .
+
+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()
diff --git a/libervia/cli/cmd_call.py b/libervia/cli/cmd_call.py
index 339bf75d..47e1cf22 100644
--- a/libervia/cli/cmd_call.py
+++ b/libervia/cli/cmd_call.py
@@ -18,199 +18,22 @@
# along with this program. If not, see .
-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):
diff --git a/pyproject.toml b/pyproject.toml
index b7b26e40..40e7d51e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -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"