pyqtsignal - Emitir señal en el hilo de python estándar

qt designer python

Question

Tengo una aplicación enhebrada donde tengo un hilo de red. La parte UI pasa una callback de callback a este hilo. El hilo es un hilo de python normal , no es QThread .

¿Es posible emitir PyQT Slot dentro de este hilo?

Answer 1

No, no es posible emitir una señal PyQt desde un hilo de Python como este.

Sin embargo, una posible solución es usar un objeto adicional compartido por ambos hilos, realizando las operaciones necesarias para finalmente emitir una señal PyQt segura para hilos.

Aquí hay una implementación de una clase "SafeConnector", haciendo uso de un par de sockets conectados y una cola para intercambiar datos entre los dos hilos, y usando un QSocketNotifier para volver al bucle de Qt. Un QObject se usa para hacer posible emitir una señal Qt adecuada:

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 thread
def emit(self, signal, args):
self._queue.put((signal, args))
self._wsock.send(''!'')
# happens in Qt''s main thread
def _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 = connector
self.daemon = True
def 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_()