18214982978

Committed 03 Oct 2025 06:39AM UTC coverage: 88.703% (+0.04%) from 88.664%

Build # 18214982978

Build Type

Pull #422

github

Committed by

web-flow

Commit Message

Merge fe026872f into 592d0cd98

Pull Request Pull Request #422: ruff fixes

Run Details

53 of 73 new or added lines in 16 files covered. (72.6%)

2 existing lines in 2 files now uncovered.

7852 of 8852 relevant lines covered (88.7%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.57

/ema_workbench/util/ema_logging.py

"""Helper functions and classes for logging in the workbench.

It is modeled on the default `logging approach that comes with
Python <https://docs.python.org/library/logging.html>`_.
This logging system will also work in case of multiprocessing.

"""

import inspect
import logging
from contextlib import contextmanager
from functools import wraps
from logging import DEBUG, INFO

# Created on 23 dec. 2010
#
# .. codeauthor:: jhkwakkel <j.h.kwakkel (at) tudelft (dot) nl>

__all__ = [
    "DEBUG",
    "DEFAULT_LEVEL",
    "INFO",
    "LOGGER_NAME",
    "get_module_logger",
    "get_rootlogger",
    "log_to_stderr",
    "method_logger",
    "temporary_filter",
]
LOGGER_NAME = "EMA"
DEFAULT_LEVEL = DEBUG
INFO = INFO  #  noqa PLW0127


def create_module_logger(name: str | None = None) -> logging.Logger:
    """Create a module logger with the given name."""
    if name is None:
        frm = inspect.stack()[1]
        mod = inspect.getmodule(frm[0])
        name = mod.__name__
    logger = logging.getLogger(f"{LOGGER_NAME}.{name}")

    _module_loggers[name] = logger
    return logger


def get_module_logger(name) -> logging.Logger:
    """Return a module logger with the given name."""
    try:
        logger = _module_loggers[name]
    except KeyError:
        logger = create_module_logger(name)

    return logger


_rootlogger = None
_module_loggers = {}
_logger = get_module_logger(__name__)

LOG_FORMAT = "[%(processName)s/%(levelname)s] %(message)s"


class TemporaryFilter(logging.Filter):
    """Helper class to temporarily log messages."""

    def __init__(self, *args, level: int = 0, func_name=None, **kwargs):
        super().__init__(*args, **kwargs)
        self.level = level
        self.func_name = func_name

    def filter(self, record) -> bool:
        """Filter out the message."""
        if self.func_name and self.func_name == record.funcName:
            return True

        return record.levelno > self.level


@contextmanager
def temporary_filter(
    name: str | list[str] = LOGGER_NAME,
    level: int | list[int] = 0,
    func_name: str | list[str] | None = None,
):
    """Temporary filter log message.

    Parameters
    ----------
    name : str or list of str, optional
           logger on which to apply the filter.
    level: int, or list of int, optional
           don't log message of this level or lower
    func_name : str or list of str, optional
            don't log message of this function

    all modules have their own unique logger
    (e.g. ema_workbench.analysis.prim)

    """
    # TODO:: probably all three should be optionally a list so you
    # might filter multiple log message from different functions
    names = [name] if isinstance(name, str) else name
    levels = [level] if isinstance(level, int) else level
    func_names = [func_name] if (isinstance(func_name, str) or func_name is None) else func_name

    # get logger
    # add filter
    max_length = max(len(names), len(levels), len(func_names))

    # make a list equal lengths?
    if len(names) < max_length:
        names = [name] * max_length
    if len(levels) < max_length:
        levels = [level] * max_length
    if len(func_names) < max_length:
        func_names = [func_name] * max_length

    filters = {}
    for name, level, func_name in zip(names, levels, func_names):
        logger = get_module_logger(name)
        filter = TemporaryFilter(
            level=level, func_name=func_name
        )  # @ReservedAssignment

        if logger == _logger:
            # root logger, add filter to handler rather than logger
            # because filters don't propagate for some unclear reason
            for handler in logger.handlers:
                handler.addFilter(filter)
                filters[filter] = handler
        else:
            logger.addFilter(filter)
            filters[filter] = logger

    yield

    for k, v in filters.items():
        v.removeFilter(k)


def method_logger(name):
    """Wrap method so that every call to it is logged."""
    logger = get_module_logger(name)
    classname = inspect.getouterframes(inspect.currentframe())[1][3]

    def real_decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # hack, because log is applied to methods, we can get
            # object instance as first arguments in args
            logger.debug(f"calling {func.__name__} on {classname}")
            res = func(*args, **kwargs)
            logger.debug(f"completed calling {func.__name__} on {classname}")
            return res

        return wrapper

    return real_decorator


def get_rootlogger() -> logging.Logger:
    """Returns root logger used by the EMA workbench.

    Returns
    -------
    the logger of the EMA workbench

    """
    global _rootlogger  # noqa PLW0603

    if not _rootlogger:
        _rootlogger = logging.getLogger(LOGGER_NAME)
        _rootlogger.handlers = []
        _rootlogger.addHandler(logging.NullHandler())
        _rootlogger.setLevel(DEBUG)

    return _rootlogger


def log_to_stderr(level=None, pass_root_logger_level=False):
    """Turn on logging and add a handler which prints to stderr.

    Parameters
    ----------
    level : int
            minimum level of the messages that will be logged
    pas_root_logger_level: bool, optional. Default False
            if true, all module loggers will be set to the
            same logging level as the root logger.
            Recommended True when using the MPIEvaluator.

    """
    if not level:
        level = DEFAULT_LEVEL

    logger = get_rootlogger()

    # avoid creation of multiple stream handlers for logging to console
    for entry in logger.handlers:
        if (isinstance(entry, logging.StreamHandler)) and (
            entry.formatter._fmt == LOG_FORMAT
        ):
            return logger

    formatter = logging.Formatter(LOG_FORMAT)
    handler = logging.StreamHandler()
    handler.setLevel(level)
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.propagate = False

    if pass_root_logger_level:
        for _, mod_logger in _module_loggers.items():
            mod_logger.setLevel(level)

    return logger

1	"""Helper functions and classes for logging in the workbench.
2
3	It is modeled on the default `logging approach that comes with
4	Python <https://docs.python.org/library/logging.html>`_.
5	This logging system will also work in case of multiprocessing.
6
7	"""
8
9	import inspect	1✔
10	import logging	1✔
11	from contextlib import contextmanager	1✔
12	from functools import wraps	1✔
13	from logging import DEBUG, INFO	1✔
14
15	# Created on 23 dec. 2010
16	#
17	# .. codeauthor:: jhkwakkel <j.h.kwakkel (at) tudelft (dot) nl>
18
19	__all__ = [	1✔
20	"DEBUG",
21	"DEFAULT_LEVEL",
22	"INFO",
23	"LOGGER_NAME",
24	"get_module_logger",
25	"get_rootlogger",
26	"log_to_stderr",
27	"method_logger",
28	"temporary_filter",
29	]
30	LOGGER_NAME = "EMA"	1✔
31	DEFAULT_LEVEL = DEBUG	1✔
32	INFO = INFO # noqa PLW0127	1✔
33
34
35	def create_module_logger(name: str \| None = None) -> logging.Logger:	1✔
36	"""Create a module logger with the given name."""
37	if name is None:	1✔
38	frm = inspect.stack()[1]	×
39	mod = inspect.getmodule(frm[0])	×
40	name = mod.__name__	×
41	logger = logging.getLogger(f"{LOGGER_NAME}.{name}")	1✔
42
43	_module_loggers[name] = logger	1✔
44	return logger	1✔
45
46
47	def get_module_logger(name) -> logging.Logger:	1✔
48	"""Return a module logger with the given name."""
49	try:	1✔
50	logger = _module_loggers[name]	1✔
51	except KeyError:	1✔
52	logger = create_module_logger(name)	1✔
53
54	return logger	1✔
55
56
57	_rootlogger = None	1✔
58	_module_loggers = {}	1✔
59	_logger = get_module_logger(__name__)	1✔
60
61	LOG_FORMAT = "[%(processName)s/%(levelname)s] %(message)s"	1✔
62
63
64	class TemporaryFilter(logging.Filter):	1✔
65	"""Helper class to temporarily log messages."""
66
67	def __init__(self, args, level: int = 0, func_name=None, *kwargs):	1✔
68	super().__init__(args, *kwargs)	1✔
69	self.level = level	1✔
70	self.func_name = func_name	1✔
71
72	def filter(self, record) -> bool:	1✔
73	"""Filter out the message."""
NEW 74	if self.func_name and self.func_name == record.funcName:	×
NEW 75	return True	×
76
77	return record.levelno > self.level	×
78
79
80	@contextmanager	1✔
81	def temporary_filter(	1✔
82	name: str \| list[str] = LOGGER_NAME,
83	level: int \| list[int] = 0,
84	func_name: str \| list[str] \| None = None,
85	):
86	"""Temporary filter log message.
87
88	Parameters
89	----------
90	name : str or list of str, optional
91	logger on which to apply the filter.
92	level: int, or list of int, optional
93	don't log message of this level or lower
94	func_name : str or list of str, optional
95	don't log message of this function
96
97	all modules have their own unique logger
98	(e.g. ema_workbench.analysis.prim)
99
100	"""
101	# TODO:: probably all three should be optionally a list so you
102	# might filter multiple log message from different functions
103	names = [name] if isinstance(name, str) else name	1✔
104	levels = [level] if isinstance(level, int) else level	1✔
105	func_names = [func_name] if (isinstance(func_name, str) or func_name is None) else func_name	1✔
106
107	# get logger
108	# add filter
109	max_length = max(len(names), len(levels), len(func_names))	1✔
110
111	# make a list equal lengths?
112	if len(names) < max_length:	1✔
113	names = [name] * max_length	×
114	if len(levels) < max_length:	1✔
115	levels = [level] * max_length	×
116	if len(func_names) < max_length:	1✔
NEW 117	func_names = [func_name] * max_length	×
118
119	filters = {}	1✔
120	for name, level, func_name in zip(names, levels, func_names):	1✔
121	logger = get_module_logger(name)	1✔
122	filter = TemporaryFilter(	1✔
123	level=level, func_name=func_name
124	) # @ReservedAssignment
125
126	if logger == _logger:	1✔
127	# root logger, add filter to handler rather than logger
128	# because filters don't propagate for some unclear reason
129	for handler in logger.handlers:	×
130	handler.addFilter(filter)	×
131	filters[filter] = handler	×
132	else:
133	logger.addFilter(filter)	1✔
134	filters[filter] = logger	1✔
135
136	yield	1✔
137
138	for k, v in filters.items():	1✔
139	v.removeFilter(k)	1✔
140
141
142	def method_logger(name):	1✔
143	"""Wrap method so that every call to it is logged."""
144	logger = get_module_logger(name)	1✔
145	classname = inspect.getouterframes(inspect.currentframe())[1][3]	1✔
146
147	def real_decorator(func):	1✔
148	@wraps(func)	1✔
149	def wrapper(args, *kwargs):	1✔
150	# hack, because log is applied to methods, we can get
151	# object instance as first arguments in args
152	logger.debug(f"calling {func.__name__} on {classname}")	1✔
153	res = func(args, *kwargs)	1✔
154	logger.debug(f"completed calling {func.__name__} on {classname}")	1✔
155	return res	1✔
156
157	return wrapper	1✔
158
159	return real_decorator	1✔
160
161
162	def get_rootlogger() -> logging.Logger:	1✔
163	"""Returns root logger used by the EMA workbench.
164
165	Returns
166	-------
167	the logger of the EMA workbench
168
169	"""
170	global _rootlogger # noqa PLW0603
171
172	if not _rootlogger:	1✔
173	_rootlogger = logging.getLogger(LOGGER_NAME)	1✔
174	_rootlogger.handlers = []	1✔
175	_rootlogger.addHandler(logging.NullHandler())	1✔
176	_rootlogger.setLevel(DEBUG)	1✔
177
178	return _rootlogger	1✔
179
180
181	def log_to_stderr(level=None, pass_root_logger_level=False):	1✔
182	"""Turn on logging and add a handler which prints to stderr.
183
184	Parameters
185	----------
186	level : int
187	minimum level of the messages that will be logged
188	pas_root_logger_level: bool, optional. Default False
189	if true, all module loggers will be set to the
190	same logging level as the root logger.
191	Recommended True when using the MPIEvaluator.
192
193	"""
194	if not level:	1✔
195	level = DEFAULT_LEVEL	1✔
196
197	logger = get_rootlogger()	1✔
198
199	# avoid creation of multiple stream handlers for logging to console
200	for entry in logger.handlers:	1✔
201	if (isinstance(entry, logging.StreamHandler)) and (	1✔
202	entry.formatter._fmt == LOG_FORMAT
203	):
204	return logger	1✔
205
206	formatter = logging.Formatter(LOG_FORMAT)	1✔
207	handler = logging.StreamHandler()	1✔
208	handler.setLevel(level)	1✔
209	handler.setFormatter(formatter)	1✔
210	logger.addHandler(handler)	1✔
211	logger.propagate = False	1✔
212
213	if pass_root_logger_level:	1✔
214	for _, mod_logger in _module_loggers.items():	×
215	mod_logger.setLevel(level)	×
216
217	return logger	1✔

quaquel / EMAworkbench / 18214982978

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous