Loading pyproject.toml +1 −1 Original line number Diff line number Diff line [tool.poetry] name = "nova-mvvm" version = "0.10.4" version = "0.11.0" description = "A Python Package for Model-View-ViewModel pattern" authors = ["Yakubov, Sergey <yakubovs@ornl.gov>"] readme = "README.md" Loading src/nova/mvvm/interface.py +57 −0 Original line number Diff line number Diff line Loading @@ -11,6 +11,58 @@ CallbackAfterUpdateType = Union[ ] class Worker: """Abstract worker class. Provides methods required to run tasks in backend. """ @abstractmethod def start(self) -> None: """Start running the task in a background thread.""" raise NotImplementedError("start() must be implemented in a subclass") @abstractmethod def connect_result(self, callback: Callable[[Any], None]) -> None: """ Register a callback to be called with the result when the task finishes. Args: callback (Callable[[Any], None]): Function called with the result. """ raise NotImplementedError("connect_result() must be implemented in a subclass") @abstractmethod def connect_error(self, callback: Callable[[Exception], None]) -> None: """ Register a callback to be called if the task raises an exception. Args: callback (Callable[[Exception], None]): Function called with the exception. """ raise NotImplementedError("connect_error() must be implemented in a subclass") @abstractmethod def connect_finished(self, callback: Callable[[], None]) -> None: """ Register a callback to be called when the task has finished (success or failure). Args: callback (Callable[[], None]): Function called with no arguments. """ raise NotImplementedError("connect_finished() must be implemented in a subclass") @abstractmethod def connect_progress(self, callback: Any) -> None: """ Register a callback to be called with progress updates (0 to 100). Args: callback (Callable[[float], None]): Function called with a float progress value. """ raise NotImplementedError("connect_progress() must be implemented in a subclass") class Communicator(ABC): """Abstract communicator class. Loading Loading @@ -97,3 +149,8 @@ class BindingInterface(ABC): ViewModel/Model variable and the GUI framework element. """ raise Exception("Please implement in a concrete class") @abstractmethod def new_worker(self, task: Callable[..., Any], *args: Any, **kwargs: Any) -> Worker: """Creates an instance of a Worker class to be used to run tasks in background.""" raise Exception("Please implement in a concrete class") src/nova/mvvm/pyqt5_binding/binding.py +19 −3 Original line number Diff line number Diff line """Binding module for PyQt5 framework.""" import inspect from typing import Any from typing import Any, Callable from PyQt5.QtCore import QObject, pyqtSignal from PyQt5.QtCore import QObject, QThreadPool, pyqtSignal from typing_extensions import override from .._internal.pyqt_communicator import PyQtCommunicator from ..interface import BindingInterface from ..interface import BindingInterface, Worker from .pyqt5_worker import PyQt5Worker def is_callable(var: Any) -> bool: Loading @@ -19,9 +21,19 @@ class PyQtObject(QObject): signal = pyqtSignal(object) class ThreadPool(QThreadPool): """ThreadPool class.""" def __init__(self) -> None: super().__init__() class PyQt5Binding(BindingInterface): """Binding Interface implementation for PyQt.""" def __init__(self) -> None: self.thread_pool = ThreadPool() def new_bind( self, linked_object: Any = None, linked_object_arguments: Any = None, callback_after_update: Any = None ) -> Any: Loading @@ -31,3 +43,7 @@ class PyQt5Binding(BindingInterface): I update and linked_object to trigger ViewModel/Model update """ return PyQtCommunicator(PyQtObject, linked_object, linked_object_arguments, callback_after_update) @override def new_worker(self, task: Callable[..., Any], *args: Any, **kwargs: Any) -> Worker: return PyQt5Worker(self.thread_pool, task, *args, **kwargs) src/nova/mvvm/pyqt5_binding/pyqt5_worker.py 0 → 100644 +69 −0 Original line number Diff line number Diff line """Worker module for PyQt5 framework.""" import sys import traceback from typing import Any, Callable from PyQt5.QtCore import QObject, QRunnable, QThreadPool, pyqtSignal, pyqtSlot from typing_extensions import override from nova.mvvm.interface import Worker class WorkerSignals(QObject): """Defines the signals available from a running worker thread.""" finished = pyqtSignal() error = pyqtSignal(tuple) progress = pyqtSignal(str, int) result = pyqtSignal(object) class PyQt5Worker(QRunnable, Worker): """Worker class that executes a function with provided arguments in a separate thread.""" def __init__(self, thread_pool: QThreadPool, task: Callable[..., Any], *args: Any, **kwargs: Any) -> None: super().__init__() self.thread_pool = thread_pool self.signals = WorkerSignals() self.task = task self.args = args self.kwargs = kwargs self.kwargs["progress"] = self._emit_progress @pyqtSlot() def run(self) -> None: try: result = self.task(*self.args, **self.kwargs) except Exception: traceback.print_exc() exctype, value = sys.exc_info()[:2] self.signals.error.emit((exctype, value, traceback.format_exc())) else: self.signals.result.emit(result) finally: self.signals.finished.emit() def _emit_progress(self, message: str, progress: int) -> None: self.signals.progress.emit(message, progress) @override def connect_error(self, callback: Callable[[Any], None]) -> None: self.signals.error.connect(callback) @override def connect_result(self, callback: Callable[[Any], None]) -> None: self.signals.result.connect(callback) @override def connect_finished(self, callback: Callable[[], None]) -> None: self.signals.finished.connect(callback) @override def connect_progress(self, callback: Callable[[str, int], None]) -> None: self.signals.progress.connect(callback) @override def start(self) -> None: self.thread_pool.start(self) src/nova/mvvm/pyqt6_binding/binding.py +19 −3 Original line number Diff line number Diff line """Binding module for PyQt framework.""" from typing import Any from typing import Any, Callable from PyQt6.QtCore import QObject, pyqtSignal # type: ignore from PyQt6.QtCore import QObject, QThreadPool, pyqtSignal # type: ignore from typing_extensions import override from .._internal.pyqt_communicator import PyQtCommunicator from ..interface import BindingInterface from ..interface import BindingInterface, Worker from .pyqt6_worker import PyQt6Worker class PyQtObject(QObject): Loading @@ -14,9 +16,19 @@ class PyQtObject(QObject): signal = pyqtSignal(object) class ThreadPool(QThreadPool): """ThreadPool class.""" def __init__(self) -> None: super().__init__() class PyQt6Binding(BindingInterface): """Binding Interface implementation for PyQt.""" def __init__(self) -> None: self.thread_pool = ThreadPool() def new_bind( self, linked_object: Any = None, linked_object_arguments: Any = None, callback_after_update: Any = None ) -> Any: Loading @@ -26,3 +38,7 @@ class PyQt6Binding(BindingInterface): I update and linked_object to trigger ViewModel/Model update """ return PyQtCommunicator(PyQtObject, linked_object, linked_object_arguments, callback_after_update) @override def new_worker(self, task: Callable[..., Any], *args: Any, **kwargs: Any) -> Worker: return PyQt6Worker(self.thread_pool, task, *args, **kwargs) Loading
pyproject.toml +1 −1 Original line number Diff line number Diff line [tool.poetry] name = "nova-mvvm" version = "0.10.4" version = "0.11.0" description = "A Python Package for Model-View-ViewModel pattern" authors = ["Yakubov, Sergey <yakubovs@ornl.gov>"] readme = "README.md" Loading
src/nova/mvvm/interface.py +57 −0 Original line number Diff line number Diff line Loading @@ -11,6 +11,58 @@ CallbackAfterUpdateType = Union[ ] class Worker: """Abstract worker class. Provides methods required to run tasks in backend. """ @abstractmethod def start(self) -> None: """Start running the task in a background thread.""" raise NotImplementedError("start() must be implemented in a subclass") @abstractmethod def connect_result(self, callback: Callable[[Any], None]) -> None: """ Register a callback to be called with the result when the task finishes. Args: callback (Callable[[Any], None]): Function called with the result. """ raise NotImplementedError("connect_result() must be implemented in a subclass") @abstractmethod def connect_error(self, callback: Callable[[Exception], None]) -> None: """ Register a callback to be called if the task raises an exception. Args: callback (Callable[[Exception], None]): Function called with the exception. """ raise NotImplementedError("connect_error() must be implemented in a subclass") @abstractmethod def connect_finished(self, callback: Callable[[], None]) -> None: """ Register a callback to be called when the task has finished (success or failure). Args: callback (Callable[[], None]): Function called with no arguments. """ raise NotImplementedError("connect_finished() must be implemented in a subclass") @abstractmethod def connect_progress(self, callback: Any) -> None: """ Register a callback to be called with progress updates (0 to 100). Args: callback (Callable[[float], None]): Function called with a float progress value. """ raise NotImplementedError("connect_progress() must be implemented in a subclass") class Communicator(ABC): """Abstract communicator class. Loading Loading @@ -97,3 +149,8 @@ class BindingInterface(ABC): ViewModel/Model variable and the GUI framework element. """ raise Exception("Please implement in a concrete class") @abstractmethod def new_worker(self, task: Callable[..., Any], *args: Any, **kwargs: Any) -> Worker: """Creates an instance of a Worker class to be used to run tasks in background.""" raise Exception("Please implement in a concrete class")
src/nova/mvvm/pyqt5_binding/binding.py +19 −3 Original line number Diff line number Diff line """Binding module for PyQt5 framework.""" import inspect from typing import Any from typing import Any, Callable from PyQt5.QtCore import QObject, pyqtSignal from PyQt5.QtCore import QObject, QThreadPool, pyqtSignal from typing_extensions import override from .._internal.pyqt_communicator import PyQtCommunicator from ..interface import BindingInterface from ..interface import BindingInterface, Worker from .pyqt5_worker import PyQt5Worker def is_callable(var: Any) -> bool: Loading @@ -19,9 +21,19 @@ class PyQtObject(QObject): signal = pyqtSignal(object) class ThreadPool(QThreadPool): """ThreadPool class.""" def __init__(self) -> None: super().__init__() class PyQt5Binding(BindingInterface): """Binding Interface implementation for PyQt.""" def __init__(self) -> None: self.thread_pool = ThreadPool() def new_bind( self, linked_object: Any = None, linked_object_arguments: Any = None, callback_after_update: Any = None ) -> Any: Loading @@ -31,3 +43,7 @@ class PyQt5Binding(BindingInterface): I update and linked_object to trigger ViewModel/Model update """ return PyQtCommunicator(PyQtObject, linked_object, linked_object_arguments, callback_after_update) @override def new_worker(self, task: Callable[..., Any], *args: Any, **kwargs: Any) -> Worker: return PyQt5Worker(self.thread_pool, task, *args, **kwargs)
src/nova/mvvm/pyqt5_binding/pyqt5_worker.py 0 → 100644 +69 −0 Original line number Diff line number Diff line """Worker module for PyQt5 framework.""" import sys import traceback from typing import Any, Callable from PyQt5.QtCore import QObject, QRunnable, QThreadPool, pyqtSignal, pyqtSlot from typing_extensions import override from nova.mvvm.interface import Worker class WorkerSignals(QObject): """Defines the signals available from a running worker thread.""" finished = pyqtSignal() error = pyqtSignal(tuple) progress = pyqtSignal(str, int) result = pyqtSignal(object) class PyQt5Worker(QRunnable, Worker): """Worker class that executes a function with provided arguments in a separate thread.""" def __init__(self, thread_pool: QThreadPool, task: Callable[..., Any], *args: Any, **kwargs: Any) -> None: super().__init__() self.thread_pool = thread_pool self.signals = WorkerSignals() self.task = task self.args = args self.kwargs = kwargs self.kwargs["progress"] = self._emit_progress @pyqtSlot() def run(self) -> None: try: result = self.task(*self.args, **self.kwargs) except Exception: traceback.print_exc() exctype, value = sys.exc_info()[:2] self.signals.error.emit((exctype, value, traceback.format_exc())) else: self.signals.result.emit(result) finally: self.signals.finished.emit() def _emit_progress(self, message: str, progress: int) -> None: self.signals.progress.emit(message, progress) @override def connect_error(self, callback: Callable[[Any], None]) -> None: self.signals.error.connect(callback) @override def connect_result(self, callback: Callable[[Any], None]) -> None: self.signals.result.connect(callback) @override def connect_finished(self, callback: Callable[[], None]) -> None: self.signals.finished.connect(callback) @override def connect_progress(self, callback: Callable[[str, int], None]) -> None: self.signals.progress.connect(callback) @override def start(self) -> None: self.thread_pool.start(self)
src/nova/mvvm/pyqt6_binding/binding.py +19 −3 Original line number Diff line number Diff line """Binding module for PyQt framework.""" from typing import Any from typing import Any, Callable from PyQt6.QtCore import QObject, pyqtSignal # type: ignore from PyQt6.QtCore import QObject, QThreadPool, pyqtSignal # type: ignore from typing_extensions import override from .._internal.pyqt_communicator import PyQtCommunicator from ..interface import BindingInterface from ..interface import BindingInterface, Worker from .pyqt6_worker import PyQt6Worker class PyQtObject(QObject): Loading @@ -14,9 +16,19 @@ class PyQtObject(QObject): signal = pyqtSignal(object) class ThreadPool(QThreadPool): """ThreadPool class.""" def __init__(self) -> None: super().__init__() class PyQt6Binding(BindingInterface): """Binding Interface implementation for PyQt.""" def __init__(self) -> None: self.thread_pool = ThreadPool() def new_bind( self, linked_object: Any = None, linked_object_arguments: Any = None, callback_after_update: Any = None ) -> Any: Loading @@ -26,3 +38,7 @@ class PyQt6Binding(BindingInterface): I update and linked_object to trigger ViewModel/Model update """ return PyQtCommunicator(PyQtObject, linked_object, linked_object_arguments, callback_after_update) @override def new_worker(self, task: Callable[..., Any], *args: Any, **kwargs: Any) -> Worker: return PyQt6Worker(self.thread_pool, task, *args, **kwargs)
