|
|
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: |
|
|
|
|
|
raise RuntimeError( |
|
|
f"Unexpected environment variable value `{var}={val}`. " |
|
|
f"Expected one of {trues + falses}") |
|
|
|
|
|
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: |
|
|
|
|
|
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 |
|
|
|