QThread example using bundled signals/slots

bundles.md

Bundles

This module provides a pair of classes that can be used to automate the construction of object interfaces using qt Signal()s and Slot()s.

Sales pitch:

• Reduced Boilerplate!

  These classes lets you define a lot of signals or slots with just a couple lines of code, making it easier to rigidly define your object's external interface, and then enforce the usage of that interface

• Add Signals to things that don't inherit from QObject!

  Signals only work correctly if they're added as a class attribute to a QObject. Apparently QObject.__init__() is responsible for some part of the signal registration process. SigBundle and SlotBundle

• Improved Clarity!

  Using SigBundle and SlotBundle organizes all of your object's signals or slots into a single object so you always know where to find them. No longer will you have to go scour an object's code to figure out it's inputs and outputs. All it's outputs are in foo.signals, all its inputs are in foo.slots.

• But wait, there's more!

  Both classes have utility methods that can even further reduce the amount of extra crap you have to type.

Example

# note that this class doesn't inherit from QObject
# ordinarily, it wouldn't be able to use signals directly
class ClassThatNeedsBundles:
    def __init__(self):
        # creates 2 signals, fizz and buzz
        self.signals = SigBundle({'fizz': [], 'buzz':[str]})
        # call signals
        self.signals.fizz.emit()
        self.signals.buzz.emit('beep boop')
        
        # create 3 slots, foo, bar, and baz
        # also creates 3 signals, on_foo, on_bar, on_baz
        # each slot emits the matching signal when it's called
        slots = {
            'foo': [], 
            'bar': [str], 
            'baz': [int]
        }
        self.slots = SlotBundle(slots)

        # connect slots to methods of 'self' with matching names
        # this creates a chain of slot -> signal -> method
        # ex:   self.slots.foo() -> 
        #       self.slots._signals.on_foo.emit() ->
        #       self.on_foo()
        self.slots.link_to(self)

        # explicitly connect slots to methods
        self.slots.link('foo', self.also_foo)

        # connect a slot to multiple methods
        self.slots.link('bar', [self.also_bar, lambda s: print(f'lambda: {s}')])

    def on_foo(self):
        print('foo happened')

    def also_foo(self):
        print('foo happened, part deux')

    def on_bar(self, s):
        print(f'bar: {s}')
        
    def also_bar(self, s):
        print(f'bar: {s}, part deux')

    def on_baz(self, i):
        print(f'baz: {i}')


if __name__ == '__main__':
    # Signals and Signals requires a Q application be running
    app = QApplication(sys.argv)

    # instantiate the class
    thing = ClassThatNeedsBundles()
    # call some slots
    thing.slots.foo()
    thing.slots.bar("some text")
    thing.slots.baz(42)
    
    # have to exec, so qt event loop runs
    sys.exit(app.exec_())

Terminal output:

foo happened
foo happened, part deux
bar: some text
bar: some text, part deux
lambda: some text
baz: 42

Long-winded Socratic Exploration of the Origins of Bundling

This is what you're here for, right? Alright. Here's our contrived, overly simplistic problem scenario:

We have a variable, count, and we want to add 1 to it once per second. Let's pretend this is hard to do and just agree that it also needs to be in seperate thread.

We're already in qt land, so we need a QThread and an instance of a QObject. We need to be able to control this counter across thread boundaries, so we need SigBundles and SlotBundles. We'll need start, stop, and reset inputs on our counter, and some feedback in the form of started, stopped, and updated signals would be useful, too. Here's a quick implementation:

class Counter(QObject):
    started = Signal()
    stopped = Signal()
    updated = Signal(int)

    def __init__(self):
        super().__init__()
        self.count = 0

    @Slot()
    def start(self):
        self.timer = QTimer()
        self.timer.timeout.connect(lambda: self.update())
        self.timer.start(1000)
        self.started.emit()

    @Slot()
    def stop(self):
        self.timer.stop()
        self.stopped.emit()

    @Slot()
    def reset(self):
        self.count = 0
        self.updated.emit(self.count)

    def update(self):
        print(self.count)
        self.count += 1
        self.updated.emit(self.count)

This is totally reasonable, but our Counter is still simple. As the class grows, it can easily get difficult to keep track of which methods are which.

class BundledCounter(QObject):
    def __init__(self):
        super().__init__()
        signals = {
            'started': [], 
            'stopped': [], 
            'updated': [int]
        }
        slots = {
            'start': [], 
            'stop': [], 
            'reset': []
        }

        self.signals = SigBundle(signals)
        self.slots = SlotBundle(slots)
        self.slots.link_to(self)

    def start(self):
        <snip>
    def on_stop(self):
        <snip>
    def on_reset(self):
        <snip>
    def update(self):
        <snip>

I've omitted the implementation details for brevity, and replaced raw Signal/Slot creation with bundles. It's a little more verbose, but the extra structure will help a lot when Counter starts growing. The entire interface is defined explicitely at the top of __init__(), so you always know where to find it. It can be defined programmatically, or common interface configurations can easily be shared between different classes.

Signal Connections, or Linking For Fun and Profit

Qt signals don't do anything unless you connect them to something. In Pyside2/PyQt5, that looks something like this:

self.signal = Signal() # simple signal

@Slot()
def slot(self): # slot to call
    print("beep")

self.signal.connect(self.slot) # connect them together

Triggering the signal will call slot(), which will print 'beep'.

>>> self.signal.emit() 
beep

More complex arrangements are possible, of course:

self.signal = Signal(str) # signals can deliver arbitary args

def slot(self, string): # works without slot decorator, thankfully
    print('slot: ' + string)

# multiple connections work just fine
self.signal.connect(self.slot)
self.signal.connect(print)
self.signal.connect(lambda s: print(f'lambda: {s}'))
self.signal.connect(lambda s: self.slot(reversed(s)))
self.signal.connect(lambda _: print("what string?"))

>>> self.signal.emit("beep")
slot: beep
beep
lambda: beep
slot: peeb
what string?

"But wait, this is just a brief tutorial about qt slots and signals to make sure the reader shares a common base of understanding with you." you interject.

You caught me.

"What does this have to do with the SigBundle and SlotBundle? And what was that line self.slots.link_to(self) about?"

Yeah, we're getting the-

"Hurry up, this document is already too long!"

Alright, fine.

This time we'll talk about linking, I promise

So, we can use SigBundle and SlotBundle to make bundles of singles and slots. For signals, all this really does is collate them all under the object.signals.foo namespace. The slot side of things is a little different. Remember back to the first example, the pure qt one? Creating a new Counter and making it count is pretty easy:

counter = Counter()
counter.start()

# if you could stdout right now, you'd see how hard this thing is counting
# trust me

The counter's start method causes it to... start counting. Very straightforward. Alright, let's look at the allegedly "improved" version:

counter = BundledCounter()
counter.slots.start() # the slot has moved

# so you still can't see stdout
# just pretend to be surprised when I tell you it's not counting
# no really, it's not doing anything

That's strange. Let's refresh our memory of the BundledCounter:

class BundledCounter(QObject):
    def __init__(self):
        super().__init__()
        # this compact version does the same thing as in the first example
        self.signals = SigBundle({'started': [], 'stopped': [], 'updated': [int]})
        self.slots = SlotBundle({'start': [], 'stop': [], 'reset': []})
        self.slots.link_to(self) # <- this might be related to the problem

    def start(self):
        <snip>
    def on_stop(self):
        <snip>
    def on_reset(self):
        <snip>
    def update(self):
        <snip>

There's that self.slots.link_to(self) line you were asking about, but what does it do? To answer that, I have to let you in on a little secret: A Slots bundle doesn't only have slots in it!

In BundledCounter, counter.slots has three slots in it: start, stop, and reset. There're created automatically based on the dict you pass into the constructor. But there's a logistical problem with this approach: A slot is a function, but a function needs a body in order to do things. Slots doesn't know anything about counting, so how could the bodies of start, stop, and reset possibly do the right thing in this new counter?

link_to() holds the answer. When you ask Slots to make a bundle of slots for you, it also makes a matching set of signals, and when one of the slots gets called, it actually just turns around and activates its matching signal. Let's illustrate by writing out something that's equivalent to self.slots in BundledCounter:

class BundledCounterSlotsSignals(QObject):
    on_start = Signal()
    on_stop = Signal()
    on_reset = Signal()

class BundledCounterSlots(QObject):
    def __init__(self):
        super().__init__()
        self.signals = BundledCounterSlotsSignals()

    @Slot()
    def start(self):
        self.on_start.emit()
    @Slot()
    def stop(self):
        self.on_stop.emit()
    @Slot()
    def reset(self):
        self.on_reset.emit()

Now there's three new, extra signals that need to be connected. Doing it all by hand would look like this:

class HandWrittenBundledCounter(QObject):
    def __init__(self):
        <snip>
        self.slots = BundledCounterSlots()
        self.slots.signals.on_start.connect(self.start)
        self.slots.signals.on_stop.connect(self.stop)
        self.slots.signals.on_reset.connect(self.reset)
        <snip>
        
    def start(self):
        <snip>
    def stop(self):
        <snip>
    def reset(self):
        <snip>
    def update(self):
        <snip>

This is where slots.link_to(self) comes in. You can pass an object to slots.link_to(), and slots will search the attributes of the provided object for methods with the same name as its internal list of signals. If it find them, it will automatically call slots.signals.<signal>.connect(object.method).

BundledCounter has four methods: start, on_stop, on_reset, update. We can ignore update because it clearly doesn't match any of the slot-signals. Of the remaining three, on_stop and on_reset match signals, so they get connected properly. The problem is start, since it doesn't match the signal on_start.

That's why there's also slots.link(). link can handle a variety of input formats:

# bare functions
slots.link(function) # function
slots.link(function1, function2) # multiple functions
slots.link([function1, function2]) # list of functions
slots.link((function1, function2)) # tuple of functions

# also supports explicit mapping of 'signal_name' -> function
slots.link('sig', func) # tuple of (sig, func)
slots.link(('sig1', func1), ('sig2', func2)) # multiple tuples
slots.link([('sig1', func1), ('sig2', func2)]) # list of tuples
slots.link({'sig1': func1, 'sig2': func2}) # dict of {sig: func}

# mixed types
slots.link(func1, func2, ('sig3', func3)) 
slots.link([func1, func2], ('sig3', func3))

# lambda
slots.link('sig', lambda: print('string'))

# multiple functions mapped to same signal
slots.link(('sig', func), ('sig', lambda: print('signal happened')))
slots.link('sig', (func1, func2))
slots.link('sig', [func1, func2])
slots.link(('sig1', [func1, func2]), ('sig2', (func3, func4)))
slots.link({'sig1': [func1, func2], 'sig2': (func3, func4)})

# also works with signal as first argument directly
slots.link('sig', function)
slots.link('sig', func1, func2)
slots.link('sig', (func1, func2))
slots.link('sig', [func1, func2])
slots.link('sig', [func1, func2], func3)
slots.link('sig', [func1, func2], (func3, func4))

# this is especially useful with lambdas
slots.link('sig', func, lambda: print('sig'))
slots.link('sig', (func, lambda: log.debug('message')))

# function chaining
slots.link(func1).link(func2).link('sig3', func3)
slots.link_to(self).link_to(other_thing).link('sig', func)
slots.link_to(self, other_thing).link('sig', func)

Additional Features

The details of an interface are quite discoverable:

>>> signals = Signals({'started': [],'stopped': [],'updated': [int]})

>>> print(self.signals.signals)
{'started': [], 'stopped': [], 'updated': [<class 'int'>]}

>>> print(self.signals.names)
['started', 'stopped', 'updated']

The first bundle example was expanded for readability, but the classes are quite flexible. The following methods are all equivalent:

def big(self):
    signals = {
        'started': [], 
        'stopped': [], 
        'updated': [int]
    }
    slots = {
        'start': [], 
        'stop': [], 
        'reset': []
    }

    self.signals = SigBundle(signals)
    self.slots = SlotBundle(slots)

    self.slots.link_to(self)
    self.slots.link('start', self.start)

def medium(self)
    signals = {'started': [], 'stopped': [], 'updated': [int]}
    slots = {'start': [], 'stop': [], 'reset': []}

    self.signals = SigBundle(signals)
    self.slots = SlotBundle(slots, link_to=self, link=('start', self.start))

def small(self):
    self.signals = SigBundle({'started': [],'stopped': [],'updated': [int]})
    self.slots = SlotBundle({'start': [],'stop': [],'reset': []}, link_to=self, link=('start', self.start))

def chained(self):
    self.signals = SigBundle({'started': [],'stopped': [],'updated': [int]})
    self.slots = SlotBundle({'start': [],'stop': [],'reset': []})
    
    # some people like this... and I guess I shouldn't judge their life choices
    self.slots.link_to(self).link('start', self.start)

bundles.py

from qt import *


class SigBundle(QObject):
    def __init__(self, signals, link_to=None, link=None):
        # signals need to be class attributes BEFORE QObject.__init__()
        for name, args in signals.items():
            setattr(self.__class__, name, Signal(*args))

        super().__init__()
        self.signals = signals
        self.names = list(signals.keys())

        if link_to:
            self.link_to(link_to)

        if link:
            self.link(link)
    
    def _link(self, slot, sig=None):
        """ Connect the given slot to a matching signal, if it exists """
        signal = sig if sig else slot.__name__
        if hasattr(self, signal):
            getattr(self, signal).connect(slot)

    def _link_to(self, obj):
        for name in self.names:
            if hasattr(obj, name):
                self._link(getattr(obj, name))

    def link_to(self, *args):
        """ Connect all of our signals to matching slots on the provided object, if they exist """
        for arg in args:
            if isinstance(arg, list):
                for obj in arg:
                    self._link_to(obj)
            else:
                self._link_to(arg)

        return self

    def _link_tuple(self, tup):
        signal = tup[0]
        if isinstance(tup[1], (list, tuple)):
            for func in tup[1]:
                self._link(func, sig=signal)
        else:
            self._link(tup[1], sig=signal)

    def link(self, *args):
        """ Connect the given slots to matching signals, if they exist """
        if isinstance(args[0], str):
            for arg in args[1:]:
                self._link_tuple((args[0], arg))
        else:
            for arg in args:
                if isinstance(arg, list):
                    for item in arg:
                        if isinstance(item, tuple):
                            self._link_tuple(item)
                        else:
                            self._link(item)

                elif isinstance(arg, tuple):
                    self._link_tuple(arg)

                elif isinstance(arg, dict):
                    for signal, func in arg.items():
                        self._link(func, sig=signal)

                else:
                    self._link(arg)
            
        return self


class SlotBundle(QObject):
    def __init__(self, slots, link_to=None, link=None, sig_fmt='on_{}'):
        super().__init__()
        self.slots = slots
        self.names = list(slots.keys())
        self.sig_fmt = sig_fmt

        # add signals
        signals = {self.sig_fmt.format(name):args for name, args in slots.items()}
        self._signals = SigBundle(signals)

        # add slot methods
        for name, args in slots.items():
            self._add_slot(name, args)

        if link_to:
            self.link_to(link_to)

        if link:
            self.link(link)

    def _add_slot(self, name, args):
        """ add a qt Slot to this object """
        @Slot(*args)
        def fn(self, *args):
            getattr(self._signals, f'on_{name}').emit(*args)

        setattr(self, name, fn.__get__(self, super()))

    def _link(self, func, sig=None):
        """ Connect the given function to a matching signal, if it exists """
        signal = sig if sig else func.__name__
        if hasattr(self._signals, signal):
            getattr(self._signals, signal).connect(func)

    def _link_to(self, obj):
        for name in self._signals.names:
            if hasattr(obj, name):
                self._link(getattr(obj, name))

    def link_to(self, *args):
        """ Connect all of our signals to matching functions on the provided object, if they exist """
        for arg in args:
            if isinstance(arg, list):
                for obj in arg:
                    self._link_to(obj)
            else:
                self._link_to(arg)

        return self

    def _link_tuple(self, tup):
        signal = self.sig_fmt.format(tup[0])
        if isinstance(tup[1], (list, tuple)):
            for func in tup[1]:
                self._link(func, sig=signal)
        else:
            self._link(tup[1], sig=signal)

    def link(self, *args):
        """ Connect the given functions to matching signals, if they exist """
        if isinstance(args[0], str):
            for arg in args[1:]:
                self._link_tuple((args[0], arg))
        else:
            for arg in args:
                if isinstance(arg, list):
                    for item in arg:
                        if isinstance(item, tuple):
                            self._link_tuple(item)
                        else:
                            self._link(item)

                elif isinstance(arg, tuple):
                    self._link_tuple(arg)

                elif isinstance(arg, dict):
                    for s, f in arg.items():
                        self._link_tuple((s, f)) 

                else:
                    self._link(arg)
            
        return self

counter.py

from qt import *
from bundles import SigBundle, SlotBundle


class Counter(QObject):
    def __init__(self):
        super().__init__()
        signals = {
            'started': [], 
            'stopped': [], 
            'updated': [int]
        }
        slots = {
            'start': [], 
            'stop': [], 
            'reset': []
        }

        self.signals = SigBundle(signals)
        self.slots = SlotBundle(slots)
        
        self.slots.link_to(self)
        self.slots.link('start', self.start)

        self.count = 0

    def start(self):
        self.timer = QTimer()
        self.timer.timeout.connect(lambda: self.update())
        self.timer.start(1000)

    def on_stop(self):
        self.timer.stop()

    def on_reset(self):
        self.count = 0
        self.signals.updated.emit(self.count)

    def update(self):
        print(self.count)
        self.count += 1
        self.signals.updated.emit(self.count)


class CounterWidget(QWidget):
    def __init__(self):
        super().__init__()
        self.thread = QThread()
        self.counter = Counter()
        self.counter.moveToThread(self.thread)
        self.thread.start()

        self.create_widgets()
        self.connect_signals()
        self.build_layout()

    def create_widgets(self):
        self.count = QLabel("?")
        self.start = QPushButton("start")
        self.stop = QPushButton("stop")
        self.reset = QPushButton("reset")

    def connect_signals(self):
        self.counter.signals.updated.connect(lambda i: self.count.setText(str(i)))
        self.start.clicked.connect(self.counter.slots.start)
        self.stop.clicked.connect(self.counter.slots.stop)
        self.reset.clicked.connect(self.counter.slots.reset)

    def build_layout(self):
        grid = QGridLayout()
        grid.setContentsMargins(0, 10, 0, 10)
        grid.setColumnStretch(2, 1)

        grid.addWidget(QLabel("multi-thread signal/slot connection example"), 0, 0, 1, 6)
        grid.addWidget(QLabel("count:"), 1, 0)
        grid.addWidget(self.count, 1, 1)
        grid.addWidget(self.start, 1, 3)
        grid.addWidget(self.stop, 1, 4)
        grid.addWidget(self.reset, 1, 5)

        self.setLayout(grid)