"""Some utility functions."""
import contextlib
import importlib
import inspect
import logging
import os.path as op
import re
import sys
import warnings
from collections.abc import Callable
from io import StringIO
from typing import Any, TypeVar
from decorator import FunctionMaker
from .docs import fill_doc, fill_doc_static
logger = logging.getLogger("mne")
logger.propagate = False
class _FrameFilter(logging.Filter):
def __init__(self):
self.add_frames = 0
def filter(self, record):
record.frame_info = "Unknown"
if self.add_frames:
frame_info = _frame_info(5 + self.add_frames)[5:][::-1]
if len(frame_info):
frame_info[-1] = (frame_info[-1] + " :").ljust(30)
if len(frame_info) > 1:
frame_info[0] = "┌" + frame_info[0]
frame_info[-1] = "└" + frame_info[-1]
for ii, info in enumerate(frame_info[1:-1], 1):
frame_info[ii] = "├" + info
record.frame_info = "\n".join(frame_info)
return True
_filter = _FrameFilter()
logger.addFilter(_filter)
_FuncT = TypeVar("_FuncT", bound=Callable[..., Any])
def verbose(function: _FuncT) -> _FuncT:
"""Verbose decorator to allow functions to override log-level.
Parameters
----------
function : callable
Function to be decorated by setting the verbosity level.
Returns
-------
dec : callable
The decorated function.
See Also
--------
set_log_level
set_config
Notes
-----
This decorator is used to set the verbose level during a function or method
call, such as :func:`mne.compute_covariance`. The `verbose` keyword
argument can be 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL', True (an
alias for 'INFO'), or False (an alias for 'WARNING'). To set the global
verbosity level for all functions, use :func:`mne.set_log_level`.
This function also serves as a docstring filler.
Examples
--------
You can use the ``verbose`` argument to set the verbose level on the fly::
>>> import mne
>>> cov = mne.compute_raw_covariance(raw, verbose='WARNING') # doctest: +SKIP
>>> cov = mne.compute_raw_covariance(raw, verbose='INFO') # doctest: +SKIP
Using up to 49 segments
Number of samples used : 5880
[done]
"""
try:
fill_doc(function)
except TypeError:
pass
return _verbose_control(function)
def verbose_static(*keys: str) -> Callable[[_FuncT], _FuncT]:
"""Verbose decorator for functions whose docstrings are statically filled.
Behaves like :func:`mne.verbose` for log-level handling, but does **not**
modify the docstring at import time. The docstring must already contain the
fully expanded text of ``docdict["verbose"]`` (and of every key in ``keys``);
this is enforced by ``tools/hooks/check_static_docs.py`` (run via
pre-commit), which can also update the docstring with ``--fix`` when the
``docdict`` entry changes.
Parameters
----------
*keys : str
Additional ``docdict`` keys whose expanded text this docstring contains
(``"verbose"`` is implied).
Returns
-------
dec : callable
The decorator.
"""
def dec(function: _FuncT) -> _FuncT:
out = _verbose_control(function)
out._static_doc_keys = ("verbose", *keys)
return out
return dec
def _verbose_control(function: _FuncT) -> _FuncT:
"""Let ``verbose`` set the log level during a call, leaving ``__doc__`` alone.
For private code that takes ``verbose`` without documenting it; documented
functions use :func:`verbose_static`, which also checks the docstring.
"""
body = """\
def %(name)s(%(signature)s):\n
try:
do_level_change = verbose is not None
except (NameError, UnboundLocalError):
raise RuntimeError('Function/method %%s does not accept verbose '
'parameter' %% (_function_,)) from None
if do_level_change:
with _use_log_level_(verbose):
return _function_(%(shortsignature)s)
else:
return _function_(%(shortsignature)s)"""
evaldict = dict(_use_log_level_=use_log_level, _function_=function)
fm = FunctionMaker(function)
attrs = dict(
__wrapped__=function,
__qualname__=function.__qualname__,
__globals__=function.__globals__,
)
return fm.make(body, evaldict, addsource=True, **attrs)
@fill_doc_static("verbose", "add_frames")
class use_log_level:
"""Context manager for logging level.
Parameters
----------
verbose : bool | str | int | None
Control verbosity of the logging output. If ``None``, use the default
verbosity level. See the :ref:`logging documentation <tut-logging>` and
:func:`mne.verbose` for details. Should only be passed as a keyword
argument.
add_frames : int | None
If int, enable (>=1) or disable (0) the printing of stack frame
information using formatting. Default (None) does not change the
formatting. This can add overhead so is meant only for debugging.
See Also
--------
mne.verbose
Notes
-----
See the :ref:`logging documentation <tut-logging>` for details.
Examples
--------
>>> from mne import use_log_level
>>> from mne.utils import logger
>>> with use_log_level(False):
... # Most MNE logger messages are "info" level, False makes them not
... # print:
... logger.info('This message will not be printed')
>>> with use_log_level(True):
... # Using verbose=True in functions, methods, or this context manager
... # will ensure they are printed
... logger.info('This message will be printed!')
This message will be printed!
"""
def __init__(self, verbose=None, *, add_frames=None):
self._level = verbose
self._add_frames = add_frames
self._old_frames = _filter.add_frames
def __enter__(self):
level = logger.level if self._level is None else self._level
self._old_level = set_log_level(
level, return_old_level=True, add_frames=self._add_frames
)
def __exit__(self, *args):
add_frames = self._old_frames if self._add_frames is not None else None
set_log_level(self._old_level, add_frames=add_frames)
_LOGGING_TYPES = dict(
DEBUG=logging.DEBUG,
INFO=logging.INFO,
WARNING=logging.WARNING,
ERROR=logging.ERROR,
CRITICAL=logging.CRITICAL,
)
@fill_doc_static("add_frames")
def set_log_level(verbose=None, return_old_level=False, add_frames=None):
"""Set the logging level.
Parameters
----------
verbose : bool, str, int, or None
The verbosity of messages to print. If a str, it can be either DEBUG,
INFO, WARNING, ERROR, or CRITICAL. Note that these are for
convenience and are equivalent to passing in logging.DEBUG, etc.
For bool, True is the same as 'INFO', False is the same as 'WARNING'.
If None, the environment variable MNE_LOGGING_LEVEL is read, and if
it doesn't exist, defaults to INFO.
return_old_level : bool
If True, return the old verbosity level.
add_frames : int | None
If int, enable (>=1) or disable (0) the printing of stack frame
information using formatting. Default (None) does not change the
formatting. This can add overhead so is meant only for debugging.
Returns
-------
old_level : int
The old level. Only returned if ``return_old_level`` is True.
"""
old_verbose = logger.level
verbose = _parse_verbose(verbose)
if verbose != old_verbose:
logger.setLevel(verbose)
if add_frames is not None:
_filter.add_frames = int(add_frames)
fmt = "%(frame_info)s " if add_frames else ""
fmt += "%(message)s"
fmt = logging.Formatter(fmt)
for handler in logger.handlers:
handler.setFormatter(fmt)
return old_verbose if return_old_level else None
def _parse_verbose(verbose):
from .check import _check_option, _validate_type
from .config import get_config
_validate_type(verbose, (bool, str, int, None), "verbose")
if verbose is None:
verbose = get_config("MNE_LOGGING_LEVEL", "INFO")
elif isinstance(verbose, bool):
if verbose is True:
verbose = "INFO"
else:
verbose = "WARNING"
if isinstance(verbose, str):
verbose = verbose.upper()
_check_option("verbose", verbose, _LOGGING_TYPES, "(when a string)")
verbose = _LOGGING_TYPES[verbose]
return verbose
def set_log_file(fname=None, output_format="%(message)s", overwrite=None):
"""Set the log to print to a file.
Parameters
----------
fname : path-like | None
Filename of the log to print to. If None, stdout is used.
To suppress log outputs, use set_log_level('WARNING').
output_format : str
Format of the output messages. See the following for examples:
https://docs.python.org/dev/howto/logging.html
e.g., "%(asctime)s - %(levelname)s - %(message)s".
overwrite : bool | None
Overwrite the log file (if it exists). Otherwise, statements
will be appended to the log (default). None is the same as False,
but additionally raises a warning to notify the user that log
entries will be appended.
"""
_remove_close_handlers(logger)
if fname is not None:
if op.isfile(fname) and overwrite is None:
warnings.warn(
"Log entries will be appended to the file. Use "
"overwrite=False to avoid this message in the "
"future.",
RuntimeWarning,
stacklevel=2,
)
overwrite = False
mode = "w" if overwrite else "a"
lh = logging.FileHandler(fname, mode=mode)
else:
"""we should just be able to do:
lh = logging.StreamHandler(sys.stdout)
but because doctests uses some magic on stdout, we have to do this:
"""
lh = logging.StreamHandler(WrapStdOut())
lh.setFormatter(logging.Formatter(output_format))
logger.addHandler(lh)
def _remove_close_handlers(logger):
for h in list(logger.handlers):
if isinstance(h, logging.FileHandler | logging.StreamHandler):
if isinstance(h, logging.FileHandler):
h.close()
logger.removeHandler(h)
class ClosingStringIO(StringIO):
"""StringIO that closes after getvalue()."""
def getvalue(self, close=True):
"""Get the value."""
out = super().getvalue()
if close:
self.close()
return out
class catch_logging:
"""Store logging.
This will remove all other logging handlers, and return the handler to
stdout when complete.
"""
def __init__(self, verbose=None):
self.verbose = verbose
def __enter__(self):
self._ctx = use_log_level(self.verbose)
self._data = ClosingStringIO()
self._lh = logging.StreamHandler(self._data)
self._lh.setFormatter(logging.Formatter("%(message)s"))
self._lh._mne_file_like = True
_remove_close_handlers(logger)
logger.addHandler(self._lh)
self._ctx.__enter__()
return self._data
def __exit__(self, *args):
self._ctx.__exit__(*args)
logger.removeHandler(self._lh)
set_log_file(None)
@contextlib.contextmanager
def _record_warnings():
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
yield w
class WrapStdOut:
"""Dynamically wrap to sys.stdout.
This makes packages that monkey-patch sys.stdout (e.g.doctest,
sphinx-gallery) work properly.
"""
def __getattr__(self, name):
if hasattr(sys.stdout, name):
return getattr(sys.stdout, name)
else:
raise AttributeError(f"'file' object has not attribute '{name}'")
_verbose_dec_re = re.compile("^<decorator-gen-[0-9]+>$")
def warn(message, category=RuntimeWarning, module="mne", ignore_namespaces=("mne",)):
"""Emit a warning with trace outside the mne namespace.
This function takes arguments like warnings.warn, and sends messages
using both ``warnings.warn`` and ``logger.warn``. Warnings can be
generated deep within nested function calls. In order to provide a
more helpful warning, this function traverses the stack until it
reaches a frame outside the ``mne`` namespace that caused the error.
Parameters
----------
message : str
Warning message.
category : instance of Warning
The warning class. Defaults to ``RuntimeWarning``.
module : str
The name of the module emitting the warning.
ignore_namespaces : list of str
Namespaces to ignore when traversing the stack.
.. versionadded:: 0.24
"""
root_dirs = [importlib.import_module(ns) for ns in ignore_namespaces]
root_dirs = [op.dirname(ns.__file__) for ns in root_dirs]
frame = None
if logger.level <= logging.WARNING:
frame = inspect.currentframe()
while frame:
fname = frame.f_code.co_filename
lineno = frame.f_lineno
if not _verbose_dec_re.search(fname):
if (
not (
any(fname.startswith(rd) for rd in root_dirs)
or ("unittest" in fname and "case" in fname)
)
or op.basename(op.dirname(fname)) == "tests"
):
break
frame = frame.f_back
del frame
warnings.warn_explicit(
message,
category,
fname,
lineno,
module,
globals().get("__warningregistry__", {}),
)
if (
any(
isinstance(h, logging.FileHandler) or getattr(h, "_mne_file_like", False)
for h in logger.handlers
)
or _filter.add_frames
):
logger.warning(message)
def _get_call_line():
"""Get the call line from within a function."""
frame = inspect.currentframe().f_back.f_back
if _verbose_dec_re.search(frame.f_code.co_filename):
frame = frame.f_back
context = inspect.getframeinfo(frame).code_context
context = "unknown" if context is None else context[0].strip()
return context
def filter_out_warnings(warn_record, category=None, match=None):
r"""Remove particular records from ``warn_record``.
This helper takes a list of :class:`warnings.WarningMessage` objects,
and remove those matching category and/or text.
Parameters
----------
category: WarningMessage type | None
class of the message to filter out
match : str | None
text or regex that matches the error message to filter out
"""
regexp = re.compile(".*" if match is None else match)
is_category = [
w.category == category if category is not None else True
for w in warn_record._list
]
is_match = [regexp.match(w.message.args[0]) is not None for w in warn_record._list]
ind = [ind for ind, (c, m) in enumerate(zip(is_category, is_match)) if c and m]
for i in reversed(ind):
warn_record._list.pop(i)
@contextlib.contextmanager
def wrapped_stdout(indent="", cull_newlines=False):
"""Wrap stdout writes to logger.info, with an optional indent prefix.
Parameters
----------
indent : str
The indentation to add.
cull_newlines : bool
If True, cull any new/blank lines at the end.
"""
orig_stdout = sys.stdout
my_out = ClosingStringIO()
sys.stdout = my_out
try:
yield
finally:
sys.stdout = orig_stdout
pending_newlines = 0
for line in my_out.getvalue().split("\n"):
if not line.strip() and cull_newlines:
pending_newlines += 1
continue
for _ in range(pending_newlines):
logger.info("\n")
logger.info(indent + line)
def _frame_info(n):
frame = inspect.currentframe()
try:
frame = frame.f_back
infos = list()
for _ in range(n):
try:
name = frame.f_globals["__name__"]
except KeyError:
pass
else:
infos.append(f"{name.removeprefix('mne.')}:{frame.f_lineno}")
frame = frame.f_back
if frame is None:
break
return infos
except Exception:
return ["unknown"]
finally:
del frame
def _verbose_safe_false(*, level="warning"):
lev = _LOGGING_TYPES[level.upper()]
return lev if logger.level <= lev else None