Source code for openmdao.utils.logger_utils

"""Miscellaneous utilities related to logging."""

import sys
import logging
from fnmatch import fnmatch


# If any method that creates handlers is called twice (e.g., setup reconfigure or during tests),
# then we need to prevent another one from being created. Since we have multiple loggers now, we
# store them in a dictionary.
_loggers = {}


def _set_handler(logger, handler, level, use_format):
    """
    Set the StreamHandler for logger.

    Parameters
    ----------
    logger : object
        Logger object.
    handler : logging handler
        handler to add to the logger
    level : int
        Logging level for this logger. Default is logging.INFO (level 20).
    use_format : bool
        Set to True to use the openmdao format "Level: message".
    """
    # set a format which is simpler for console use
    if use_format:
        formatter = logging.Formatter('%(levelname)s: %(message)s')
        handler.setFormatter(formatter)

    handler.setLevel(level)
    logger.addHandler(handler)


[docs]def get_logger(name='default_logger', level=logging.INFO, use_format=False, out_stream='stdout', out_file=None, lock=None): """ Return a logger that writes to an I/O stream. Parameters ---------- name : str Name of the logger to be returned, will be created if it doesn't exist. level : int Logging level for this logger. Default is logging.INFO (level 20). (applied only when creating a new logger or setting a new stream). use_format : bool Set to True to use the openmdao format "Level: message". (applied only when creating a new logger or setting a new stream). out_stream : 'stdout', 'stderr' or file-like Output stream to which logger output will be directed. out_file : str or None If not None, add a FileHandler to write to this file. lock : bool If True, do not allow the handler to be changed until unlocked. If False, unlock the handler for the logger. Returns ------- <logging.Logger> Logger that writes to a stream and adheres to requested settings. """ if out_stream == 'stdout': out_stream = sys.stdout elif out_stream == 'stderr': out_stream = sys.stderr if name in _loggers: # use existing logger info = _loggers[name] logger = info['logger'] stream = info['stream'] ofile = info['file'] locked = info['locked'] unlock = lock is False # redirect log to new stream (if not locked) if (out_stream != stream or ofile != out_file) and (not locked or unlock): for handler in logger.handlers: logger.removeHandler(handler) if out_stream: _set_handler(logger, logging.StreamHandler(out_stream), level, use_format) if out_file: _set_handler(logger, logging.FileHandler(out_file, mode='w'), level, use_format) info['stream'] = out_stream info['file'] = out_file # update locked status info['locked'] = lock else: # create new logger logger = logging.getLogger(name) logger.setLevel(level) if out_stream: _set_handler(logger, logging.StreamHandler(out_stream), level, use_format) if out_file: _set_handler(logger, logging.FileHandler(out_file, mode='w'), level, use_format) _loggers[name] = { 'logger': logger, 'stream': out_stream, 'file': out_file, 'locked': lock } return logger
[docs]class TestLogger(object): """ A logger that saves messages in lists. Attributes ---------- _msgs : dict Stores lists of messages under 'error', 'warning', and 'info' keys. """
[docs] def __init__(self): """ Initialize the message dict. """ self._msgs = {'error': [], 'warning': [], 'info': []}
[docs] def error(self, msg): """ Collect an error message. Parameters ---------- msg : str An error message. """ self._msgs['error'].append(msg)
[docs] def warning(self, msg): """ Collect a warning message. Parameters ---------- msg : str A warning message. """ self._msgs['warning'].append(msg)
[docs] def info(self, msg): """ Collect an informational message. Parameters ---------- msg : str An informational message. """ self._msgs['info'].append(msg)
[docs] def get(self, typ): """ Return all stored messages of a specific type. Parameters ---------- typ : str Type of messages ('error', 'warning', 'info') to be returned. Returns ------- list of str Any messages of that type that have been written to the logger. """ return self._msgs[typ]
[docs] def contains(self, typ, message): """ Do any of the stored messages of a specific type equal the given message. Parameters ---------- typ : str Type of messages ('error', 'warning', 'info') to be returned. message : str The message to match. Returns ------- bool True if any of the lines of stored messages of a specific type equal the line. """ for s in self._msgs[typ]: if s == message: return True return False
[docs] def find_in(self, typ, message): """ Find the given message among the stored messages. Raises an exception if the given message isn't found. Parameters ---------- typ : str Type of messages ('error', 'warning', 'info') to be searched. message : str The message to match. """ if not self.contains(typ, message): msgs = ',\n'.join(self._msgs[typ]) raise RuntimeError(f'Message "{message}" not found in:\n{msgs}.')
[docs] def find_match_in(self, typ, pattern): """ Find a message with the given pattern among the stored messages. Raises an exception if a message with the given pattern isn't found. Parameters ---------- typ : str Type of messages ('error', 'warning', 'info') to be searched. pattern : str The pattern to match. """ for msg in self._msgs[typ]: if fnmatch(msg, pattern): return msgs = ',\n'.join(self._msgs[typ]) raise RuntimeError(f'Message with pattern "{pattern}" not found in:\n{msgs}')