5

i'm programming an application in python using twisted and PyQt . the problem that i'm facing is that when a function in my twisted code is executed i have to print a line in the GUI, i'm trying to achieve this by emiting a signal (Non PyQt class). This does not seem to work, i have a doubt that the twisted event loop is screwing things for PyQt. Because the closeEvent signal is not being trapped by the program.

Here's the code snippet:

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()
4

4 回答 4

4

这是我在我的一个代码中用于从 QGraphicsItems 发送信号的内容（因为它们不是从 QObject 派生的，默认情况下不能发送/接收信号）。它基本上是 Radio- 答案的简化版本。

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 类中，您添加 aSenderObject作为sender变量。在使用非 QObject 类的任何地方，您都可以将信号从 连接sender到您需要的任何东西。

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

更新

完整的代码是这样的，应该打印出“发生了什么事......”：

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)

这意味着您还需要获取protocol实例self.pf（顺便说一句，您是否也导入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： self.sender.registered.emit()

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

class SenderObject(QC.QObject):
    registered = QC.pyqtSignal()
于 2013-07-05T07:50:32.727 回答
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
于 2017-07-20T01:28:28.843 回答
0

两个基本问题是：

1）必须知道信号的发送者和接收者

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

2) 信号必须源自 QObject。

话虽如此，我不确定规范的实现是什么。这是一种方法，使用您在之前的一篇文章中提出的 Bridge 的想法，以及 Protocol 内部的特殊 Emitter 类。当调用 protocol.register() 时，运行此代码将简单地打印“正在工作”。

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
于 2013-07-04T08:06:55.757 回答
0

我稍微改进了@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)并照常继续。

于 2021-10-18T00:35:53.290 回答