Source code for rta_to_lstchain_formator.utils.rta_logging

import logging
import sys
from enum import StrEnum
from pathlib import Path

# Enum class of logging levels, usefull to constraint the variable type in configuration json schema.
LOGGING_LEVELS = StrEnum("LOGGING_LEVELS", {v: v for v in logging.getLevelNamesMapping().keys()})


[docs] def log_uncaught_exceptions(): """Makes all uncaught exception to be logged by the default logger. Keyboard exceptions and children classes are not logged so one can kill the program with ctr+C. """ def handle_exception(exc_type, exc_value, exc_traceback): if not issubclass(exc_type, KeyboardInterrupt): logging.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback)) sys.__excepthook__(exc_type, exc_value, exc_traceback) return sys.excepthook = handle_exception
[docs] def init_logging(log_header_str: str, log_filename: Path | None, log_level: LOGGING_LEVELS): """(Re-)initialize all loggers""" logging.captureWarnings(True) # log all warnings from the warnings module. log_uncaught_exceptions() # log all uncaught exceptions as well logging_format = "%(asctime)s %(levelname)s {} %(pathname)s:%(lineno)s:%(funcName)s %(message)s".format( log_header_str ) handlers = [] if log_filename is not None: # write log to file handlers.append(logging.FileHandler(log_filename)) else: # write log to standard output and error # WARNING: this could cause freezing due to output pipes been full if this is called as a subprocess ! handlers.append(logging.StreamHandler()) logging.basicConfig( level=log_level.name, format=logging_format, handlers=handlers, force=True, ) logging.info("Logging configured - start logging")