# Licensed under a 3-clause BSD style license - see LICENSE.rst
from __future__ import absolute_import, division, print_function
import os
import sys
import logging
import logging.config
import yaml
import fermipy
[docs]def log_level(level):
"""This is a function that returns a python like
level from a HEASOFT like level.
"""
levels_dict = {0: 50,
1: 40,
2: 30,
3: 20,
4: 10}
if not isinstance(level, int):
level = int(level)
if level > 4:
level = 4
return levels_dict[level]
[docs]class Logger(object):
"""This class provides helper functions which facilitate creating
instances of the built-in logger class."""
[docs] @staticmethod
def setup(config=None, logfile=None):
"""This method sets up the default configuration of the
logger. Once this method is called all subsequent instances
Logger instances will inherit this configuration."""
if config is None:
configpath = os.path.join(fermipy.PACKAGE_ROOT, 'config',
'logging.yaml')
with open(configpath, 'r') as f:
config = yaml.load(f)
# Update configuration
if logfile:
for name, h in config['handlers'].items():
if 'file_handler' in name:
config['handlers'][name]['filename'] = logfile
logging.config.dictConfig(config)
# get the root logger
# logger = logging.getLogger()
# for h in logger.handlers:
# if 'file_handler' in h.name:
# print(h.name)
[docs]class StreamLogger(object):
"""File-like object to log stdout/stderr using the `logging` module."""
def __init__(self, name='stdout', logfile=None, quiet=True):
self.logger = logging.getLogger(name)
self.logger.setLevel(logging.DEBUG)
self.format = ('%(asctime)s - %(name)s - %(funcName)s - '
'%(levelname)s - %(message)s')
self.datefmt = '%Y-%m-%d %H:%M:%S'
self.stdout = sys.stdout
# logging.basicConfig(level=logging.DEBUG,
# format=self.format, filename=file)
self.formatter = logging.Formatter(self.format, self.datefmt)
fhdlr = logging.FileHandler(logfile)
fhdlr.setFormatter(self.formatter)
self.logger.addHandler(fhdlr)
if not quiet and name == 'stdout':
shdlr = logging.StreamHandler(sys.stdout)
shdlr.setFormatter(self.formatter)
self.logger.addHandler(shdlr)
elif not quiet and name == 'stderr':
shdlr = logging.StreamHandler(sys.stderr)
shdlr.setFormatter(self.formatter)
self.logger.addHandler(shdlr)
sys.stdout = self
def __del__(self):
self.close()
[docs] def close(self):
self.flush()
sys.stdout = self.stdout
self.logger.handlers = []
[docs] def write(self, msg, level=logging.DEBUG):
msg = msg.rstrip('\n')
if len(msg) > 0:
self.logger.log(level, msg)
[docs] def flush(self):
for handler in self.logger.handlers:
handler.flush()