This is a longer post than usual, but it’s pulling several different topics together.
While working on a personal project, I felt the need to mock up PyQt signals. If you’re not familiar with signals, they are an implementation of the observer pattern used to pass events in the Qt system. In the C++ implementation, they are very powerful and expressive, as the Qt MOC does a lot of work to make them easy to use and overcome a lot of work you’d need to do to get flexible callbacks in pure C++. In Python, they are not as cool, only because Python’s dynamic binding and duck-typing make implementing arbitrary callbacks a snap.
To use a signal in PyQt, you instantiate a QObject with a signal and connect it to a slot on another QObject. All of the Qt widgets come with signals, and often you will use them or implement your own. A sample custom signal/slot system is seen below:
from PySide import QtCore
class MySource(QtCore.QObject):
mySignal = QtCore.Signal(object, int)
def __init__(self, parent=None):
QtCore.QObject.__init__(self, parent)
class MyDest(QtCore.QObject):
def __init__(self, parent=None):
QtCore.QObject.__init__(self, parent)
def mySlot(self, obj, i):
print('Object: %s, Int: %s' % (obj.__class__.__name__, i))
if __name__ == '__main__':
a = MySource()
b = MyDest()
a.mySignal.connect(b.mySlot)
a.mySignal.emit(a, 12)
Exec’ing this file yields:
Object: MySource, Int: 12
I was working with PySide and getting some errors, so I decided I would make classes that would function like PyQt objects with signals, get the code working, then swap out the underlying objects with their Qt equivalents. I’ve always liked Qt’s signals, and ended up using the Python implementation a fair amount in my project, and thought it would be a good exercise to create them from scratch.
To emulate the calling style of a PyQt signal, the Python descriptor is a natural choice. Descriptors normally get or set a value. In this case, we want the __get__ to return an object that has a connect method that will create the signal connection. The connection is an entry in a WeakKeyDictionary. The weak key dict will cleanup the signals when an object is deleted by the Python garbage collector – this prevents signal connections from keeping an object from being deleted.
The _Sig object is responsible for mapping the class instance to the target instances. It stores the args of the signal, which would be the types that are going to be sent when the signal’s emit function is called. WeakKeyDictionary is used throughout to prevent circular references from keeping objects alive – this is strictly an observer pattern.
It would be nice to keep a reference only to the methods, but I was unsuccessful storing those in a WeakSet, so the object itself is needed. The _Sig class will only work with class instances, although it should be possible to adapt to trigger static functions.
import weakref,inspect
class _Sig(object):
"""Maps objects to create Signal connection."""
def __init__(self, pr, args):
"""
Args:
pr (object): An instance of a class.
args (list or tuple): A sequence of arg types.
Attributes:
_ob (object): Instance of the class that would be the
sender.
_args (list or tuple): The sequence of arg types to use
for the Signal.
_conn (WeakKeyDictionary): Maps the target item to the
target method.
"""
self._ob = pr
self._args = args
self._conn = weakref.WeakKeyDictionary()
def emit(self, *args):
"""Emits the signal
Args:
The values to emit - the types should match _args.
Note:
If _ob has a method called 'signalsBlocked', it will be
called. If the method returns True, the signal is not
emitted.
Raises:
TypeError: If one of the emitted args is not an instance
of _args, TypeError is raised.
"""
if hasattr(self._ob, 'signalsBlocked') and self._ob.signalsBlocked():
return
idx = 0
for ik,k in zip(self._args, args):
if not isinstance(k, ik):
msg = 'Pos %s: Expected %s, got %s' % (idx, ik, k)
raise TypeError(msg)
idx += 1
for val in self._conn.values():
for ck in val:
ck(*args)
def connect(self, method):
"""Create connection to method.
Args:
method (instanceMethod): The 'slot' method to be called
when Signal is emitted.
Raises:
TypeError: Method must be a method on a class instance.
"""
if not inspect.ismethod(method):
raise TypeError('Expected method, got %s' % method.__class__)
self._conn.setdefault(method.__self__, []).append(method)
def disconnect(self, method):
"""Disconnect a method from the signal.
Args:
method (instanceMethod): The 'slot' method to be disconnected.
"""
current = self._conn.get(method.__self__, [])
current = [x for x in current if x.im_func is not method.im_func]
assert(method not in current)
if current:
self._conn[method.__self__] = current
else:
self._conn.pop(method.__self__, None)
def getOutputs(self):
"""Get a list of output methods.
Returns:
A list of methods that will be triggered when this signal
is emitted.
"""
result = []
for val in self._conn.values():
result.extend(val)
return result
By itself, the _Sig does nothing – it needs a descriptor to access it. The Signal descriptor uses a __set__ method that raises a RuntimeError, which keeps the user from assigning a value to the attribute and thus overwriting it.
class Signal(object):
"""The actual Signal object.
This descriptor does the work of providing access to the
underlying _Sig object.
Attributes:
_map (dict): Map of Signal instances to WeakKeyDictionaries.
"""
# Map this signal instance to some data.
_map = {}
def __init__(self, *args):
"""Initialize the Signal
Args:
List of types - these types are used to match the types
emitted by the Signal.
"""
self._args = args
def __get__(self, obj, typ=None):
"""Return a _Sig object mapped to obj.
We use the signal as the key so that other signals do not
collide. Each Signal instance is unique and located in a class
definition.
A weak key dict maps the actual instance to the _Sig object.
Args:
obj (object): Instance of for the descriptor.
typ (class): The owner.
Returns:
A _Sig object initialized to obj.
"""
tmp = self._map.setdefault(self, weakref.WeakKeyDictionary())
return tmp.setdefault(obj, _Sig(obj, self._args))
def __set__(self, obj, val):
"""Raise an exception.
This prevents a user from setting an attribute on the instance
that would replace the signal.
Args:
obj (object): Instance of class.
val (object): Value to set to.
Raises:
RuntimeError: Always raise this.
"""
raise RuntimeError('Cannot assign to Signal')
To use the signal, put it in a class definition and connect it from an instance of the class. Notice that Python class instances should be new-style classes derived from object.
class Source(object):
test = Signal(int, float, object)
class Target(object):
def slot(self, i, f, o):
print('slot(%s, %s, %s)' % (i, f, o))
src = Source()
trg = Target()
src.test.connect(trg.slot)
src.test.emit(12, 33.4, src.__class__.__name__)
Executing the code above yields:
slot(12, 33.4, Source)
This implementation of the Signal checks the types of the args sent, which was it’s goal since it is emulating a Qt style signal. Different implementations could omit the type-checking, although I find it handy in larger systems. There is no concept of the sender() method – this is a method on QObjects that interfaces with the Signals. It would be possible to add the sender as an argument when the signal is emitted. For Python, one could also add optional keyword arguments. That will be left as an exercise for the reader.
There is also no checking for cycles – one could create an endless cycle of signals to slots. I may address this in a future post if it comes up.
