PySide Signal cannot overload Python classes

Question

PySide claims that Signals can be defined using the QtCore.Signal() class. Python types and C types can be passed as parameters to it. If you need to overload it just pass the types as tuples or lists. From PySide Doc Signals and Slots in PySide, they showed a way to create multiple signals at once. That line is:

# create two new signals on the fly: one will handle
# int type, the other will handle strings
speak = QtCore.Signal((int,), (str,))

I went a bit far and created something like this (for learning purposes):

speak = QtCore.Signal((int,), (str,), (tuple,), (list,), (A,), (B,), (Exception,), (unicode,), (float,))

A and B are two different dummy new style classes I created. Then I investigated how many signals were actually created by printing them out:

print someone.speak
print someone.speak[int]
print someone.speak[str]
print someone.speak[tuple]
print someone.speak[list]
print someone.speak[A]
print someone.speak[B]
print someone.speak[Exception]
print someone.speak[unicode]
print someone.speak[float]

And I got these:

<PySide.QtCore.SignalInstance object at 0x02179BA0>
<PySide.QtCore.SignalInstance object at 0x02179BA0>
<PySide.QtCore.SignalInstance object at 0x02179BB0>
<PySide.QtCore.SignalInstance object at 0x02179BC0>
<PySide.QtCore.SignalInstance object at 0x02179BC0>
<PySide.QtCore.SignalInstance object at 0x02179BC0>
<PySide.QtCore.SignalInstance object at 0x02179BC0>
<PySide.QtCore.SignalInstance object at 0x02179BC0>
<PySide.QtCore.SignalInstance object at 0x02179BB0>
<PySide.QtCore.SignalInstance object at 0x02179C00>

Obersevation: tuple, list, Exception and my custom A and B got the same signal. I know that tuple and list are a bit general and used by PySide to overload stuff. But why my A and B also get the same signal? It looks like only those really primitive types like int, float and str in Python get new signals.

Can anybody explain this weird behavior?

Big thanks in advance!

[UPDATE]

I did above exploration when I encountered a problem in a real PySide project. I simplified the real version into this:

#!/usr/bin/env python
import sys
from PySide import QtCore

class A(object):
    def __init__(self, msg):
        self.msg = msg

class B(object):
    def __init__(self, msg):
        self.msg = msg

@QtCore.Slot(A)
def sayA(a):
    print 'From sayA:', a.msg

@QtCore.Slot(B)
def sayB(b):
    print 'From sayB:', b.msg

class Communicate(QtCore.QObject):
    speak = QtCore.Signal((A,), (B,))

someone = Communicate()

someone.speak[A].connect(sayA)
someone.speak[B].connect(sayB)

someone.speak[A].emit(A('A is good'))
someone.speak[B].emit(B('B is bad'))

It will print out:

From sayA: A is good
From sayB: A is good
From sayA: B is bad
From sayB: B is bad

I expected A is good to be printed from sayA only.

Answer 1

See this answer to a question related to your question:

Yes is possible to define signal with python objects, you could use:

signal_node_selected = QtCore.Signal(object)

I assume that passing any custom Python class is considered as passing object. There is no way to make signals distinguish them.

Answer 2

Thanks for your research! I had the same issue and worked around it by chaining signals and using alternate names to carry the different data:

class King(QtGui.QWidget):
    summon = QtCore.Signal(str)
    summon_knights = QtCore.Signal(object)

    def __init__(self, parent=None):
        super(King, self).__init__(parent=parent)

        summon.connect(lambda x: self.summon_knights.emit(self.get_available_knights()))

    def get_available_knights(self):
        return ["Lancelot", "Galahad"]

The lambda was in order to handle or ignore the str carried by the summon signal.

This was my less elegant scratch work to prove to myself that it wasn't behaving like I wanted. The sleeps are there to force the results into a more chronological order since the asynchronous signal/slot behavior causes it to get pretty jumbled otherwise (which is good for speed, but difficult to interpret).

from PySide import QtCore
from PySide import QtGui
import sys
from collections import OrderedDict
from time import sleep

class Analyst(QtCore.QObject):

    finished = QtCore.Signal((object,), (list,))

    def __init__(self, parent=None):
        super(Analyst, self).__init__(parent=parent)

        self.number = 10000.0

    def analyze(self):
        print "Analyst working!"
        result = OrderedDict()
        result["a"] = self.number / 100.0
        result["b"] = self.number / 200.0

        sleep(1)

        report = ['Analysis Report',
                  ' a: {0}'.format(result["a"]),
                  ' b: {0}'.format(result["b"])
                  ]

        print "Analyst done!"

        self.finished[object].emit(result)
        sleep(1)
        self.finished[list].emit(report)

class Manager(QtCore.QObject):

    announceStartWork = QtCore.Signal()
    allDone = QtCore.Signal()

    def __init__(self, parent=None):
        super(Manager, self).__init__(parent=parent)

        self.analyst = Analyst(self)
        self.analyst.finished[object].connect(self.post_process)
        self.analyst.finished[list].connect(self.report_result)
        self.announceStartWork.connect(self.analyst.analyze)
        self.reportsDone = False
        self.resultsDone = False
        self.allDone.connect(self.exitWhenReady)

    def directAnalyst(self):
        print "Telling analyst to start"
        self.announceStartWork.emit()

    def post_process(self, results):

        print "Results type (expecting OrderedDict): {0}".format(type(results))

        if issubclass(type(results), dict):
            summation = 0
            for value in results.values():
                summation += value

            print "Sum of Results: {0}".format(summation)
            self.resultsDone = True
            self.allDone.emit()
        else:
            print "*** WRONG TYPE! DICT slot got the LIST!"


    def report_result(self, report):

        print "Report type (expecting list): {0}".format(type(report))

        if issubclass(type(report), list):
            report_str = '\n'.join(report)
            print "Report of original result: \n{0}".format(report_str)

            self.reportsDone = True
            self.allDone.emit()
        else:
            print "*** WRONG TYPE! LIST slot got the DICT!"


    def exitWhenReady(self):
        tasksCompleted = [self.reportsDone, self.resultsDone]
        if all(tasksCompleted):
            print "All tasks completed"
            app.exit()
        else:
            print "Not all tasks completed yet"


if __name__ == "__main__":

    app = QtCore.QCoreApplication([])
    manager = Manager()
    manager.directAnalyst()
    sys.exit(app.exec_())

Here's the output:

Telling analyst to start
Analyst working!
Analyst done!
Results type (expecting OrderedDict): <class 'collections.OrderedDict'>
Sum of Results: 150.0
Not all tasks completed yet
Report type (expecting list): <class 'collections.OrderedDict'>
*** WRONG TYPE! LIST slot got the DICT!
Results type (expecting OrderedDict): <type 'list'>
*** WRONG TYPE! DICT slot got the LIST!
Report type (expecting list): <type 'list'>
Report of original result: 
Analysis Report
 a: 100.0
 b: 50.0
All tasks completed

Okay, so I implemented my own "fix" on the Manager, Analyst example above and wound up with this:

from PySide import QtCore
from PySide import QtGui
import sys
from collections import OrderedDict
from time import sleep

class Analyst(QtCore.QObject):

    finished_result = QtCore.Signal(object)
    finished_report = QtCore.Signal(object)
    _finished = QtCore.Signal(object)

    def __init__(self, parent=None):
        super(Analyst, self).__init__(parent=parent)

        self.number = 10000.0
        self._finished.connect(self._make_signal)

    def analyze(self):
        print "Analyst working!"
        result = OrderedDict()
        result["a"] = self.number / 100.0
        result["b"] = self.number / 200.0

        sleep(1)

        report = ['Analysis Report',
                  ' a: {0}'.format(result["a"]),
                  ' b: {0}'.format(result["b"])
                  ]

        print "Analyst done!"

        self._finished.emit(("result", result))
        sleep(1)
        self._finished.emit(("report", report))

    def _make_signal(self, data):

        if data[0] == "result":
            self.finished_result.emit(data[1])
        elif data[0] == "report":
            self.finished_report.emit(data[1])


class Manager(QtCore.QObject):

    announceStartWork = QtCore.Signal()
    allDone = QtCore.Signal()

    def __init__(self, parent=None):
        super(Manager, self).__init__(parent=parent)

        self.analyst = Analyst(self)
        self.analyst.finished_result.connect(self.post_process)
        self.analyst.finished_report.connect(self.report_result)
        self.announceStartWork.connect(self.analyst.analyze)
        self.reportsDone = False
        self.resultsDone = False
        self.allDone.connect(self.exitWhenReady)

    def directAnalyst(self):
        print "Telling analyst to start"
        self.announceStartWork.emit()

    def post_process(self, results):

        print "Results type (expecting OrderedDict): {0}".format(type(results))

        if issubclass(type(results), dict):
            summation = 0
            for value in results.values():
                summation += value

            print "Sum of Results: {0}".format(summation)
            self.resultsDone = True
            self.allDone.emit()
        else:
            print "*** WRONG TYPE! DICT slot got the LIST!"

    def report_result(self, report):

        print "Report type (expecting list): {0}".format(type(report))

        if issubclass(type(report), list):
            report_str = '\n'.join(report)
            print "Report of original result: \n{0}".format(report_str)

            self.reportsDone = True
            self.allDone.emit()
        else:
            print "*** WRONG TYPE! LIST slot got the DICT!"


    def exitWhenReady(self):
        tasksCompleted = [self.reportsDone, self.resultsDone]
        if all(tasksCompleted):
            print "All tasks completed"
            app.exit()
        else:
            print "Not all tasks completed yet"


if __name__ == "__main__":

    app = QtCore.QCoreApplication([])
    manager = Manager()
    manager.directAnalyst()
    sys.exit(app.exec_())

Which has the following output:

Telling analyst to start
Analyst working!
Analyst done!
Results type (expecting OrderedDict): <class 'collections.OrderedDict'>
Sum of Results: 150.0
Not all tasks completed yet
Report type (expecting list): <type 'list'>
Report of original result: 
Analysis Report
 a: 100.0
 b: 50.0
All tasks completed