feat: group ui related packages under a ui directory
This commit is contained in:
Brian McGee
@@ -1,28 +0,0 @@
|
||||
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
|
||||
@@ -1,6 +0,0 @@
|
||||
import sys
|
||||
|
||||
from . import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,200 +0,0 @@
|
||||
# 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
|
||||
@@ -1,82 +0,0 @@
|
||||
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
|
||||
@@ -1,7 +0,0 @@
|
||||
from pathlib import Path
|
||||
|
||||
loc: Path = Path(__file__).parent
|
||||
|
||||
|
||||
def get_asset(name: str | Path) -> Path:
|
||||
return loc / name
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 3.1 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 375 B |
Binary file not shown.
|
Before Width: | Height: | Size: 717 B |
Binary file not shown.
|
Before Width: | Height: | Size: 717 B |
Binary file not shown.
|
Before Width: | Height: | Size: 1.5 KiB |
@@ -1,118 +0,0 @@
|
||||
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()
|
||||
@@ -1,237 +0,0 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user