




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


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


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





def 		__A		(	)      ->       List[Any]:
	a										=							"""https://pypi.org/pypi/diffusers/json"""
	a										=							json.loads(request.urlopen(__lowerCamelCase    ).read()    )["""releases"""].keys()
	return sorted(__lowerCamelCase							,			key=lambda __lowerCamelCase    : version.Version(__lowerCamelCase    )    )





def 		__A		(	)      ->       Union[str, Any]:
	# 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    )
	a										=							Path(__lowerCamelCase    ) / """__init__.py"""
	if not init_path.exists():
		init_path.touch()





def 		__A		(	__lowerCamelCase    )      ->       int:
	init_hf_modules()
	a										=							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    )
	a										=							dynamic_module_path / """__init__.py"""
	if not init_path.exists():
		init_path.touch()





def 		__A		(	__lowerCamelCase    )      ->       Optional[int]:
	with open(__lowerCamelCase							,			"""r"""							,			encoding="""utf-8"""    ) as f:
		a										=							f.read()

	# Imports of the form `import .xxx`
	a										=							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    )      ->       Dict:
	a										=							False
	a										=							[module_file]
	a										=							[]

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

		a										=							Path(__lowerCamelCase    ).parent
		a										=							[str(module_path / m    ) for m in new_imports]
		a										=							[f for f in new_import_files if f not in all_relative_imports]
		a										=							[f'{f}.py' for f in new_import_files]

		a										=							len(__lowerCamelCase    ) == 0
		all_relative_imports.extend(__lowerCamelCase    )

	return all_relative_imports





def 		__A		(	__lowerCamelCase    )      ->       List[Any]:
	with open(__lowerCamelCase							,			"""r"""							,			encoding="""utf-8"""    ) as f:
		a										=							f.read()

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

	# Unique-ify and test we got them all
	a										=							list(set(__lowerCamelCase    )    )
	a										=							[]
	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    )      ->       int:
	a										=							module_path.replace(os.path.sep							,			"""."""    )
	a										=							importlib.import_module(__lowerCamelCase    )

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





def 		__A		(	__lowerCamelCase    )      ->       List[Any]:
	from ..pipelines import DiffusionPipeline

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

	a										=							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}.'    )
			a										=							cls

	return pipeline_class





def 		__A		(	__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase = None							,			__lowerCamelCase = False							,			__lowerCamelCase = False							,			__lowerCamelCase = None							,			__lowerCamelCase = None							,			__lowerCamelCase = None							,			__lowerCamelCase = False							,			)      ->       List[Any]:
	a										=							str(__lowerCamelCase    )

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

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

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

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

		a										=							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.
		a										=							submodule_path / commit_hash
		a										=							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							,			)      ->       List[str]:
	a										=							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"""							,			""""""    )    )







# flake8: noqa
# Lint as: python3

from typing import Dict, List, Optional, Type

from .. import config
from ..utils import logging
from .formatting import (
    ArrowFormatter,
    CustomFormatter,
    Formatter,
    PandasFormatter,
    PythonFormatter,
    TensorFormatter,
    format_table,
    query_table,
)
from .np_formatter import NumpyFormatter


__UpperCamelCase					:       List[str]    					=      logging.get_logger(__name__)

__UpperCamelCase					:       Dict[Optional[str], Type[Formatter]]    					=      {}
__UpperCamelCase					:       Dict[Optional[str], str]    					=      {}
__UpperCamelCase					:       Dict[Optional[str], Exception]    					=      {}





def 		__A		(	__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase = None							,			)      ->       Optional[int]:
	a										=							aliases if aliases is not None else []
	if format_type in _FORMAT_TYPES:
		logger.warning(
		    f'Overwriting format type \'{format_type}\' ({_FORMAT_TYPES[format_type].__name__} -> {formatter_cls.__name__})'    )
	a										=							formatter_cls
	for alias in set(aliases + [format_type]    ):
		if alias in _FORMAT_TYPES_ALIASES:
			logger.warning(
			    f'Overwriting format type alias \'{alias}\' ({_FORMAT_TYPES_ALIASES[alias]} -> {format_type})'    )
		a										=							format_type





def 		__A		(	__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase = None    )      ->       List[str]:
	a										=							aliases if aliases is not None else []
	for alias in set(aliases + [format_type]    ):
		a										=							unavailable_error


# Here we define all the available formatting functions that can be used by `Dataset.set_format`
_register_formatter(PythonFormatter, None, aliases=["python"])
_register_formatter(ArrowFormatter, "arrow", aliases=["pa", "pyarrow"])
_register_formatter(NumpyFormatter, "numpy", aliases=["np"])
_register_formatter(PandasFormatter, "pandas", aliases=["pd"])
_register_formatter(CustomFormatter, "custom")

if config.TORCH_AVAILABLE:
		from .torch_formatter import TorchFormatter

		_register_formatter(TorchFormatter, "torch", aliases=["pt", "pytorch"])
else:
		__UpperCamelCase					:       str    					=      ValueError("PyTorch needs to be installed to be able to return PyTorch tensors.")
		_register_unavailable_formatter(_torch_error, "torch", aliases=["pt", "pytorch"])

if config.TF_AVAILABLE:
		from .tf_formatter import TFFormatter

		_register_formatter(TFFormatter, "tensorflow", aliases=["tf"])
else:
		__UpperCamelCase					:       List[str]    					=      ValueError("Tensorflow needs to be installed to be able to return Tensorflow tensors.")
		_register_unavailable_formatter(_tf_error, "tensorflow", aliases=["tf"])

if config.JAX_AVAILABLE:
		from .jax_formatter import JaxFormatter

		_register_formatter(JaxFormatter, "jax", aliases=[])
else:
		__UpperCamelCase					:       List[str]    					=      ValueError("JAX needs to be installed to be able to return JAX arrays.")
		_register_unavailable_formatter(_jax_error, "jax", aliases=[])





def 		__A		(	__lowerCamelCase    )      ->       Optional[str]:
	if format_type in _FORMAT_TYPES_ALIASES:
		return _FORMAT_TYPES_ALIASES[format_type]
	else:
		return format_type





def 		__A		(	__lowerCamelCase							,			**__lowerCamelCase    )      ->       Formatter:
	a										=							get_format_type_from_alias(__lowerCamelCase    )
	if format_type in _FORMAT_TYPES:
		return _FORMAT_TYPES[format_type](**__lowerCamelCase    )
	if format_type in _FORMAT_TYPES_ALIASES_UNAVAILABLE:
		raise _FORMAT_TYPES_ALIASES_UNAVAILABLE[format_type]
	else:
		raise ValueError(
		    f'Return type should be None or selected in {list(type for type in _FORMAT_TYPES.keys() if type != None    )}, but got \'{format_type}\''    )

