



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


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


_snake_case				=					logging.get_logger(__name__)  # pylint: disable=invalid-name
def     _a							(     )     ->						Dict:

							"""simple docstring"""





							__UpperCamelCase          =  'https://pypi.org/pypi/diffusers/json'
							__UpperCamelCase          =  json.loads(request.urlopen(_lowerCAmelCase		).read()		)['releases'].keys()
							return sorted(_lowerCAmelCase				, key=lambda __lowercase		: version.Version(_lowerCAmelCase		)		)
def     _a							(     )     ->						Tuple:

							"""simple docstring"""





							if HF_MODULES_CACHE in sys.path:
														return

							sys.path.append(_lowerCAmelCase		)
							os.makedirs(_lowerCAmelCase				, exist_ok=_lowerCAmelCase		)
							__UpperCamelCase          =  Path(_lowerCAmelCase		) / '__init__.py'
							if not init_path.exists():
														init_path.touch()
def     _a							(     __lowercase		)     ->						int:

							"""simple docstring"""





							init_hf_modules()
							__UpperCamelCase          =  Path(_lowerCAmelCase		) / 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(_lowerCAmelCase				, exist_ok=_lowerCAmelCase		)
							__UpperCamelCase          =  dynamic_module_path / '__init__.py'
							if not init_path.exists():
														init_path.touch()
def     _a							(     __lowercase		)     ->						str:

							"""simple docstring"""





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

							# Imports of the form `import .xxx`
							__UpperCamelCase          =  re.findall('^\s*import\s+\.(\S+)\s*$'				, _lowerCAmelCase				, flags=re.MULTILINE		)
							# Imports of the form `from .xxx import yyy`
							relative_imports += re.findall('^\s*from\s+\.(\S+)\s+import'				, _lowerCAmelCase				, flags=re.MULTILINE		)
							# Unique-ify
							return list(set(_lowerCAmelCase		)		)
def     _a							(     __lowercase		)     ->						Dict:

							"""simple docstring"""





							__UpperCamelCase          =  False
							__UpperCamelCase          =  [module_file]
							__UpperCamelCase          =  []

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

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

														__UpperCamelCase          =  len(_lowerCAmelCase		) == 0
														all_relative_imports.extend(_lowerCAmelCase		)

							return all_relative_imports
def     _a							(     __lowercase		)     ->						Any:

							"""simple docstring"""





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

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

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

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

							return get_relative_imports(_lowerCAmelCase		)
def     _a							(     __lowercase				, __lowercase		)     ->						Union[str, Any]:

							"""simple docstring"""





							__UpperCamelCase          =  module_path.replace(os.path.sep				, '.'		)
							__UpperCamelCase          =  importlib.import_module(_lowerCAmelCase		)

							if class_name is None:
														return find_pipeline_class(_lowerCAmelCase		)
							return getattr(_lowerCAmelCase				, _lowerCAmelCase		)
def     _a							(     __lowercase		)     ->						Optional[int]:

							"""simple docstring"""





							from ..pipelines import DiffusionPipeline

							__UpperCamelCase          =  dict(inspect.getmembers(_lowerCAmelCase				, inspect.isclass		)		)

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

							return pipeline_class
def     _a							(     __lowercase				, __lowercase				, __lowercase = None				, __lowercase = False				, __lowercase = False				, __lowercase = None				, __lowercase = None				, __lowercase = None				, __lowercase = False				, )     ->						List[Any]:

							"""simple docstring"""





							__UpperCamelCase          =  str(_lowerCAmelCase		)

							__UpperCamelCase          =  os.path.join(_lowerCAmelCase				, _lowerCAmelCase		)

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

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

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

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

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

def     _a							(     __lowercase				, __lowercase				, __lowercase = None				, __lowercase = None				, __lowercase = False				, __lowercase = False				, __lowercase = None				, __lowercase = None				, __lowercase = None				, __lowercase = False				, **__lowercase				, )     ->						str:

							"""simple docstring"""





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







from datasets.utils.patching import _PatchedModuleObj, patch_submodule

from . import _test_patching





def 						_lowerCAmelCase							(     )    ->						Tuple:


						'''simple docstring'''






						import os as original_os
						from os import path as original_path
						from os import rename as original_rename
						from os.path import dirname as original_dirname
						from os.path import join as original_join

						assert _test_patching.os is original_os
						assert _test_patching.path is original_path
						assert _test_patching.join is original_join

						assert _test_patching.renamed_os is original_os
						assert _test_patching.renamed_path is original_path
						assert _test_patching.renamed_join is original_join

						__snake_case				=		"__test_patch_submodule_mock__"
						with patch_submodule(_test_patching   ,						"os.path.join"   ,						_lowerCAmelCase			):
												# Every way to access os.path.join must be patched, and the rest must stay untouched

												# check os.path.join
												assert isinstance(_test_patching.os   ,						_PatchedModuleObj			)
												assert isinstance(_test_patching.os.path   ,						_PatchedModuleObj			)
												assert _test_patching.os.path.join is mock

												# check path.join
												assert isinstance(_test_patching.path   ,						_PatchedModuleObj			)
												assert _test_patching.path.join is mock

												# check join
												assert _test_patching.join is mock

												# check that the other attributes are untouched
												assert _test_patching.os.rename is original_rename
												assert _test_patching.path.dirname is original_dirname
												assert _test_patching.os.path.dirname is original_dirname

												# Even renamed modules or objects must be patched

												# check renamed_os.path.join
												assert isinstance(_test_patching.renamed_os   ,						_PatchedModuleObj			)
												assert isinstance(_test_patching.renamed_os.path   ,						_PatchedModuleObj			)
												assert _test_patching.renamed_os.path.join is mock

												# check renamed_path.join
												assert isinstance(_test_patching.renamed_path   ,						_PatchedModuleObj			)
												assert _test_patching.renamed_path.join is mock

												# check renamed_join
												assert _test_patching.renamed_join is mock

												# check that the other attributes are untouched
												assert _test_patching.renamed_os.rename is original_rename
												assert _test_patching.renamed_path.dirname is original_dirname
												assert _test_patching.renamed_os.path.dirname is original_dirname

						# check that everthing is back to normal when the patch is over

						assert _test_patching.os is original_os
						assert _test_patching.path is original_path
						assert _test_patching.join is original_join

						assert _test_patching.renamed_os is original_os
						assert _test_patching.renamed_path is original_path
						assert _test_patching.renamed_join is original_join





def 						_lowerCAmelCase							(     )    ->						int:


						'''simple docstring'''






						assert _test_patching.open is open

						__snake_case				=		"__test_patch_submodule_builtin_mock__"
						# _test_patching has "open" in its globals
						assert _test_patching.open is open
						with patch_submodule(_test_patching   ,						"open"   ,						_lowerCAmelCase			):
												assert _test_patching.open is mock

						# check that everthing is back to normal when the patch is over

						assert _test_patching.open is open





def 						_lowerCAmelCase							(     )    ->						Optional[Any]:


						'''simple docstring'''






						__snake_case				=		"__test_patch_submodule_missing_mock__"
						with patch_submodule(_test_patching   ,						"pandas.read_csv"   ,						_lowerCAmelCase			):
												pass





def 						_lowerCAmelCase							(     )    ->						Optional[Any]:


						'''simple docstring'''






						__snake_case				=		"__test_patch_submodule_missing_builtin_mock__"
						# _test_patching doesn't have "len" in its globals
						assert getattr(_test_patching   ,						"len"   ,						_lowerCAmelCase			) is None
						with patch_submodule(_test_patching   ,						"len"   ,						_lowerCAmelCase			):
												assert _test_patching.len is mock
						assert _test_patching.len is len





def 						_lowerCAmelCase							(     )    ->						Optional[Any]:


						'''simple docstring'''






						__snake_case				=		"__test_patch_submodule_start_and_stop_mock__"
						__snake_case				=		patch_submodule(_test_patching   ,						"open"   ,						_lowerCAmelCase			)
						assert _test_patching.open is open
						patch.start()
						assert _test_patching.open is mock
						patch.stop()
						assert _test_patching.open is open





def 						_lowerCAmelCase							(     )    ->						Any:


						'''simple docstring'''






						from os import rename as original_rename
						from os.path import dirname as original_dirname
						from os.path import join as original_join

						__snake_case				=		"__test_patch_submodule_successive_join__"
						__snake_case				=		"__test_patch_submodule_successive_dirname__"
						__snake_case				=		"__test_patch_submodule_successive_rename__"
						assert _test_patching.os.path.join is original_join
						assert _test_patching.os.path.dirname is original_dirname
						assert _test_patching.os.rename is original_rename

						with patch_submodule(_test_patching   ,						"os.path.join"   ,						_lowerCAmelCase			):
												with patch_submodule(_test_patching   ,						"os.rename"   ,						_lowerCAmelCase			):
																		with patch_submodule(_test_patching   ,						"os.path.dirname"   ,						_lowerCAmelCase			):
																								assert _test_patching.os.path.join is mock_join
																								assert _test_patching.os.path.dirname is mock_dirname
																								assert _test_patching.os.rename is mock_rename

    # try another order
						with patch_submodule(_test_patching   ,						"os.rename"   ,						_lowerCAmelCase			):
												with patch_submodule(_test_patching   ,						"os.path.join"   ,						_lowerCAmelCase			):
																		with patch_submodule(_test_patching   ,						"os.path.dirname"   ,						_lowerCAmelCase			):
																								assert _test_patching.os.path.join is mock_join
																								assert _test_patching.os.path.dirname is mock_dirname
																								assert _test_patching.os.rename is mock_rename

						assert _test_patching.os.path.join is original_join
						assert _test_patching.os.path.dirname is original_dirname
						assert _test_patching.os.rename is original_rename





def 						_lowerCAmelCase							(     )    ->						str:


						'''simple docstring'''






						__snake_case				=		"__test_patch_submodule_doesnt_exist_mock__"
						with patch_submodule(_test_patching   ,						"__module_that_doesn_exist__.__attribute_that_doesn_exist__"   ,						_lowerCAmelCase			):
												pass
						with patch_submodule(_test_patching   ,						"os.__attribute_that_doesn_exist__"   ,						_lowerCAmelCase			):
												pass





