Source code for spinn_utilities.log

# Copyright (c) 2017-2019 The University of Manchester
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <>.

import atexit
import logging
import re
import sys
from inspect import getfullargspec
from .overrides import overrides

    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL,

[docs]class ConfiguredFilter(object): """ Allow a parent logger to filter a child logger. """ __slots__ = [ "_default_level", "_levels"] def __init__(self, conf): self._levels = ConfiguredFormatter.construct_logging_parents(conf) self._default_level = _LEVELS[conf.get("Logging", "default")]
[docs] def filter(self, record): """ Get the level for the deepest parent, and filter appropriately. """ level = ConfiguredFormatter.level_of_deepest_parent( self._levels, if level is None: return record.levelno >= self._default_level return record.levelno >= level
[docs]class ConfiguredFormatter(logging.Formatter): """ Defines the logging format for the SpiNNaker host software. """ # Precompile this RE; it gets used quite a few times __last_component = re.compile(r'\.[^.]+$') def __init__(self, conf): if conf.get("Logging", "default") == "debug": fmt = "%(asctime)-15s %(levelname)s: %(pathname)s: %(message)s" else: fmt = "%(asctime)-15s %(levelname)s: %(message)s" super().__init__(fmt=fmt, datefmt="%Y-%m-%d %H:%M:%S")
[docs] @staticmethod def construct_logging_parents(conf): """ Create a dictionary of module names and logging levels. """ # Construct the dictionary _levels = {} if not conf.has_section("Logging"): return _levels for label, level in _LEVELS.items(): if conf.has_option("Logging", label): modules = [s.strip() for s in conf.get('Logging', label).split(',')] if '' not in modules: _levels.update(dict((m, level) for m in modules)) return _levels
[docs] @staticmethod def deepest_parent(parents, child): """ Greediest match between child and parent. """ # TODO: this can almost certainly be neater! # Repeatedly strip elements off the child until we match an item in # parents match = child while '.' in match and match not in parents: match = ConfiguredFormatter.__last_component.sub('', match) # If no match then return None, there is no deepest parent if match not in parents: match = None return match
[docs] @staticmethod def level_of_deepest_parent(parents, child): """ The logging level of the greediest match between child and parent. """ # child = re.sub( r'^pacman103\.', '', child ) parent = ConfiguredFormatter.deepest_parent(parents.keys(), child) if parent is None: return None return parents[parent]
class _BraceMessage(object): """ A message that converts a Python format string to a string """ __slots__ = [ "args", "fmt", "kwargs"] def __init__(self, fmt, args, kwargs): self.fmt = fmt self.args = args self.kwargs = kwargs def __str__(self): return str(self.fmt).format(*self.args, **self.kwargs)
[docs]class LogLevelTooHighException(Exception): """ An Exception throw when the System tries to log at a level where an Exception is a better option. """
[docs]class FormatAdapter(logging.LoggerAdapter): """ An adaptor for logging with messages that uses Python format strings. Example:: log = FormatAdapter(logging.getLogger(__name__))"this message has {} inside {}", 123, 'itself') # --> INFO: this message has 123 inside itself """ __kill_level = logging.CRITICAL + 1 __repeat_at_end = logging.WARNING __repeat_messages = [] __write_normal = True __report_file = None
[docs] @classmethod def set_kill_level(cls, level=None): """ Allow system to change the level at which a log is changed to an Exception Static so effects all log messages :param int level: The level to set. The values in :py:mod:`logging` are recommended. """ if level is None: cls.__kill_level = logging.CRITICAL + 1 else: cls.__kill_level = level
[docs] @classmethod def set_report_File(cls, report_file): """ :param report_file: :param write_normal: :return: """ cls.__report_file = report_file level = logging.getLevelName(cls.__repeat_at_end) with open(report_file, "a") as report_file: report_file.write( "This is a record of all logged messages at level {} or " "above\n".format(level))
def __init__(self, logger, extra=None): if extra is None: extra = {} super().__init__(logger, extra) self.do_log = logger._log # pylint: disable=protected-access
[docs] @overrides(logging.LoggerAdapter.log, extend_doc=False) def log(self, level, msg, *args, **kwargs): """ Delegate a log call to the underlying logger, applying appropriate\ transformations to allow the log message to be written using\ Python format string, rather than via `%`-substitutions. """ if level >= FormatAdapter.__kill_level: raise LogLevelTooHighException(_BraceMessage(msg, args, kwargs)) message = _BraceMessage(msg, args, kwargs) if level >= FormatAdapter.__repeat_at_end: FormatAdapter.__repeat_messages.append((message)) if self.__report_file: with open(self.__report_file, "a") as report_file: report_file.write(message.fmt) report_file.write("\n") if self.isEnabledFor(level): msg, log_kwargs = self.process(msg, kwargs) if "exc_info" in kwargs: log_kwargs["exc_info"] = kwargs["exc_info"] self.do_log(level, message, (), **log_kwargs)
[docs] @overrides(logging.LoggerAdapter.process, extend_doc=False) def process(self, msg, kwargs): """ Process the logging message and keyword arguments passed in to a\ logging call to insert contextual information. You can either\ manipulate the message itself, the keyword arguments or both.\ Return the message and *kwargs* modified (or not) to suit your\ needs. """ # pylint: disable=deprecated-method return msg, { key: kwargs[key] for key in getfullargspec(self.do_log).args[1:] if key in kwargs}
@classmethod def _atexit_handler(cls): messages = cls._repeat_log() if messages: level = logging.getLevelName(cls.__repeat_at_end) if cls.__report_file: print("\nWARNING: {} log messages were generated at " "level {} or above.".format(len(messages), level), file=sys.stderr) print("This may mean that the results are invalid.", file=sys.stderr) print("You are advised to check the details of these here: {}" "".format(cls.__report_file), file=sys.stderr) else: print("\nThese log messages where generated at level {} or " "above".format(level), file=sys.stderr) for message in messages: print(message, file=sys.stderr) @classmethod def _repeat_log(cls): """ Returns the log of messages to print on exit and \ *clears that log*. .. note:: Should only be called externally from test code! """ try: return cls.__repeat_messages finally: cls.__repeat_messages = []