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


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


def     __lowerCAmelCase						(       )      ->	str:





     '''simple docstring'''





     SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       "https://pypi.org/pypi/diffusers/json"
     SCREAMING_SNAKE_CASE					:			List[Any]    	=       json.loads(request.urlopen(_lowerCamelCase		).read()		)["releases"].keys()
     return sorted(_lowerCamelCase       ,      key=lambda a_		: version.Version(_lowerCamelCase		)		)


def     __lowerCAmelCase						(       )      ->	Any:





     '''simple docstring'''





     if HF_MODULES_CACHE in sys.path:
          return

     sys.path.append(_lowerCamelCase		)
     os.makedirs(_lowerCamelCase       ,      exist_ok=_lowerCamelCase		)
     SCREAMING_SNAKE_CASE					:			List[Any]    	=       Path(_lowerCamelCase		) / "__init__.py"
     if not init_path.exists():
          init_path.touch()


def     __lowerCAmelCase						(       a_		)      ->	Tuple:





     '''simple docstring'''





     init_hf_modules()
     SCREAMING_SNAKE_CASE					:			Tuple    	=       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		)
     SCREAMING_SNAKE_CASE					:			Optional[int]    	=       dynamic_module_path / "__init__.py"
     if not init_path.exists():
          init_path.touch()


def     __lowerCAmelCase						(       a_		)      ->	List[str]:





     '''simple docstring'''





     with open(_lowerCamelCase       ,      'r'       ,      encoding='utf-8'		) as f:
          SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       f.read()

     # Imports of the form `import .xxx`
     SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       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						(       a_		)      ->	str:





     '''simple docstring'''





     SCREAMING_SNAKE_CASE					:			Optional[int]    	=       False
     SCREAMING_SNAKE_CASE					:			Tuple    	=       [module_file]
     SCREAMING_SNAKE_CASE					:			Optional[int]    	=       []

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

          SCREAMING_SNAKE_CASE					:			Optional[Any]    	=       Path(_lowerCamelCase		).parent
          SCREAMING_SNAKE_CASE					:			List[str]    	=       [str(module_path / m		) for m in new_imports]
          SCREAMING_SNAKE_CASE					:			Optional[Any]    	=       [f for f in new_import_files if f not in all_relative_imports]
          SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       [f"""{f}.py""" for f in new_import_files]

          SCREAMING_SNAKE_CASE					:			Tuple    	=       len(_lowerCamelCase		) == 0
          all_relative_imports.extend(_lowerCamelCase		)

     return all_relative_imports


def     __lowerCAmelCase						(       a_		)      ->	Optional[int]:





     '''simple docstring'''





     with open(_lowerCamelCase       ,      'r'       ,      encoding='utf-8'		) as f:
          SCREAMING_SNAKE_CASE					:			Dict    	=       f.read()

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

     # Unique-ify and test we got them all
     SCREAMING_SNAKE_CASE					:			Any    	=       list(set(_lowerCamelCase		)		)
     SCREAMING_SNAKE_CASE					:			Tuple    	=       []
     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						(       a_       ,      a_		)      ->	Any:





     '''simple docstring'''





     SCREAMING_SNAKE_CASE					:			int    	=       module_path.replace(os.path.sep       ,      '.'		)
     SCREAMING_SNAKE_CASE					:			Optional[Any]    	=       importlib.import_module(_lowerCamelCase		)

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


def     __lowerCAmelCase						(       a_		)      ->	List[str]:





     '''simple docstring'''





     from ..pipelines import DiffusionPipeline

     SCREAMING_SNAKE_CASE					:			int    	=       dict(inspect.getmembers(_lowerCamelCase       ,      inspect.isclass		)		)

     SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       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}."""		)
               SCREAMING_SNAKE_CASE					:			Any    	=       cls

     return pipeline_class


def     __lowerCAmelCase						(       a_       ,      a_       ,      a_ = None       ,      a_ = False       ,      a_ = False       ,      a_ = None       ,      a_ = None       ,      a_ = None       ,      a_ = False       ,      )      ->	str:





     '''simple docstring'''





     SCREAMING_SNAKE_CASE					:			List[Any]    	=       str(_lowerCamelCase		)

     SCREAMING_SNAKE_CASE					:			Any    	=       os.path.join(_lowerCamelCase       ,      _lowerCamelCase		)

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

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

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

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





     '''simple docstring'''





     SCREAMING_SNAKE_CASE					:			int    	=       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'       ,      ''		)		)




'''simple docstring'''





def     __lowerCAmelCase						(       a_       ,      a_		)      ->	str:





     '''simple docstring'''





     if number < 0 or shift_amount < 0:
          raise ValueError('both inputs must be positive integers'		)

     SCREAMING_SNAKE_CASE					:			str    	=       str(bin(a_		)		)
     binary_number += "0" * shift_amount
     return binary_number


def     __lowerCAmelCase						(       a_       ,      a_		)      ->	str:





     '''simple docstring'''





     if number < 0 or shift_amount < 0:
          raise ValueError('both inputs must be positive integers'		)

     SCREAMING_SNAKE_CASE					:			Optional[Any]    	=       str(bin(a_		)		)[2:]
     if shift_amount >= len(a_		):
          return "0b0"
     SCREAMING_SNAKE_CASE					:			Dict    	=       binary_number[: len(a_		) - shift_amount]
     return "0b" + shifted_binary_number


def     __lowerCAmelCase						(       a_       ,      a_		)      ->	str:





     '''simple docstring'''





     if number >= 0:  # Get binary representation of positive number
          SCREAMING_SNAKE_CASE					:			Tuple    	=       '0' + str(bin(a_		)		).strip('-'		)[2:]
     else:  # Get binary (2's complement) representation of negative number
          SCREAMING_SNAKE_CASE					:			Union[str, Any]    	=       len(bin(a_		)[3:]		)  # Find 2's complement of number
          SCREAMING_SNAKE_CASE					:			Any    	=       bin(abs(a_		) - (1 << binary_number_length)		)[3:]
          SCREAMING_SNAKE_CASE					:			Optional[Any]    	=       (
              '1' + '0' * (binary_number_length - len(a_		)) + binary_number
          )

     if shift_amount >= len(a_		):
          return "0b" + binary_number[0] * len(a_		)
     return (
         "0b"
         + binary_number[0] * shift_amount
         + binary_number[: len(a_		) - shift_amount]
     )


if __name__ == "__main__":
 import doctest

 doctest.testmod()



