




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_		(       )     ->       Tuple:





				"""simple docstring"""


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







def 			lowerCamelCase_		(       )     ->       Union[str, Any]:





				"""simple docstring"""


				if HF_MODULES_CACHE in sys.path:
								return

				sys.path.append(UpperCamelCase__      )
				os.makedirs(UpperCamelCase__ ,     exist_ok=UpperCamelCase__      )
				__lowerCamelCase		       =    Path(UpperCamelCase__      ) / '__init__.py'
				if not init_path.exists():
								init_path.touch()







def 			lowerCamelCase_		(       UpperCamelCase__				:      Union[str, os.PathLike]      )     ->       Any:





				"""simple docstring"""


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







def 			lowerCamelCase_		(       UpperCamelCase__				:      Optional[Any]      )     ->       Optional[int]:





				"""simple docstring"""


				with open(UpperCamelCase__ ,     'r' ,     encoding='utf-8'      ) as f:
								__lowerCamelCase		       =    f.read()

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







def 			lowerCamelCase_		(       UpperCamelCase__				:      Dict      )     ->       Optional[Any]:





				"""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(UpperCamelCase__      )      )

								__lowerCamelCase		       =    Path(UpperCamelCase__      ).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(UpperCamelCase__      ) == 0
								all_relative_imports.extend(UpperCamelCase__      )

				return all_relative_imports







def 			lowerCamelCase_		(       UpperCamelCase__				:      Optional[Any]      )     ->       List[str]:





				"""simple docstring"""


				with open(UpperCamelCase__ ,     'r' ,     encoding='utf-8'      ) as f:
								__lowerCamelCase		       =    f.read()

				# Imports of the form `import xxx`
				__lowerCamelCase		       =    re.findall('^\s*import\s+(\S+)\s*$' ,     UpperCamelCase__ ,     flags=re.MULTILINE      )
				# Imports of the form `from xxx import yyy`
				imports += re.findall('^\s*from\s+(\S+)\s+import' ,     UpperCamelCase__ ,     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(UpperCamelCase__      )      )
				__lowerCamelCase		       =    []
				for imp in imports:
								try:
												importlib.import_module(UpperCamelCase__      )
								except ImportError:
												missing_packages.append(UpperCamelCase__      )

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

				return get_relative_imports(UpperCamelCase__      )







def 			lowerCamelCase_		(       UpperCamelCase__				:      int ,     UpperCamelCase__				:      Union[str, Any]      )     ->       int:





				"""simple docstring"""


				__lowerCamelCase		       =    module_path.replace(os.path.sep ,     '.'      )
				__lowerCamelCase		       =    importlib.import_module(UpperCamelCase__      )

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







def 			lowerCamelCase_		(       UpperCamelCase__				:      str      )     ->       Tuple:





				"""simple docstring"""


				from ..pipelines import DiffusionPipeline

				__lowerCamelCase		       =    dict(inspect.getmembers(UpperCamelCase__ ,     inspect.isclass      )      )

				__lowerCamelCase		       =    None
				for cls_name, cls in cls_members.items():
								if (
								    cls_name != DiffusionPipeline.__name__
								    and issubclass(cls ,     UpperCamelCase__      )
								    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_		(       UpperCamelCase__				:      Union[str, os.PathLike] ,     UpperCamelCase__				:      str ,     UpperCamelCase__				:      Optional[Union[str, os.PathLike]] = None ,     UpperCamelCase__				:      bool = False ,     UpperCamelCase__				:      bool = False ,     UpperCamelCase__				:      Optional[Dict[str, str]] = None ,     UpperCamelCase__				:      Optional[Union[bool, str]] = None ,     UpperCamelCase__				:      Optional[str] = None ,     UpperCamelCase__				:      bool = False ,     )     ->       Tuple:





				"""simple docstring"""


				__lowerCamelCase		       =    str(UpperCamelCase__      )

				__lowerCamelCase		       =    os.path.join(UpperCamelCase__ ,     UpperCamelCase__      )

				if os.path.isfile(UpperCamelCase__      ):
								__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=UpperCamelCase__ ,     pipeline=UpperCamelCase__      )
								try:
												__lowerCamelCase		       =    cached_download(
												    UpperCamelCase__ ,     cache_dir=UpperCamelCase__ ,     force_download=UpperCamelCase__ ,     proxies=UpperCamelCase__ ,     resume_download=UpperCamelCase__ ,     local_files_only=UpperCamelCase__ ,     use_auth_token=UpperCamelCase__ ,     )
												__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(
												    UpperCamelCase__ ,     UpperCamelCase__ ,     cache_dir=UpperCamelCase__ ,     force_download=UpperCamelCase__ ,     proxies=UpperCamelCase__ ,     resume_download=UpperCamelCase__ ,     local_files_only=UpperCamelCase__ ,     use_auth_token=UpperCamelCase__ ,     )
												__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(UpperCamelCase__      )

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

								__lowerCamelCase		       =    model_info(UpperCamelCase__ ,     revision=UpperCamelCase__ ,     token=UpperCamelCase__      ).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(UpperCamelCase__      )

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







def 			lowerCamelCase_		(       UpperCamelCase__				:      Union[str, os.PathLike] ,     UpperCamelCase__				:      str ,     UpperCamelCase__				:      Optional[str] = None ,     UpperCamelCase__				:      Optional[Union[str, os.PathLike]] = None ,     UpperCamelCase__				:      bool = False ,     UpperCamelCase__				:      bool = False ,     UpperCamelCase__				:      Optional[Dict[str, str]] = None ,     UpperCamelCase__				:      Optional[Union[bool, str]] = None ,     UpperCamelCase__				:      Optional[str] = None ,     UpperCamelCase__				:      bool = False ,     **UpperCamelCase__				:      List[str] ,     )     ->       str:





				"""simple docstring"""


				__lowerCamelCase		       =    get_cached_module_file(
				    UpperCamelCase__ ,     UpperCamelCase__ ,     cache_dir=UpperCamelCase__ ,     force_download=UpperCamelCase__ ,     resume_download=UpperCamelCase__ ,     proxies=UpperCamelCase__ ,     use_auth_token=UpperCamelCase__ ,     revision=UpperCamelCase__ ,     local_files_only=UpperCamelCase__ ,     )
				return get_class_in_module(UpperCamelCase__ ,     final_module.replace('.py' ,     ''      )      )








from ..utils import DummyObject, requires_backends







class     __lowerCAmelCase (       metaclass=__magic_name__ ):

			"""simple docstring"""







			snake_case_   		=       ['''onnx''']


			def __init__(      self			,						*lowerCamelCase__			,						**lowerCamelCase__							)    ->      Optional[Any]:



							'''simple docstring'''
							requires_backends(self			,						['onnx']							)


			@classmethod
			def       lowercase_    (      cls			,						*lowerCamelCase__			,						**lowerCamelCase__							)    ->      Dict:



							'''simple docstring'''
							requires_backends(cls			,						['onnx']							)






			@classmethod
			def       lowercase_    (      cls			,						*lowerCamelCase__			,						**lowerCamelCase__							)    ->      List[str]:



							'''simple docstring'''
							requires_backends(cls			,						['onnx']							)


