Coverage for C:\Repos\leo-editor\leo\core\signal_manager.py: 21%

85 statements  

« prev     ^ index     » next       coverage.py v6.4, created at 2022-05-24 10:21 -0500

1#@+leo-ver=5-thin 

2#@+node:tbrown.20171028115541.1: * @file signal_manager.py 

3""" 

4signal_manager.py - SignalManager - light weight signal management 

5 

6Extremely light weight. No enforcement of signal arguments, or 

7even explicit listing of which signals exist. 

8 

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: 

15 

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 

41 

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 

51 

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) 

59 

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""" 

100 

101 # simple use 

102 

103 

104 class Emitter(SignalManager): 

105 

106 def some_emission(self): 

107 self.emit('the_emission', 12, [1, 2, 3]) 

108 

109 def hear_emit(n, l): 

110 print(f"Got {n} {l}") 

111 

112 emitter = Emitter() 

113 emitter.connect('the_emission', hear_emit) 

114 emitter.some_emission() 

115 

116 # use with proxy and locking 

117 

118 

119 class Tester: 

120 

121 def __init__(self, name, relay): 

122 self.name = name 

123 self.relay = relay 

124 connect(self.relay, 'work_done', self.check_work) 

125 

126 def do_work(self): 

127 emit(self.relay, 'work_done', 4.2, animal='otters', _sig_lock=self) 

128 

129 def check_work(self, num, animal='eels'): 

130 if is_locked(self): 

131 return 

132 print(f"{self.name} heard about {num} {animal}") 

133 

134 

135 class SomeProxy: 

136 """Like a public notice board""" 

137 pass 

138 

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() 

148 

149#@@language python 

150#@@tabwidth -4 

151#@-leo