[PyQt] Any way to run code before each QThread?

David Boddie david at boddie.org.uk
Mon Jul 16 14:33:06 BST 2018


Just in case anyone finds it useful or interesting, this is more or less what
I tried to do by applying an event filter to the QApplication object:

class ThreadFilter(QObject):

    report = pyqtSignal(object)
    
    def __init__(self, parent = None):
        QObject.__init__(self, parent)
        self.report.connect(self.collect, Qt.QueuedConnection)
    
    def eventFilter(self, obj, event):
    
        if event.type() == QEvent.ThreadChange:
        
            if not isinstance(obj, QThread):
                print("This thread:", QThread.currentThread())
                print("Object", obj, "in thread:", obj.thread())
                self.report.emit(obj)
        
        return False
    
    def collect(self, obj):
    
        thread = obj.thread()
        print(obj, thread)
        thread.started.connect(self.settrace_start)
        thread.finished.connect(self.settrace_finish)
    
    def settrace_start(self):

        print("Starting thread:", self.sender(), self.sender().isRunning())
    
    def settrace_finish(self):

        print("Finishing thread:", self.sender(), self.sender().isRunning())

The idea is that the filter intercepts the ThreadChange event, which tells us
that an object is being moved between threads, but only tells us about the
old thread. This is handled in the eventFilter method.

However, we have previouly (in the __init__ method) connected a custom signal
(report) to a slot using a QueuedConnection, which means that the signal will
be sent through via an event queue (as I understand it) and not through a
direct call to the slot.

This allows the report signal to be emitted in the eventFilter method but
received after the object has been moved to a new thread (in theory), so that
we can obtain the new thread and connect its signals to other slots to track
its running time.

The event filter can be applied to an QApplication instance, so it can be
applied to QApplication.instance() although this has to occur before the
application creates and runs any new threads, obviously.

    thread_filter = ThreadFilter()
    QApplication.instance().installEventFilter(thread_filter)

Note that this trick only seems to work with recent versions of Qt that
define QEvent::Event::ThreadChange (in C++, QEvent.ThreadChange in PyQt)
and only for applications that use QObject.moveToThread. It doesn't seem to
work for those that subclass QThread and reimplement its run() method.

Hope this inspires someone to build on or improve the concept.

David


More information about the PyQt mailing list