






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__      :     Dict       =					(
    'https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py'
)


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






def     a     (		):






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






def     a     (		):






				'''simple docstring'''
				# This function has already been executed if HF_MODULES_CACHE already is in the Python path.
				if HF_MODULES_CACHE in sys.path:
								return

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






def     a     (		lowerCamelCase_							):






				'''simple docstring'''
				init_hf_modules()
				lowercase__  					=     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_							)
				lowercase__  					=     dynamic_module_path / '''__init__.py'''
				if not init_path.exists():
								init_path.touch()






def     a     (		lowerCamelCase_							):






				'''simple docstring'''
				with open(lowerCamelCase_						,  '''r'''						,  encoding='''utf-8'''							) as f:
								lowercase__  					=     f.read()

				# Imports of the form `import .xxx`
				lowercase__  					=     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     (		lowerCamelCase_							):






				'''simple docstring'''
				lowercase__  					=     False
				lowercase__  					=     [module_file]
				lowercase__  					=     []

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

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

								lowercase__  					=     len(lowerCamelCase_							) == 0
								all_relative_imports.extend(lowerCamelCase_							)

				return all_relative_imports






def     a     (		lowerCamelCase_							):






				'''simple docstring'''
				with open(lowerCamelCase_						,  '''r'''						,  encoding='''utf-8'''							) as f:
								lowercase__  					=     f.read()

				# Imports of the form `import xxx`
				lowercase__  					=     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
				lowercase__  					=     [imp.split('''.'''							)[0] for imp in imports if not imp.startswith('''.'''							)]

				# Unique-ify and test we got them all
				lowercase__  					=     list(set(lowerCamelCase_							)							)
				lowercase__  					=     []
				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     (		lowerCamelCase_						,  lowerCamelCase_							):






				'''simple docstring'''
				lowercase__  					=     module_path.replace(os.path.sep						,  '''.'''							)
				lowercase__  					=     importlib.import_module(lowerCamelCase_							)

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






def     a     (		lowerCamelCase_							):






				'''simple docstring'''
				from ..pipelines import DiffusionPipeline

				lowercase__  					=     dict(inspect.getmembers(lowerCamelCase_						,  inspect.isclass							)							)

				lowercase__  					=     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}."""							)
												lowercase__  					=     cls

				return pipeline_class






def     a     (		lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_ = None						,  lowerCamelCase_ = False						,  lowerCamelCase_ = False						,  lowerCamelCase_ = None						,  lowerCamelCase_ = None						,  lowerCamelCase_ = None						,  lowerCamelCase_ = False						,  ):






				'''simple docstring'''
				lowercase__  					=     str(lowerCamelCase_							)

				lowercase__  					=     os.path.join(lowerCamelCase_						,  lowerCamelCase_							)

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

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

				# Now we move the module inside our cached dynamic modules.
				lowercase__  					=     DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
				create_dynamic_module(lowerCamelCase_							)
				lowercase__  					=     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:
												lowercase__  					=     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_							):
												lowercase__  					=     use_auth_token
								elif use_auth_token is True:
												lowercase__  					=     HfFolder.get_token()
								else:
												lowercase__  					=     None

								lowercase__  					=     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.
								lowercase__  					=     submodule_path / commit_hash
								lowercase__  					=     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     (		lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_ = None						,  lowerCamelCase_ = None						,  lowerCamelCase_ = False						,  lowerCamelCase_ = False						,  lowerCamelCase_ = None						,  lowerCamelCase_ = None						,  lowerCamelCase_ = None						,  lowerCamelCase_ = False						,  **lowerCamelCase_						,  ):






				'''simple docstring'''
				lowercase__  					=     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 typing import Dict

from .base import GenericTensor, Pipeline
class  _UpperCAmelCase   (			A__      ):

				"""simple docstring"""




				def 							lowercase__  (	self		:       Any,    lowerCamelCase		:       Tuple=None,    lowerCamelCase		:       Union[str, Any]=None,    lowerCamelCase		:       Union[str, Any]=None,    **lowerCamelCase		:       List[Any]    ):

								'''simple docstring'''
								if tokenize_kwargs is None:
												lowercase__  					=     {}

								if truncation is not None:
												if "truncation" in tokenize_kwargs:
																raise ValueError(
																    '''truncation parameter defined twice (given as keyword argument as well as in tokenize_kwargs)'''    )
												lowercase__  					=     truncation

								lowercase__  					=     tokenize_kwargs

								lowercase__  					=     {}
								if return_tensors is not None:
												lowercase__  					=     return_tensors

								return preprocess_params, {}, postprocess_params




				def 							lowercase__  (	self		:       Tuple,    lowerCamelCase		:       Optional[int],    **lowerCamelCase		:       List[Any]    ):

								'''simple docstring'''
								lowercase__  					=     self.framework
								lowercase__  					=     self.tokenizer(lowerCamelCase,    return_tensors=lowerCamelCase,    **lowerCamelCase    )
								return model_inputs




				def 							lowercase__  (	self		:       Union[str, Any],    lowerCamelCase		:       Dict    ):

								'''simple docstring'''
								lowercase__  					=     self.model(**lowerCamelCase    )
								return model_outputs




				def 							lowercase__  (	self		:       Union[str, Any],    lowerCamelCase		:       Dict,    lowerCamelCase		:       Tuple=False    ):

								'''simple docstring'''
								# [0] is the first available tensor, logits or last_hidden_state.
								if return_tensors:
												return model_outputs[0]
								if self.framework == "pt":
												return model_outputs[0].tolist()
								elif self.framework == "tf":
												return model_outputs[0].numpy().tolist()




				def __call__(	self		:       List[str],    *lowerCamelCase		:       List[Any],    **lowerCamelCase		:       str    ):

								'''simple docstring'''
								return super().__call__(*lowerCamelCase,    **lowerCamelCase    )




