import logging
import os
import sys
from logging.handlers import TimedRotatingFileHandler
logger = logging.getLogger(__name__)
SUBMITTED = os.getenv('FSLSUBALREADYRUN', default='') == 'true'
[docs]def get_level():
"""
Returns the requested log level
0: no logging
1: warning+
2: info+
3: debug+
4: all+
"""
levels = (
None,
logging.WARNING,
logging.INFO,
logging.DEBUG,
0,
)
if 'MCLOG' in os.environ:
idx = int(os.environ['MCLOG'])
else:
idx = 3 if SUBMITTED else 2
return levels[idx]
LOGSET = False
[docs]def setup_log(filename=None, replace=True):
"""
Sets up the logger to log to the given filename
Sets up logging to:
- If .log directory exists:
- ~/.log/python.info (INFO and higher)
- ~/.log/python.debug (DEBUG and higher)
- stdout (INFO and higher if debug is False, DEBUG and higher otherwise)
- debug defaults to True for submitted job, False otherwise
- `filename` if provided
:param filename: filename used in the logging (default: only set up logging to stdout/stderr and ~/.log directory)
:param replace: if True replaces any previous logging
"""
global LOGSET
logging.root.setLevel(logging.DEBUG)
if replace:
logger.info('Removing previous logging')
for handler in tuple(logging.root.handlers):
logging.root.removeHandler(handler)
LOGSET = False
if get_level() is None:
return
if filename is not None:
handler = logging.FileHandler(filename, mode='a')
handler.setLevel(logging.DEBUG)
handler.setFormatter(formatter())
logging.root.addHandler(handler)
if not LOGSET:
# sets up stdout and stderr
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setLevel(get_level())
stdout_handler.setFormatter(formatter())
logging.root.addHandler(stdout_handler)
if SUBMITTED:
stderr_handler = logging.StreamHandler(sys.stderr)
stderr_handler.setLevel(logging.WARNING)
stderr_handler.setFormatter(formatter())
logging.root.addHandler(stderr_handler)
log_directory = os.path.expanduser('~/.log')
if os.path.isdir(log_directory):
try:
main_info = TimedRotatingFileHandler(os.path.join(log_directory, 'python.info'), when='midnight')
main_info.setLevel(logging.INFO)
main_info.setFormatter(formatter())
logging.root.addHandler(main_info)
except OSError:
print(f"Failed to open {os.path.join(log_directory, 'python.info')}")
try:
main_debug = TimedRotatingFileHandler(os.path.join(log_directory, 'python.debug'), when='midnight')
main_debug.setLevel(logging.DEBUG)
main_debug.setFormatter(formatter())
logging.root.addHandler(main_debug)
except OSError:
print(f"Failed to open {os.path.join(log_directory, 'python.info')}")
else:
logger.info('Log directory %s not found; Skipping setup of global logging', log_directory)
LOGSET = True
logger.info('Initial logging setup to %s complete', filename)
else:
logger.info('Additional logging to %s setup', filename)
[docs]def log_function(verbose=True, include_parameters=False, include_result=False):
"""Log the time spent in a function.
When verbose: log both entering and leaving of function, otherwise only when leaving function.
"""
def towrap(func):
name = getattr(func, '__module___', repr(func))
logger = getLogger(name)
@wraps(func)
def wrapped(*args, **kwargs):
if verbose:
if include_parameters:
logger.info('Calling %s(%s, %s)' % (func.__name__, str(args)[1:-1], ', '.join(['%s = %s' % item for item in kwargs.items()])))
else:
logger.info('Calling %s' % func)
time_start = time.time()
result = func(*args, **kwargs)
total_time = time.time() - time_start
msg = 'Done %s%s%s in %.3f sec' % (("%s = " % result if include_result else ""), func.__name__, ("(%s, %s)" % (str(args)[1:-1], ', '.join(['%s = %s' % item for item in kwargs.items()])) if include_parameters else ""), total_time)
logger.info(msg)
return result
return wrapped
return towrap