




import importlib
import inspect
import json
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Dict, Optional, Union
from urllib import request

from huggingface_hub import HfFolder, cached_download, hf_hub_download, model_info
from packaging import version

from .. import __version__
from . import DIFFUSERS_DYNAMIC_MODULE_NAME, HF_MODULES_CACHE, logging


lowercase__				:     Dict 		=						(
    "https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py"
)


lowercase__				:     Any 		=						logging.get_logger(__name__)  # pylint: disable=invalid-name


def 						lowerCamelCase__    (		):







	'''simple docstring'''





	snake_case_		    =     "https://pypi.org/pypi/diffusers/json"
	snake_case_		    =     json.loads(request.urlopen(_A							).read()							)["releases"].keys()
	return sorted(_A     ,	key=lambda _A							: version.Version(_A							)							)


def 						lowerCamelCase__    (		):







	'''simple docstring'''





	if HF_MODULES_CACHE in sys.path:
		return

	sys.path.append(_A							)
	os.makedirs(_A     ,	exist_ok=_A							)
	snake_case_		    =     Path(_A							) / "__init__.py"
	if not init_path.exists():
		init_path.touch()


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	init_hf_modules()
	snake_case_		    =     Path(_A							) / name
	# If the parent module does not exist yet, recursively create it.
	if not dynamic_module_path.parent.exists():
		create_dynamic_module(dynamic_module_path.parent							)
	os.makedirs(_A     ,	exist_ok=_A							)
	snake_case_		    =     dynamic_module_path / "__init__.py"
	if not init_path.exists():
		init_path.touch()


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	with open(_A     ,	"r"     ,	encoding="utf-8"							) as f:
		snake_case_		    =     f.read()

	# Imports of the form `import .xxx`
	snake_case_		    =     re.findall("^\s*import\s+\.(\S+)\s*$"     ,	_A     ,	flags=re.MULTILINE							)
	# Imports of the form `from .xxx import yyy`
	relative_imports += re.findall("^\s*from\s+\.(\S+)\s+import"     ,	_A     ,	flags=re.MULTILINE							)
	# Unique-ify
	return list(set(_A							)							)


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	snake_case_		    =     False
	snake_case_		    =     [module_file]
	snake_case_		    =     []

	# Let's recurse through all relative imports
	while not no_change:
		snake_case_		    =     []
		for f in files_to_check:
			new_imports.extend(get_relative_imports(_A							)							)

		snake_case_		    =     Path(_A							).parent
		snake_case_		    =     [str(module_path / m							) for m in new_imports]
		snake_case_		    =     [f for f in new_import_files if f not in all_relative_imports]
		snake_case_		    =     [f"{f}.py" for f in new_import_files]

		snake_case_		    =     len(_A							) == 0
		all_relative_imports.extend(_A							)

	return all_relative_imports


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	with open(_A     ,	"r"     ,	encoding="utf-8"							) as f:
		snake_case_		    =     f.read()

	# Imports of the form `import xxx`
	snake_case_		    =     re.findall("^\s*import\s+(\S+)\s*$"     ,	_A     ,	flags=re.MULTILINE							)
	# Imports of the form `from xxx import yyy`
	imports += re.findall("^\s*from\s+(\S+)\s+import"     ,	_A     ,	flags=re.MULTILINE							)
	# Only keep the top-level module
	snake_case_		    =     [imp.split("."							)[0] for imp in imports if not imp.startswith("."							)]

	# Unique-ify and test we got them all
	snake_case_		    =     list(set(_A							)							)
	snake_case_		    =     []
	for imp in imports:
		try:
			importlib.import_module(_A							)
		except ImportError:
			missing_packages.append(_A							)

	if len(_A							) > 0:
		raise ImportError(
		    "This modeling file requires the following packages that were not found in your environment: "
		    f"{', '.join(_A							)}. Run `pip install {' '.join(_A							)}`"							)

	return get_relative_imports(_A							)


def 						lowerCamelCase__    (		_A     ,	_A							):







	'''simple docstring'''





	snake_case_		    =     module_path.replace(os.path.sep     ,	"."							)
	snake_case_		    =     importlib.import_module(_A							)

	if class_name is None:
		return find_pipeline_class(_A							)
	return getattr(_A     ,	_A							)


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	from ..pipelines import DiffusionPipeline

	snake_case_		    =     dict(inspect.getmembers(_A     ,	inspect.isclass							)							)

	snake_case_		    =     None
	for cls_name, cls in cls_members.items():
		if (
		    cls_name != DiffusionPipeline.__name__
		    and issubclass(cls     ,	_A							)
		    and cls.__module__.split("."							)[0] != "diffusers"
		):
			if pipeline_class is not None:
				raise ValueError(
				    f"Multiple classes that inherit from {DiffusionPipeline.__name__} have been found:"
				    f" {pipeline_class.__name__}, and {cls_name}. Please make sure to define only one in"
				    f" {loaded_module}."							)
			snake_case_		    =     cls

	return pipeline_class


def 						lowerCamelCase__    (		_A     ,	_A     ,	_A = None     ,	_A = False     ,	_A = False     ,	_A = None     ,	_A = None     ,	_A = None     ,	_A = False     ,	):







	'''simple docstring'''





	snake_case_		    =     str(_A							)

	snake_case_		    =     os.path.join(_A     ,	_A							)

	if os.path.isfile(_A							):
		snake_case_		    =     module_file_or_url
		snake_case_		    =     "local"
	elif pretrained_model_name_or_path.count("/"							) == 0:
		snake_case_		    =     get_diffusers_versions()
		# cut ".dev0"
		snake_case_		    =     "v" + ".".join(__version__.split("."							)[:3]							)

		# retrieve github version that matches
		if revision is None:
			snake_case_		    =     latest_version if latest_version[1:] in available_versions else "main"
			logger.info(f"Defaulting to latest_version: {revision}."							)
		elif revision in available_versions:
			snake_case_		    =     f"v{revision}"
		elif revision == "main":
			snake_case_		    =     revision
		else:
			raise ValueError(
			    f"`custom_revision`: {revision} does not exist. Please make sure to choose one of"
			    f" {', '.join(available_versions + ['main']							)}."							)

		# community pipeline on GitHub
		snake_case_		    =     COMMUNITY_PIPELINES_URL.format(revision=_A     ,	pipeline=_A							)
		try:
			snake_case_		    =     cached_download(
			    _A     ,	cache_dir=_A     ,	force_download=_A     ,	proxies=_A     ,	resume_download=_A     ,	local_files_only=_A     ,	use_auth_token=_A     ,	)
			snake_case_		    =     "git"
			snake_case_		    =     pretrained_model_name_or_path + ".py"
		except EnvironmentError:
			logger.error(f"Could not locate the {module_file} inside {pretrained_model_name_or_path}."							)
			raise
	else:
		try:
			# Load from URL or cache if already cached
			snake_case_		    =     hf_hub_download(
			    _A     ,	_A     ,	cache_dir=_A     ,	force_download=_A     ,	proxies=_A     ,	resume_download=_A     ,	local_files_only=_A     ,	use_auth_token=_A     ,	)
			snake_case_		    =     os.path.join("local"     ,	"--".join(pretrained_model_name_or_path.split("/"							)							)							)
		except EnvironmentError:
			logger.error(f"Could not locate the {module_file} inside {pretrained_model_name_or_path}."							)
			raise

    # Check we have all the requirements in our environment
	snake_case_		    =     check_imports(_A							)

	# Now we move the module inside our cached dynamic modules.
	snake_case_		    =     DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
	create_dynamic_module(_A							)
	snake_case_		    =     Path(_A							) / full_submodule
	if submodule == "local" or submodule == "git":
		# We always copy local files (we could hash the file to see if there was a change, and give them the name of
		# that hash, to only copy when there is a modification but it seems overkill for now).
		# The only reason we do the copy is to avoid putting too many folders in sys.path.
		shutil.copy(_A     ,	submodule_path / module_file							)
		for module_needed in modules_needed:
			snake_case_		    =     f"{module_needed}.py"
			shutil.copy(os.path.join(_A     ,	_A							)     ,	submodule_path / module_needed							)
	else:
		# Get the commit hash
		# TODO: we will get this info in the etag soon, so retrieve it from there and not here.
		if isinstance(_A     ,	_A							):
			snake_case_		    =     use_auth_token
		elif use_auth_token is True:
			snake_case_		    =     HfFolder.get_token()
		else:
			snake_case_		    =     None

		snake_case_		    =     model_info(_A     ,	revision=_A     ,	token=_A							).sha

		# The module file will end up being placed in a subfolder with the git hash of the repo. This way we get the
		# benefit of versioning.
		snake_case_		    =     submodule_path / commit_hash
		snake_case_		    =     full_submodule + os.path.sep + commit_hash
		create_dynamic_module(_A							)

		if not (submodule_path / module_file).exists():
			shutil.copy(_A     ,	submodule_path / module_file							)
		# Make sure we also have every file with relative
		for module_needed in modules_needed:
			if not (submodule_path / module_needed).exists():
				get_cached_module_file(
				    _A     ,	f"{module_needed}.py"     ,	cache_dir=_A     ,	force_download=_A     ,	resume_download=_A     ,	proxies=_A     ,	use_auth_token=_A     ,	revision=_A     ,	local_files_only=_A     ,	)
	return os.path.join(_A     ,	_A							)






def 						lowerCamelCase__    (		_A     ,	_A     ,	_A = None     ,	_A = None     ,	_A = False     ,	_A = False     ,	_A = None     ,	_A = None     ,	_A = None     ,	_A = False     ,	**_A     ,	):







	'''simple docstring'''





	snake_case_		    =     get_cached_module_file(
	    _A     ,	_A     ,	cache_dir=_A     ,	force_download=_A     ,	resume_download=_A     ,	proxies=_A     ,	use_auth_token=_A     ,	revision=_A     ,	local_files_only=_A     ,	)
	return get_class_in_module(_A     ,	final_module.replace(".py"     ,	""							)							)






from torch import nn


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	if act_fn in ["swish", "silu"]:
		return nn.SiLU()
	elif act_fn == "mish":
		return nn.Mish()
	elif act_fn == "gelu":
		return nn.GELU()
	else:
		raise ValueError(f"Unsupported activation function: {act_fn}"							)
