Coverage for bim2sim/kernel/log.py: 98%

1import logging 

2from pathlib import Path 

3from typing import List, Tuple 

4 

5# from bim2sim.project import Project 

6 

7USER = 'user' 

8 

9user = {'audience': USER} 

10 

11 

12# TODO: check log calls 

13# TODO: fix errors exposed by log messages 

14 

15 

16class AudienceFilter(logging.Filter): 

17 def __init__(self, audience, name=''): 

18 super().__init__(name) 

19 self.audience = audience 

20 

21 def filter(self, record: logging.LogRecord) -> bool: 

22 audience = getattr(record, 'audience', None) 

23 return audience == self.audience 

24 

25 

26class ThreadLogFilter(logging.Filter): 

27 """This filter only show log entries for specified thread name.""" 

28 

29 def __init__(self, thread_name, *args, **kwargs): 

30 super().__init__(*args, **kwargs) 

31 self.thread_name = thread_name 

32 

33 def filter(self, record): 

34 return record.threadName == self.thread_name 

35 

36 

37def get_user_logger(name): 

38 return logging.LoggerAdapter(logging.getLogger(name), user) 

39 

40 

41class BufferedHandler(logging.Handler): 

42 """Handler to buffer messages and don't loose them.""" 

43 def __init__(self, level=logging.NOTSET): 

44 super().__init__(level) 

45 self.buffer = [] 

46 

47 def emit(self, record): 

48 self.buffer.append(record) 

49 

50 def flush_buffer(self, file_handler): 

51 for record in self.buffer: 

52 file_handler.emit(record) 

53 self.buffer.clear() 

54 

55 

56def initial_logging_setup(level=logging.DEBUG): 

57 """Initial setup before project folder exists. 

58 

59 This is the first of the two-step logging setup. It makes sure that logging 

60 messages before project folder creation are still prompted to stream 

61 handler and saved to BufferHandler to get complete log files later. 

62 """ 

63 general_logger = logging.getLogger('bim2sim', ) 

64 general_logger.setLevel(level) 

65 general_log_stream_handler = logging.StreamHandler() 

66 general_log_stream_handler.setFormatter(dev_formatter) 

67 # general_log_stream_handler.addFilter(AudienceFilter(USER)) 

68 general_logger.addHandler(general_log_stream_handler) 

69 

70 # buffered handler to maintain messages that are logged before project 

71 # folder and therefore storage place for file handler exists 

72 buffered_handler = BufferedHandler() 

73 general_logger.addHandler(buffered_handler) 

74 

75 return general_logger 

76 

77 

78def project_logging_setup( 

79 prj=None, 

80) -> Tuple[List[logging.Handler], List[ThreadLogFilter]]: 

81 """Setup project logging 

82 

83 This is the second step of the two-step logging setup. After project folder 

84 creation this step adds completes the setup and adds the filder handlers 

85 that store the logs. 

86 

87 This creates the following: 

88 * the file output file bim2sim.log where the logs are stored 

89 * the logger quality_logger which stores all information about the quality 

90 of existing information of the BIM model. This logger is only on file to 

91 keep the logs cleaner 

92 """ 

93 # get general_logger from initial_logging_setup 

94 general_logger = logging.getLogger('bim2sim') 

95 

96 prj_log_path = prj.paths.log 

97 

98 # get existing stream handler and buffer handler from initial_logging_setup 

99 handlers = [] 

100 general_log_stream_handler = None 

101 buffered_handler = None 

102 for handler in general_logger.handlers: 

103 if isinstance(handler, logging.StreamHandler): 

104 general_log_stream_handler = handler 

105 handlers.append(general_log_stream_handler) 

106 if isinstance(handler, BufferedHandler): 

107 buffered_handler = handler 

108 handlers.append(buffered_handler) 

109 

110 # add file handler for general log and flush logs from buffer log to keep 

111 # all log messages between FolderStructure creation and project init 

112 general_log_file_handler = logging.FileHandler( 

113 prj_log_path / 'bim2sim.log') 

114 general_log_file_handler.setFormatter(dev_formatter) 

115 general_log_file_handler.addFilter(AudienceFilter(audience=None)) 

116 general_logger.addHandler(general_log_file_handler) 

117 if buffered_handler: 

118 buffered_handler.flush_buffer(general_log_file_handler) 

119 handlers.append(general_log_file_handler) 

120 

121 # add quality logger with stream and file handler 

122 quality_logger = logging.getLogger('bim2sim.QualityReport') 

123 # do not propagate messages to main bim2sim logger to keep logs cleaner 

124 quality_logger.propagate = False 

125 quality_handler = logging.FileHandler( 

126 Path(prj_log_path / "IFCQualityReport.log")) 

127 quality_handler.setFormatter(quality_formatter) 

128 quality_logger.addHandler(quality_handler) 

129 handlers.append(quality_handler) 

130 

131 # set thread log filter 

132 thread_log_filter = ThreadLogFilter(prj.thread_name) 

133 for handler in handlers: 

134 handler.addFilter(thread_log_filter) 

135 

136 return handlers, [thread_log_filter] 

137 

138 

139def teardown_loggers(): 

140 """Closes and removes all handlers from loggers in the 'bim2sim' hierarchy. 

141 

142 Iterates through all existing loggers and cleans up those that start 

143 with 'bim2sim'. 

144 For each matching logger, all handlers are properly closed and removed. 

145 Errors during file handler closure (e.g., due to already deleted l 

146 og files) are silently ignored. 

147 

148 """ 

149 # Get all loggers from logging manager hierarchy 

150 logger_dict = logging.Logger.manager.loggerDict 

151 

152 # Iterate through all loggers that start with 'bim2sim' 

153 for logger_name, logger_instance in logger_dict.items(): 

154 if isinstance(logger_instance, 

155 logging.Logger) and logger_name.startswith('bim2sim'): 

156 for handler in logger_instance.handlers[:]: 

157 try: 

158 handler.close() 

159 except (PermissionError, FileNotFoundError): 

160 pass 

161 logger_instance.removeHandler(handler) 

162 

163 

164class CustomFormatter(logging.Formatter): 

165 """Custom logging design based on 

166 https://stackoverflow.com/questions/384076/how-can-i-color-python 

167 -logging-output""" 

168 

169 def __init__(self, fmt): 

170 super().__init__() 

171 self._fmt = fmt 

172 

173 def format(self, record): 

174 grey = "\x1b[37;20m" 

175 green = "\x1b[32;20m" 

176 yellow = "\x1b[33;20m" 

177 red = "\x1b[31;20m" 

178 bold_red = "\x1b[31;1m" 

179 reset = "\x1b[0m" 

180 # format = "[%(levelname)s] %(name)s: %(message)s" 

181 

182 FORMATS = { 

183 logging.DEBUG: grey + self._fmt + reset, 

184 logging.INFO: green + self._fmt + reset, 

185 logging.WARNING: yellow + self._fmt + reset, 

186 logging.ERROR: red + self._fmt + reset, 

187 logging.CRITICAL: bold_red + self._fmt + reset 

188 } 

189 log_fmt = FORMATS.get(record.levelno) 

190 formatter = logging.Formatter(log_fmt) 

191 return formatter.format(record) 

192 

193 

194quality_formatter = CustomFormatter('[QUALITY-%(levelname)s] %(name)s:' 

195 ' %(message)s') 

196user_formatter = CustomFormatter('[USER-%(levelname)s]:' 

197 ' %(message)s') 

198dev_formatter = CustomFormatter('[DEV-%(levelname)s] -' 

199 ' %(asctime)s %(name)s.%(funcName)s:' 

200 ' %(message)s')