File size: 5,229 Bytes
838f737 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 |
import importlib.util
import os
import warnings
from functools import partial, wraps
from typing import Optional
def eval_env(var, default):
"""Check if environment varable has True-y value"""
if var not in os.environ:
return default
val = os.environ.get(var, "0")
trues = ["1", "true", "TRUE", "on", "ON", "yes", "YES"]
falses = ["0", "false", "FALSE", "off", "OFF", "no", "NO"]
if val in trues:
return True
if val not in falses:
# fmt: off
raise RuntimeError(
f"Unexpected environment variable value `{var}={val}`. "
f"Expected one of {trues + falses}")
# fmt: on
return False
def is_module_available(*modules: str) -> bool:
r"""Returns if a top-level module with :attr:`name` exists *without**
importing it. This is generally safer than try-catch block around a
`import X`. It avoids third party libraries breaking assumptions of some of
our tests, e.g., setting multiprocessing start method when imported
(see librosa/#747, torchvision/#544).
"""
return all(importlib.util.find_spec(m) is not None for m in modules)
def requires_module(*modules: str):
"""Decorate function to give error message if invoked without required optional modules.
This decorator is to give better error message to users rather
than raising ``NameError: name 'module' is not defined`` at random places.
"""
missing = [m for m in modules if not is_module_available(m)]
if not missing:
# fall through. If all the modules are available, no need to decorate
def decorator(func):
return func
else:
req = f"module: {missing[0]}" if len(missing) == 1 else f"modules: {missing}"
def decorator(func):
@wraps(func)
def wrapped(*args, **kwargs):
raise RuntimeError(f"{func.__module__}.{func.__name__} requires {req}")
return wrapped
return decorator
UNSUPPORTED = []
def wrap_deprecated(func, name, direction: str, version: Optional[str] = None, remove: bool = False):
@wraps(func)
def wrapped(*args, **kwargs):
message = f"{name} has been deprecated. {direction}"
if remove:
message += f' It will be removed from {"a future" if version is None else "the " + str(version)} release. '
warnings.warn(message, stacklevel=2)
return func(*args, **kwargs)
return wrapped
def deprecated(direction: str, version: Optional[str] = None, remove: bool = False):
"""Decorator to add deprecation message
Args:
direction (str): Migration steps to be given to users.
version (str or int): The version when the object will be removed
remove (bool): If enabled, append future removal message.
"""
def decorator(func):
wrapped = wrap_deprecated(func, f"{func.__module__}.{func.__name__}", direction, version=version, remove=remove)
message = "This function has been deprecated. "
if remove:
message += f'It will be removed from {"future" if version is None else version} release. '
wrapped.__doc__ = f"""DEPRECATED
.. warning::
{message}
{direction}
{func.__doc__}
"""
return wrapped
return decorator
DEPRECATION_MSG = (
"This deprecation is part of a large refactoring effort to transition TorchAudio into a maintenance phase. "
"Please see https://github.com/pytorch/audio/issues/3902 for more information."
)
IO_DEPRECATION_MSG = (
"This deprecation is part of a large refactoring effort to transition TorchAudio into a maintenance phase. "
"The decoding and encoding capabilities of PyTorch for both audio"
" and video are being consolidated into TorchCodec. "
"Please see https://github.com/pytorch/audio/issues/3902 for more information."
)
dropping_support = deprecated(DEPRECATION_MSG, version="2.9", remove=True)
def dropping_class_support(c, msg=DEPRECATION_MSG):
c.__init__ = wrap_deprecated(c.__init__, f"{c.__module__}.{c.__name__}", msg, version="2.9", remove=True)
c.__doc__ = f"""DEPRECATED
.. warning::
This class is deprecated from version 2.8. It will be removed in the 2.9 release.
{msg}
{c.__doc__}
"""
UNSUPPORTED.append(c)
return c
def dropping_const_support(c, msg=DEPRECATION_MSG, name=None):
c.__doc__ = f"""[DEPRECATED]
.. warning::
This object is deprecated deprecated from version 2.8. It will be removed in the 2.9 release.
{msg}
{c.__doc__}
"""
return c
dropping_class_io_support = partial(dropping_class_support, msg=IO_DEPRECATION_MSG)
dropping_io_support = deprecated(IO_DEPRECATION_MSG, version="2.9", remove=True)
def fail_with_message(message):
"""Generate decorator to give users message about missing TorchAudio extension."""
def decorator(func):
@wraps(func)
def wrapped(*args, **kwargs):
raise RuntimeError(f"{func.__module__}.{func.__name__} {message}")
return wrapped
return decorator
def no_op(func):
"""Op-op decorator. Used in place of fail_with_message when a functionality that requires extension works fine."""
return func
|