Coverage for src/pycse/supyrvisor.py: 0.00%

1"""A supervisor decorator to rerun functions if they can be fixed. 

2 

3Functions can work but fail, or fail by raising exceptions. If you can examine 

4the ouput of a function and algorithmically propose a new set of arguments to 

5fix it, then supervisor can help you automate this. The idea is to write check 

6and exception functions for this, and then decorate your function to use them. 

7 

8This is an alternative to Custodian 

9(http://materialsproject.github.io/custodian/) which is pretty awesome, but also 

10heavy-weight IMO. 

11 

12This library was a proof of concept in making a decorator for this purpose. In 

13the end, I am not sure it is less heavyweight than custodian. 

14 

15[2023-09-20 Wed] Lightly tested on some examples. 

16[2023-09-21 Thu] Added the manager. 

17""" 

18 

19import functools 

20import inspect 

21 

22 

23def supervisor(check_funcs=(), exception_funcs=(), max_errors=5, verbose=False): 

24 """Decorator to supervise a function. After the function is run, each 

25 function in CHECK_FUNCS is run on the result. Each checker function has the 

26 signature check(args, kwargs, result). If the function should be rerun, then 

27 the check function should return a new args, kwargs to rerun the function 

28 with. Otherwise, it should return None 

29 

30 If there is an exception in the function, then each function in 

31 EXCEPTION_FUNCS will be run. Each exception function has the signature 

32 func(args, kwargs, exc). If one of them can fix the issue, it should return 

33 a new (args, kwargs) to rerun the function with, and otherwise return None 

34 which indicates there is no fix. 

35 

36 MAX_ERRORS is the maximum number of issues to try to fix. A value of -1 

37 means try forever. 

38 """ 

39 

40 def decorator(func): 

41 @functools.wraps(func) 

42 def wrapper(*args, **kwargs): 

43 nerrors = 0 

44 run = args, kwargs 

45 

46 while run and (nerrors < max_errors): 

47 try: 

48 result = func(*run[0], **run[1]) 

49 for checker in check_funcs: 

50 # run is None if everything checks out 

51 # or run is (args, kwargs) if it needs to be run again 

52 args, kwargs = run 

53 run = checker(args, kwargs, result) 

54 if run: 

55 if verbose: 

56 s = getattr(checker, "__name__", checker) 

57 print(f"Proposed fix in {s}: {run}") 

58 nerrors += 1 

59 # short-circuit break because we need to run it now. 

60 # this is a sequential fix, and does not allow a way 

61 # to choose what to fix if there is more than one 

62 # error 

63 break 

64 # After all the checks, run is None if they all passed, that 

65 # means we should return 

66 if not check_funcs or run is None: 

67 return result 

68 # Now should be returning to the while loop with new params 

69 # in run 

70 except Exception as e: 

71 if not exception_funcs: 

72 raise (e) # no fixer funcs defined, so we re-raise 

73 

74 for exc in exception_funcs: 

75 run = exc(run[0], run[1], e) 

76 if run: 

77 if verbose: 

78 s = getattr(exc, "__name__", exc) 

79 print(f"Proposed fix in {s}: {run}") 

80 nerrors += 1 

81 break # break out as soon as we get a fix 

82 

83 if run is None: 

84 # no new thing to try, reraise 

85 raise (e) 

86 

87 # if run is not None, this goes back to the while loop with 

88 # new params in run 

89 

90 # after the loop, we should raise if we got too many errors 

91 if nerrors == max_errors: 

92 raise Exception("Too many errors found") 

93 

94 return wrapper 

95 

96 return decorator 

97 

98 

99# The manager version 

100 

101 

102def check_result(func): 

103 """Decorator for functions to check the function result.""" 

104 

105 # This code defines a wrapper for a callable class, or a function. It feels 

106 # weird, but I could not find a way to inspect the func to see if it is a 

107 # class method any other way. inspect.ismethod did not work here. 

108 if func.__name__ == "__call__": 

109 

110 def wrapper(self, arguments, result): 

111 if isinstance(result, Exception): 

112 return None 

113 else: 

114 return func(self, arguments, result) 

115 

116 else: 

117 

118 def wrapper(arguments, result): 

119 if isinstance(result, Exception): 

120 return None 

121 else: 

122 return func(arguments, result) 

123 

124 return wrapper 

125 

126 

127def check_exception(func): 

128 """Decorator for functions to fix exceptions.""" 

129 if func.__name__ == "__call__": 

130 

131 def wrapper(self, arguments, result): 

132 if isinstance(result, Exception): 

133 return func(self, arguments, result) 

134 

135 else: 

136 

137 def wrapper(arguments, result): 

138 if isinstance(result, Exception): 

139 return func(arguments, result) 

140 else: 

141 return None 

142 

143 return wrapper 

144 

145 

146def manager(checkers=(), max_errors=5, verbose=False): 

147 """Decorator to manage a function. After the function is run, each function 

148 in CHECKERS is run on the result. Each checker function has the signature 

149 check(arguments, result). arguments will always be a dictionary of kwargs, 

150 including the default values. If the function should be rerun, then the 

151 checker function should return a new arguments dictionary to rerun the 

152 function with. Otherwise, it should return None. 

153 

154 The checker functions should be decorated with check_results or 

155 check_exception to indicate which one they handle. 

156 

157 MAX_ERRORS is the maximum number of issues to try to fix. A value of -1 

158 means try forever. 

159 

160 """ 

161 

162 def decorator(func): 

163 @functools.wraps(func) 

164 def wrapper(*args, **kwargs): 

165 nerrors = 0 

166 

167 # build the kwargs representation 

168 # this converts args to kwargs 

169 sig = inspect.signature(func) 

170 normalized_args = sig.bind(*args, **kwargs) 

171 normalized_args.apply_defaults() 

172 runargs = normalized_args.arguments 

173 

174 while runargs and (nerrors < max_errors): 

175 try: 

176 result = func(**runargs) 

177 for checker in checkers: 

178 # run is None if everything checks out 

179 # or run is (args, kwargs) if it needs to be run again 

180 rerun_args = checker(runargs, result) 

181 if rerun_args: 

182 runargs = rerun_args 

183 if verbose: 

184 s = getattr(checker, "__name__", checker) 

185 print(f"Proposed fix in {s}: {runargs}") 

186 nerrors += 1 

187 # short-circuit break because we need to run it now. 

188 # this is a sequential fix, and does not allow a way 

189 # to choose what to fix if there is more than one 

190 # error 

191 break 

192 # After all the checks, run is None if they all passed, that 

193 # means we should return 

194 if not checkers or rerun_args is None: 

195 return result 

196 # Now should be returning to the while loop with new params 

197 # in run 

198 

199 except Exception as e: 

200 for checker in checkers: 

201 rerun_args = checker(runargs, e) 

202 if rerun_args: 

203 runargs = rerun_args 

204 if verbose: 

205 s = getattr(checker, "__name__", checker) 

206 print(f"Proposed fix in {s}: {runargs}") 

207 nerrors += 1 

208 break # break out as soon as we get a fix 

209 

210 if rerun_args is None: 

211 # no new arguments to rerun with were found 

212 # so nothing can be fixed. 

213 raise (e) 

214 

215 # if runargs is not None, this goes back to the while loop 

216 # with new params in run 

217 

218 # after the loop, we should raise if we got too many errors 

219 if nerrors == max_errors: 

220 raise Exception("Too many errors found") 

221 

222 return wrapper 

223 

224 return decorator