Coverage for gws-app/gws/core/debug.py: 20%

1"""Debugging and profiling tools.""" 

2 

3import re 

4import socket 

5import sys 

6import time 

7import traceback 

8 

9from . import log 

10 

11 

12def inspect(arg, max_depth=1, all_props=False): 

13 """Inspect a value up to the given depth. 

14 

15 Args: 

16 arg: The value. 

17 max_depth: Maximum nesting depth. 

18 all_props: If ``True``, list all attributes including callables, and expand dates. 

19 

20 Yields: 

21 Description lines. 

22 """ 

23 

24 yield from _dump(arg, None, 0, max_depth or 1, all_props, []) 

25 

26 

27def p(*args, lines=False, stack=False, d=3, all=False): 

28 """Log an inspection of the arguments at the debug level. 

29 

30 Args: 

31 *args: Values to inspect. 

32 lines: If ``True``, log the string value of each argument line by line, with line numbers. 

33 stack: If ``True``, log the current call stack instead. 

34 d: Maximum nesting depth. 

35 all: If ``True``, list all attributes including callables. 

36 """ 

37 

38 sep = '-' * 60 

39 

40 if lines: 

41 for arg in args: 

42 for s in enumerate(str(arg).split('\n'), 1): 

43 log.debug('%06d:%s' % s, stacklevel=2) 

44 log.debug(sep, stacklevel=2) 

45 return 

46 

47 if stack: 

48 for s in traceback.format_stack()[:-1]: 

49 log.debug(s.replace('\n', ' '), stacklevel=2) 

50 return 

51 

52 for arg in args: 

53 for s in inspect(arg, max_depth=d, all_props=all): 

54 log.debug(s, stacklevel=2) 

55 log.debug(sep, stacklevel=2) 

56 

57 

58_TIME_STACK = [] 

59 

60 

61def time_start(label=None): 

62 """Start a timer. 

63 

64 Timers are nested; ``time_end`` stops the most recently started one. 

65 

66 Args: 

67 label: Timer label for the log message. 

68 """ 

69 _TIME_STACK.append((time.time(), label or 'default')) 

70 

71 

72def time_end(): 

73 """Stop the most recently started timer and log the elapsed time at the debug level.""" 

74 if _TIME_STACK: 

75 t2 = time.time() 

76 t1, label = _TIME_STACK.pop() 

77 log.debug(f'@PROFILE {label} :: {t2 - t1:.2f}', stacklevel=2) 

78 

79 

80def pycharm_debugger_check(path_to_pycharm_debug_egg, host, port, suspend=False): 

81 """Connect to a PyCharm debugger if it is listening. 

82 

83 Attempt to open the debugger socket first and return that socket when pydevd asks for it. 

84 If there is no socket, then IDEA is not listening, return quickly. 

85 

86 Args: 

87 path_to_pycharm_debug_egg: Path to the pydevd egg file or to the directory containing it. 

88 host: Debugger host. 

89 port: Debugger port. 

90 suspend: If ``True``, suspend execution after connecting. 

91 

92 Returns: 

93 ``True`` if the debugger was connected, ``False`` otherwise. 

94 """ 

95 

96 sock = None 

97 try: 

98 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

99 sock.connect((host, port)) 

100 except: 

101 sock = None 

102 

103 if not sock: 

104 return False 

105 

106 if not path_to_pycharm_debug_egg.endswith('.egg'): 

107 if sys.version_info >= (3, 0): 

108 path_to_pycharm_debug_egg += '/pycharm-debug-py3k.egg' 

109 else: 

110 path_to_pycharm_debug_egg += '/pycharm-debug.egg' 

111 

112 sys.path.append(path_to_pycharm_debug_egg) 

113 pydevd = __import__('pydevd') 

114 

115 pydevd.StartClient = lambda h, p: sock 

116 pydevd.start_client = lambda h, p: sock 

117 pydevd.settrace(host, port=port, stdoutToServer=True, stderrToServer=True, suspend=suspend) 

118 

119 return True 

120 

121 

122## 

123 

124 

125_noexpand = { 

126 "<class 'datetime.datetime'>", 

127 "<class 'datetime.date'>", 

128 "<class 'memoryview'>", 

129} 

130 

131 

132def _should_list(k, v): 

133 if k.startswith('__'): 

134 return False 

135 if isinstance(v, type): 

136 return True 

137 if callable(v): 

138 return False 

139 return True 

140 

141 

142_MAX_REPR_LEN = 255 

143 

144 

145def _repr(x): 

146 s = repr(x) 

147 if len(s) > _MAX_REPR_LEN: 

148 s = s[:_MAX_REPR_LEN] + '...' 

149 return s 

150 

151 

152def _dump(x, name, depth, max_depth, all_props, seen): 

153 pfx = ' ' * depth 

154 

155 if name: 

156 pfx += str(name) + ': ' 

157 

158 t = str(type(x)) 

159 m = re.match(r"^<(type|class) '(.+?)'>", t) 

160 pfx += m.group(2) if m else t 

161 

162 try: 

163 pfx += '(%d)' % len(x) 

164 except (TypeError, AttributeError): 

165 pass 

166 

167 head = pfx + ' = ' + _repr(x) 

168 

169 if not x or isinstance(x, (bool, int, float, str, bytes)): 

170 yield head 

171 return 

172 

173 if not all_props and str(type(x)) in _noexpand: 

174 yield head 

175 return 

176 

177 try: 

178 if x in seen: 

179 yield head 

180 return 

181 seen.append(x) 

182 except: 

183 pass 

184 

185 if depth >= max_depth: 

186 yield pfx + '...' 

187 return 

188 

189 if isinstance(x, dict): 

190 yield pfx + ' = ' 

191 for k in x: 

192 yield from _dump(x[k], repr(k), depth + 1, max_depth, all_props, seen) 

193 return 

194 

195 if isinstance(x, (set, list, tuple)): 

196 yield pfx + ' = ' 

197 for k, v in enumerate(x): 

198 yield from _dump(v, str(k), depth + 1, max_depth, all_props, seen) 

199 return 

200 

201 yield head 

202 

203 for k in dir(x): 

204 try: 

205 v = getattr(x, k) 

206 except Exception as e: 

207 v = '?' + repr(e) 

208 if all_props or _should_list(k, v): 

209 yield from _dump(v, k, depth + 1, max_depth, all_props, seen)