
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


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


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


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


def 				SCREAMING_SNAKE_CASE_   (      ):
      # 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(UpperCamelCase__		)
      os.makedirs(UpperCamelCase__			,		exist_ok=UpperCamelCase__		)
      UpperCamelCase__     : Union[str, Any]											=					Path(UpperCamelCase__		) / '''__init__.py'''
      if not init_path.exists():
            init_path.touch()


def 				SCREAMING_SNAKE_CASE_   (      UpperCamelCase__		):
      init_hf_modules()
      UpperCamelCase__     : Union[str, Any]											=					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__		)
      UpperCamelCase__     : Optional[Any]											=					dynamic_module_path / '''__init__.py'''
      if not init_path.exists():
            init_path.touch()


def 				SCREAMING_SNAKE_CASE_   (      UpperCamelCase__		):
      with open(UpperCamelCase__			,		'''r'''			,		encoding='''utf-8'''		) as f:
            UpperCamelCase__     : List[str]											=					f.read()

      # Imports of the form `import .xxx`
      UpperCamelCase__     : Optional[int]											=					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 				SCREAMING_SNAKE_CASE_   (      UpperCamelCase__		):
      UpperCamelCase__     : List[Any]											=					False
      UpperCamelCase__     : Optional[int]											=					[module_file]
      UpperCamelCase__     : Tuple											=					[]

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

            UpperCamelCase__     : Tuple											=					Path(UpperCamelCase__		).parent
            UpperCamelCase__     : Tuple											=					[str(module_path / m		) for m in new_imports]
            UpperCamelCase__     : Union[str, Any]											=					[f for f in new_import_files if f not in all_relative_imports]
            UpperCamelCase__     : Tuple											=					[f'''{f}.py''' for f in new_import_files]

            UpperCamelCase__     : List[str]											=					len(UpperCamelCase__		) == 0
            all_relative_imports.extend(UpperCamelCase__		)

      return all_relative_imports


def 				SCREAMING_SNAKE_CASE_   (      UpperCamelCase__		):
      with open(UpperCamelCase__			,		'''r'''			,		encoding='''utf-8'''		) as f:
            UpperCamelCase__     : List[str]											=					f.read()

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

      # Unique-ify and test we got them all
      UpperCamelCase__     : List[Any]											=					list(set(UpperCamelCase__		)		)
      UpperCamelCase__     : List[Any]											=					[]
      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 				SCREAMING_SNAKE_CASE_   (      UpperCamelCase__			,		UpperCamelCase__		):
      UpperCamelCase__     : Dict											=					module_path.replace(os.path.sep			,		'''.'''		)
      UpperCamelCase__     : List[Any]											=					importlib.import_module(UpperCamelCase__		)

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


def 				SCREAMING_SNAKE_CASE_   (      UpperCamelCase__		):
      from ..pipelines import DiffusionPipeline

      UpperCamelCase__     : Optional[int]											=					dict(inspect.getmembers(UpperCamelCase__			,		inspect.isclass		)		)

      UpperCamelCase__     : Dict											=					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}.'''		)
                  UpperCamelCase__     : Tuple											=					cls

      return pipeline_class


def 				SCREAMING_SNAKE_CASE_   (      UpperCamelCase__			,		UpperCamelCase__			,		UpperCamelCase__ = None			,		UpperCamelCase__ = False			,		UpperCamelCase__ = False			,		UpperCamelCase__ = None			,		UpperCamelCase__ = None			,		UpperCamelCase__ = None			,		UpperCamelCase__ = False			,		):
      UpperCamelCase__     : int											=					str(UpperCamelCase__		)

      UpperCamelCase__     : str											=					os.path.join(UpperCamelCase__			,		UpperCamelCase__		)

      if os.path.isfile(UpperCamelCase__		):
            UpperCamelCase__     : Union[str, Any]											=					module_file_or_url
            UpperCamelCase__     : List[str]											=					'''local'''
      elif pretrained_model_name_or_path.count('''/'''		) == 0:
            UpperCamelCase__     : Union[str, Any]											=					get_diffusers_versions()
            # cut ".dev0"
            UpperCamelCase__     : Tuple											=					'''v''' + '''.'''.join(__version__.split('''.'''		)[:3]		)

            # retrieve github version that matches
            if revision is None:
                  UpperCamelCase__     : Optional[int]											=					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__     : List[Any]											=					f'''v{revision}'''
            elif revision == "main":
                  UpperCamelCase__     : str											=					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__     : List[str]											=					COMMUNITY_PIPELINES_URL.format(revision=UpperCamelCase__			,		pipeline=UpperCamelCase__		)
            try:
                  UpperCamelCase__     : List[str]											=					cached_download(
                      UpperCamelCase__			,		cache_dir=UpperCamelCase__			,		force_download=UpperCamelCase__			,		proxies=UpperCamelCase__			,		resume_download=UpperCamelCase__			,		local_files_only=UpperCamelCase__			,		use_auth_token=UpperCamelCase__			,		)
                  UpperCamelCase__     : Optional[int]											=					'''git'''
                  UpperCamelCase__     : Union[str, Any]											=					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__     : str											=					hf_hub_download(
                      UpperCamelCase__			,		UpperCamelCase__			,		cache_dir=UpperCamelCase__			,		force_download=UpperCamelCase__			,		proxies=UpperCamelCase__			,		resume_download=UpperCamelCase__			,		local_files_only=UpperCamelCase__			,		use_auth_token=UpperCamelCase__			,		)
                  UpperCamelCase__     : Tuple											=					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__     : Dict											=					check_imports(UpperCamelCase__		)

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

            UpperCamelCase__     : List[str]											=					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.
            UpperCamelCase__     : str											=					submodule_path / commit_hash
            UpperCamelCase__     : Optional[Any]											=					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 				SCREAMING_SNAKE_CASE_   (      UpperCamelCase__			,		UpperCamelCase__			,		UpperCamelCase__ = None			,		UpperCamelCase__ = None			,		UpperCamelCase__ = False			,		UpperCamelCase__ = False			,		UpperCamelCase__ = None			,		UpperCamelCase__ = None			,		UpperCamelCase__ = None			,		UpperCamelCase__ = False			,		**UpperCamelCase__			,		):
      UpperCamelCase__     : List[Any]											=					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'''			,		''''''		)		)




# limitations under the License.

# NOTE: This file is deprecated and will be removed in a future version.
# It only exists so that temporarely `from diffusers.pipelines import DiffusionPipeline` works

from .pipelines import DiffusionPipeline, ImagePipelineOutput  # noqa: F401
from .utils import deprecate


deprecate(
    "pipelines_utils",
    "0.22.0",
    "Importing `DiffusionPipeline` or `ImagePipelineOutput` from diffusers.pipeline_utils is deprecated. Please import from diffusers.pipelines.pipeline_utils instead.",
    standard_warn=False,
    stacklevel=3,
)


