"""Module containing decorators"""
import functools
import logging
import time
import warnings
[docs]
def manual_run(turn_into_run=True, store_meta_data=True, clean_up=True):
"""Can be used to decorate a function as a manual run function.
This can be helpful if you want the run functionality without using an environment.
:param turn_into_run:
If the trajectory should become a `single run` with more specialized functionality
during a single run.
:param store_meta_data:
If meta-data like runtime should be automatically stored
:param clean_up:
If all data added during the single run should be removed, only works
if ``turn_into_run=True``.
"""
def wrapper(func):
@functools.wraps(func)
def new_func(traj, *args, **kwargs):
do_wrap = not traj._run_by_environment
if do_wrap:
traj.f_start_run(turn_into_run=turn_into_run)
result = func(traj, *args, **kwargs)
if do_wrap:
traj.f_finalize_run(store_meta_data=store_meta_data, clean_up=clean_up)
return result
return new_func
return wrapper
def deprecated(msg=""):
"""This is a decorator which can be used to mark functions
as deprecated. It will result in a warning being emitted
when the function is used.
:param msg:
Additional message added to the warning.
"""
def wrapper(func):
@functools.wraps(func)
def new_func(*args, **kwargs):
warning_string = f"Call to deprecated function or property `{func.__name__}`."
warning_string = warning_string + " " + msg
warnings.warn(
warning_string,
category=DeprecationWarning,
)
return func(*args, **kwargs)
return new_func
return wrapper
def copydoc(fromfunc, sep="\n"):
"""Decorator: Copy the docstring of `fromfunc`
If the doc contains a line with the keyword `ABSTRACT`,
like `ABSTRACT: Needs to be defined in subclass`, this line and the line after are removed.
"""
def _decorator(func):
sourcedoc = fromfunc.__doc__
# Remove the ABSTRACT line:
split_doc = sourcedoc.split("\n")
split_doc_no_abstract = [line for line in split_doc if "ABSTRACT" not in line]
# If the length is different we have found an ABSTRACT line
# Finally we want to remove the final blank line, otherwise
# we would have three blank lines at the end
if len(split_doc) != len(split_doc_no_abstract):
sourcedoc = "\n".join(split_doc_no_abstract[:-1])
if func.__doc__ is None:
func.__doc__ = sourcedoc
else:
func.__doc__ = sep.join([sourcedoc, func.__doc__])
return func
return _decorator
def kwargs_mutual_exclusive(param1_name, param2_name, map2to1=None):
"""If there exist mutually exclusive parameters checks for them and maps param2 to 1."""
def wrapper(func):
@functools.wraps(func)
def new_func(*args, **kwargs):
if param2_name in kwargs:
if param1_name in kwargs:
raise ValueError(
f"You cannot specify `{param1_name}` and `{param2_name}` at the same time, "
f"they are mutually exclusive."
)
param2 = kwargs.pop(param2_name)
if map2to1 is not None:
param1 = map2to1(param2)
else:
param1 = param2
kwargs[param1_name] = param1
return func(*args, **kwargs)
return new_func
return wrapper
def kwargs_api_change(old_name, new_name=None):
"""This is a decorator which can be used if a kwarg has changed
its name over versions to also support the old argument name.
Issues a warning if the old keyword argument is detected and
converts call to new API.
:param old_name:
Old name of the keyword argument
:param new_name:
New name of keyword argument
"""
def wrapper(func):
@functools.wraps(func)
def new_func(*args, **kwargs):
if old_name in kwargs:
if new_name is None:
warning_string = (
f"Using deprecated keyword argument `{old_name}` in function `{func.__name__}`. "
"This keyword is no longer supported, please don`t use it "
"anymore."
)
else:
warning_string = (
f"Using deprecated keyword argument `{old_name}` in function `{func.__name__}`, "
f"please use keyword `{new_name}` instead."
)
warnings.warn(warning_string, category=DeprecationWarning)
value = kwargs.pop(old_name)
if new_name is not None:
kwargs[new_name] = value
return func(*args, **kwargs)
return new_func
return wrapper
def not_in_run(func):
"""This is a decorator that signaling that a function is not available during a single run."""
doc = func.__doc__
na_string = """\nATTENTION: This function is not available during a single run!\n"""
if doc is not None:
func.__doc__ = "\n".join([doc, na_string])
func._not_in_run = True
@functools.wraps(func)
def new_func(self, *args, **kwargs):
if self._is_run:
raise TypeError(f"Function `{func.__name__}` is not available during a single run.")
return func(self, *args, **kwargs)
return new_func
def with_open_store(func):
"""This is a decorator that signaling that a function is only available if the storage is open."""
doc = func.__doc__
na_string = """\nATTENTION: This function can only be used if the store is open!\n"""
if doc is not None:
func.__doc__ = "\n".join([doc, na_string])
func._with_open_store = True
@functools.wraps(func)
def new_func(self, *args, **kwargs):
if not self.traj.v_storage_service.is_open:
raise TypeError(f"Function `{func.__name__}` is only available if the storage is open.")
return func(self, *args, **kwargs)
return new_func
def retry(n, errors, wait=0.0, logger_name=None):
"""This is a decorator that retries a function.
Tries `n` times and catches a given tuple of `errors`.
If the `n` retries are not enough, the error is reraised.
If desired `waits` some seconds.
Optionally takes a 'logger_name' of a given logger to print the caught error.
"""
def wrapper(func):
@functools.wraps(func)
def new_func(*args, **kwargs):
retries = 0
while True:
try:
result = func(*args, **kwargs)
if retries and logger_name:
logger = logging.getLogger(logger_name)
logger.debug(f"Retry of `{func.__name__}` successful")
return result
except errors:
if retries >= n:
if logger_name:
logger = logging.getLogger(logger_name)
logger.exception(
f"I could not execute `{func.__name__}` with args {args} and kwargs {kwargs}, "
f"starting next try. "
)
raise
elif logger_name:
logger = logging.getLogger(logger_name)
logger.debug(
f"I could not execute `{func.__name__}` with args {args} and kwargs {kwargs}, "
f"starting next try. "
)
retries += 1
if wait:
time.sleep(wait)
return new_func
return wrapper
def _prfx_getattr_(obj, item):
"""Replacement of __getattr__"""
if item.startswith("f_") or item.startswith("v_"):
return getattr(obj, item[2:])
raise AttributeError(f"`{obj.__class__.__name__}` object has no attribute `{item}`")
def _prfx_setattr_(obj, item, value):
"""Replacement of __setattr__"""
if item.startswith("v_"):
return setattr(obj, item[2:], value)
else:
return super(obj.__class__, obj).__setattr__(item, value)
def prefix_naming(cls):
"""Decorate that adds the prefix naming scheme"""
if hasattr(cls, "__getattr__"):
raise TypeError("__getattr__ already defined")
cls.__getattr__ = _prfx_getattr_
cls.__setattr__ = _prfx_setattr_
return cls