Implementing signals and slots

[nikhil-sethi]

I'm trying to implement a custom signal and slot but there seem to be a few errors. I can't inherit from QObject(which seems to be the recommended way) as it doesn't exist. This example throws: AttributeError: module 'PythonQt.QtCore' has no attribute 'QObject' The discussion here doesn't seem to be resolved either.

If I do it without classes like so:

from PythonQt.QtCore import Signal
def slot():
    print("something")
sig = Signal("QVariant") # also tried without QVariant
sig.connect(slot)
sig.emit()

It results in a crash due to segmentation fault. I'm new to Qt itself so would love any suggestions/right way of doing things. Thank You.

[Adi1413]

In short:

Use signals from real Qt objects → works fine.

Making your own signals in Python → not supported yet, causes crash.

Contributing → help add this feature or update documentation.

example:- from PythonQt.QtCore import QTimer

def my_slot(): print("Timer triggered!")

make a timer

timer = QTimer()

connect its signal to our function

timer.timeout.connect(my_slot)

start the timer every 1000 ms (1 second)

timer.start(1000)

[usiems]

Here is example code that we use in MeVisLab to test signals/slots defined in Python, and communication between threads through signal/slot connections:

# classes for implementing custom signal/slot communication
class MySender(QtCore.QObject):

    emitProgress = QtCore.Signal(float)  # this is actually a double argument in C++


class MyReceiver(QtCore.QObject):

    @QtCore.Slot(float)
    def progress(self, value):
        # update some progress field
        ctx.field("progress").value = value  # this is MeVisLab specific


# do some work in another thread
class MyWorkerThread(QtCore.QThread):

    def __init__(self):
        super().__init__()
        self.totalSum = 0
        self.max = 1000000
        # set up communication
        self.messenger = MySender()
        self.receiver = MyReceiver()
        self.messenger.moveToThread(self)
        self.messenger.connect("emitProgress(double)", self.receiver, "progress(double)")

    def run(self):
        # this is executed in another thread
        _sum = 0
        for i in range(self.max):
            _sum += i
            if (i + 1) % 1000 == 0:
                # Update progress from time to time
                self.messenger.emitProgress((i + 1) / self.max)
                # Give up the GIL (global interpreter lock),
                # so that Python code can run in the main thread:
                time.sleep(0.001)
        self.totalSum = _sum


myThread = None


def workFinished():
    # This is called when the thread finishes
    global myThread
    print("Thread done")
    print(f"Sum: {myThread.totalSum}")
    # Delete the QThread object
    myThread = None


def testQThread():
    global myThread
    myThread = MyWorkerThread()
    myThread.finished.connect(workFinished)
    myThread.start()