rename ui to clan-app and move clan-app one layer up

pkgs/clan-app/clan_app/__init__.py

@@ -0,0 +1,28 @@
+import argparse
+import logging
+import sys
+
+from clan_cli.profiler import profile
+
+log = logging.getLogger(__name__)
+
+from clan_app.app import ClanAppOptions, app_run
+
+
+@profile
+def main(argv: list[str] = sys.argv) -> int:
+    parser = argparse.ArgumentParser(description="Clan App")
+    parser.add_argument(
+        "--content-uri", type=str, help="The URI of the content to display"
+    )
+    parser.add_argument("--debug", action="store_true", help="Enable debug mode")
+    args = parser.parse_args(argv[1:])
+
+    app_opts = ClanAppOptions(content_uri=args.content_uri, debug=args.debug)
+    try:
+        app_run(app_opts)
+    except KeyboardInterrupt:
+        log.info("Keyboard interrupt received, exiting...")
+        return 0
+
+    return 0

pkgs/clan-app/clan_app/__main__.py

@@ -0,0 +1,6 @@
+import sys
+
+from . import main
+
+if __name__ == "__main__":
+    sys.exit(main())

pkgs/clan-app/clan_app/api/file_gtk.py

@@ -0,0 +1,200 @@
+# ruff: noqa: N801
+import gi
+
+gi.require_version("Gtk", "4.0")
+
+import logging
+import time
+from pathlib import Path
+from typing import Any
+
+from clan_lib.api import ApiError, ErrorDataClass, SuccessDataClass
+from clan_lib.api.directory import FileRequest
+from gi.repository import Gio, GLib, Gtk
+
+log = logging.getLogger(__name__)
+
+
+def remove_none(_list: list) -> list:
+    return [i for i in _list if i is not None]
+
+
+RESULT: dict[str, SuccessDataClass[list[str] | None] | ErrorDataClass] = {}
+
+
+def open_file(
+    file_request: FileRequest, *, op_key: str
+) -> SuccessDataClass[list[str] | None] | ErrorDataClass:
+    GLib.idle_add(gtk_open_file, file_request, op_key)
+
+    while RESULT.get(op_key) is None:
+        time.sleep(0.2)
+    response = RESULT[op_key]
+    del RESULT[op_key]
+    return response
+
+
+def gtk_open_file(file_request: FileRequest, op_key: str) -> bool:
+    def returns(data: SuccessDataClass | ErrorDataClass) -> None:
+        global RESULT
+        RESULT[op_key] = data
+
+    def on_file_select(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
+        try:
+            gfile = file_dialog.open_finish(task)
+            if gfile:
+                selected_path = remove_none([gfile.get_path()])
+                returns(
+                    SuccessDataClass(
+                        op_key=op_key, data=selected_path, status="success"
+                    )
+                )
+        except Exception as e:
+            log.exception("Error opening file")
+            returns(
+                ErrorDataClass(
+                    op_key=op_key,
+                    status="error",
+                    errors=[
+                        ApiError(
+                            message=e.__class__.__name__,
+                            description=str(e),
+                            location=["open_file"],
+                        )
+                    ],
+                )
+            )
+
+    def on_file_select_multiple(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
+        try:
+            gfiles: Any = file_dialog.open_multiple_finish(task)
+            if gfiles:
+                selected_paths = remove_none([gfile.get_path() for gfile in gfiles])
+                returns(
+                    SuccessDataClass(
+                        op_key=op_key, data=selected_paths, status="success"
+                    )
+                )
+            else:
+                returns(SuccessDataClass(op_key=op_key, data=None, status="success"))
+        except Exception as e:
+            log.exception("Error opening file")
+            returns(
+                ErrorDataClass(
+                    op_key=op_key,
+                    status="error",
+                    errors=[
+                        ApiError(
+                            message=e.__class__.__name__,
+                            description=str(e),
+                            location=["open_file"],
+                        )
+                    ],
+                )
+            )
+
+    def on_folder_select(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
+        try:
+            gfile = file_dialog.select_folder_finish(task)
+            if gfile:
+                selected_path = remove_none([gfile.get_path()])
+                returns(
+                    SuccessDataClass(
+                        op_key=op_key, data=selected_path, status="success"
+                    )
+                )
+            else:
+                returns(SuccessDataClass(op_key=op_key, data=None, status="success"))
+        except Exception as e:
+            log.exception("Error opening file")
+            returns(
+                ErrorDataClass(
+                    op_key=op_key,
+                    status="error",
+                    errors=[
+                        ApiError(
+                            message=e.__class__.__name__,
+                            description=str(e),
+                            location=["open_file"],
+                        )
+                    ],
+                )
+            )
+
+    def on_save_finish(file_dialog: Gtk.FileDialog, task: Gio.Task) -> None:
+        try:
+            gfile = file_dialog.save_finish(task)
+            if gfile:
+                selected_path = remove_none([gfile.get_path()])
+                returns(
+                    SuccessDataClass(
+                        op_key=op_key, data=selected_path, status="success"
+                    )
+                )
+            else:
+                returns(SuccessDataClass(op_key=op_key, data=None, status="success"))
+        except Exception as e:
+            log.exception("Error opening file")
+            returns(
+                ErrorDataClass(
+                    op_key=op_key,
+                    status="error",
+                    errors=[
+                        ApiError(
+                            message=e.__class__.__name__,
+                            description=str(e),
+                            location=["open_file"],
+                        )
+                    ],
+                )
+            )
+
+    dialog = Gtk.FileDialog()
+
+    if file_request.title:
+        dialog.set_title(file_request.title)
+
+    if file_request.filters:
+        filters = Gio.ListStore.new(Gtk.FileFilter)
+        file_filters = Gtk.FileFilter()
+
+        if file_request.filters.title:
+            file_filters.set_name(file_request.filters.title)
+
+        if file_request.filters.mime_types:
+            for mime in file_request.filters.mime_types:
+                file_filters.add_mime_type(mime)
+                filters.append(file_filters)
+
+        if file_request.filters.patterns:
+            for pattern in file_request.filters.patterns:
+                file_filters.add_pattern(pattern)
+
+        if file_request.filters.suffixes:
+            for suffix in file_request.filters.suffixes:
+                file_filters.add_suffix(suffix)
+
+        filters.append(file_filters)
+        dialog.set_filters(filters)
+
+    if file_request.initial_file:
+        p = Path(file_request.initial_file).expanduser()
+        f = Gio.File.new_for_path(str(p))
+        dialog.set_initial_file(f)
+
+    if file_request.initial_folder:
+        p = Path(file_request.initial_folder).expanduser()
+        f = Gio.File.new_for_path(str(p))
+        dialog.set_initial_folder(f)
+
+    # if select_folder
+    if file_request.mode == "select_folder":
+        dialog.select_folder(callback=on_folder_select)
+    if file_request.mode == "open_multiple_files":
+        dialog.open_multiple(callback=on_file_select_multiple)
+    elif file_request.mode == "open_file":
+        dialog.open(callback=on_file_select)
+    elif file_request.mode == "save":
+        dialog.save(callback=on_save_finish)
+
+    return GLib.SOURCE_REMOVE

pkgs/clan-app/clan_app/app.py

@@ -0,0 +1,82 @@
+import logging
+
+from clan_cli.profiler import profile
+
+log = logging.getLogger(__name__)
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+from clan_cli.custom_logger import setup_logging
+from clan_lib.api import API, ErrorDataClass, SuccessDataClass
+
+from clan_app.api.file_gtk import open_file
+from clan_app.deps.webview.webview import Size, SizeHint, Webview
+
+
+@dataclass
+class ClanAppOptions:
+    content_uri: str
+    debug: bool
+
+
+@profile
+def app_run(app_opts: ClanAppOptions) -> int:
+    if app_opts.debug:
+        setup_logging(logging.DEBUG, root_log_name=__name__.split(".")[0])
+        setup_logging(logging.DEBUG, root_log_name="clan_cli")
+    else:
+        setup_logging(logging.INFO, root_log_name=__name__.split(".")[0])
+        setup_logging(logging.INFO, root_log_name="clan_cli")
+
+    log.debug("Debug mode enabled")
+
+    if app_opts.content_uri:
+        content_uri = app_opts.content_uri
+    else:
+        site_index: Path = Path(os.getenv("WEBUI_PATH", ".")).resolve() / "index.html"
+        content_uri = f"file://{site_index}"
+
+    webview = Webview(debug=app_opts.debug)
+    webview.title = "Clan App"
+    # This seems to call the gtk api correctly but and gtk also seems to our icon, but somehow the icon is not loaded.
+    webview.icon = "clan-white"
+
+    def cancel_task(
+        task_id: str, *, op_key: str
+    ) -> SuccessDataClass[None] | ErrorDataClass:
+        """Cancel a task by its op_key."""
+        log.debug(f"Cancelling task with op_key: {task_id}")
+        future = webview.threads.get(task_id)
+        if future:
+            future.stop_event.set()
+            log.debug(f"Task {task_id} cancelled.")
+        else:
+            log.warning(f"Task {task_id} not found.")
+        return SuccessDataClass(
+            op_key=op_key,
+            data=None,
+            status="success",
+        )
+
+    def list_tasks(
+        *,
+        op_key: str,
+    ) -> SuccessDataClass[list[str]] | ErrorDataClass:
+        """List all tasks."""
+        log.debug("Listing all tasks.")
+        tasks = list(webview.threads.keys())
+        return SuccessDataClass(
+            op_key=op_key,
+            data=tasks,
+            status="success",
+        )
+
+    API.overwrite_fn(list_tasks)
+    API.overwrite_fn(open_file)
+    API.overwrite_fn(cancel_task)
+    webview.bind_jsonschema_api(API)
+    webview.size = Size(1280, 1024, SizeHint.NONE)
+    webview.navigate(content_uri)
+    webview.run()
+    return 0

pkgs/clan-app/clan_app/assets/__init__.py

@@ -0,0 +1,7 @@
+from pathlib import Path
+
+loc: Path = Path(__file__).parent
+
+
+def get_asset(name: str | Path) -> Path:
+    return loc / name

pkgs/clan-app/clan_app/deps/webview/_webview_ffi.py

@@ -0,0 +1,118 @@
+import ctypes
+import ctypes.util
+import os
+import platform
+from ctypes import CFUNCTYPE, c_char_p, c_int, c_void_p
+from pathlib import Path
+
+
+def _encode_c_string(s: str) -> bytes:
+    return s.encode("utf-8")
+
+
+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() -> list[str]:
+    """Get platform-specific library names."""
+    system = platform.system().lower()
+    machine = platform.machine().lower()
+
+    if system == "windows":
+        if machine == "amd64" or machine == "x86_64":
+            return ["webview.dll", "WebView2Loader.dll"]
+        if machine == "arm64":
+            msg = "arm64 is not supported on Windows"
+            raise RuntimeError(msg)
+        msg = f"Unsupported architecture: {machine}"
+        raise RuntimeError(msg)
+    if system == "darwin":
+        if machine == "arm64":
+            return ["libwebview.dylib"]
+        msg = "Not supported"
+        raise RuntimeError(msg)
+    # linux
+    return ["libwebview.so"]
+
+
+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_p = Path(lib_dir)
+    lib_names = _get_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()]
+    if not missing_libs:
+        return lib_paths
+    return None
+
+
+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:
+            self.lib = ctypes.cdll.LoadLibrary(str(library_paths[0]))
+        except Exception as e:
+            print(f"Failed to load webview library: {e}")
+            raise
+
+        # Define FFI functions
+        self.webview_create = self.lib.webview_create
+        self.webview_create.argtypes = [c_int, c_void_p]
+        self.webview_create.restype = c_void_p
+
+        self.webview_destroy = self.lib.webview_destroy
+        self.webview_destroy.argtypes = [c_void_p]
+
+        self.webview_run = self.lib.webview_run
+        self.webview_run.argtypes = [c_void_p]
+
+        self.webview_terminate = self.lib.webview_terminate
+        self.webview_terminate.argtypes = [c_void_p]
+
+        self.webview_set_title = self.lib.webview_set_title
+        self.webview_set_title.argtypes = [c_void_p, c_char_p]
+
+        self.webview_set_icon = self.lib.webview_set_icon
+        self.webview_set_icon.argtypes = [c_void_p, c_char_p]
+
+        self.webview_set_size = self.lib.webview_set_size
+        self.webview_set_size.argtypes = [c_void_p, c_int, c_int, c_int]
+
+        self.webview_navigate = self.lib.webview_navigate
+        self.webview_navigate.argtypes = [c_void_p, c_char_p]
+
+        self.webview_init = self.lib.webview_init
+        self.webview_init.argtypes = [c_void_p, c_char_p]
+
+        self.webview_eval = self.lib.webview_eval
+        self.webview_eval.argtypes = [c_void_p, c_char_p]
+
+        self.webview_bind = self.lib.webview_bind
+        self.webview_bind.argtypes = [c_void_p, c_char_p, c_void_p, c_void_p]
+
+        self.webview_unbind = self.lib.webview_unbind
+        self.webview_unbind.argtypes = [c_void_p, c_char_p]
+
+        self.webview_return = self.lib.webview_return
+        self.webview_return.argtypes = [c_void_p, c_char_p, c_int, c_char_p]
+
+        self.CFUNCTYPE = CFUNCTYPE
+
+
+_webview_lib = _WebviewLibrary()

pkgs/clan-app/clan_app/deps/webview/webview.py

@@ -0,0 +1,237 @@
+import ctypes
+import functools
+import json
+import logging
+import threading
+from collections.abc import Callable
+from dataclasses import dataclass
+from enum import IntEnum
+from typing import Any
+
+from clan_cli.async_run import set_should_cancel
+from clan_lib.api import (
+    ApiError,
+    ErrorDataClass,
+    MethodRegistry,
+    dataclass_to_dict,
+    from_dict,
+)
+
+from ._webview_ffi import _encode_c_string, _webview_lib
+
+log = logging.getLogger(__name__)
+
+
+class SizeHint(IntEnum):
+    NONE = 0
+    MIN = 1
+    MAX = 2
+    FIXED = 3
+
+
+class FuncStatus(IntEnum):
+    SUCCESS = 0
+    FAILURE = 1
+
+
+class Size:
+    def __init__(self, width: int, height: int, hint: SizeHint) -> None:
+        self.width = width
+        self.height = height
+        self.hint = hint
+
+
+@dataclass
+class WebThread:
+    thread: threading.Thread
+    stop_event: threading.Event
+
+
+class Webview:
+    def __init__(
+        self, debug: bool = False, size: Size | None = None, window: int | None = None
+    ) -> None:
+        self._handle = _webview_lib.webview_create(int(debug), window)
+        self._callbacks: dict[str, Callable[..., Any]] = {}
+        self.threads: dict[str, WebThread] = {}
+
+        if size:
+            self.size = size
+
+    def api_wrapper(
+        self,
+        api: MethodRegistry,
+        method_name: str,
+        wrap_method: Callable[..., Any],
+        op_key_bytes: bytes,
+        request_data: bytes,
+        arg: int,
+    ) -> None:
+        op_key = op_key_bytes.decode()
+        args = json.loads(request_data.decode())
+        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"] = op_key
+        # TODO: We could remove the wrapper in the MethodRegistry
+        # and just call the method directly
+
+        def thread_task(stop_event: threading.Event) -> None:
+            try:
+                set_should_cancel(lambda: stop_event.is_set())
+                result = wrap_method(**reconciled_arguments)
+
+                serialized = json.dumps(
+                    dataclass_to_dict(result), indent=4, ensure_ascii=False
+                )
+
+                log.debug(f"Result for {method_name}: {serialized}")
+                self.return_(op_key, FuncStatus.SUCCESS, serialized)
+            except Exception as e:
+                log.exception(f"Error while handling result of {method_name}")
+                result = ErrorDataClass(
+                    op_key=op_key,
+                    status="error",
+                    errors=[
+                        ApiError(
+                            message="An internal error occured",
+                            description=str(e),
+                            location=["bind_jsonschema_api", method_name],
+                        )
+                    ],
+                )
+                serialized = json.dumps(
+                    dataclass_to_dict(result), indent=4, ensure_ascii=False
+                )
+                self.return_(op_key, FuncStatus.FAILURE, serialized)
+            finally:
+                del self.threads[op_key]
+
+        stop_event = threading.Event()
+        thread = threading.Thread(
+            target=thread_task, args=(stop_event,), name="WebviewThread"
+        )
+        thread.start()
+        self.threads[op_key] = WebThread(thread=thread, stop_event=stop_event)
+
+    def __enter__(self) -> "Webview":
+        return self
+
+    @property
+    def size(self) -> Size:
+        return self._size
+
+    @size.setter
+    def size(self, value: Size) -> None:
+        _webview_lib.webview_set_size(
+            self._handle, value.width, value.height, value.hint
+        )
+        self._size = value
+
+    @property
+    def title(self) -> str:
+        return self._title
+
+    @title.setter
+    def title(self, value: str) -> None:
+        _webview_lib.webview_set_title(self._handle, _encode_c_string(value))
+        self._title = value
+
+    @property
+    def icon(self) -> str:
+        return self._icon
+
+    @icon.setter
+    def icon(self, value: str) -> None:
+        _webview_lib.webview_set_icon(self._handle, _encode_c_string(value))
+        self._icon = value
+
+    def destroy(self) -> None:
+        for name in list(self._callbacks.keys()):
+            self.unbind(name)
+        _webview_lib.webview_terminate(self._handle)
+        _webview_lib.webview_destroy(self._handle)
+        self._handle = None
+
+    def navigate(self, url: str) -> None:
+        _webview_lib.webview_navigate(self._handle, _encode_c_string(url))
+
+    def run(self) -> None:
+        _webview_lib.webview_run(self._handle)
+        log.info("Shutting down webview...")
+        self.destroy()
+
+    def bind_jsonschema_api(self, api: MethodRegistry) -> None:
+        for name, method in api.functions.items():
+            wrapper = functools.partial(
+                self.api_wrapper,
+                api,
+                name,
+                method,
+            )
+            c_callback = _webview_lib.CFUNCTYPE(
+                None, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_void_p
+            )(wrapper)
+
+            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())
+            try:
+                result = callback(*args)
+                success = True
+            except Exception as e:
+                result = str(e)
+                success = False
+            self.return_(seq.decode(), 0 if success else 1, json.dumps(result))
+
+        c_callback = _webview_lib.CFUNCTYPE(
+            None, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_void_p
+        )(wrapper)
+        self._callbacks[name] = c_callback
+        _webview_lib.webview_bind(
+            self._handle, _encode_c_string(name), c_callback, None
+        )
+
+    def unbind(self, name: str) -> None:
+        if name in self._callbacks:
+            _webview_lib.webview_unbind(self._handle, _encode_c_string(name))
+            del self._callbacks[name]
+
+    def return_(self, seq: str, status: int, result: str) -> None:
+        _webview_lib.webview_return(
+            self._handle, _encode_c_string(seq), status, _encode_c_string(result)
+        )
+
+    def eval(self, source: str) -> None:
+        _webview_lib.webview_eval(self._handle, _encode_c_string(source))
+
+    def init(self, source: str) -> None:
+        _webview_lib.webview_init(self._handle, _encode_c_string(source))
+
+
+if __name__ == "__main__":
+    wv = Webview()
+    wv.title = "Hello, World!"
+    wv.navigate("https://www.google.com")
+    wv.run()