Source code for brian2.utils.logger

"""
Brian's logging module.

Preferences
-----------
.. document_brian_prefs:: logging
"""

import atexit
import logging
import logging.handlers
import os
import shutil
import sys
import tempfile
import time
from logging.handlers import RotatingFileHandler
from warnings import warn

import numpy

try:
    import scipy
except ImportError:
    scipy = None
import sympy

import brian2
from brian2.core.preferences import BrianPreference, prefs

from .environment import running_from_ipython

__all__ = ["get_logger", "BrianLogger", "std_silent"]

# ===============================================================================
# Logging preferences
# ===============================================================================


[docs] def log_level_validator(log_level): log_levels = ("CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "DIAGNOSTIC") return log_level.upper() in log_levels
#: Our new log level for more detailed debug output (mostly useful for debugging #: Brian itself, not for user scripts) DIAGNOSTIC = 5 #: Translation from string representation to number LOG_LEVELS = { "CRITICAL": logging.CRITICAL, "ERROR": logging.ERROR, "WARNING": logging.WARNING, "INFO": logging.INFO, "DEBUG": logging.DEBUG, "DIAGNOSTIC": DIAGNOSTIC, } logging.addLevelName(DIAGNOSTIC, "DIAGNOSTIC") if "logging" not in prefs.pref_register: # Duplicate import of this module can happen when the documentation is built prefs.register_preferences( "logging", "Logging system preferences", delete_log_on_exit=BrianPreference( default=True, docs=""" Whether to delete the log and script file on exit. If set to ``True`` (the default), log files (and the copy of the main script) will be deleted after the brian process has exited, unless an uncaught exception occurred. If set to ``False``, all log files will be kept. """, ), file_log_level=BrianPreference( default="DEBUG", docs=""" What log level to use for the log written to the log file. In case file logging is activated (see `logging.file_log`), which log level should be used for logging. Has to be one of CRITICAL, ERROR, WARNING, INFO, DEBUG or DIAGNOSTIC. """, validator=log_level_validator, ), console_log_level=BrianPreference( default="INFO", docs=""" What log level to use for the log written to the console. Has to be one of CRITICAL, ERROR, WARNING, INFO, DEBUG or DIAGNOSTIC. """, validator=log_level_validator, ), file_log=BrianPreference( default=True, docs=""" Whether to log to a file or not. If set to ``True`` (the default), logging information will be written to a file. The log level can be set via the `logging.file_log_level` preference. """, ), file_log_max_size=BrianPreference( default=10000000, docs=""" The maximum size for the debug log before it will be rotated. If set to any value ``> 0``, the debug log will be rotated once this size is reached. Rotating the log means that the old debug log will be moved into a file in the same directory but with suffix ``".1"`` and the a new log file will be created with the same pathname as the original file. Only one backup is kept; if a file with suffix ``".1"`` already exists when rotating, it will be overwritten. If set to ``0``, no log rotation will be applied. The default setting rotates the log file after 10MB. """, ), save_script=BrianPreference( default=True, docs=""" Whether to save a copy of the script that is run. If set to ``True`` (the default), a copy of the currently run script is saved to a temporary location. It is deleted after a successful run (unless `logging.delete_log_on_exit` is ``False``) but is kept after an uncaught exception occured. This can be helpful for debugging, in particular when several simulations are running in parallel. """, ), std_redirection=BrianPreference( default=True, docs=""" Whether or not to redirect stdout/stderr to null at certain places. This silences a lot of annoying compiler output, but will also hide error messages making it harder to debug problems. You can always temporarily switch it off when debugging. If `logging.std_redirection_to_file` is set to ``True`` as well, then the output is saved to a file and if an error occurs the name of this file will be printed. """, ), std_redirection_to_file=BrianPreference( default=True, docs=""" Whether to redirect stdout/stderr to a file. If both ``logging.std_redirection`` and this preference are set to ``True``, all standard output/error (most importantly output from the compiler) will be stored in files and if an error occurs the name of this file will be printed. If `logging.std_redirection` is ``True`` and this preference is ``False``, then all standard output/error will be completely suppressed, i.e. neither be displayed nor stored in a file. The value of this preference is ignore if `logging.std_redirection` is set to ``False``. """, ), display_brian_error_message=BrianPreference( default=True, docs=""" Whether to display a text for uncaught errors, mentioning the location of the log file, the mailing list and the github issues. Defaults to ``True``.""", ), ) # =============================================================================== # Initial setup # =============================================================================== def _encode(text): """Small helper function to encode unicode strings as UTF-8.""" return text.encode("UTF-8") UNHANDLED_ERROR_MESSAGE = ( "Brian 2 encountered an unexpected error. " "If you think this is a bug in Brian 2, please report this issue either to the " "discourse forum at <http://brian.discourse.group/>, " "or to the issue tracker at <https://github.com/brian-team/brian2/issues>." )
[docs] def brian_excepthook(exc_type, exc_obj, exc_tb): """ Display a message mentioning the debug log in case of an uncaught exception. """ # Do not catch Ctrl+C if exc_type == KeyboardInterrupt: return logger = logging.getLogger("brian2") BrianLogger.exception_occured = True if not prefs["logging.display_brian_error_message"]: # Put the exception message in the log file, but do not log to the # console if BrianLogger.console_handler is not None: logger.removeHandler(BrianLogger.console_handler) logger.exception("An exception occured", exc_info=(exc_type, exc_obj, exc_tb)) if BrianLogger.console_handler is not None: logger.addHandler(BrianLogger.console_handler) # Run the default except hook return sys.__excepthook__(exc_type, exc_obj, exc_tb) message = UNHANDLED_ERROR_MESSAGE if BrianLogger.tmp_log is not None: message += ( " Please include this file with debug information in your " f"report: {BrianLogger.tmp_log} " ) if BrianLogger.tmp_script is not None: message += ( " Additionally, you can also include a copy " "of the script that was run, available " f"at: {BrianLogger.tmp_script}" ) if hasattr(std_silent, "dest_fname_stdout"): stdout = std_silent.dest_fname_stdout stderr = std_silent.dest_fname_stderr message += ( " You can also include a copy of the " "redirected std stream outputs, available at " f"'{stdout}' and '{stderr}'." ) message += " Thanks!" # very important :) logger.error(message, exc_info=(exc_type, exc_obj, exc_tb))
[docs] def clean_up_logging(): """ Shutdown the logging system and delete the debug log file if no error occured. """ logging.shutdown() if not BrianLogger.exception_occured and prefs["logging.delete_log_on_exit"]: if BrianLogger.tmp_log is not None: try: os.remove(BrianLogger.tmp_log) except OSError as exc: warn(f"Could not delete log file: {exc}") # Remove log files that have been rotated (currently only one) rotated_log = f"{BrianLogger.tmp_log}.1" if os.path.exists(rotated_log): try: os.remove(rotated_log) except OSError as exc: warn(f"Could not delete log file: {exc}") if BrianLogger.tmp_script is not None: try: os.remove(BrianLogger.tmp_script) except OSError as exc: warn(f"Could not delete copy of script file: {exc}") std_silent.close()
atexit.register(clean_up_logging)
[docs] class HierarchyFilter: """ A class for suppressing all log messages in a subtree of the name hierarchy. Does exactly the opposite as the `logging.Filter` class, which allows messages in a certain name hierarchy to *pass*. Parameters ---------- name : str The name hiearchy to suppress. See `BrianLogger.suppress_hierarchy` for details. """ def __init__(self, name): self.orig_filter = logging.Filter(name)
[docs] def filter(self, record): """ Filter out all messages in a subtree of the name hierarchy. """ # do the opposite of what the standard filter class would do return not self.orig_filter.filter(record)
[docs] class NameFilter: """ A class for suppressing log messages ending with a certain name. Parameters ---------- name : str The name to suppress. See `BrianLogger.suppress_name` for details. """ def __init__(self, name): self.name = name
[docs] def filter(self, record): """ Filter out all messages ending with a certain name. """ # The last part of the name record_name = record.name.split(".")[-1] return self.name != record_name
[docs] class RemoveBrian2Filter(logging.Filter): """ A class for removing the ``brian2`` prefix from log messages. Will be used for extension packages that use the Brian logging system. """
[docs] def filter(self, record): assert record.name.startswith("brian2."), record.name record.name = record.name[7:] return True
[docs] class BrianLogger: """ Convenience object for logging. Call `get_logger` to get an instance of this class. Parameters ---------- name : str The name used for logging, normally the name of the module. If the name does not start with ``brian2.``, it will be prepended automatically when interacting with the `logging` module. This means that from the logging's system point of view, it will use the configuration for the logger in the ``brian2`` hierachy. However, when displaying the name in the log messages, the ``brian2.`` prefix will be removed. This is useful for extension modules, which can use the Brian logging system, but will be displayed as ``myextension`` instead of ``brian2.myextension``. This is also used in Brian's test suite, which only considers log messages starting with ``brian2``. """ #: Class attribute to remember whether any exception occured exception_occured = False #: Class attribute for remembering log messages that should only be #: displayed once _log_messages = set() #: The name of the temporary log file (by default deleted after the run if #: no exception occurred), if any tmp_log = None #: The `logging.FileHandler` responsible for logging to the temporary log #: file file_handler = None #: The `logging.StreamHandler` responsible for logging to the console console_handler = None #: The name of the temporary copy of the main script file (by default #: deleted after the run if no exception occurred), if any tmp_script = None #: The pid of the process that initialized the logger – used to switch off file logging in #: multiprocessing contexts _pid = None def __init__(self, name): if not name.startswith("brian2."): self.name = "brian2." + name self.filter_name = True else: self.name = name self.filter_name = False def _log(self, log_level, msg, name_suffix, once): """ Log an entry. Parameters ---------- log_level : {'debug', 'info', 'warn', 'error'} The level with which to log the message. msg : str The log message. name_suffix : str A suffix that will be added to the logger name. once : bool Whether to suppress identical messages if they are logged again. """ name = self.name if name_suffix: name += f".{name_suffix}" # Switch off file logging when using multiprocessing if BrianLogger.tmp_log is not None and BrianLogger._pid != os.getpid(): BrianLogger.tmp_log = None logging.getLogger("brian2").removeHandler(BrianLogger.file_handler) BrianLogger.file_handler = None if once: # Check whether this exact message has already been displayed log_tuple = (name, log_level, msg) if log_tuple in BrianLogger._log_messages: return else: BrianLogger._log_messages.add(log_tuple) the_logger = logging.getLogger(name) if self.filter_name: filter = RemoveBrian2Filter() the_logger.addFilter(filter) the_logger.log(LOG_LEVELS[log_level], msg) if self.filter_name: the_logger.removeFilter(filter)
[docs] def diagnostic(self, msg, name_suffix=None, once=False): """ Log a diagnostic message. Parameters ---------- msg : str The message to log. name_suffix : str, optional A suffix to add to the name, e.g. a class or function name. once : bool, optional Whether this message should be logged only once and not repeated if sent another time. """ self._log("DIAGNOSTIC", msg, name_suffix, once)
[docs] def debug(self, msg, name_suffix=None, once=False): """ Log a debug message. Parameters ---------- msg : str The message to log. name_suffix : str, optional A suffix to add to the name, e.g. a class or function name. once : bool, optional Whether this message should be logged only once and not repeated if sent another time. """ self._log("DEBUG", msg, name_suffix, once)
[docs] def info(self, msg, name_suffix=None, once=False): """ Log an info message. Parameters ---------- msg : str The message to log. name_suffix : str, optional A suffix to add to the name, e.g. a class or function name. once : bool, optional Whether this message should be logged only once and not repeated if sent another time. """ self._log("INFO", msg, name_suffix, once)
[docs] def warn(self, msg, name_suffix=None, once=False): """ Log a warn message. Parameters ---------- msg : str The message to log. name_suffix : str, optional A suffix to add to the name, e.g. a class or function name. once : bool, optional Whether this message should be logged only once and not repeated if sent another time. """ self._log("WARNING", msg, name_suffix, once)
[docs] def error(self, msg, name_suffix=None, once=False): """ Log an error message. Parameters ---------- msg : str The message to log. name_suffix : str, optional A suffix to add to the name, e.g. a class or function name. once : bool, optional Whether this message should be logged only once and not repeated if sent another time. """ self._log("ERROR", msg, name_suffix, once)
@staticmethod def _suppress(filterobj, filter_log_file): """ Apply a filter object to log messages. Parameters ---------- filterobj : `logging.Filter` A filter object to apply to log messages. filter_log_file : bool Whether the filter also applies to log messages in the log file. """ BrianLogger.console_handler.addFilter(filterobj) if filter_log_file: BrianLogger.file_handler.addFilter(filterobj)
[docs] @staticmethod def suppress_hierarchy(name, filter_log_file=False): """ Suppress all log messages in a given hiearchy. Parameters ---------- name : str Suppress all log messages in the given `name` hierarchy. For example, specifying ``'brian2'`` suppresses all messages logged by Brian, specifying ``'brian2.codegen'`` suppresses all messages generated by the code generation modules. filter_log_file : bool, optional Whether to suppress the messages also in the log file. Defaults to ``False`` meaning that suppressed messages are not displayed on the console but are still saved to the log file. """ suppress_filter = HierarchyFilter(name) BrianLogger._suppress(suppress_filter, filter_log_file)
[docs] @staticmethod def suppress_name(name, filter_log_file=False): """ Suppress all log messages with a given name. Parameters ---------- name : str Suppress all log messages ending in the given `name`. For example, specifying ``'resolution_conflict'`` would suppress messages with names such as ``brian2.equations.codestrings.CodeString.resolution_conflict`` or ``brian2.equations.equations.Equations.resolution_conflict``. filter_log_file : bool, optional Whether to suppress the messages also in the log file. Defaults to ``False`` meaning that suppressed messages are not displayed on the console but are still saved to the log file. """ suppress_filter = NameFilter(name) BrianLogger._suppress(suppress_filter, filter_log_file)
[docs] @staticmethod def log_level_diagnostic(): """ Set the log level to "diagnostic". """ BrianLogger.console_handler.setLevel(DIAGNOSTIC)
[docs] @staticmethod def log_level_debug(): """ Set the log level to "debug". """ BrianLogger.console_handler.setLevel(logging.DEBUG)
[docs] @staticmethod def log_level_info(): """ Set the log level to "info". """ BrianLogger.console_handler.setLevel(logging.INFO)
[docs] @staticmethod def log_level_warn(): """ Set the log level to "warn". """ BrianLogger.console_handler.setLevel(logging.WARN)
[docs] @staticmethod def log_level_error(): """ Set the log level to "error". """ BrianLogger.console_handler.setLevel(logging.ERROR)
[docs] @staticmethod def initialize(): """ Initialize Brian's logging system. This function will be called automatically when Brian is imported. """ # get the main logger logger = logging.getLogger("brian2") logger.propagate = False logger.setLevel(LOG_LEVELS["DIAGNOSTIC"]) # Log to a file if prefs["logging.file_log"]: try: # Temporary filename used for logging with tempfile.NamedTemporaryFile( prefix="brian_debug_", suffix=".log", delete=False ) as tmp_f: BrianLogger.tmp_log = tmp_f.name # Remove any previously existing file handler if BrianLogger.file_handler is not None: BrianLogger.file_handler.close() logger.removeHandler(BrianLogger.file_handler) # Rotate log file after prefs['logging.file_log_max_size'] bytes and keep one copy BrianLogger.file_handler = RotatingFileHandler( BrianLogger.tmp_log, mode="a", maxBytes=prefs["logging.file_log_max_size"], backupCount=1, encoding="utf-8", ) BrianLogger.file_handler.setLevel( LOG_LEVELS[prefs["logging.file_log_level"].upper()] ) BrianLogger.file_handler.setFormatter( logging.Formatter( "%(asctime)s %(levelname)-10s %(name)s: %(message)s" ) ) logger.addHandler(BrianLogger.file_handler) BrianLogger._pid = os.getpid() except OSError as ex: warn(f"Could not create log file: {ex}") # Save a copy of the script BrianLogger.tmp_script = None if prefs["logging.save_script"]: if ( len(sys.argv[0]) and not running_from_ipython() and os.path.isfile(sys.argv[0]) ): try: tmp_file = tempfile.NamedTemporaryFile( prefix="brian_script_", suffix=".py", delete=False ) with tmp_file: # Timestamp tmp_file.write(_encode(f"# {time.asctime()}\n")) # Command line arguments tmp_file.write(_encode(f"# Run as: {' '.join(sys.argv)}\n\n")) # The actual script file # TODO: We are copying the script file as it is, this might clash # with the encoding we used for the comments added above with open(os.path.abspath(sys.argv[0]), "rb") as script_file: shutil.copyfileobj(script_file, tmp_file) BrianLogger.tmp_script = tmp_file.name except OSError as ex: warn(f"Could not copy script file to temp directory: {ex}") if BrianLogger.console_handler is not None: logger.removeHandler(BrianLogger.console_handler) # create console handler with a higher log level BrianLogger.console_handler = logging.StreamHandler() BrianLogger.console_handler.setLevel( LOG_LEVELS[prefs["logging.console_log_level"]] ) BrianLogger.console_handler.setFormatter( logging.Formatter("%(levelname)-10s %(message)s [%(name)s]") ) # add the handler to the logger logger.addHandler(BrianLogger.console_handler) # We want to log all warnings logging.captureWarnings(True) # pylint: disable=E1101 # Manually connect to the warnings logger so that the warnings end up in # the log file. Note that connecting to the console handler here means # duplicated warning messages in the ipython notebook, but not doing so # would mean that they are not displayed at all in the standard ipython # interface... warn_logger = logging.getLogger("py.warnings") warn_logger.addHandler(BrianLogger.console_handler) if BrianLogger.file_handler is not None: warn_logger.addHandler(BrianLogger.file_handler) # Put some standard info into the log file logger.log( logging.DEBUG, f"Logging to file: {BrianLogger.tmp_log}, copy of main script saved as:" f" {BrianLogger.tmp_script}", ) logger.log(logging.DEBUG, f"Python interpreter: {sys.executable}") logger.log(logging.DEBUG, f"Platform: {sys.platform}") version_infos = { "brian": brian2.__version__, "numpy": numpy.__version__, "scipy": scipy.__version__ if scipy else "not installed", "sympy": sympy.__version__, "python": sys.version, } for _name, _version in version_infos.items(): logger.log(logging.DEBUG, f"{_name} version is: {str(_version)}") # Handle uncaught exceptions sys.excepthook = brian_excepthook
[docs] def get_logger(module_name="brian2"): """ Get an object that can be used for logging. Parameters ---------- module_name : str The name used for logging, should normally be the module name as returned by ``__name__``. Returns ------- logger : `BrianLogger` """ return BrianLogger(module_name)
[docs] class catch_logs: """ A context manager for catching log messages. Use this for testing the messages that are logged. Defaults to catching warning/error messages and this is probably the only real use case for testing. Note that while this context manager is active, *all* log messages are suppressed. Using this context manager returns a list of (log level, name, message) tuples. Parameters ---------- log_level : int or str, optional The log level above which messages are caught. only_from : list, optional A list of module names from which messages are caught. Defaults to the ``brian2`` module. Examples -------- >>> logger = get_logger('brian2.logtest') >>> logger.warn('An uncaught warning') # doctest: +SKIP WARNING brian2.logtest: An uncaught warning >>> with catch_logs() as l: ... logger.warn('a caught warning') ... print('l contains: %s' % l) ... l contains: [('WARNING', 'brian2.logtest', 'a caught warning')] """ _entered = False def __init__(self, log_level=logging.WARN, only_from=("brian2",)): self.log_list = [] self.handler = LogCapture(self.log_list, log_level, only_from=only_from) self._entered = False def __enter__(self): if self._entered: raise RuntimeError(f"Cannot enter {self!r} twice") self._entered = True return self.log_list def __exit__(self, *exc_info): if not self._entered: raise RuntimeError(f"Cannot exit {self!r} without entering first") self.handler.uninstall()
[docs] class LogCapture(logging.Handler): """ A class for capturing log warnings. This class is used by `~brian2.utils.logger.catch_logs` to allow testing in a similar way as with `warnings.catch_warnings`. """ def __init__(self, log_list, log_level=logging.WARN, only_from=("brian2",)): logging.Handler.__init__(self, level=log_level) self.log_list = log_list self.only_from = only_from # make a copy of the previous handlers self.handlers = list(logging.getLogger("brian2").handlers) self.install()
[docs] def emit(self, record): # Append a tuple consisting of (level, name, msg) to the list of # warnings if any(record.name.startswith(name) for name in self.only_from): self.log_list.append((record.levelname, record.name, record.msg))
[docs] def install(self): """ Install this handler to catch all warnings. Temporarily disconnect all other handlers. """ the_logger = logging.getLogger("brian2") for handler in self.handlers: the_logger.removeHandler(handler) the_logger.addHandler(self)
[docs] def uninstall(self): """ Uninstall this handler and re-connect the previously installed handlers. """ the_logger = logging.getLogger("brian2") for handler in self.handlers: the_logger.addHandler(handler)
# See http://stackoverflow.com/questions/26126160/redirecting-standard-out-in-err-back-after-os-dup2 # for an explanation of how this function works. Note that 1 and 2 are the file # numbers for stdout and stderr
[docs] class std_silent: """ Context manager that temporarily silences stdout and stderr but keeps the output saved in a temporary file and writes it if an exception is raised. """ dest_stdout = None dest_stderr = None def __init__(self, alwaysprint=False): self.alwaysprint = alwaysprint or not prefs["logging.std_redirection"] self.redirect_to_file = prefs["logging.std_redirection_to_file"] if ( not self.alwaysprint and self.redirect_to_file and std_silent.dest_stdout is None ): std_silent.dest_fname_stdout = tempfile.NamedTemporaryFile( prefix="brian_stdout_", suffix=".log", delete=False ).name std_silent.dest_fname_stderr = tempfile.NamedTemporaryFile( prefix="brian_stderr_", suffix=".log", delete=False ).name std_silent.dest_stdout = open(std_silent.dest_fname_stdout, "w") std_silent.dest_stderr = open(std_silent.dest_fname_stderr, "w") def __enter__(self): if not self.alwaysprint and self.redirect_to_file: sys.stdout.flush() sys.stderr.flush() self.orig_out_fd = os.dup(1) self.orig_err_fd = os.dup(2) os.dup2(std_silent.dest_stdout.fileno(), 1) os.dup2(std_silent.dest_stderr.fileno(), 2) def __exit__(self, exc_type, exc_value, traceback): if not self.alwaysprint and self.redirect_to_file: std_silent.dest_stdout.flush() std_silent.dest_stderr.flush() if exc_type is not None: with open(std_silent.dest_fname_stdout) as f: out = f.read() with open(std_silent.dest_fname_stderr) as f: err = f.read() os.dup2(self.orig_out_fd, 1) os.dup2(self.orig_err_fd, 2) os.close(self.orig_out_fd) os.close(self.orig_err_fd) if exc_type is not None: sys.stdout.write(out) sys.stderr.write(err)
[docs] @classmethod def close(cls): if std_silent.dest_stdout is not None: std_silent.dest_stdout.close() if prefs["logging.delete_log_on_exit"]: try: os.remove(std_silent.dest_fname_stdout) except OSError: # TODO: this happens quite frequently - why? # The file objects are closed as far as Python is concerned, # but maybe Windows is still hanging on to them? pass if std_silent.dest_stderr is not None: std_silent.dest_stderr.close() if prefs["logging.delete_log_on_exit"]: try: os.remove(std_silent.dest_fname_stderr) except OSError: pass