Source code for ess.sans.logging
# SPDX-License-Identifier: BSD-3-Clause
# Copyright (c) 2023 Scipp contributors (https://github.com/scipp)
# @author Jan-Lukas Wynen
"""
Utilities for logging.
All code in the ess package should log auxiliary information
through this module.
Loggers should be obtained through :py:func:`ess.logging.get_logger`
and pass a name that reflects the current context.
E.g., in the loki package, pass ``'loki'`` (all lowercase).
Logging can be configured using :py:func:`ess.logging.configure` or
:py:func:`ess.logging.configure_workflow`.
Use the latter at the beginning of workflow notebooks.
"""
import functools
import inspect
import logging
import logging.config
from collections.abc import Callable, Sequence
from copy import copy
from os import PathLike
from typing import Any
import scipp as sc
import scippneutron as scn
from scipp.utils import running_in_jupyter
[docs]
def get_logger(subname: str | None = None) -> logging.Logger:
"""Return one of ess's loggers.
Parameters
----------
subname:
Name of an instrument, technique, or workflow.
If given, return the logger with the given name
as a child of the ess logger.
Otherwise, return the general ess logger.
Returns
-------
:
The requested logger.
"""
name = 'scipp.ess' + ('.' + subname if subname else '')
return logging.getLogger(name)
[docs]
def log_call(
*, instrument: str, message: str | None = None, level: int | str = logging.INFO
):
"""
Decorator that logs a message every time the function is called.
"""
level = logging.getLevelName(level) if isinstance(level, str) else level
def deco(f: Callable):
@functools.wraps(f)
def impl(*args, **kwargs):
if message is not None:
get_logger(instrument).log(level, message)
else:
get_logger(instrument).log(level, 'Calling %s', _function_name(f))
return f(*args, **kwargs)
return impl
return deco
[docs]
class Formatter(logging.Formatter):
"""
Logging formatter that indents messages and optionally shows threading information.
"""
[docs]
def __init__(self, show_thread: bool, show_process: bool):
"""
Initialize the formatter.
The formatting is mostly fixed.
Only printing of thread and processor names can be toggled using the
corresponding arguments.
Times are always printed in ISO 8601 format.
"""
fmt_proc = '%(processName)s' if show_process else ''
fmt_thread = '%(threadName)s' if show_thread else ''
if show_process:
fmt_proc_thread = fmt_proc
if show_thread:
fmt_proc_thread += ',' + fmt_thread
elif show_thread:
fmt_proc_thread = fmt_thread
else:
fmt_proc_thread = ''
fmt_pre = '[%(asctime)s] %(levelname)-8s '
fmt_post = '<%(name)s> : %(message)s'
fmt = (
fmt_pre
+ ('{' + fmt_proc_thread + '} ' if fmt_proc_thread else '')
+ fmt_post
)
super().__init__(fmt, datefmt='%Y-%m-%dT%H:%M:%S%z')
[docs]
def format(self, record: logging.LogRecord) -> str:
record = copy(record)
record.msg = '\n ' + record.msg.replace('\n', '\n ')
return super().format(record)
[docs]
def default_loggers_to_configure() -> list[logging.Logger]:
"""
Return a list of all loggers that get configured by ess by default.
"""
import pooch
return [
sc.get_logger(),
logging.getLogger('Mantid'),
pooch.get_logger(),
]
[docs]
def configure(
*,
filename: str | PathLike | None = 'scipp.ess.log',
file_level: str | int = logging.INFO,
stream_level: str | int = logging.WARNING,
widget_level: str | int = logging.INFO,
show_thread: bool = False,
show_process: bool = False,
loggers: Sequence[str | logging.Logger] | None = None,
):
"""Set up logging for the ess package.
This function is meant as a helper for application (or notebook) developers.
It configures the loggers of ess, scippneutron, scipp, and some
third party packages.
*Calling it from a library should be avoided*
because it can mess up a user's setup.
Up to 3 handlers are configured:
- *File Handler* Writes files to a file with given name.
Can be disabled using the `filename` argument.
- *Stream Handler* Writes to `sys.stderr`.
- *Widget Handler* Writes to a :py:class:`scipp.logging.LogWidget`
if Python is running in a Jupyter notebook.
Parameters
----------
filename:
Name of the log file. Overwrites existing files.
Setting this to `None` disables logging to file.
file_level:
Log level for the file handler.
stream_level:
Log level for the stream handler.
widget_level:
Log level for the widget handler.
show_thread:
If `True`, log messages include the name of the thread
the message originates from.
show_process:
If `True`, log messages include the name of the process
the message originates from.
loggers:
Collection of loggers or names of loggers to configure.
If not given, uses :py:func:`default_loggers_to_configure`.
See Also
--------
ess.logging.configure_workflow:
Configure logging and do some additional setup for a reduction workflow.
"""
if configure.is_configured:
get_logger().warning(
'Called `logging.configure` but logging is already configured'
)
return
handlers = _make_handlers(
filename, file_level, stream_level, widget_level, show_thread, show_process
)
base_level = _base_level([file_level, stream_level, widget_level])
loggers = {
logging.getLogger(logger) if isinstance(logger, str) else logger
for logger in (default_loggers_to_configure() if loggers is None else loggers)
}
for logger in loggers:
_configure_logger(logger, handlers, base_level)
if any(logger.name == 'Mantid' for logger in loggers):
_configure_mantid_logging('notice')
configure.is_configured = True
configure.is_configured = False
[docs]
def configure_workflow(
workflow_name: str | None = None, *, display: bool | None = None, **kwargs
) -> logging.Logger:
"""Configure logging for a reduction workflow.
Configures loggers, logs a greeting message, sets up a logger for a workflow,
and optionally creates and displays a log widget.
Parameters
----------
workflow_name:
Used as the name of the returned logger.
display:
If `True`, show a :py:class:`scipp.logging.LogWidget`
in the outputs of the current cell.
Defaults to `True` in Jupyter and `False` otherwise.
kwargs:
Forwarded to :py:func:`ess.logging.configure`.
Refer to that function for details.
Returns
-------
:
A logger for use in the workflow.
See Also
--------
ess.logging.configure:
General logging setup.
"""
configure(**kwargs)
greet()
if (display is None and running_in_jupyter()) or display:
sc.display_logs()
return get_logger(workflow_name)
[docs]
def greet():
"""Log a message showing the versions of important packages."""
# Import here so we don't import from a partially built package.
from . import __version__
msg = f'''Software Versions:
ess: {__version__} (https://scipp.github.io/ess)
scippneutron: {scn.__version__} (https://scipp.github.io/scippneutron)
scipp: {sc.__version__} (https://scipp.github.io)'''
mantid_version = _mantid_version()
if mantid_version:
msg += f'\n Mantid: {mantid_version} (https://www.mantidproject.org)'
get_logger().info(msg)
_INSTRUMENTS = [
'amor',
'beer',
'bifrost',
'cspec',
'dream',
'estia',
'freia',
'heimdal',
'loki',
'magic',
'miracles',
'nmx',
'odin',
'skadi',
'trex',
'v20',
'vespa',
]
def _deduce_instrument_name(f: Any) -> str | None:
# Assumes package name: ess.<instrument>[.subpackage]
package = inspect.getmodule(f).__package__
components = package.split('.', 2)
try:
if components[0] == 'ess':
candidate = components[1]
if candidate in _INSTRUMENTS:
return candidate
except IndexError:
pass
return None
def _function_name(f: Callable) -> str:
if hasattr(f, '__module__'):
return f'{f.__module__}.{f.__name__}'
return f.__name__
def _make_stream_handler(
level: str | int, show_thread: bool, show_process: bool
) -> logging.StreamHandler:
handler = logging.StreamHandler()
handler.setLevel(level)
handler.setFormatter(Formatter(show_thread, show_process))
return handler
def _make_file_handler(
filename: str | PathLike,
level: str | int,
show_thread: bool,
show_process: bool,
) -> logging.FileHandler:
handler = logging.FileHandler(filename, mode='w')
handler.setLevel(level)
handler.setFormatter(Formatter(show_thread, show_process))
return handler
def _make_handlers(
filename: str | PathLike | None,
file_level: str | int,
stream_level: str | int,
widget_level: str | int,
show_thread: bool,
show_process: bool,
) -> list[logging.Handler]:
handlers = [_make_stream_handler(stream_level, show_thread, show_process)]
if filename is not None:
handlers.append(
_make_file_handler(filename, file_level, show_thread, show_process)
)
if running_in_jupyter():
handlers.append(sc.logging.make_widget_handler())
return handlers
def _configure_logger(
logger: logging.Logger, handlers: list[logging.Handler], level: str | int
):
for handler in handlers:
logger.addHandler(handler)
logger.setLevel(level)
def _configure_mantid_logging(level: str):
try:
from mantid.utils.logging import log_to_python
log_to_python(level)
except ImportError:
pass
def _base_level(levels: list[str | int]) -> int:
return min(
logging.getLevelName(level) if isinstance(level, str) else level
for level in levels
)
def _mantid_version() -> str | None:
try:
import mantid
return mantid.__version__
except ImportError:
return None