Coverage for C:\leo.repo\leo-editor\leo\core\signal_manager.py: 21%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#@+leo-ver=5-thin
2#@+node:tbrown.20171028115541.1: * @file signal_manager.py
3"""
4signal_manager.py - SignalManager - light weight signal management
6Extremely light weight. No enforcement of signal arguments, or
7even explicit listing of which signals exist.
9Terry Brown, terrynbrown@gmail.com, Thu Mar 23 21:13:38 2017
10"""
11from collections import defaultdict
12#@+others
13#@+node:tbrown.20171028115601.2: ** class SignalData
14class SignalData:
16 def __init__(self):
17 self.listeners = defaultdict(list)
18 self.emitters = []
19 self.locked = False
20#@+node:tbrown.20171028115601.4: ** class MsgSignalHandled
21class MsgSignalHandled:
22 """A listener can return SignalManager.MsgSignalHandled to prevent
23 other listeners from being called
24 """
25 pass
26#@+node:tbrown.20171028115601.5: ** _setup
27def _setup(obj):
28 if not hasattr(obj, '_signal_data'):
29 obj._signal_data = SignalData()
30#@+node:tbrown.20171028115601.6: ** emit
31def emit(source, signal_name, *args, **kwargs):
32 """Emit signal to all listeners"""
33 if not hasattr(source, '_signal_data'):
34 return
35 if '_sig_lock' in kwargs:
36 obj_to_lock = kwargs.pop('_sig_lock')
37 _setup(obj_to_lock)
38 obj_to_lock._signal_data.locked = True
39 else:
40 obj_to_lock = None
42 for listener in source._signal_data.listeners[signal_name]:
43 try:
44 if listener.__self__._signal_data.locked:
45 continue
46 except AttributeError:
47 pass
48 response = listener(*args, **kwargs)
49 if response == MsgSignalHandled:
50 break
52 if obj_to_lock is not None:
53 obj_to_lock._signal_data.locked = False
54#@+node:tbrown.20171028115601.7: ** connect
55def connect(source, signal_name, listener):
56 """Connect to signal"""
57 _setup(source)
58 source._signal_data.listeners[signal_name].append(listener)
60 if hasattr(listener, '__self__'):
61 obj = listener.__self__
62 _setup(obj)
63 obj._signal_data.emitters.append(source)
64#@+node:tbrown.20171028115601.8: ** disconnect_all
65def disconnect_all(listener):
66 """Disconnect from all signals"""
67 for emitter in listener._signal_data.emitters:
68 for signal in emitter._signal_data.listeners:
69 emitter._signal_data.listeners[signal] = [
70 i for i in emitter._signal_data.listeners[signal]
71 if getattr(i, '__self__', None) != listener
72 ]
73#@+node:tbrown.20171028115601.9: ** is_locked
74def is_locked(obj):
75 return hasattr(obj, '_signal_data') and obj._signal_data.locked
76#@+node:tbrown.20171028115601.10: ** lock
77def lock(obj):
78 _setup(obj)
79 obj._signal_data.locked = True
80#@+node:tbrown.20171028115601.11: ** unlock
81def unlock(obj):
82 _setup(obj)
83 obj._signal_data.locked = False
84#@+node:tbrown.20171028115601.12: ** class SignalManager
85class SignalManager:
86 """SignalManager - light weight signal management mixin."""
87 #@+others
88 #@+node:tbrown.20171028115601.13: *3* emit
89 def emit(self, signal_name, *args, **kwargs):
90 """Emit signal to all listeners"""
91 emit(self, signal_name, *args, **kwargs)
92 #@+node:tbrown.20171028115601.14: *3* connect
93 def connect(self, signal_name, listener):
94 """Connect to signal"""
95 connect(self, signal_name, listener)
96 #@-others
97#@+node:tbrown.20171028115601.15: ** main
98def main():
99 """test of SignalManager"""
101 # simple use
104 class Emitter(SignalManager):
106 def some_emission(self):
107 self.emit('the_emission', 12, [1, 2, 3])
109 def hear_emit(n, l):
110 print(f"Got {n} {l}")
112 emitter = Emitter()
113 emitter.connect('the_emission', hear_emit)
114 emitter.some_emission()
116 # use with proxy and locking
119 class Tester:
121 def __init__(self, name, relay):
122 self.name = name
123 self.relay = relay
124 connect(self.relay, 'work_done', self.check_work)
126 def do_work(self):
127 emit(self.relay, 'work_done', 4.2, animal='otters', _sig_lock=self)
129 def check_work(self, num, animal='eels'):
130 if is_locked(self):
131 return
132 print(f"{self.name} heard about {num} {animal}")
135 class SomeProxy:
136 """Like a public notice board"""
137 pass
139 relay = SomeProxy()
140 a = Tester('A', relay)
141 a.do_work()
142 b = Tester('B', relay)
143 a.do_work()
144 b.do_work()
145#@-others
146if __name__ == '__main__':
147 main()
149#@@language python
150#@@tabwidth -4
151#@-leo