Coverage for src/sensai/util/logging.py: 41%
202 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-11-29 18:29 +0000
1import atexit
2import logging as lg
3import sys
4import time
5from abc import ABC, abstractmethod
6from datetime import datetime
7from io import StringIO
8from logging import *
9from typing import List, Callable, Optional, TypeVar, TYPE_CHECKING, Generic
11from .time import format_duration
13if TYPE_CHECKING:
14 import pandas as pd
17log = getLogger(__name__)
19LOG_DEFAULT_FORMAT = '%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s'
20T = TypeVar("T")
21THandler = TypeVar("THandler", bound=Handler)
23# Holds the log format that is configured by the user (using function `configure`), such
24# that it can be reused in other places
25_logFormat = LOG_DEFAULT_FORMAT
27# User-configured callback which is called after logging is configured via function `configure`
28_configureCallbacks: List[Callable[[], None]] = []
31def remove_log_handlers():
32 """
33 Removes all current log handlers
34 """
35 logger = getLogger()
36 while logger.hasHandlers():
37 logger.removeHandler(logger.handlers[0])
40def remove_log_handler(handler):
41 getLogger().removeHandler(handler)
44def is_log_handler_active(handler):
45 return handler in getLogger().handlers
48def is_enabled() -> bool:
49 """
50 :return: True if logging is enabled (at least one log handler exists)
51 """
52 return getLogger().hasHandlers()
55def set_configure_callback(callback: Callable[[], None], append: bool = True) -> None:
56 """
57 Configures a function to be called when logging is configured, e.g. through :func:`configure, :func:`run_main` or
58 :func:`run_cli`.
59 A typical use for the callback is to configure the logging behaviour of packages, setting appropriate log levels.
61 :param callback: the function to call
62 :param append: whether to append to the list of callbacks; if False, any existing callbacks will be removed
63 """
64 global _configureCallbacks
65 if not append:
66 _configureCallbacks = []
67 _configureCallbacks.append(callback)
70# noinspection PyShadowingBuiltins
71def configure(format=LOG_DEFAULT_FORMAT, level=lg.DEBUG):
72 """
73 Configures logging to stdout with the given format and log level,
74 also configuring the default log levels of some overly verbose libraries as well as some pandas output options.
76 :param format: the log format
77 :param level: the minimum log level
78 """
79 global _logFormat
80 _logFormat = format
81 remove_log_handlers()
82 basicConfig(level=level, format=format, stream=sys.stdout)
83 getLogger("matplotlib").setLevel(lg.INFO)
84 getLogger("urllib3").setLevel(lg.INFO)
85 getLogger("msal").setLevel(lg.INFO)
86 try:
87 import pandas as pd
88 pd.set_option('display.max_colwidth', 255)
89 except ImportError:
90 pass
91 for callback in _configureCallbacks:
92 callback()
95# noinspection PyShadowingBuiltins
96def run_main(main_fn: Callable[..., T], format=LOG_DEFAULT_FORMAT, level=lg.DEBUG) -> T:
97 """
98 Configures logging with the given parameters, ensuring that any exceptions that occur during
99 the execution of the given function are logged.
100 Logs two additional messages, one before the execution of the function, and one upon its completion.
102 :param main_fn: the function to be executed
103 :param format: the log message format
104 :param level: the minimum log level
105 :return: the result of `main_fn`
106 """
107 configure(format=format, level=level)
108 log.info("Starting")
109 try:
110 result = main_fn()
111 log.info("Done")
112 return result
113 except Exception as e:
114 log.error("Exception during script execution", exc_info=e)
117# noinspection PyShadowingBuiltins
118def run_cli(main_fn: Callable[..., T], format: str = LOG_DEFAULT_FORMAT, level: int = lg.DEBUG) -> Optional[T]:
119 """
120 Configures logging with the given parameters and runs the given main function as a
121 CLI using `jsonargparse` (which is configured to also parse attribute docstrings, such
122 that dataclasses can be used as function arguments).
123 Using this function requires that `jsonargparse` and `docstring_parser` be available.
124 Like `run_main`, two additional log messages will be logged (at the beginning and end
125 of the execution), and it is ensured that all exceptions will be logged.
127 :param main_fn: the function to be executed
128 :param format: the log message format
129 :param level: the minimum log level
130 :return: the result of `main_fn`
131 """
132 from jsonargparse import set_docstring_parse_options, CLI
134 set_docstring_parse_options(attribute_docstrings=True)
135 return run_main(lambda: CLI(main_fn), format=format, level=level)
138def datetime_tag() -> str:
139 """
140 :return: a string tag for use in log file names which contains the current date and time (compact but readable)
141 """
142 return datetime.now().strftime('%Y%m%d-%H%M%S')
145_fileLoggerPaths: List[str] = []
146_isAtExitReportFileLoggerRegistered = False
149def _at_exit_report_file_logger():
150 for path in _fileLoggerPaths:
151 print(f"A log file was saved to {path}")
154def add_file_logger(path, append=True, register_atexit=True) -> FileHandler:
155 global _isAtExitReportFileLoggerRegistered
156 log.info(f"Logging to {path} ...")
157 mode = "a" if append else "w"
158 handler = FileHandler(path, mode=mode)
159 handler.setFormatter(Formatter(_logFormat))
160 Logger.root.addHandler(handler)
161 _fileLoggerPaths.append(path)
162 if not _isAtExitReportFileLoggerRegistered and register_atexit:
163 atexit.register(_at_exit_report_file_logger)
164 _isAtExitReportFileLoggerRegistered = True
165 return handler
168class MemoryStreamHandler(StreamHandler):
169 def __init__(self, stream: StringIO):
170 super().__init__(stream)
172 def get_log(self) -> str:
173 stream: StringIO = self.stream
174 return stream.getvalue()
177def add_memory_logger() -> MemoryStreamHandler:
178 """
179 Adds an in-memory logger, i.e. all log statements are written to a memory buffer which can be retrieved
180 using the handler's `get_log` method.
181 """
182 handler = MemoryStreamHandler(StringIO())
183 handler.setFormatter(Formatter(_logFormat))
184 Logger.root.addHandler(handler)
185 return handler
188class StopWatch:
189 """
190 Represents a stop watch for timing an execution. Constructing an instance starts the stopwatch.
191 """
192 def __init__(self, start=True):
193 self.start_time = time.time()
194 self._elapsed_secs = 0.0
195 self.is_running = start
197 def reset(self, start=True):
198 """
199 Resets the stopwatch, setting the elapsed time to zero.
201 :param start: whether to start the stopwatch immediately
202 """
203 self.start_time = time.time()
204 self._elapsed_secs = 0.0
205 self.is_running = start
207 def restart(self):
208 """
209 Resets the stopwatch (setting the elapsed time to zero) and restarts it immediately
210 """
211 self.reset(start=True)
213 def _get_elapsed_time_since_last_start(self):
214 if self.is_running:
215 return time.time() - self.start_time
216 else:
217 return 0
219 def pause(self):
220 """
221 Pauses the stopwatch. It can be resumed via method 'resume'.
222 """
223 self._elapsed_secs += self._get_elapsed_time_since_last_start()
224 self.is_running = False
226 def resume(self):
227 """
228 Resumes the stopwatch (assuming that it is currently paused). If the stopwatch is not paused,
229 the method has no effect (and a warning is logged).
230 """
231 if not self.is_running:
232 self.start_time = time.time()
233 self.is_running = True
234 else:
235 log.warning("Stopwatch is already running (resume has not effect)")
237 def get_elapsed_time_secs(self) -> float:
238 """
239 Gets the total elapsed time, in seconds, on this stopwatch.
241 :return: the elapsed time in seconds
242 """
243 return self._elapsed_secs + self._get_elapsed_time_since_last_start()
245 def get_elapsed_timedelta(self) -> "pd.Timedelta":
246 """
247 :return: the elapsed time as a pandas.Timedelta object
248 """
249 return pd.Timedelta(self.get_elapsed_time_secs(), unit="s")
251 def get_elapsed_time_string(self) -> str:
252 """
253 :return: a string representation of the elapsed time
254 """
255 secs = self.get_elapsed_time_secs()
256 if secs < 60:
257 return f"{secs:.3f} seconds"
258 else:
259 try:
260 import pandas as pd
261 return str(pd.Timedelta(secs, unit="s"))
262 except ImportError:
263 return format_duration(secs)
266class StopWatchManager:
267 """
268 A singleton which manages a pool of named stopwatches, such that executions to be timed by referring to a name only -
269 without the need for a limited scope.
270 """
271 _instance = None
273 @classmethod
274 def get_instance(cls):
275 if cls._instance is None:
276 cls._instance = StopWatchManager(42)
277 return cls._instance
279 def __init__(self, secret):
280 if secret != 42:
281 raise Exception("Use only the singleton instance via getInstance")
282 self._stopWatches = {}
284 def start(self, name):
285 self._stopWatches[name] = time.time()
287 def stop(self, name) -> float:
288 """
289 :param name: the name of the stopwatch
290 :return: the time that has passed in seconds
291 """
292 time_passed_secs = time.time() - self._stopWatches[name]
293 del self._stopWatches[name]
294 return time_passed_secs
296 def is_running(self, name):
297 return name in self._stopWatches
300class LogTime:
301 """
302 An execution time logger which can be conveniently applied using a with-statement - in order to log the executing time of the respective
303 with-block.
304 """
306 def __init__(self, name, enabled=True, logger: Logger = None):
307 """
308 :param name: the name of the event whose time is to be logged upon completion as "<name> completed in <time>"
309 :param enabled: whether the logging is actually enabled; can be set to False to disable logging without necessitating
310 changes to client code
311 :param logger: the logger to use; if None, use the logger of LogTime's module
312 """
313 self.name = name
314 self.enabled = enabled
315 self.stopwatch = None
316 self.logger = logger if logger is not None else log
318 def start(self):
319 """
320 Starts the stopwatch
321 """
322 self.stopwatch = StopWatch()
323 if self.enabled:
324 self.logger.info(f"{self.name} starting ...")
326 def stop(self):
327 """
328 Stops the stopwatch and logs the time taken (if enabled)
329 """
330 if self.stopwatch is not None and self.enabled:
331 self.logger.info(f"{self.name} completed in {self.stopwatch.get_elapsed_time_string()}")
333 def __exit__(self, exc_type, exc_value, traceback):
334 self.stop()
336 def __enter__(self):
337 self.start()
338 return self
341class LoggerContext(Generic[THandler], ABC):
342 """
343 Base class for context handlers to be used in conjunction with Python's `with` statement.
344 """
346 def __init__(self, enabled=True):
347 """
348 :param enabled: whether to actually perform any logging.
349 This switch allows the with statement to be applied regardless of whether logging shall be enabled.
350 """
351 self.enabled = enabled
352 self._log_handler = None
354 @abstractmethod
355 def _create_log_handler(self) -> THandler:
356 pass
358 def __enter__(self) -> Optional[THandler]:
359 if self.enabled:
360 self._log_handler = self._create_log_handler()
361 return self._log_handler
363 def __exit__(self, exc_type, exc_value, traceback):
364 if self._log_handler is not None:
365 remove_log_handler(self._log_handler)
368class FileLoggerContext(LoggerContext[FileHandler]):
369 """
370 A context handler to be used in conjunction with Python's `with` statement which enables file-based logging.
371 """
373 def __init__(self, path: str, append=True, enabled=True):
374 """
375 :param path: the path to the log file
376 :param append: whether to append in case the file already exists; if False, always create a new file.
377 :param enabled: whether to actually perform any logging.
378 This switch allows the with statement to be applied regardless of whether logging shall be enabled.
379 """
380 self.path = path
381 self.append = append
382 super().__init__(enabled=enabled)
384 def _create_log_handler(self) -> FileHandler:
385 return add_file_logger(self.path, append=self.append, register_atexit=False)
388class MemoryLoggerContext(LoggerContext[MemoryStreamHandler]):
389 """
390 A context handler to be used in conjunction with Python's `with` statement which enables in-memory logging.
391 """
393 def _create_log_handler(self) -> MemoryStreamHandler:
394 return add_memory_logger()
397class LoggingDisabledContext:
398 """
399 A context manager that will temporarily disable logging
400 """
401 def __init__(self, highest_level=CRITICAL):
402 """
403 :param highest_level: the highest level to disable
404 """
405 self._highest_level = highest_level
406 self._previous_level = None
408 def __enter__(self):
409 self._previous_level = lg.root.manager.disable
410 lg.disable(self._highest_level)
412 def __exit__(self, exc_type, exc_value, traceback):
413 lg.disable(self._previous_level)
416# noinspection PyShadowingBuiltins
417class FallbackHandler(Handler):
418 """
419 A log handler which applies the given handler only if the root logger has no defined handlers (or no handlers other than
420 this fallback handler itself)
421 """
422 def __init__(self, handler: Handler, level=NOTSET):
423 """
424 :param handler: the handler to which messages shall be passed on
425 :param level: the minimum level to output; if NOTSET, pass on everything
426 """
427 super().__init__(level=level)
428 self._handler = handler
430 def emit(self, record):
431 root_handlers = getLogger().handlers
432 if len(root_handlers) == 0 or (len(root_handlers) == 1 and root_handlers[0] == self):
433 self._handler.emit(record)