Source code for PartSeg.common_backend.progress_thread
import inspect
from typing import Optional, Union
from qtpy.QtCore import QThread, Signal
[docs]
class ProgressTread(QThread):
range_changed = Signal(int, int)
step_changed = Signal(int)
error_signal = Signal(Exception)
[docs]
def info_function(self, label: str, val: int):
if label == "max":
self.range_changed.emit(0, val)
elif label == "step":
self.step_changed.emit(val)
[docs]
class ExecuteFunctionThread(ProgressTread):
"""Generic Thread to execute"""
def __init__(self, fun, args: Optional[Union[list, tuple]] = None, kwargs: Optional[dict] = None):
super().__init__()
self.args = args or []
self.kwargs = kwargs or {}
self.function = fun
self.result = None
[docs]
def run(self):
try:
if "callback_function" in inspect.signature(self.function).parameters:
self.result = self.function(*self.args, **self.kwargs, callback_function=self.info_function)
elif (
"range_changed" in inspect.signature(self.function).parameters
and "step_changed" in inspect.signature(self.function).parameters
):
self.result = self.function(
*self.args,
**self.kwargs,
range_changed=self.range_changed.emit,
step_changed=self.step_changed.emit,
)
else:
self.result = self.function(*self.args, **self.kwargs)
except Exception as e: # pylint: disable=broad-except
self.error_signal.emit(e)