Coverage for src/km3pipe/logger.py: 50%

1# Filename: logger.py 

2# pylint: disable=locally-disabled,C0103 

3""" 

4The logging facility. 

5 

6""" 

7from hashlib import sha256 

8from inspect import getframeinfo, stack 

9import socket 

10import logging 

11import logging.handlers 

12 

13from .tools import colored, supports_color 

14 

15__author__ = "Tamas Gal" 

16__copyright__ = "Copyright 2016, Tamas Gal and the KM3NeT collaboration." 

17__credits__ = [] 

18__license__ = "MIT" 

19__maintainer__ = "Tamas Gal" 

20__email__ = "tgal@km3net.de" 

21__status__ = "Development" 

22 

23loggers = {} # this holds all the registered loggers 

24# logging.basicConfig() 

25 

26DEPRECATION = 45 

27logging.addLevelName(DEPRECATION, "DEPRECATION") 

28ONCE = 46 

29logging.addLevelName(ONCE, "ONCE") 

30 

31 

32def deprecation(self, message, *args, **kws): 

33 """Show a deprecation warning.""" 

34 self._log(DEPRECATION, message, args, **kws) 

35 

36 

37def once(self, message, *args, **kws): 

38 """Show a message only once, determined by position in source or identifer. 

39 

40 This will not work in IPython or Jupyter notebooks if no identifier is 

41 specified, since then the determined position in source contains the 

42 execution number of the input (cell), which changes every time. 

43 Set a unique ``identifier=X``, otherwise the message will be printed every 

44 time. 

45 

46 """ 

47 identifier = kws.pop("identifier", None) 

48 

49 if identifier is None: 

50 caller = getframeinfo(stack()[1][0]) 

51 identifier = "%s:%d" % (caller.filename, caller.lineno) 

52 if not hasattr(self, "once_dict"): 

53 self.once_dict = {} 

54 if identifier in self.once_dict: 

55 return 

56 self.once_dict[identifier] = True 

57 self._log(ONCE, message, args, **kws) 

58 

59 

60logging.Logger.deprecation = deprecation 

61logging.Logger.once = once 

62 

63if supports_color(): 

64 logging.addLevelName( 

65 logging.INFO, "\033[1;32m%s\033[1;0m" % logging.getLevelName(logging.INFO) 

66 ) 

67 logging.addLevelName( 

68 logging.DEBUG, "\033[1;34m%s\033[1;0m" % logging.getLevelName(logging.DEBUG) 

69 ) 

70 logging.addLevelName( 

71 logging.WARNING, "\033[1;33m%s\033[1;0m" % logging.getLevelName(logging.WARNING) 

72 ) 

73 logging.addLevelName( 

74 logging.ERROR, "\033[1;31m%s\033[1;0m" % logging.getLevelName(logging.ERROR) 

75 ) 

76 logging.addLevelName( 

77 logging.CRITICAL, 

78 "\033[1;101m%s\033[1;0m" % logging.getLevelName(logging.CRITICAL), 

79 ) 

80 logging.addLevelName(DEPRECATION, "\033[1;35m%s\033[1;0m" % "DEPRECATION") 

81 logging.addLevelName(ONCE, "\033[1;36m%s\033[1;0m" % "ONCE") 

82 

83 

84class LogIO(object): 

85 """Read/write logging information.""" 

86 

87 def __init__(self, node, stream, url="pi2089.physik.uni-erlangen.de", port=28777): 

88 self.node = node 

89 self.stream = stream 

90 self.url = url 

91 self.port = port 

92 self.sock = None 

93 self.connect() 

94 

95 def send(self, message, level="info"): 

96 message_string = "+log|{0}|{1}|{2}|{3}\r\n".format( 

97 self.stream, self.node, level, message 

98 ) 

99 try: 

100 self.sock.send(message_string) 

101 except socket.error: 

102 print("Lost connection, reconnecting...") 

103 self.connect() 

104 self.sock.send(message_string) 

105 

106 def connect(self): 

107 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

108 self.sock.connect((self.url, self.port)) 

109 

110 

111def get_logger(name, filename=None, stream_loglevel="INFO", file_loglevel="DEBUG"): 

112 """Helper function to get a logger""" 

113 if name in loggers: 

114 return loggers[name] 

115 logger = logging.getLogger(name) 

116 logger.propagate = False 

117 

118 with_color = supports_color() 

119 

120 pre1, suf1 = hash_coloured_escapes(name) if with_color else ("", "") 

121 pre2, suf2 = hash_coloured_escapes(name + "salt") if with_color else ("", "") 

122 formatter = logging.Formatter( 

123 "%(asctime)s %(levelname)s {}+{}+{} " 

124 "%(name)s: %(message)s".format(pre1, pre2, suf1), 

125 datefmt="%Y-%m-%d %H:%M:%S", 

126 ) 

127 if filename is not None: 

128 ch_file = logging.handlers.RotatingFileHandler( 

129 filename, maxBytes=5 * 1024 * 1024, backupCount=10 

130 ) 

131 ch_file.setLevel(file_loglevel) 

132 ch_file.setFormatter(formatter) 

133 logger.addHandler(ch_file) 

134 ch = logging.StreamHandler() 

135 ch.setLevel(stream_loglevel) 

136 ch.setFormatter(formatter) 

137 logger.addHandler(ch) 

138 

139 loggers[name] = logger 

140 

141 logger.once_dict = {} 

142 

143 return logger 

144 

145 

146def available_loggers(): 

147 """Return a list of avialable logger names""" 

148 return list(logging.Logger.manager.loggerDict.keys()) 

149 

150 

151def set_level(log_or_name, level): 

152 """Set the log level for given logger and all handlers""" 

153 if isinstance(log_or_name, str): 

154 log = get_logger(log_or_name) 

155 else: 

156 log = log_or_name 

157 log.setLevel(level) 

158 for handler in log.handlers: 

159 handler.setLevel(level) 

160 

161 

162def get_printer(name, color=None, ansi_code=None, force_color=False): 

163 """Return a function which prints a message with a coloured name prefix""" 

164 

165 if force_color or supports_color(): 

166 if color is None and ansi_code is None: 

167 cpre_1, csuf_1 = hash_coloured_escapes(name) 

168 cpre_2, csuf_2 = hash_coloured_escapes(name + "salt") 

169 name = cpre_1 + "+" + cpre_2 + "+" + csuf_1 + " " + name 

170 else: 

171 name = colored(name, color=color, ansi_code=ansi_code) 

172 

173 prefix = name + ": " 

174 

175 def printer(text): 

176 print(prefix + str(text)) 

177 

178 return printer 

179 

180 

181def hash_coloured(text): 

182 """Return a ANSI coloured text based on its hash""" 

183 ansi_code = int(sha256(text.encode("utf-8")).hexdigest(), 16) % 230 

184 return colored(text, ansi_code=ansi_code) 

185 

186 

187def hash_coloured_escapes(text): 

188 """Return the ANSI hash colour prefix and suffix for a given text""" 

189 ansi_code = int(sha256(text.encode("utf-8")).hexdigest(), 16) % 230 

190 prefix, suffix = colored("SPLIT", ansi_code=ansi_code).split("SPLIT") 

191 return prefix, suffix