
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


_A   =     (
    'https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py'
)


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






def    lowerCamelCase__				(				):




			"""simple docstring"""


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






def    lowerCamelCase__				(				):




			"""simple docstring"""


			if HF_MODULES_CACHE in sys.path:
						return

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






def    lowerCamelCase__				(				__lowerCAmelCase  :  Tuple				):




			"""simple docstring"""


			init_hf_modules()
			lowerCAmelCase_						=	Path(__snake_case				) / 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(__snake_case    ,							exist_ok=__snake_case				)
			lowerCAmelCase_						=	dynamic_module_path / "__init__.py"
			if not init_path.exists():
						init_path.touch()






def    lowerCamelCase__				(				__lowerCAmelCase  :  Dict				):




			"""simple docstring"""


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

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






def    lowerCamelCase__				(				__lowerCAmelCase  :  Dict				):




			"""simple docstring"""


			lowerCAmelCase_						=	False
			lowerCAmelCase_						=	[module_file]
			lowerCAmelCase_						=	[]

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

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

						lowerCAmelCase_						=	len(__snake_case				) == 0
						all_relative_imports.extend(__snake_case				)

			return all_relative_imports






def    lowerCamelCase__				(				__lowerCAmelCase  :  Any				):




			"""simple docstring"""


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

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

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

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

			return get_relative_imports(__snake_case				)






def    lowerCamelCase__				(				__lowerCAmelCase  :  Tuple    ,							__lowerCAmelCase  :  Dict				):




			"""simple docstring"""


			lowerCAmelCase_						=	module_path.replace(os.path.sep    ,							"."				)
			lowerCAmelCase_						=	importlib.import_module(__snake_case				)

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






def    lowerCamelCase__				(				__lowerCAmelCase  :  List[Any]				):




			"""simple docstring"""


			from ..pipelines import DiffusionPipeline

			lowerCAmelCase_						=	dict(inspect.getmembers(__snake_case    ,							inspect.isclass				)				)

			lowerCAmelCase_						=	None
			for cls_name, cls in cls_members.items():
						if (
						    cls_name != DiffusionPipeline.__name__
						    and issubclass(cls    ,							__snake_case				)
						    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}."""				)
									lowerCAmelCase_						=	cls

			return pipeline_class






def    lowerCamelCase__				(				__lowerCAmelCase  :  Dict    ,							__lowerCAmelCase  :  Any    ,							__lowerCAmelCase  :  int = None    ,							__lowerCAmelCase  :  int = False    ,							__lowerCAmelCase  :  Any = False    ,							__lowerCAmelCase  :  List[Any] = None    ,							__lowerCAmelCase  :  List[str] = None    ,							__lowerCAmelCase  :  int = None    ,							__lowerCAmelCase  :  int = False    ,							):




			"""simple docstring"""


			lowerCAmelCase_						=	str(__snake_case				)

			lowerCAmelCase_						=	os.path.join(__snake_case    ,							__snake_case				)

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

						# retrieve github version that matches
						if revision is None:
									lowerCAmelCase_						=	latest_version if latest_version[1:] in available_versions else "main"
									logger.info(F"""Defaulting to latest_version: {revision}."""				)
						elif revision in available_versions:
									lowerCAmelCase_						=	F"""v{revision}"""
						elif revision == "main":
									lowerCAmelCase_						=	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
						lowerCAmelCase_						=	COMMUNITY_PIPELINES_URL.format(revision=__snake_case    ,							pipeline=__snake_case				)
						try:
									lowerCAmelCase_						=	cached_download(
									    __snake_case    ,							cache_dir=__snake_case    ,							force_download=__snake_case    ,							proxies=__snake_case    ,							resume_download=__snake_case    ,							local_files_only=__snake_case    ,							use_auth_token=__snake_case    ,							)
									lowerCAmelCase_						=	"git"
									lowerCAmelCase_						=	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
									lowerCAmelCase_						=	hf_hub_download(
									    __snake_case    ,							__snake_case    ,							cache_dir=__snake_case    ,							force_download=__snake_case    ,							proxies=__snake_case    ,							resume_download=__snake_case    ,							local_files_only=__snake_case    ,							use_auth_token=__snake_case    ,							)
									lowerCAmelCase_						=	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
			lowerCAmelCase_						=	check_imports(__snake_case				)

			# Now we move the module inside our cached dynamic modules.
			lowerCAmelCase_						=	DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
			create_dynamic_module(__snake_case				)
			lowerCAmelCase_						=	Path(__snake_case				) / 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(__snake_case    ,							submodule_path / module_file				)
						for module_needed in modules_needed:
									lowerCAmelCase_						=	F"""{module_needed}.py"""
									shutil.copy(os.path.join(__snake_case    ,							__snake_case				)    ,							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(__snake_case    ,							__snake_case				):
									lowerCAmelCase_						=	use_auth_token
						elif use_auth_token is True:
									lowerCAmelCase_						=	HfFolder.get_token()
						else:
									lowerCAmelCase_						=	None

						lowerCAmelCase_						=	model_info(__snake_case    ,							revision=__snake_case    ,							token=__snake_case				).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.
						lowerCAmelCase_						=	submodule_path / commit_hash
						lowerCAmelCase_						=	full_submodule + os.path.sep + commit_hash
						create_dynamic_module(__snake_case				)

						if not (submodule_path / module_file).exists():
									shutil.copy(__snake_case    ,							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(
												    __snake_case    ,							F"""{module_needed}.py"""    ,							cache_dir=__snake_case    ,							force_download=__snake_case    ,							resume_download=__snake_case    ,							proxies=__snake_case    ,							use_auth_token=__snake_case    ,							revision=__snake_case    ,							local_files_only=__snake_case    ,							)
			return os.path.join(__snake_case    ,							__snake_case				)






def    lowerCamelCase__				(				__lowerCAmelCase  :  Union[str, Any]    ,							__lowerCAmelCase  :  str    ,							__lowerCAmelCase  :  str = None    ,							__lowerCAmelCase  :  Optional[int] = None    ,							__lowerCAmelCase  :  Dict = False    ,							__lowerCAmelCase  :  Optional[Any] = False    ,							__lowerCAmelCase  :  List[str] = None    ,							__lowerCAmelCase  :  List[str] = None    ,							__lowerCAmelCase  :  str = None    ,							__lowerCAmelCase  :  str = False    ,							**__lowerCAmelCase  :  List[str]    ,							):




			"""simple docstring"""


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








"""simple docstring"""






from ..utils import DummyObject, requires_backends



class 			A__   (  metaclass=__SCREAMING_SNAKE_CASE							):




	'''simple docstring'''





	SCREAMING_SNAKE_CASE          =	['torch', 'torchsde']

	def __init__(     self:		int       ,			*_SCREAMING_SNAKE_CASE:		Optional[Any]       ,			**_SCREAMING_SNAKE_CASE:		str)				->       Dict:

				"""simple docstring"""

				requires_backends(self       ,			["torch", "torchsde"])

	@classmethod
	def 							_SCREAMING_SNAKE_CASE   (     cls:		Optional[Any]       ,			*_SCREAMING_SNAKE_CASE:		Optional[int]       ,			**_SCREAMING_SNAKE_CASE:		Optional[int])				->       Optional[Any]:

				"""simple docstring"""

				requires_backends(cls       ,			["torch", "torchsde"])







	@classmethod
	def 							_SCREAMING_SNAKE_CASE   (     cls:		Dict       ,			*_SCREAMING_SNAKE_CASE:		List[Any]       ,			**_SCREAMING_SNAKE_CASE:		Any)				->       Optional[int]:

				"""simple docstring"""

				requires_backends(cls       ,			["torch", "torchsde"])