Source code for thetis.log

"""
Loggers for Thetis

Creates two logger instances, one for general model output and one for debug,
warning, error etc. messages.

To print to the model output stream, use :func:`~.print_output`.

Debug, warning etc. messages are issued with :func:`~.debug`, :func:`~.info`,
:func:`~.warning`, :func:`~.error`, :func:`~.critical` methods.
"""
from thetis.utility import COMM_WORLD
import logging
from logging import DEBUG, INFO, WARNING, ERROR, CRITICAL
import os
import io

__all__ = ('logger', 'output_logger',
           'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL',
           'log', 'debug', 'info', 'warning', 'error', 'critical',
           'print_output', 'set_thetis_loggers', 'thetis_log_level',
           'set_log_directory')

logger_format = {
    'thetis': '%(name)s:%(levelname)s %(message)s',
    'thetis_output': '%(message)s',
}


class ThetisLogConfig:
    """Module-wide config object"""
    filename = None
    mem_buffer = None


thetis_log_config = ThetisLogConfig()


class BufferHandler(logging.StreamHandler):
    pass


[docs] def set_thetis_loggers(comm=COMM_WORLD): """Set stream handlers for log messages. :kwarg comm: The communicator the handler should be collective over. Only rank-0 on that communicator will write to the log, other ranks will use a :class:`logging.NullHandler`. """ def add_stream_handler(buffered=False): if comm is None or comm.rank == 0: if buffered: handler = BufferHandler(thetis_log_config.mem_buffer) else: handler = logging.StreamHandler() handler.setFormatter(logging.Formatter(fmt=fmt)) else: handler = logging.NullHandler() logger.addHandler(handler) if thetis_log_config.mem_buffer is None: thetis_log_config.mem_buffer = io.StringIO() for name, fmt in logger_format.items(): logger = logging.getLogger(name) for handler in logger.handlers: if isinstance(handler, logging.StreamHandler): logger.removeHandler(handler) add_stream_handler() add_stream_handler(buffered=True)
[docs] def set_log_directory(output_directory, comm=COMM_WORLD, mode='w'): """ Forward all log output to `output_directory/log` file. When called, a new empty log file is created. If called twice with a different `output_directory`, a warning is raised, and the new log file location is assigned. The old log file or the `output_directory` are not removed. :arg output_directory: the directory where log file is stored :kwarg comm: The communicator the handler should be collective over. Only rank-0 on that communicator will write to the log, other ranks will use a :class:`logging.NullHandler`. :kwarg mode: write mode, 'w' removes previous log file (if any), otherwise appends to it. Default: 'w'. """ def create_directory(dir): if comm.rank == 0: if os.path.exists(dir): if not os.path.isdir(dir): raise IOError('file with same name exists', dir) else: os.makedirs(dir) def rm_handlers(cls=logging.FileHandler): for name in logger_format: logger = logging.getLogger(name) for handler in logger.handlers: if isinstance(handler, cls): logger.removeHandler(handler) def rm_file_handlers(): rm_handlers(cls=logging.FileHandler) def rm_buf_handlers(): rm_handlers(cls=BufferHandler) def assign_file_handler(logfile): for name, fmt in logger_format.items(): logger = logging.getLogger(name) if comm.rank == 0: new_handler = logging.FileHandler(logfile, mode='a') new_handler.setFormatter(logging.Formatter(fmt)) else: new_handler = logging.NullHandler() logger.addHandler(new_handler) logfile = os.path.join(output_directory, 'log') if thetis_log_config.filename == logfile: # no change return different_file = thetis_log_config.filename is not None if different_file: old_file = str(thetis_log_config.filename) rm_file_handlers() if mode == 'w' and os.path.isfile(logfile): # silently remove previous log if comm.rank == 0: os.remove(logfile) create_directory(output_directory) if comm.rank == 0: buffer_content = thetis_log_config.mem_buffer.getvalue() with open(logfile, 'w') as f: f.write(buffer_content) rm_buf_handlers() thetis_log_config.filename = logfile assign_file_handler(logfile) if different_file: msg = (f'Setting a log file "{logfile}" that differs from previous ' f'"{old_file}", removing old handler') warning(msg) # to new log
[docs] def thetis_log_level(level): """Set the log level for Thetis logger. This controls what level of logging messages are printed to stderr. The higher the level, the fewer the number of messages. :arg level: The level to use, one of 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'. """ logger = logging.getLogger('thetis') logger.setLevel(level)
# logger for error, warning etc messages logger = logging.getLogger('thetis') log = logger.log debug = logger.debug info = logger.info warning = logger.warning error = logger.error critical = logger.critical # logger for model output messages with no prefix, used with print_output output_logger = logging.getLogger('thetis_output') output_logger.setLevel(INFO) print_output = output_logger.info