mirror of
https://github.com/aclist/dztui.git
synced 2026-08-28 02:37:20 +02:00
98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
import inspect
|
|
import logging
|
|
import threading
|
|
from typing import Any
|
|
|
|
from functools import wraps
|
|
from typing import Callable
|
|
from dzgui.views.dialogs.generic import WaitDialog
|
|
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "3.0")
|
|
from gi.repository import Gtk, GLib # noqa E402
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def call_on_thread(dialog_str: str) -> Callable:
|
|
def decorator(func: Callable) -> Callable:
|
|
@wraps(func)
|
|
def wrapper(*args: Any, **kwargs: Any) -> Callable:
|
|
self = args[0]
|
|
stored = StoredFunc(func, *args, **kwargs)
|
|
if not hasattr(self, "thread_man"):
|
|
raise AttributeError
|
|
if type(self.thread_man) is not ThreadingManager:
|
|
raise TypeError(
|
|
"Attribute 'thread_man' must be of type 'ThreadingManager'"
|
|
)
|
|
self.thread_man.call_on_thread(dialog_str, stored)
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
class StoredFunc:
|
|
def __init__(self, func: Callable, *args, **kwargs) -> None:
|
|
sig = inspect.signature(func)
|
|
self.func = func
|
|
self.bindings = sig.bind(*args, **kwargs)
|
|
|
|
def call(self) -> None:
|
|
self.func(*self.bindings.args, *self.bindings.kwargs)
|
|
|
|
|
|
class ThreadingManager:
|
|
def __init__(self, controller: "Controller") -> None:
|
|
self.controller = controller
|
|
self.jobs = 1
|
|
self.cleanup_func = None
|
|
self.destroy_first = False
|
|
|
|
def call_on_thread(self, dialog_str: str, func: StoredFunc) -> None:
|
|
def callback() -> None:
|
|
func.call()
|
|
GLib.idle_add(self._destroy_on_idle)
|
|
|
|
self.wait_dialog = WaitDialog(self.controller, dialog_str, jobs=self.jobs)
|
|
self.wait_dialog.show_all()
|
|
thread = threading.Thread(target=callback)
|
|
thread.start()
|
|
|
|
def set_job_count(self, jobs: int) -> None:
|
|
self.jobs = jobs
|
|
|
|
def increment_dialog(self) -> None:
|
|
GLib.idle_add(self.wait_dialog.increment)
|
|
|
|
def increment_dialog_with_str(self, text: str) -> None:
|
|
GLib.idle_add(lambda: self.wait_dialog.increment(text))
|
|
|
|
def set_cleanup_func(self, func: StoredFunc, destroy_first: bool = False) -> None:
|
|
if type(func) not in (StoredFunc, type(None)):
|
|
msg = f"Callback function '{func}' is not of type StoredFunc or None"
|
|
logger.critical(msg)
|
|
raise TypeError(msg)
|
|
self.destroy_first = destroy_first
|
|
self.cleanup_func = func
|
|
|
|
def get_cleanup_func(self) -> StoredFunc:
|
|
return self.cleanup_func
|
|
|
|
def _destroy_on_idle(self) -> GLib.SOURCE_REMOVE:
|
|
if self.destroy_first:
|
|
self.wait_dialog.destroy()
|
|
|
|
func = self.get_cleanup_func()
|
|
if func is not None:
|
|
func.call()
|
|
self.set_cleanup_func(None)
|
|
if not self.destroy_first:
|
|
self.wait_dialog.destroy()
|
|
|
|
return GLib.SOURCE_REMOVE
|
|
|
|
def get_wait_dialog(self) -> WaitDialog:
|
|
return self.wait_dialog
|