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
Python, 336 lines
Download
Copy to clipboard1 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")
|
