util.py 2.95 KB
Newer Older
1

Dan Pascu's avatar
Dan Pascu committed
2 3
from PyQt5.QtCore import QObject, QThread, QTimer
from PyQt5.QtWidgets import QApplication
4
from application.python.decorator import decorator, preserve_signature
5
from application.python.descriptor import classproperty
6
from application.python.types import Singleton
7 8 9
from functools import partial
from threading import Event
from sys import exc_info
10

11 12
from blink.event import CallFunctionEvent

13

14 15 16
__all__ = ['QSingleton', 'call_in_gui_thread', 'call_later', 'run_in_gui_thread']


17 18 19 20
class QSingleton(Singleton, type(QObject)):
    """A metaclass for making Qt objects singletons"""


21 22 23 24
def call_later(interval, function, *args, **kw):
    QTimer.singleShot(int(interval*1000), lambda: function(*args, **kw))


25
def call_in_gui_thread(function, *args, **kw):
26 27
    application = Application.instance
    if QThread.currentThread() is Application.gui_thread:
28
        return function(*args, **kw)
29 30
    else:
        application.postEvent(application, CallFunctionEvent(function, args, kw))
31 32


33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68
@decorator
def run_in_gui_thread(function=None, wait=False):
    if function is not None:
        @preserve_signature(function)
        def function_wrapper(*args, **kw):
            application = Application.instance
            if QThread.currentThread() is Application.gui_thread:
                return function(*args, **kw)
            else:
                if wait:
                    executor = FunctionExecutor(function)
                    application.postEvent(application, CallFunctionEvent(executor, args, kw))
                    return executor.wait()
                else:
                    application.postEvent(application, CallFunctionEvent(function, args, kw))
        return function_wrapper
    else:
        return partial(run_in_gui_thread, wait=wait)


class Application(object):
    __attributes__ = {}

    @classproperty
    def instance(cls):
        try:
            return cls.__attributes__['instance']
        except KeyError:
            return cls.__attributes__.setdefault('instance', QApplication.instance())

    @classproperty
    def gui_thread(cls):
        try:
            return cls.__attributes__['gui_thread']
        except KeyError:
            return cls.__attributes__.setdefault('gui_thread', cls.instance.thread())
69 70


71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93
class FunctionExecutor(object):
    __slots__ = 'function', 'event', 'result', 'exception', 'traceback'

    def __init__(self, function):
        self.function = function
        self.event = Event()
        self.result = None
        self.exception = None
        self.traceback = None

    def __call__(self, *args, **kw):
        try:
            self.result = self.function(*args, **kw)
        except BaseException as exception:
            self.exception = exception
            self.traceback = exc_info()[2]
        finally:
            self.event.set()

    def wait(self):
        self.event.wait()
        if self.exception is not None:
            raise type(self.exception), self.exception, self.traceback
94
        else:
95
            return self.result
96 97