如何从非PyQt类中发出信号？

Question

我正在使用Twisted和PyQt编写Python应用程序。我遇到的问题是，在执行我的Twisted代码中的函数时，我必须在GUI中打印一行。我试图通过发射一个信号（非PyQt类）来实现这一点。但是这似乎不起作用，我怀疑Twisted事件循环对PyQt造成了影响。因为closeEvent信号没有被程序捕获。

以下是代码片段：

from PyQt4 import QtGui, QtCore
import sys
from twisted.internet.protocol import Factory, Protocol
from twisted.protocols import amp
import qt4reactor

class register_procedure(amp.Command):
    arguments = [('MAC',amp.String()),
                        ('IP',amp.String()),
                        ('Computer_Name',amp.String()),
                        ('OS',amp.String())
                        ]
    response = [('req_status', amp.String()),
         ('ALIGN_FUNCTION', amp.String()),
                         ('ALIGN_Confirmation', amp.Integer()),
                         ('Callback_offset',amp.Integer())
                        ]

class Ui_MainWindow(QtGui.QMainWindow):

    def __init__(self,reactor, parent=None):
        super(Ui_MainWindow,self).__init__(parent)
        self.reactor=reactor
        self.pf = Factory()
        self.pf.protocol = Protocol
        self.reactor.listenTCP(3610, self.pf) # listen on port 1234

        def setupUi(self,MainWindow):
        MainWindow.setObjectName(_fromUtf8("MainWindow"))
        MainWindow.resize(903, 677)
        self.centralwidget = QtGui.QWidget(MainWindow)
        sizePolicy = QtGui.QSizePolicy(QtGui.QSizePolicy.Expanding, QtGui.QSizePolicy.Expanding)
        sizePolicy.setHorizontalStretch(0)
        sizePolicy.setVerticalStretch(0)
        sizePolicy.setHeightForWidth(self.centralwidget.sizePolicy().hasHeightForWidth())
        self.centralwidget.setSizePolicy(sizePolicy)

        self.create_item()


        self.retranslateUi(MainWindow)
        self.connect(self, QtCore.SIGNAL('triggered()'), self.closeEvent)
        QtCore.QObject.connect(self,QtCore.SIGNAL('registered()'),self.registered)
        QtCore.QMetaObject.connectSlotsByName(MainWindow)

    def retranslateUi(self, MainWindow):
        MainWindow.setWindowTitle(_translate("MainWindow", "MainWindow", None))
        self.pushButton_4.setText(_translate("MainWindow", "Delete System ", None))
        self.pushButton.setText(_translate("MainWindow", "Add System", None))
        self.label_2.setText(_translate("MainWindow", "SYSTEM STATUS", None))
        self.label.setText(_translate("MainWindow", "Monitoring Output", None))


    def registered(self):# this function is not being triggered
        print "check" 
        self.textbrowser.append()

    def closeEvent(self, event):#neither is this being triggered
        print "asdf"
        self.rector.stop()
        MainWindow.close()
        event.accept()


class Protocol(amp.AMP):
    @register_procedure.responder
    def register_procedure(self,MAC,IP,Computer_Name,OS):
        self.bridge_conn=bridge()
        cursor_device.execute("""select * FROM devices where MAC = ?;""",[(MAC)])
        exists_val=cursor_device.fetchone()
        cursor_device.fetchone()
        print "register"
        if not exists_val== "":
            cursor_device.execute("""update devices set IP= ? , Computer_name= ? , OS = ?  where MAC= ?;""",[IP,Computer_Name,OS,MAC])
            QtCore.QObject.emit( QtCore.SIGNAL('registered')) # <--emits signal
            return {'req_status': "done" ,'ALIGN_FUNCTION':'none','ALIGN_Confirmation':0,'Callback_offset':call_offset(1)}
        else:
            cursor_device.execute("""INSERT INTO devices(Mac,Ip,Computer_name,Os) values (?,?,?,?);""",[MAC,IP,Computer_Name,OS])
            QtCore.QObject.emit( QtCore.SIGNAL('registered'))#<--emits signal
            return {'req_status': "done" ,'ALIGN_FUNCTION':'main_loop()','ALIGN_Confirmation':0,'Callback_offset':0}



if  __name__ == "__main__":
    app = QtGui.QApplication(sys.argv)

    try:
        import qt4reactor
    except ImportError:
        from twisted.internet import qt4reactor
    qt4reactor.install()

    from twisted.internet import reactor
    MainWindow = QtGui.QMainWindow() # <-- Instantiate QMainWindow object.
    ui = Ui_MainWindow(reactor)
    ui.setupUi(MainWindow)
    MainWindow.show()
    reactor.run()

- thecreator232

Answer 1

这是我在一个代码中用来发送QGraphicsItems信号的方法（因为它们不是从QObject派生的，不能默认地发送/接收信号）。它基本上是Radio-'s答案的简化版。

from PyQt4 import QtGui as QG
from PyQt4 import QtCore as QC

class SenderObject(QC.QObject):
    something_happened = QC.pyqtSignal()

SenderObject是一个继承自QObject的小类，您可以在其中放置需要发出的所有信号。在这种情况下，仅定义了一个信号。

class SnapROIItem(QG.QGraphicsRectItem):
    def __init__(self, parent = None):
        super(SnapROIItem, self).__init__(parent)
        self.sender = SenderObject()
    def do_something_and_emit(self):
        ...
        self.sender.something_happened.emit()

在非 QObject 类中，您可以将一个名为 SenderObject 的对象添加为 sender 变量。在任何使用非 QObject 类的地方，您都可以将来自 sender 的信号连接到您所需的任何内容。

class ROIManager(QC.QObject):
    def add_snaproi(self, snaproi):
        snaproi.sender.something_happened.connect(...)

更新

完整的代码如下，应该会打印出 "Something happened..."：

from PyQt4 import QtGui as QG
from PyQt4 import QtCore as QC

class SenderObject(QC.QObject):
    something_happened = QC.pyqtSignal()

class SnapROIItem(QG.QGraphicsItem):
    def __init__(self, parent = None):
        super(SnapROIItem, self).__init__(parent)
        self.sender = SenderObject()
    def do_something_and_emit(self):
        self.sender.something_happened.emit()

class ROIManager(QC.QObject):
    def __init__(self, parent=None):
        super(ROIManager,self).__init__(parent)
    def add_snaproi(self, snaproi):
        snaproi.sender.something_happened.connect(self.new_roi)
    def new_roi(self):
        print 'Something happened in ROI!'

if __name__=="__main__":)
    roimanager = ROIManager()
    snaproi = SnapROIItem()
    roimanager.add_snaproi(snaproi)
    snaproi.do_something_and_emit()

更新2

不是

QtCore.QObject.connect(self,QtCore.SIGNAL('registered()'),self.registered) 

您需要：

protocol.sender.registered.connect(self.registered)

这意味着您还需要获取self.pf中的protocol实例（顺便问一下，您是导入Protocol然后再自己定义吗？）。

在协议类中，而不是...

QtCore.QObject.emit( QtCore.SIGNAL('registered')

首先，您需要在协议中实例化一个 SenderObject。

class Protocol(amp.AMP):
    def __init__( self, *args, **kw ):
       super(Protocol, self).__init__(*args, **kw)
       self.sender = SenderObject()

然后，在register_procedure中通过sender发出信号：

为了使所有这些工作正常，您必须将SenderObject定义为：

class SenderObject(QC.QObject):
    registered = QC.pyqtSignal()

- numentar

---

@adroi：我尝试了这种方法，但它给了我一个错误提示“QtCore.pyqtSignal: 没有 connect() 属性”。 - thecreator232

请尝试使用完整的代码，看看是否出现相同的错误。 - numentar

如果您看我的代码，会发现有两个函数，Ui_MainWindow是PyQt类，而Protocol不是。我正在尝试让Protocol类与Ui_mainWindow通信。 - thecreator232

Answer 2

这是一篇旧文章，但对我很有帮助。以下是我的版本。其中一个项目不是QObject信号，而是通知另外两个非QObject运行它们的方法。

from PyQt4 import QtGui, QtCore

class Signal(object):
    class Emitter(QtCore.QObject):
        registered = QtCore.pyqtSignal()
        def __init__(self):
            super(Signal.Emitter, self).__init__()

    def __init__(self):
        self.emitter = Signal.Emitter()

    def register(self):
        self.emitter.registered.emit()

    def connect(self, signal, slot):
        signal.emitter.registered.connect(slot)

class item(object):
    def __init__(self, name):
        self.name = name
        self.signal = Signal()

    def something(self):
        print self.name, ' says something'

>>> itemA = item('a')
>>> itemB = item('b')
>>> itemC = item('c')
>>> itemA.signal.connect(itemA.signal, itemB.something)
>>> itemA.signal.connect(itemA.signal, itemC.something)
>>> itemA.signal.register()

b  says something
c  says something

- Ryan Trowbridge

Answer 3

我对 @Ryan Trowbridge 的解决方案进行了一些改进，以处理通用类型。我希望使用 Generic[T]，但结果太复杂了。

class GenSignal:
    def __init__(self,typ):
        Emitter = type('Emitter', (QtCore.QObject,), {'signal': Signal(typ)})
        self.emitter = Emitter()

    def emit(self,*args,**kw):
        self.emitter.signal.emit(*args,**kw)

    def connect(self,  slot):
        self.emitter.signal.connect(slot)

使用时，x=GenSignal(int)，然后像往常一样继续操作。

- eyal0931

---

https://dev59.com/Jqr2oIgBc1ULPQZFIQ5a?noredirect=1#comment123016708_69594116 有相关讨论。 - eyal0931

Answer 4

两个基本问题是：

1）某些东西必须知道信号的发送者和接收者。

考虑在Qt中更常见的情况，您可能有多个按钮，每个按钮都有一个“clicked”信号。插槽需要知道哪个按钮被点击，因此捕获通用信号没有太多意义。

2）信号必须来自QObject。

---

话虽如此，我不确定标准实现是什么。这里有一种方法可以实现它，使用您早期帖子中提到的桥接（Bridge）思想，以及协议内部的特殊发射器（Emitter）类。运行此代码将在调用protocol.register()时简单地打印出“Working it”。

from PyQt4 import QtGui, QtCore
import sys

class Ui_MainWindow(QtGui.QMainWindow):
    def __init__(self):
        super(Ui_MainWindow, self).__init__()

    def work(self):
        print "Working it"

class Protocol(object):
    class Emitter(QtCore.QObject):
        registered = QtCore.pyqtSignal()
        def __init__(self):
            super(Protocol.Emitter, self).__init__()

    def __init__(self):
        self.emitter = Protocol.Emitter()

    def register(self):
        self.emitter.registered.emit()

class Bridge(QtCore.QObject):
    def __init__(self, gui, protocol):
        super(Bridge, self).__init__()
        self.gui = gui
        self.protocol = protocol

    def bridge(self):
        self.protocol.emitter.registered.connect(self.gui.work)

app = QtGui.QApplication(sys.argv)
gui = Ui_MainWindow()
protocol = Protocol()
bridge = Bridge(gui, protocol)
bridge.bridge()
#protocol.register() #uncomment to see 'Working it' printed to the console

- Radio-

---

请确认以上给出的代码中，Protocol类是一个非PyQt类？是吗？ - thecreator232

是的，这只是一个通用的Python 'object'。 - Radio-