Coverage for src/sensai/util/logging.py: 39%
185 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-08-13 22:17 +0000
1import atexit
2import logging as lg
3import sys
4import time
5from datetime import datetime
6from io import StringIO
7from logging import *
8from typing import List, Callable, Optional, TypeVar, TYPE_CHECKING
10from .time import format_duration
12if TYPE_CHECKING:
13 import pandas as pd
16log = getLogger(__name__)
18LOG_DEFAULT_FORMAT = '%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s'
19T = TypeVar("T")
21# Holds the log format that is configured by the user (using function `configure`), such
22# that it can be reused in other places
23_logFormat = LOG_DEFAULT_FORMAT
25# User-configured callback which is called after logging is configured via function `configure`
26_configureCallbacks: List[Callable[[], None]] = []
29def remove_log_handlers():
30 """
31 Removes all current log handlers
32 """
33 logger = getLogger()
34 while logger.hasHandlers():
35 logger.removeHandler(logger.handlers[0])
38def remove_log_handler(handler):
39 getLogger().removeHandler(handler)
42def is_log_handler_active(handler):
43 return handler in getLogger().handlers
46def is_enabled() -> bool:
47 """
48 :return: True if logging is enabled (at least one log handler exists)
49 """
50 return getLogger().hasHandlers()
53def set_configure_callback(callback: Callable[[], None], append: bool = True) -> None:
54 """
55 Configures a function to be called when logging is configured, e.g. through :func:`configure, :func:`run_main` or
56 :func:`run_cli`.
57 A typical use for the callback is to configure the logging behaviour of packages, setting appropriate log levels.
59 :param callback: the function to call
60 :param append: whether to append to the list of callbacks; if False, any existing callbacks will be removed
61 """
62 global _configureCallbacks
63 if not append:
64 _configureCallbacks = []
65 _configureCallbacks.append(callback)
68# noinspection PyShadowingBuiltins
69def configure(format=LOG_DEFAULT_FORMAT, level=lg.DEBUG):
70 """
71 Configures logging to stdout with the given format and log level,
72 also configuring the default log levels of some overly verbose libraries as well as some pandas output options.
74 :param format: the log format
75 :param level: the minimum log level
76 """
77 global _logFormat
78 _logFormat = format
79 remove_log_handlers()
80 basicConfig(level=level, format=format, stream=sys.stdout)
81 getLogger("matplotlib").setLevel(lg.INFO)
82 getLogger("urllib3").setLevel(lg.INFO)
83 getLogger("msal").setLevel(lg.INFO)
84 try:
85 import pandas as pd
86 pd.set_option('display.max_colwidth', 255)
87 except ImportError:
88 pass
89 for callback in _configureCallbacks:
90 callback()
93# noinspection PyShadowingBuiltins
94def run_main(main_fn: Callable[..., T], format=LOG_DEFAULT_FORMAT, level=lg.DEBUG) -> T:
95 """
96 Configures logging with the given parameters, ensuring that any exceptions that occur during
97 the execution of the given function are logged.
98 Logs two additional messages, one before the execution of the function, and one upon its completion.
100 :param main_fn: the function to be executed
101 :param format: the log message format
102 :param level: the minimum log level
103 :return: the result of `main_fn`
104 """
105 configure(format=format, level=level)
106 log.info("Starting")
107 try:
108 result = main_fn()
109 log.info("Done")
110 return result
111 except Exception as e:
112 log.error("Exception during script execution", exc_info=e)
115# noinspection PyShadowingBuiltins
116def run_cli(main_fn: Callable[..., T], format: str = LOG_DEFAULT_FORMAT, level: int = lg.DEBUG) -> Optional[T]:
117 """
118 Configures logging with the given parameters and runs the given main function as a
119 CLI using `jsonargparse` (which is configured to also parse attribute docstrings, such
120 that dataclasses can be used as function arguments).
121 Using this function requires that `jsonargparse` and `docstring_parser` be available.
122 Like `run_main`, two additional log messages will be logged (at the beginning and end
123 of the execution), and it is ensured that all exceptions will be logged.
125 :param main_fn: the function to be executed
126 :param format: the log message format
127 :param level: the minimum log level
128 :return: the result of `main_fn`
129 """
130 from jsonargparse import set_docstring_parse_options, CLI
132 set_docstring_parse_options(attribute_docstrings=True)
133 return run_main(lambda: CLI(main_fn), format=format, level=level)
136def datetime_tag() -> str:
137 """
138 :return: a string tag for use in log file names which contains the current date and time (compact but readable)
139 """
140 return datetime.now().strftime('%Y%m%d-%H%M%S')
143_fileLoggerPaths: List[str] = []
144_isAtExitReportFileLoggerRegistered = False
145_memoryLogStream: Optional[StringIO] = None
148def _at_exit_report_file_logger():
149 for path in _fileLoggerPaths:
150 print(f"A log file was saved to {path}")
153def add_file_logger(path, register_atexit=True):
154 global _isAtExitReportFileLoggerRegistered
155 log.info(f"Logging to {path} ...")
156 handler = FileHandler(path)
157 handler.setFormatter(Formatter(_logFormat))
158 Logger.root.addHandler(handler)
159 _fileLoggerPaths.append(path)
160 if not _isAtExitReportFileLoggerRegistered and register_atexit:
161 atexit.register(_at_exit_report_file_logger)
162 _isAtExitReportFileLoggerRegistered = True
163 return handler
166def add_memory_logger() -> None:
167 """
168 Enables in-memory logging (if it is not already enabled), i.e. all log statements are written to a memory buffer and can later be
169 read via function `get_memory_log()`
170 """
171 global _memoryLogStream
172 if _memoryLogStream is not None:
173 return
174 _memoryLogStream = StringIO()
175 handler = StreamHandler(_memoryLogStream)
176 handler.setFormatter(Formatter(_logFormat))
177 Logger.root.addHandler(handler)
180def get_memory_log():
181 return _memoryLogStream.getvalue()
184class StopWatch:
185 """
186 Represents a stop watch for timing an execution. Constructing an instance starts the stopwatch.
187 """
188 def __init__(self, start=True):
189 self.start_time = time.time()
190 self._elapsed_secs = 0.0
191 self.is_running = start
193 def reset(self, start=True):
194 """
195 Resets the stopwatch, setting the elapsed time to zero.
197 :param start: whether to start the stopwatch immediately
198 """
199 self.start_time = time.time()
200 self._elapsed_secs = 0.0
201 self.is_running = start
203 def restart(self):
204 """
205 Resets the stopwatch (setting the elapsed time to zero) and restarts it immediately
206 """
207 self.reset(start=True)
209 def _get_elapsed_time_since_last_start(self):
210 if self.is_running:
211 return time.time() - self.start_time
212 else:
213 return 0
215 def pause(self):
216 """
217 Pauses the stopwatch. It can be resumed via method 'resume'.
218 """
219 self._elapsed_secs += self._get_elapsed_time_since_last_start()
220 self.is_running = False
222 def resume(self):
223 """
224 Resumes the stopwatch (assuming that it is currently paused). If the stopwatch is not paused,
225 the method has no effect (and a warning is logged).
226 """
227 if not self.is_running:
228 self.start_time = time.time()
229 self.is_running = True
230 else:
231 log.warning("Stopwatch is already running (resume has not effect)")
233 def get_elapsed_time_secs(self) -> float:
234 """
235 Gets the total elapsed time, in seconds, on this stopwatch.
237 :return: the elapsed time in seconds
238 """
239 return self._elapsed_secs + self._get_elapsed_time_since_last_start()
241 def get_elapsed_timedelta(self) -> "pd.Timedelta":
242 """
243 :return: the elapsed time as a pandas.Timedelta object
244 """
245 return pd.Timedelta(self.get_elapsed_time_secs(), unit="s")
247 def get_elapsed_time_string(self) -> str:
248 """
249 :return: a string representation of the elapsed time
250 """
251 secs = self.get_elapsed_time_secs()
252 if secs < 60:
253 return f"{secs:.3f} seconds"
254 else:
255 try:
256 import pandas as pd
257 return str(pd.Timedelta(secs, unit="s"))
258 except ImportError:
259 return format_duration(secs)
262class StopWatchManager:
263 """
264 A singleton which manages a pool of named stopwatches, such that executions to be timed by referring to a name only -
265 without the need for a limited scope.
266 """
267 _instance = None
269 @classmethod
270 def get_instance(cls):
271 if cls._instance is None:
272 cls._instance = StopWatchManager(42)
273 return cls._instance
275 def __init__(self, secret):
276 if secret != 42:
277 raise Exception("Use only the singleton instance via getInstance")
278 self._stopWatches = {}
280 def start(self, name):
281 self._stopWatches[name] = time.time()
283 def stop(self, name) -> float:
284 """
285 :param name: the name of the stopwatch
286 :return: the time that has passed in seconds
287 """
288 time_passed_secs = time.time() - self._stopWatches[name]
289 del self._stopWatches[name]
290 return time_passed_secs
292 def is_running(self, name):
293 return name in self._stopWatches
296class LogTime:
297 """
298 An execution time logger which can be conveniently applied using a with-statement - in order to log the executing time of the respective
299 with-block.
300 """
302 def __init__(self, name, enabled=True, logger: Logger = None):
303 """
304 :param name: the name of the event whose time is to be logged upon completion as "<name> completed in <time>"
305 :param enabled: whether the logging is actually enabled; can be set to False to disable logging without necessitating
306 changes to client code
307 :param logger: the logger to use; if None, use the logger of LogTime's module
308 """
309 self.name = name
310 self.enabled = enabled
311 self.stopwatch = None
312 self.logger = logger if logger is not None else log
314 def start(self):
315 """
316 Starts the stopwatch
317 """
318 self.stopwatch = StopWatch()
319 if self.enabled:
320 self.logger.info(f"{self.name} starting ...")
322 def stop(self):
323 """
324 Stops the stopwatch and logs the time taken (if enabled)
325 """
326 if self.stopwatch is not None and self.enabled:
327 self.logger.info(f"{self.name} completed in {self.stopwatch.get_elapsed_time_string()}")
329 def __exit__(self, exc_type, exc_value, traceback):
330 self.stop()
332 def __enter__(self):
333 self.start()
334 return self
337class FileLoggerContext:
338 """
339 A context handler to be used in conjunction with Python's `with` statement which enables file-based logging.
340 """
341 def __init__(self, path: str, enabled=True):
342 """
343 :param path: the path to the log file
344 :param enabled: whether to actually perform any logging.
345 This switch allows the with statement to be applied regardless of whether logging shall be enabled.
346 """
347 self.enabled = enabled
348 self.path = path
349 self._log_handler = None
351 def __enter__(self):
352 if self.enabled:
353 self._log_handler = add_file_logger(self.path, register_atexit=False)
355 def __exit__(self, exc_type, exc_value, traceback):
356 if self._log_handler is not None:
357 remove_log_handler(self._log_handler)
360class LoggingDisabledContext:
361 """
362 A context manager that will temporarily disable logging
363 """
364 def __init__(self, highest_level=CRITICAL):
365 """
366 :param highest_level: the highest level to disable
367 """
368 self._highest_level = highest_level
369 self._previous_level = None
371 def __enter__(self):
372 self._previous_level = lg.root.manager.disable
373 lg.disable(self._highest_level)
375 def __exit__(self, exc_type, exc_value, traceback):
376 lg.disable(self._previous_level)
379# noinspection PyShadowingBuiltins
380class FallbackHandler(Handler):
381 """
382 A log handler which applies the given handler only if the root logger has no defined handlers (or no handlers other than
383 this fallback handler itself)
384 """
385 def __init__(self, handler: Handler, level=NOTSET):
386 """
387 :param handler: the handler to which messages shall be passed on
388 :param level: the minimum level to output; if NOTSET, pass on everything
389 """
390 super().__init__(level=level)
391 self._handler = handler
393 def emit(self, record):
394 root_handlers = getLogger().handlers
395 if len(root_handlers) == 0 or (len(root_handlers) == 1 and root_handlers[0] == self):
396 self._handler.emit(record)