Coverage for enpt/utils/logging.py: 71%

1# -*- coding: utf-8 -*- 

2 

3# EnPT, EnMAP Processing Tool - A Python package for pre-processing of EnMAP Level-1B data 

4# 

5# Copyright (C) 2018-2024 Karl Segl (GFZ Potsdam, segl@gfz-potsdam.de), Daniel Scheffler 

6# (GFZ Potsdam, danschef@gfz-potsdam.de), Niklas Bohn (GFZ Potsdam, nbohn@gfz-potsdam.de), 

7# Stéphane Guillaso (GFZ Potsdam, stephane.guillaso@gfz-potsdam.de) 

8# 

9# This software was developed within the context of the EnMAP project supported 

10# by the DLR Space Administration with funds of the German Federal Ministry of 

11# Economic Affairs and Energy (on the basis of a decision by the German Bundestag: 

12# 50 EE 1529) and contributions from DLR, GFZ and OHB System AG. 

13# 

14# This program is free software: you can redistribute it and/or modify it under 

15# the terms of the GNU General Public License as published by the Free Software 

16# Foundation, either version 3 of the License, or (at your option) any later 

17# version. Please note the following exception: `EnPT` depends on tqdm, which 

18# is distributed under the Mozilla Public Licence (MPL) v2.0 except for the files 

19# "tqdm/_tqdm.py", "setup.py", "README.rst", "MANIFEST.in" and ".gitignore". 

20# Details can be found here: https://github.com/tqdm/tqdm/blob/master/LICENCE. 

21# 

22# This program is distributed in the hope that it will be useful, but WITHOUT 

23# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS 

24# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more 

25# details. 

26# 

27# You should have received a copy of the GNU Lesser General Public License along 

28# with this program. If not, see <https://www.gnu.org/licenses/>. 

29 

30"""EnPT logging module containing logging related classes and functions.""" 

31 

32import logging 

33import os 

34import warnings 

35import sys 

36from io import StringIO # Python 3 only 

37 

38__author__ = 'Daniel Scheffler' 

39 

40 

41class EnPT_Logger(logging.Logger): 

42 """Class for the EnPT logger.""" 

43 

44 def __init__(self, 

45 name_logfile: str, 

46 fmt_suffix: any = None, 

47 path_logfile: str = None, 

48 log_level: any = 'INFO', 

49 append: bool = True) -> None: 

50 """Return a logging.logger instance pointing to the given logfile path. 

51 

52 :param name_logfile: 

53 :param fmt_suffix: if given, it will be included into log formatter 

54 :param path_logfile: if no path is given, only a StreamHandler is created 

55 :param log_level: the logging level to be used (choices: 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'; 

56 default: 'INFO') 

57 :param append: <bool> whether to append the log message to an existing logfile (1) 

58 or to create a new logfile (0); default=1 

59 """ 

60 # private attributes 

61 self._captured_stream = '' 

62 

63 # attributes that need to be present in order to unpickle the logger via __setstate_ 

64 self.name_logfile = name_logfile 

65 self.fmt_suffix = fmt_suffix 

66 self.path_logfile = path_logfile 

67 self.log_level = log_level 

68 

69 super(EnPT_Logger, self).__init__(name_logfile) 

70 

71 self.path_logfile = path_logfile 

72 self.formatter_fileH = logging.Formatter('%(asctime)s' + (' [%s]' % fmt_suffix if fmt_suffix else '') + 

73 ' %(levelname)s: %(message)s', datefmt='%Y/%m/%d %H:%M:%S') 

74 self.formatter_ConsoleH = logging.Formatter('%(asctime)s' + (' [%s]' % fmt_suffix if fmt_suffix else '') + 

75 ': %(message)s', datefmt='%Y/%m/%d %H:%M:%S') 

76 

77 if path_logfile: 

78 # create output directory 

79 while not os.path.isdir(os.path.dirname(path_logfile)): 

80 try: 

81 os.makedirs(os.path.dirname(path_logfile)) 

82 except OSError as e: 

83 if e.errno != 17: 

84 raise 

85 else: 

86 pass 

87 

88 # create FileHandler 

89 fileHandler = logging.FileHandler(path_logfile, mode='a' if append else 'w') 

90 fileHandler.setFormatter(self.formatter_fileH) 

91 fileHandler.setLevel(log_level) 

92 else: 

93 fileHandler = None 

94 

95 # create StreamHandler 

96 self.streamObj = StringIO() 

97 streamHandler = logging.StreamHandler(stream=self.streamObj) 

98 streamHandler.setFormatter(self.formatter_fileH) 

99 streamHandler.set_name('StringIO handler') 

100 

101 # create ConsoleHandler for logging levels DEGUG and INFO -> logging to sys.stdout 

102 consoleHandler_out = logging.StreamHandler(stream=sys.stdout) # by default it would go to sys.stderr 

103 consoleHandler_out.setFormatter(self.formatter_ConsoleH) 

104 consoleHandler_out.set_name('console handler stdout') 

105 consoleHandler_out.setLevel(log_level) 

106 consoleHandler_out.addFilter(LessThanFilter(logging.WARNING)) 

107 

108 # create ConsoleHandler for logging levels WARNING, ERROR, CRITICAL -> logging to sys.stderr 

109 consoleHandler_err = logging.StreamHandler(stream=sys.stderr) 

110 consoleHandler_err.setFormatter(self.formatter_ConsoleH) 

111 consoleHandler_err.setLevel(logging.WARNING) 

112 consoleHandler_err.set_name('console handler stderr') 

113 

114 self.setLevel(log_level) 

115 

116 if not self.handlers: 

117 if fileHandler: 

118 self.addHandler(fileHandler) 

119 self.addHandler(streamHandler) 

120 self.addHandler(consoleHandler_out) 

121 self.addHandler(consoleHandler_err) 

122 

123 def __getstate__(self): 

124 self.close() 

125 return self.__dict__ 

126 

127 def __setstate__(self, ObjDict): 

128 """Define how the attributes of EnPT_Logger are unpickled.""" 

129 self.__init__(ObjDict['name_logfile'], fmt_suffix=ObjDict['fmt_suffix'], path_logfile=ObjDict['path_logfile'], 

130 log_level=ObjDict['log_level'], append=True) 

131 ObjDict = self.__dict__ 

132 return ObjDict 

133 

134 @property 

135 def captured_stream(self) -> str: 

136 """Return the already captured logging stream. 

137 

138 NOTE: 

139 - set self.captured_stream: 

140 self.captured_stream = 'any string' 

141 """ 

142 if not self._captured_stream: 

143 self._captured_stream = self.streamObj.getvalue() 

144 

145 return self._captured_stream 

146 

147 @captured_stream.setter 

148 def captured_stream(self, string: str): 

149 assert isinstance(string, str), "'captured_stream' can only be set to a string. Got %s." % type(string) 

150 self._captured_stream = string 

151 

152 def close(self): 

153 """Close all logging handlers.""" 

154 # update captured_stream and flush stream 

155 self.captured_stream += self.streamObj.getvalue() 

156 

157 for handler in self.handlers[:]: 

158 try: 

159 if handler.get_name() == 'StringIO handler': 

160 self.streamObj.flush() 

161 self.removeHandler(handler) # if not called with '[:]' the StreamHandlers are left open 

162 handler.flush() 

163 handler.close() 

164 except PermissionError: 

165 warnings.warn('Could not properly close logfile due to a PermissionError: %s' % sys.exc_info()[1]) 

166 except ValueError as e: 

167 if str(e) != 'I/O operation on closed file': 

168 raise 

169 

170 if self.handlers[:]: 

171 warnings.warn('Not all logging handlers could be closed. Remaining handlers: %s' % self.handlers[:]) 

172 

173 def view_logfile(self): 

174 """View the log file written to disk.""" 

175 with open(self.path_logfile) as inF: 

176 print(inF.read()) 

177 

178 def __del__(self): 

179 self.close() 

180 

181 def __enter__(self): 

182 return self 

183 

184 def __exit__(self, exc_type, exc_val, exc_tb): 

185 self.close() 

186 return True if exc_type is None else False 

187 

188 

189def close_logger(logger): 

190 """Close the handlers of the given logging.Logger instance. 

191 

192 :param logger: logging.Logger instance or subclass instance 

193 """ 

194 if logger and hasattr(logger, 'handlers'): 

195 for handler in logger.handlers[:]: # if not called with '[:]' the StreamHandlers are left open 

196 try: 

197 logger.removeHandler(handler) 

198 handler.flush() 

199 handler.close() 

200 except PermissionError: 

201 warnings.warn('Could not properly close logfile due to a PermissionError: %s' % sys.exc_info()[1]) 

202 

203 if logger.handlers[:]: 

204 warnings.warn('Not all logging handlers could be closed. Remaining handlers: %s' % logger.handlers[:]) 

205 

206 

207def shutdown_loggers(): 

208 """Shutdown any currently active loggers.""" 

209 logging.shutdown() 

210 

211 

212class LessThanFilter(logging.Filter): 

213 """Filter class to filter log messages by a maximum log level. 

214 

215 Based on http://stackoverflow.com/questions/2302315/ 

216 how-can-info-and-debug-logging-message-be-sent-to-stdout-and-higher-level-messag 

217 """ 

218 

219 def __init__(self, exclusive_maximum, name=""): 

220 """Get an instance of LessThanFilter. 

221 

222 :param exclusive_maximum: maximum log level, e.g., logger.WARNING 

223 :param name: 

224 """ 

225 super(LessThanFilter, self).__init__(name) 

226 self.max_level = exclusive_maximum 

227 

228 def filter(self, record): 

229 """Filter funtion. 

230 

231 NOTE: Returns True if logging level of the given record is below the maximum log level. 

232 

233 :param record: 

234 :return: bool 

235 """ 

236 # non-zero return means we log this message 

237 return True if record.levelno < self.max_level else False