Welcome, guest | Sign In | My Account | Store | Cart

Simple signal library similar to PyQT signals. Signals helps to decouple code in GUI applications. This code could be used in Tkinter applications for example.

Inspired and based in these other modules:

https://github.com/shaunduncan/smokesignal

https://github.com/dgovil/PySignal

https://github.com/jek/blinker

Python, 336 lines
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
import contextlib

__all__ = ["Signal", "SignalFactory", "NotRregisteredSignal", "InvalidSlot"]

class InvalidSlot(Exception):
    """Slot is not longer valid"""

class NotRregisteredSignal(ValueError):
    """Signal not registered in factory beforehand"""

class Slot(object):
    """Base class of slots"""
    def __init__(self, signal, func, max_calls=None):
        if not callable(func):
            raise ValueError("objects is not non-callable")

        self._signal = signal
        self._max_calls = max_calls

        self._func = func

    @property
    def func(self):
        return self._func

    @property
    def signal(self):
        return self._signal

    @property
    def max_calls(self):
        return self._max_calls

    def __call__(self, *args, **kwargs):
        if self._max_calls is not None:
            if self._max_calls == 0:
                raise InvalidSlot("Not possible to call more times callback")
            else:
                self._max_calls -= 1
            
        self._func(*args, **kwargs)
    
class Signal(object):
    """The Signal is the core object that handles connection and emission ."""

    def __init__(self, name=""):
        self._blocked = False
        self._slots = []

        self._name = str(name)
        
    @property
    def name(self):
        return self._name

    def emit(self, *args, **kwargs):
        """Calls all the connected slots with the provided args and kwargs unless block is activated"""

        if self._blocked: return

        for slot_index, slot in enumerate(list(self._slots)):
            slot(*args, **kwargs)
            
            if slot.max_calls == 0:
                del self._slots[slot_index]

    def on(self, callback, max_calls=None):
        if not callable(callback):
            raise ValueError("Connection to non-callable object failed")
        
        if max_calls is not None and not isinstance(max_calls, int):
            raise ValueError("max_calls should be None or integer")
            
        slot = Slot(self, callback, max_calls)
        self._slots.append(slot)
        
        return slot
    
    def once(self, callback):
        """Registers a callback that will respond to an event at most one time"""

        return self.on(callback, max_calls=1, weak_ref=weak_ref)

    def disconnect(self, slot):
        """Disconnects the slot from the signal"""

        """
        if not isinstance(slot, Slot):
            raise ValueError("Arguments is not slot")
        """
        if slot.signal != self: return

        found = False
        for index, other_slot in enumerate(self._slots):
            if other_slot == slot:
                found = True
                break

        if not found:
            raise Exception("Not valid slot.")
        
        del self._slots[index]

    def clear(self):
        """Clears the signal of all connected slots"""

        self._slots = []

    def block(self, isBlocked):
        """Sets blocking of the signal"""

        self._blocked = bool(isBlocked)
    
    @property    
    def is_blocked(self):
        return self._blocked
        
    @property
    def number_of_slots(self):
        return len(self._slots)

class Signal_Factory(object):
    """The Signal Factory object lets you handle signals by a string based name instead of by objects."""

    def __init__(self, mandatory_registry=False):
        self._signal_registry = dict()
        self._mandatory_registry = mandatory_registry

    def _check_signal_is_registered(self, signal):
        if signal not in self._signal_registry:
            if self._mandatory_registry:
                raise NotRregisteredSignal
            else:
                self._signal_registry[signal] = Signal(name=signal)

    def add_signal(self, signal_obj):
        self._signal_registry[signal_obj.name] = signal_obj
        
    def register_signal(self, signal):
        if signal in self._signal_registry:
            raise ValueError("Signal already registered")

        self._signal_registry[signal] = Signal()
        
    def unregister_signal(self, signal):
        if signal not in self._signal_registry:
            raise ValueError("Not posisble to unregistered not registered signal: %s"%signal)
        
        self._signal_registry[signal].clear()
        del self._signal_registry[signal]

    def signal_names(self):
        return self._signal_registry.keys()

    def emit(self, signal, *args, **kwargs):
        """Emits a signal by name. Any additional args or kwargs are passed to the signal"""

        self._check_signal_is_registered(signal)
        self._signal_registry[signal].emit(*args, **kwargs)

    def on(self, signal, callback=None, max_calls=None):
        """
        Registers a single callback for receiving an event. Optionally, can specify a maximum number 
        of times the callback should receive a signal. This method works as both a function and a decorator.
        """
        
        self._check_signal_is_registered(signal)

        if callback is None:
            # decorator
            def decorator(callback):
                self.on(signal, callback, max_calls=max_calls)
                
            return decorator
        else:
            return self._signal_registry[signal].on(callback, max_calls=max_calls)
        
    def once(self, signal, callback=None, weak_ref=True):
        self._check_signal_is_registered(signal)

        if callback is None:
            # decorator
            def decorator(callback):
                return self.once(signal, callback, max_calls=max_calls)
                
            return decorator
        else:
            return self._signal_registry[signal].once(callback)

    def block(self, signal=None, isBlocked=True):
        """Sets the block on provided signals, or to all signals"""

        if signal is None:
            for signal in self._signal_registry.keys():
                self._signal_registry[signal].block(isBlocked)
        else:
            self._check_signal_is_registered(signal)

            self._signal_registry[signal].block(isBlocked)
        
    def unblock(self, signal=None):
        if signal is None:
            for signal in self._signal_registry.keys():
                self._signal_registry[signal].block(False)
        else:
            self._check_signal_is_registered(signal)

            self._signal_registry[signal].block(False)
        
    def is_blocked(self, signal):
        self._check_signal_is_registered(signal)

        return self._signal_registry[signal].is_blocked()

    def disconnect_from(self, signal, slot):
        """Remove slot from receiver list if it responds to the signal"""
        
        self._check_signal_is_registered(signal)
        self._signal_registry[signal].disconnect(slot)

    def disconnect(self, slot):
        if not isinstance(slot, Slot):
            raise ValueError("Argument not a slot")

        self._signal_registry[slot.signal.name].disconnect(slot)
    
    def clear_all(self):
        """
        Clears all callbacks for all signals
        """
        for signal_object in self._signal_registry.values():
            signal_object.clear()
            
    def clear(self, *signals):
        """Clears all callbacks for a particular signal or signals"""

        if signals:
            for signal in signals:
                self._check_signal_is_registered(signal)
        else:
            signals = self._signal_registry.keys()

        for signal in signals:
            self._signal_registry[signal].clear()

    @contextlib.contextmanager
    def emitting(self, exit, enter=None):
        """
        Context manager for emitting signals either on enter or on exit of a context.
        By default, if this context manager is created using a single arg-style argument,
        it will emit a signal on exit. Otherwise, keyword arguments indicate signal points
        """
        
        self._check_signal_is_registered(exit)

        if enter is not None:
            self._check_signal_is_registered(enter)
            self.emit(enter)

        try:
            yield
        finally:
            self.emit(exit)
    
    def signal(self, signal):
        self._check_signal_is_registered(signal)
        return self._signal_registry[signal]

    def __contains__(self, signal):
        return signal in self._signal_registry
        
    exists_signal = __contains__

    def number_of_slots(self, signal):
        if signal in self._signal_registry:
            return self._signal_registry[signal].number_of_slots
        else:
            raise NotRregisteredSignal

    def __getitem__(self, signal):
        return self._signal_registry[signal]

    def __delitem__(self, signal):
        del self._signal_registry[signal]

if __name__ == "__main__":
    def greet1(name):
        print("Hello,", name)
        
    def greet2(name):
        print("Hi,", name)

    connection = Signal_Factory()
    connection.on('Greet', greet1)
    slot = connection.on('Greet', greet2)

    connection.emit('Greet', "John")

    connection.disconnect_from("Greet", slot)
    connection.emit('Greet', "Albert")
    
    print("------------------")
    connection.clear('Greet')
    
    print ("Greet has %d slots"%connection.signal("Greet").number_of_slots)
    slot = connection.on('Greet', greet2)
    # It's possible to disconnect directly without indicating the name of signal.
    connection.disconnect(slot)

    print("No output because there is no handler..")
    connection.emit('Greet', "Albert")
    
    print("------------------")
    connection2 = Signal_Factory()
    
    @connection2.on('foo')
    def my_callback1():
        print("called my_callback1()")

    @connection2.on('foo', max_calls=2)
    def my_callback2():
        print("called my_callback2()")
        
    with connection2.emitting('foo'):
        print("inside first with statement")
        
    print("------------------")
    
    def exit_func():
        print("exit of context manager")

    connection2.on('bar', exit_func)

    # 'foo' emitted on enter, 'bar' emitted on exit
    with connection2.emitting(enter='foo', exit='bar'):
        print("inside second with statement")