yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

clan-app: working js<->python api bridge

Browse Source
This commit is contained in:
Qubasa
2025-01-04 20:02:43 +01:00
parent d60cd27097
commit bed51fc324
15 changed files with 103 additions and 613 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
pkgs/clan-app/clan_app/__init__.py
View File

@@ -9,9 +9,10 @@ import argparse
import os
from pathlib import Path
from clan_cli.api import API
from clan_cli.custom_logger import setup_logging
from clan_app.deps.webview.webview import Webview
from clan_app.deps.webview.webview import Size, SizeHint, Webview
@profile
@@ -37,9 +38,10 @@ def main(argv: list[str] = sys.argv) -> int:
site_index: Path = Path(os.getenv("WEBUI_PATH", ".")).resolve() / "index.html"
content_uri = f"file://{site_index}"
site_index: Path = Path(os.getenv("WEBUI_PATH", ".")).resolve() / "index.html"
content_uri = f"file://{site_index}"
webview = Webview(debug=args.debug)
webview.bind_jsonschema_api(API)
webview = Webview()
webview.size = Size(1280, 1024, SizeHint.NONE)
webview.navigate(content_uri)
webview.run()
return 0

143
pkgs/clan-app/clan_app/app.py
View File

@@ -1,143 +0,0 @@
#!/usr/bin/env python3
import logging
import os
from typing import Any, ClassVar
import gi
from clan_app import assets
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
from pathlib import Path
from clan_cli.custom_logger import setup_logging
from gi.repository import Adw, Gdk, Gio, GLib, GObject, Gtk
from clan_app.components.interfaces import ClanConfig
from .windows.main_window import MainWindow
log = logging.getLogger(__name__)
class MainApplication(Adw.Application):
"""
This class is initialized every time the app is started
Only the Adw.ApplicationWindow is a singleton.
So don't use any singletons in the Adw.Application class.
"""
__gsignals__: ClassVar = {
"join_request": (GObject.SignalFlags.RUN_FIRST, None, [str]),
}
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(
application_id="org.clan.app",
flags=Gio.ApplicationFlags.HANDLES_COMMAND_LINE,
)
self.add_main_option(
"debug",
ord("d"),
GLib.OptionFlags.NONE,
GLib.OptionArg.NONE,
"enable debug mode",
None,
)
self.add_main_option(
"content-uri",
GLib.OptionFlags.NONE,
GLib.OptionFlags.NONE,
GLib.OptionArg.STRING,
"set the webview content uri",
None,
)
site_index: Path = Path(os.getenv("WEBUI_PATH", ".")).resolve() / "index.html"
self.content_uri = f"file://{site_index}"
self.window: MainWindow | None = None
self.connect("activate", self.on_activate)
self.connect("shutdown", self.on_shutdown)
def on_shutdown(self, source: "MainApplication") -> None:
log.debug("Shutting down Adw.Application")
if self.get_windows() == []:
log.debug("No windows to destroy")
if self.window:
# TODO: Doesn't seem to raise the destroy signal. Need to investigate
# self.get_windows() returns an empty list. Desync between window and application?
self.window.close()
else:
log.error("No window to destroy")
def do_command_line(self, command_line: Any) -> int:
options = command_line.get_options_dict()
# convert GVariantDict -> GVariant -> dict
options = options.end().unpack()
if "debug" in options and self.window is None:
setup_logging(logging.DEBUG, root_log_name=__name__.split(".")[0])
setup_logging(logging.DEBUG, root_log_name="clan_cli")
elif self.window is None:
setup_logging(logging.INFO, root_log_name=__name__.split(".")[0])
log.debug("Debug logging enabled")
if "content-uri" in options:
self.content_uri = options["content-uri"]
log.debug(f"Setting content uri to {self.content_uri}")
args = command_line.get_arguments()
self.activate()
# Check if there are arguments that are not inside the options
if len(args) > 1:
non_option_args = [arg for arg in args[1:] if arg not in options.values()]
if non_option_args:
uri = non_option_args[0]
self.emit("join_request", uri)
return 0
def on_window_hide_unhide(self, *_args: Any) -> None:
if not self.window:
log.error("No window to hide/unhide")
return
if self.window.is_visible():
self.window.hide()
else:
self.window.present()
def dummy_menu_entry(self) -> None:
log.info("Dummy menu entry called")
def on_activate(self, source: "MainApplication") -> None:
if not self.window:
self.init_style()
self.window = MainWindow(
config=ClanConfig(initial_view="webview", content_uri=self.content_uri)
)
self.window.set_application(self)
self.window.show()
# TODO: For css styling
def init_style(self) -> None:
resource_path = assets.loc / "style.css"
log.debug(f"Style css path: {resource_path}")
css_provider = Gtk.CssProvider()
css_provider.load_from_path(str(resource_path))
display = Gdk.Display.get_default()
assert display is not None
Gtk.StyleContext.add_provider_for_display(
display,
css_provider,
Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION,
)

0
pkgs/clan-app/clan_app/components/__init__.py
View File

127
pkgs/clan-app/clan_app/components/executor.py
View File

@@ -1,127 +0,0 @@
import dataclasses
import logging
import multiprocessing as mp
import os
import signal
import sys
import traceback
from collections.abc import Callable
from pathlib import Path
from typing import Any
log = logging.getLogger(__name__)
# Kill the new process and all its children by sending a SIGTERM signal to the process group
def _kill_group(proc: mp.Process) -> None:
pid = proc.pid
if proc.is_alive() and pid:
os.killpg(pid, signal.SIGTERM)
else:
log.warning(f"Process '{proc.name}' with pid '{pid}' is already dead")
@dataclasses.dataclass(frozen=True)
class MPProcess:
name: str
proc: mp.Process
out_file: Path
# Kill the new process and all its children by sending a SIGTERM signal to the process group
def kill_group(self) -> None:
_kill_group(proc=self.proc)
def _set_proc_name(name: str) -> None:
if sys.platform != "linux":
return
import ctypes
# Define the prctl function with the appropriate arguments and return type
libc = ctypes.CDLL("libc.so.6")
prctl = libc.prctl
prctl.argtypes = [
ctypes.c_int,
ctypes.c_char_p,
ctypes.c_ulong,
ctypes.c_ulong,
ctypes.c_ulong,
]
prctl.restype = ctypes.c_int
# Set the process name to "my_process"
prctl(15, name.encode(), 0, 0, 0)
def _init_proc(
func: Callable,
out_file: Path,
proc_name: str,
on_except: Callable[[Exception, mp.process.BaseProcess], None] | None,
**kwargs: Any,
) -> None:
# Create a new process group
os.setsid()
# Open stdout and stderr
with out_file.open("w") as out_fd:
os.dup2(out_fd.fileno(), sys.stdout.fileno())
os.dup2(out_fd.fileno(), sys.stderr.fileno())
# Print some information
pid = os.getpid()
gpid = os.getpgid(pid=pid)
# Set the process name
_set_proc_name(proc_name)
# Close stdin
sys.stdin.close()
linebreak = "=" * 5
# Execute the main function
print(linebreak + f" {func.__name__}:{pid} " + linebreak, file=sys.stderr)
try:
func(**kwargs)
except Exception as ex:
traceback.print_exc()
if on_except is not None:
on_except(ex, mp.current_process())
# Kill the new process and all its children by sending a SIGTERM signal to the process group
pid = os.getpid()
gpid = os.getpgid(pid=pid)
print(f"Killing process group pid={pid} gpid={gpid}", file=sys.stderr)
os.killpg(gpid, signal.SIGTERM)
sys.exit(1)
# Don't use a finally block here, because we want the exitcode to be set to
# 0 if the function returns normally
def spawn(
*,
out_file: Path,
on_except: Callable[[Exception, mp.process.BaseProcess], None] | None,
func: Callable,
**kwargs: Any,
) -> MPProcess:
# Decouple the process from the parent
if mp.get_start_method(allow_none=True) is None:
mp.set_start_method(method="forkserver")
# Set names
proc_name = f"MPExec:{func.__name__}"
# Start the process
proc = mp.Process(
target=_init_proc,
args=(func, out_file, proc_name, on_except),
name=proc_name,
kwargs=kwargs,
)
proc.start()
# Return the process
mp_proc = MPProcess(name=proc_name, proc=proc, out_file=out_file)
return mp_proc

11
pkgs/clan-app/clan_app/components/interfaces.py
View File

@@ -1,11 +0,0 @@
from dataclasses import dataclass
import gi
gi.require_version("Gtk", "4.0")
@dataclass
class ClanConfig:
initial_view: str
content_uri: str

25
pkgs/clan-app/clan_app/deps/webview/_webview_ffi.py
View File

@@ -10,12 +10,12 @@ def _encode_c_string(s: str) -> bytes:
return s.encode("utf-8")
def _get_webview_version():
def _get_webview_version() -> str:
"""Get webview version from environment variable or use default"""
return os.getenv("WEBVIEW_VERSION", "0.8.1")
def _get_lib_names():
def _get_lib_names() -> list[str]:
"""Get platform-specific library names."""
system = platform.system().lower()
machine = platform.machine().lower()
@@ -25,8 +25,9 @@ def _get_lib_names():
return ["webview.dll", "WebView2Loader.dll"]
if machine == "arm64":
msg = "arm64 is not supported on Windows"
raise Exception(msg)
return None
raise RuntimeError(msg)
msg = f"Unsupported architecture: {machine}"
raise RuntimeError(msg)
if system == "darwin":
if machine == "arm64":
return ["libwebview.aarch64.dylib"]
@@ -35,16 +36,16 @@ def _get_lib_names():
return ["libwebview.so"]
def _be_sure_libraries():
def _be_sure_libraries() -> list[Path] | None:
"""Ensure libraries exist and return paths."""
lib_dir = os.environ.get("WEBVIEW_LIB_DIR")
if not lib_dir:
msg = "WEBVIEW_LIB_DIR environment variable is not set"
raise RuntimeError(msg)
lib_dir = Path(lib_dir)
lib_dir_p = Path(lib_dir)
lib_names = _get_lib_names()
lib_paths = [lib_dir / lib_name for lib_name in lib_names]
lib_paths = [lib_dir_p / lib_name for lib_name in lib_names]
# Check if any library is missing
missing_libs = [path for path in lib_paths if not path.exists()]
@@ -56,10 +57,14 @@ def _be_sure_libraries():
class _WebviewLibrary:
def __init__(self) -> None:
lib_names = _get_lib_names()
library_path = ctypes.util.find_library(lib_names[0])
if not library_path:
library_paths = _be_sure_libraries()
if not library_paths:
msg = f"Failed to find required library: {lib_names}"
raise RuntimeError(msg)
try:
library_path = ctypes.util.find_library(lib_names[0])
if not library_path:
library_paths = _be_sure_libraries()
self.lib = ctypes.cdll.LoadLibrary(str(library_paths[0]))
except Exception as e:
print(f"Failed to load webview library: {e}")

73
pkgs/clan-app/clan_app/deps/webview/webview.py
View File

@@ -1,11 +1,17 @@
import ctypes
import json
import logging
import threading
from collections.abc import Callable
from enum import IntEnum
from typing import Any
from clan_cli.api import MethodRegistry, dataclass_to_dict, from_dict
from ._webview_ffi import _encode_c_string, _webview_lib
log = logging.getLogger(__name__)
class SizeHint(IntEnum):
NONE = 0
@@ -26,7 +32,7 @@ class Webview:
self, debug: bool = False, size: Size | None = None, window: int | None = None
) -> None:
self._handle = _webview_lib.webview_create(int(debug), window)
self._callbacks = {}
self._callbacks: dict[str, Callable[..., Any]] = {}
if size:
self.size = size
@@ -65,6 +71,71 @@ class Webview:
_webview_lib.webview_run(self._handle)
self.destroy()
def bind_jsonschema_api(self, api: MethodRegistry) -> None:
for name, method in api.functions.items():
def wrapper(
seq: bytes,
req: bytes,
arg: int,
wrap_method: Callable[..., Any] = method,
method_name: str = name,
) -> None:
def thread_task() -> None:
args = json.loads(req.decode())
try:
log.debug(f"Calling {method_name}({args[0]})")
# Initialize dataclasses from the payload
reconciled_arguments = {}
for k, v in args[0].items():
# Some functions expect to be called with dataclass instances
# But the js api returns dictionaries.
# Introspect the function and create the expected dataclass from dict dynamically
# Depending on the introspected argument_type
arg_class = api.get_method_argtype(method_name, k)
# TODO: rename from_dict into something like construct_checked_value
# from_dict really takes Anything and returns an instance of the type/class
reconciled_arguments[k] = from_dict(arg_class, v)
reconciled_arguments["op_key"] = seq.decode()
# TODO: We could remove the wrapper in the MethodRegistry
# and just call the method directly
result = wrap_method(**reconciled_arguments)
success = True
except Exception as e:
log.exception(f"Error calling {method_name}")
result = str(e)
success = False
try:
serialized = json.dumps(
dataclass_to_dict(result), indent=4, ensure_ascii=False
)
except TypeError:
log.exception(f"Error serializing result for {method_name}")
raise
log.debug(f"Result for {method_name}: {serialized}")
self.return_(seq.decode(), 0 if success else 1, serialized)
thread = threading.Thread(target=thread_task)
thread.start()
c_callback = _webview_lib.CFUNCTYPE(
None, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_void_p
)(wrapper)
log.debug(f"Binding {name} to {method}")
if name in self._callbacks:
msg = f"Callback {name} already exists. Skipping binding."
raise RuntimeError(msg)
self._callbacks[name] = c_callback
_webview_lib.webview_bind(
self._handle, _encode_c_string(name), c_callback, None
)
def bind(self, name: str, callback: Callable[..., Any]) -> None:
def wrapper(seq: bytes, req: bytes, arg: int) -> None:
args = json.loads(req.decode())

0
pkgs/clan-app/clan_app/singletons/__init__.py
View File

118
pkgs/clan-app/clan_app/singletons/toast.py
View File

@@ -1,118 +0,0 @@
import logging
from collections.abc import Callable
from typing import Any
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
from gi.repository import Adw
log = logging.getLogger(__name__)
class ToastOverlay:
"""
The ToastOverlay is a class that manages the display of toasts
It should be used as a singleton in your application to prevent duplicate toasts
Usage
"""
# For some reason, the adw toast overlay cannot be subclassed
# Thats why it is added as a class property
overlay: Adw.ToastOverlay
active_toasts: set[str]
_instance: "None | ToastOverlay" = None
def __init__(self) -> None:
msg = "Call use() instead"
raise RuntimeError(msg)
@classmethod
def use(cls: Any) -> "ToastOverlay":
if cls._instance is None:
cls._instance = cls.__new__(cls)
cls.overlay = Adw.ToastOverlay()
cls.active_toasts = set()
return cls._instance
def add_toast_unique(self, toast: Adw.Toast, key: str) -> None:
if key not in self.active_toasts:
self.active_toasts.add(key)
self.overlay.add_toast(toast)
toast.connect("dismissed", lambda toast: self.active_toasts.remove(key))
class WarningToast:
toast: Adw.Toast
def __init__(self, message: str, persistent: bool = False) -> None:
super().__init__()
self.toast = Adw.Toast.new(
f"<span foreground='orange'>⚠ Warning </span> {message}"
)
self.toast.set_use_markup(True)
self.toast.set_priority(Adw.ToastPriority.NORMAL)
if persistent:
self.toast.set_timeout(0)
class InfoToast:
toast: Adw.Toast
def __init__(self, message: str, persistent: bool = False) -> None:
super().__init__()
self.toast = Adw.Toast.new(f"<span>❕</span> {message}")
self.toast.set_use_markup(True)
self.toast.set_priority(Adw.ToastPriority.NORMAL)
if persistent:
self.toast.set_timeout(0)
class SuccessToast:
toast: Adw.Toast
def __init__(self, message: str, persistent: bool = False) -> None:
super().__init__()
self.toast = Adw.Toast.new(f"<span foreground='green'>✅</span> {message}")
self.toast.set_use_markup(True)
self.toast.set_priority(Adw.ToastPriority.NORMAL)
if persistent:
self.toast.set_timeout(0)
class LogToast:
toast: Adw.Toast
def __init__(
self,
message: str,
on_button_click: Callable[[], None],
button_label: str = "More",
persistent: bool = False,
) -> None:
super().__init__()
self.toast = Adw.Toast.new(
f"""Logs are available <span weight="regular">{message}</span>"""
)
self.toast.set_use_markup(True)
self.toast.set_priority(Adw.ToastPriority.NORMAL)
if persistent:
self.toast.set_timeout(0)
self.toast.set_button_label(button_label)
self.toast.connect(
"button-clicked",
lambda _: on_button_click(),
)

37
pkgs/clan-app/clan_app/singletons/use_views.py
View File

@@ -1,37 +0,0 @@
from typing import Any
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
from gi.repository import Adw
class ViewStack:
"""
This is a singleton.
It is initialized with the first call of use()
Usage:
ViewStack.use().set_visible()
ViewStack.use() can also be called before the data is needed. e.g. to eliminate/reduce waiting time.
"""
_instance: "None | ViewStack" = None
view: Adw.ViewStack
# Make sure the VMS class is used as a singleton
def __init__(self) -> None:
msg = "Call use() instead"
raise RuntimeError(msg)
@classmethod
def use(cls: Any) -> "ViewStack":
if cls._instance is None:
cls._instance = cls.__new__(cls)
cls.view = Adw.ViewStack()
return cls._instance

0
pkgs/clan-app/clan_app/windows/__init__.py
View File

51
pkgs/clan-app/clan_app/windows/main_window.py
View File

@@ -1,51 +0,0 @@
import logging
import os
import gi
from clan_cli.api import API
from clan_app.components.interfaces import ClanConfig
from clan_app.singletons.toast import ToastOverlay
from clan_app.singletons.use_views import ViewStack
from clan_app.views.webview import WebExecutor
gi.require_version("Adw", "1")
from gi.repository import Adw, Gio
log = logging.getLogger(__name__)
class MainWindow(Adw.ApplicationWindow):
def __init__(self, config: ClanConfig) -> None:
super().__init__()
self.set_title("Clan App")
self.set_default_size(1280, 1024)
# Overlay for GTK side exclusive toasts
overlay = ToastOverlay.use().overlay
view = Adw.ToolbarView()
overlay.set_child(view)
self.set_content(overlay)
header = Adw.HeaderBar()
view.add_top_bar(header)
app = Gio.Application.get_default()
assert app is not None
stack_view = ViewStack.use().view
webexec = WebExecutor(jschema_api=API, content_uri=config.content_uri)
stack_view.add_named(webexec.get_webview(), "webview")
stack_view.set_visible_child_name(config.initial_view)
view.set_content(stack_view)
self.connect("destroy", self.on_destroy)
def on_destroy(self, source: "Adw.ApplicationWindow") -> None:
log.debug("Destroying Adw.ApplicationWindow")
os._exit(0)