


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


UpperCAmelCase   :    Union[str, Any]     =		logging.get_logger(__name__)  # pylint: disable=invalid-name






def 	__lowerCamelCase						(							):





			'''simple docstring'''




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






def 	__lowerCamelCase						(							):





			'''simple docstring'''




			if HF_MODULES_CACHE in sys.path:
						return

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






def 	__lowerCamelCase						(							lowerCamelCase__		:						Union[str, os.PathLike]      ):





			'''simple docstring'''




			init_hf_modules()
			lowerCamelCase    		=  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__      )
			lowerCamelCase    		=  dynamic_module_path / """__init__.py"""
			if not init_path.exists():
						init_path.touch()






def 	__lowerCamelCase						(							lowerCamelCase__		:						Any      ):





			'''simple docstring'''




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

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

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

			return all_relative_imports






def 	__lowerCamelCase						(							lowerCamelCase__		:						int      ):





			'''simple docstring'''




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

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

			# Unique-ify and test we got them all
			lowerCamelCase    		=  list(set(lowerCamelCase__      )      )
			lowerCamelCase    		=  []
			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 	__lowerCamelCase						(							lowerCamelCase__		:						List[str]  ,							lowerCamelCase__		:						Optional[Any]      ):





			'''simple docstring'''




			lowerCamelCase    		=  module_path.replace(os.path.sep  ,							"""."""      )
			lowerCamelCase    		=  importlib.import_module(lowerCamelCase__      )

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






def 	__lowerCamelCase						(							lowerCamelCase__		:						List[str]      ):





			'''simple docstring'''




			from ..pipelines import DiffusionPipeline

			lowerCamelCase    		=  dict(inspect.getmembers(lowerCamelCase__  ,							inspect.isclass      )      )

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

			return pipeline_class






def 	__lowerCamelCase						(							lowerCamelCase__		:						Union[str, os.PathLike]  ,							lowerCamelCase__		:						str  ,							lowerCamelCase__		:						Optional[Union[str, os.PathLike]] = None  ,							lowerCamelCase__		:						bool = False  ,							lowerCamelCase__		:						bool = False  ,							lowerCamelCase__		:						Optional[Dict[str, str]] = None  ,							lowerCamelCase__		:						Optional[Union[bool, str]] = None  ,							lowerCamelCase__		:						Optional[str] = None  ,							lowerCamelCase__		:						bool = False  ,							):





			'''simple docstring'''




			lowerCamelCase    		=  str(lowerCamelCase__      )

			lowerCamelCase    		=  os.path.join(lowerCamelCase__  ,							lowerCamelCase__      )

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

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

						lowerCamelCase    		=  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.
						lowerCamelCase    		=  submodule_path / commit_hash
						lowerCamelCase    		=  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 	__lowerCamelCase						(							lowerCamelCase__		:						Union[str, os.PathLike]  ,							lowerCamelCase__		:						str  ,							lowerCamelCase__		:						Optional[str] = None  ,							lowerCamelCase__		:						Optional[Union[str, os.PathLike]] = None  ,							lowerCamelCase__		:						bool = False  ,							lowerCamelCase__		:						bool = False  ,							lowerCamelCase__		:						Optional[Dict[str, str]] = None  ,							lowerCamelCase__		:						Optional[Union[bool, str]] = None  ,							lowerCamelCase__		:						Optional[str] = None  ,							lowerCamelCase__		:						bool = False  ,							**lowerCamelCase__		:						Dict  ,							):





			'''simple docstring'''




			lowerCamelCase    		=  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"""  ,							""""""      )      )





# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse

from .config import config_command_parser
from .config_args import default_config_file, load_config_from_file  # noqa: F401
from .default import default_command_parser
from .update import update_command_parser






def 	__lowerCamelCase						(							lowerCamelCase__		:						Any=None      ):





			'''simple docstring'''




			lowerCamelCase    		=  argparse.ArgumentParser(add_help=lowerCamelCase__  ,							allow_abbrev=lowerCamelCase__      )
			# The main config parser
			lowerCamelCase    		=  config_command_parser(lowerCamelCase__      )
			# The subparser to add commands to
			lowerCamelCase    		=  config_parser.add_subparsers(title="""subcommands"""  ,							dest="""subcommand"""      )

			# Then add other parsers with the parent parser
			default_command_parser(lowerCamelCase__  ,							parents=[parent_parser]      )
			update_command_parser(lowerCamelCase__  ,							parents=[parent_parser]      )

			return config_parser






def 	__lowerCamelCase						(							):





			'''simple docstring'''




			lowerCamelCase    		=  get_config_parser()
			lowerCamelCase    		=  config_parser.parse_args()

			if not hasattr(lowerCamelCase__  ,							"""func"""      ):
						config_parser.print_help()
						exit(1      )

			# Run
			args.func(lowerCamelCase__      )


if __name__ == "__main__":
		main()

