I have a threaded application where I do have a network thread. The UI-part passes a callback
to this thread. The thread is a normal python thread - it's NO QThread
.
Is it possible to emit PyQT Slot within this thread?
I have a threaded application where I do have a network thread. The UI-part passes a callback
to this thread. The thread is a normal python thread - it's NO QThread
.
Is it possible to emit PyQT Slot within this thread?
No, it is not possible to emit a PyQt signal from a python thread like this.
However, a possible solution is to use an additional object shared by both threads, making the necessary operations to finally emit a thread-safe PyQt signal.
Here is an implementation of a "SafeConnector" class, making use of a pair of connected sockets and a Queue to exchange data between the two threads, and using a QSocketNotifier to get back in Qt's loop. A QObject is used to make it possible to emit a proper Qt signal:
from PyQt4 import Qt, QtCore, QtGui
import threading
import socket
import Queue
import time# Object of this class has to be shared between
# the two threads (Python and Qt one).
# Qt thread calls 'connect',
# Python thread calls 'emit'.
# The slot corresponding to the emitted signal
# will be called in Qt's thread.
class SafeConnector:def __init__(self):self._rsock, self._wsock = socket.socketpair()self._queue = Queue.Queue()self._qt_object = QtCore.QObject()self._notifier = QtCore.QSocketNotifier(self._rsock.fileno(),QtCore.QSocketNotifier.Read)self._notifier.activated.connect(self._recv)def connect(self, signal, receiver):QtCore.QObject.connect(self._qt_object, signal, receiver)# should be called by Python threaddef emit(self, signal, args):self._queue.put((signal, args))self._wsock.send('!')# happens in Qt's main threaddef _recv(self):self._rsock.recv(1)signal, args = self._queue.get()self._qt_object.emit(signal, args)class PythonThread(threading.Thread):def __init__(self, connector, *args, **kwargs):threading.Thread.__init__(self, *args, **kwargs)self.connector = connectorself.daemon = Truedef emit_signal(self):self.connector.emit(QtCore.SIGNAL("test"), str(time.time()))def run(self):while True:time.sleep(1)self.emit_signal()if __name__ == '__main__':app = QtGui.QApplication([])mainwin = QtGui.QMainWindow()label = QtGui.QLabel(mainwin)mainwin.setCentralWidget(label)connector = SafeConnector()python_thread = PythonThread(connector)connector.connect(QtCore.SIGNAL("test"), label.setText)python_thread.start()mainwin.show()app.exec_()