global _package_map

pkg_exists = importlib.util.find_spec(pkg_name) is not None

pkg_version = "N/A"

if pkg_exists:

if _package_map is None:

_package_map = defaultdict(list)

try:

# Fallback for Python < 3.10

for dist in importlib_metadata.distributions():

_top_level_declared = (dist.read_text("top_level.txt") or "").split()

# Infer top-level package names from file structure

_inferred_opt_names = {

f.parts[0] if len(f.parts) > 1 else inspect.getmodulename(f) for f in (dist.files or [])

} - {None}

_top_level_inferred = filter(lambda name: "." not in name, _inferred_opt_names)

for pkg in _top_level_declared or _top_level_inferred:

_package_map[pkg].append(dist.metadata["Name"])

except Exception as _:

pass

try:

if get_dist_name and pkg_name in _package_map and _package_map[pkg_name]:

if len(_package_map[pkg_name]) > 1:

logger.warning(

f"Multiple distributions found for package {pkg_name}. Picked distribution: {_package_map[pkg_name][0]}"

)

pkg_name = _package_map[pkg_name][0]

pkg_version = importlib_metadata.version(pkg_name)

logger.debug(f"Successfully imported {pkg_name} version {pkg_version}")

except (ImportError, importlib_metadata.PackageNotFoundError):

pkg_exists = False

[

("bs4", (is_bs4_available, BS4_IMPORT_ERROR)),

("inflect", (is_inflect_available, INFLECT_IMPORT_ERROR)),

("onnx", (is_onnx_available, ONNX_IMPORT_ERROR)),

("opencv", (is_opencv_available, OPENCV_IMPORT_ERROR)),

("scipy", (is_scipy_available, SCIPY_IMPORT_ERROR)),

("torch", (is_torch_available, PYTORCH_IMPORT_ERROR)),

("transformers", (is_transformers_available, TRANSFORMERS_IMPORT_ERROR)),

("unidecode", (is_unidecode_available, UNIDECODE_IMPORT_ERROR)),

("librosa", (is_librosa_available, LIBROSA_IMPORT_ERROR)),

("note_seq", (is_note_seq_available, NOTE_SEQ_IMPORT_ERROR)),

("wandb", (is_wandb_available, WANDB_IMPORT_ERROR)),

("tensorboard", (is_tensorboard_available, TENSORBOARD_IMPORT_ERROR)),

("compel", (is_compel_available, COMPEL_IMPORT_ERROR)),

("ftfy", (is_ftfy_available, FTFY_IMPORT_ERROR)),

("torchsde", (is_torchsde_available, TORCHSDE_IMPORT_ERROR)),

("invisible_watermark", (is_invisible_watermark_available, INVISIBLE_WATERMARK_IMPORT_ERROR)),

("peft", (is_peft_available, PEFT_IMPORT_ERROR)),

("safetensors", (is_safetensors_available, SAFETENSORS_IMPORT_ERROR)),

("bitsandbytes", (is_bitsandbytes_available, BITSANDBYTES_IMPORT_ERROR)),

("sentencepiece", (is_sentencepiece_available, SENTENCEPIECE_IMPORT_ERROR)),

("imageio", (is_imageio_available, IMAGEIO_IMPORT_ERROR)),

("gguf", (is_gguf_available, GGUF_IMPORT_ERROR)),

("torchao", (is_torchao_available, TORCHAO_IMPORT_ERROR)),

("quanto", (is_optimum_quanto_available, QUANTO_IMPORT_ERROR)),

("sdnq", (is_sdnq_available, SDNQ_IMPORT_ERROR)),

("pytorch_retinaface", (is_pytorch_retinaface_available, PYTORCH_RETINAFACE_IMPORT_ERROR)),

("better_profanity", (is_better_profanity_available, BETTER_PROFANITY_IMPORT_ERROR)),

("nltk", (is_nltk_available, NLTK_IMPORT_ERROR)),

("torch_neuronx", (is_torch_neuronx_available, TORCH_NEURONX_IMPORT_ERROR)),

]

if not isinstance(backends, (list, tuple)):

backends = [backends]

name = obj.__name__ if hasattr(obj, "__name__") else obj.__class__.__name__

checks = (BACKENDS_MAPPING[backend] for backend in backends)

failed = [msg.format(name) for available, msg in checks if not available()]

if failed:

raise ImportError("".join(failed))

if name in [

"VersatileDiffusionTextToImagePipeline",

"VersatileDiffusionPipeline",

"VersatileDiffusionDualGuidedPipeline",

"StableDiffusionImageVariationPipeline",

"UnCLIPPipeline",

] and is_transformers_version("<", "4.25.0"):

raise ImportError(

f"You need to install `transformers>=4.25` in order to use {name}: \n```\n pip install"

" --upgrade transformers \n```"

)

if name in ["StableDiffusionDepth2ImgPipeline", "StableDiffusionPix2PixZeroPipeline"] and is_transformers_version(

"<", "4.26.0"

):

raise ImportError(

f"You need to install `transformers>=4.26` in order to use {name}: \n```\n pip install"

" --upgrade transformers \n```"

"""

def __getattr__(cls, key):

if key.startswith("_") and key not in ["_load_connected_pipes", "_is_onnx"]:

return super().__getattr__(cls, key)

"""

if operation not in STR_OPERATION_TO_FUNC.keys():

raise ValueError(f"`operation` must be one of {list(STR_OPERATION_TO_FUNC.keys())}, received {operation}")

operation = STR_OPERATION_TO_FUNC[operation]

if isinstance(library_or_version, str):

library_or_version = parse(importlib_metadata.version(library_or_version))

"""

"""

if not is_torch_xla_available:

return False

"""

if not _transformers_available:

return False

"""

if not _kernels_available:

return False

"""

if not _hf_hub_available:

return False

"""

if not _accelerate_available:

return False

"""

if not _peft_available:

return False

"""

if not _bitsandbytes_available:

return False

"""

if not _gguf_available:

return False

"""

if not _sdnq_available:

return False

"""

if not _torchao_available:

return False

"""

if not _optimum_quanto_available:

return False

"""

if not _nvidia_modelopt_available:

return False

"""

if not _xformers_available:

return False

"""

if not _sageattention_available:

return False

"""

if not _flash_attn_available:

return False

"""

objects = {}

for name in dir(module):

if name.startswith("_"):

continue

objects[name] = getattr(module, name)

"""

# Very heavily inspired by optuna.integration._IntegrationModule

# https://github.com/optuna/optuna/blob/master/optuna/integration/__init__.py

def __init__(self, name, module_file, import_structure, module_spec=None, extra_objects=None):

super().__init__(name)

self._modules = set(import_structure.keys())

self._class_to_module = {}

for key, values in import_structure.items():

for value in values:

self._class_to_module[value] = key

# Needed for autocompletion in an IDE

self.__all__ = list(import_structure.keys()) + list(chain(*import_structure.values()))

self.__file__ = module_file

self.__spec__ = module_spec

self.__path__ = [os.path.dirname(module_file)]

self._objects = {} if extra_objects is None else extra_objects

self._name = name

self._import_structure = import_structure

# Needed for autocompletion in an IDE

def __dir__(self):

result = super().__dir__()

# The elements of self.__all__ that are submodules may or may not be in the dir already, depending on whether

# they have been accessed or not. So we only add the elements of self.__all__ that are not already in the dir.

for attr in self.__all__:

if attr not in result:

result.append(attr)

return result

def __getattr__(self, name: str) -> Any:

if name in self._objects: