





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


__lowerCamelCase       :     Optional[Any]   =     logging.get_logger(__name__)  # pylint: disable=invalid-name



def 		A_				(     )						-> Union[str, Any]:
 UpperCamelCase	:					List[Any]       	=   "https://pypi.org/pypi/diffusers/json"
 UpperCamelCase	:					Union[str, Any]       	=   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     )
 UpperCamelCase	:					List[str]       	=   Path(_lowerCAmelCase     ) / "__init__.py"
 if not init_path.exists():
  init_path.touch()



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



def 		A_				(     _lowerCAmelCase     )						-> List[str]:
 with open(_lowerCAmelCase							,   "r"							,   encoding="utf-8"     ) as f:
  UpperCamelCase	:					Tuple       	=   f.read()

 # Imports of the form `import .xxx`
 UpperCamelCase	:					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 		A_				(     _lowerCAmelCase     )						-> Optional[int]:
 UpperCamelCase	:					Any       	=   False
 UpperCamelCase	:					Dict       	=   [module_file]
 UpperCamelCase	:					Dict       	=   []

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

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

  UpperCamelCase	:					str       	=   len(_lowerCAmelCase     ) == 0
  all_relative_imports.extend(_lowerCAmelCase     )

 return all_relative_imports



def 		A_				(     _lowerCAmelCase     )						-> int:
 with open(_lowerCAmelCase							,   "r"							,   encoding="utf-8"     ) as f:
  UpperCamelCase	:					Tuple       	=   f.read()

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

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

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



def 		A_				(     _lowerCAmelCase     )						-> Tuple:
 from ..pipelines import DiffusionPipeline

 UpperCamelCase	:					Union[str, Any]       	=   dict(inspect.getmembers(_lowerCAmelCase							,   inspect.isclass     )     )

 UpperCamelCase	:					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}."""     )
   UpperCamelCase	:					int       	=   cls

 return pipeline_class



def 		A_				(     _lowerCAmelCase							,   _lowerCAmelCase							,   _lowerCAmelCase = None							,   _lowerCAmelCase = False							,   _lowerCAmelCase = False							,   _lowerCAmelCase = None							,   _lowerCAmelCase = None							,   _lowerCAmelCase = None							,   _lowerCAmelCase = False							,   )						-> Optional[Any]:
 UpperCamelCase	:					Tuple       	=   str(_lowerCAmelCase     )

 UpperCamelCase	:					List[str]       	=   os.path.join(_lowerCAmelCase							,   _lowerCAmelCase     )

 if os.path.isfile(_lowerCAmelCase     ):
  UpperCamelCase	:					Tuple       	=   module_file_or_url
  UpperCamelCase	:					int       	=   "local"
 elif pretrained_model_name_or_path.count("/"     ) == 0:
  UpperCamelCase	:					Optional[int]       	=   get_diffusers_versions()
  # cut ".dev0"
  UpperCamelCase	:					Any       	=   "v" + ".".join(__version__.split("."     )[:3]     )

  # retrieve github version that matches
  if revision is None:
   UpperCamelCase	:					List[str]       	=   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	:					Union[str, Any]       	=   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	:					Optional[Any]       	=   COMMUNITY_PIPELINES_URL.format(revision=_lowerCAmelCase							,   pipeline=_lowerCAmelCase     )
  try:
   UpperCamelCase	:					List[Any]       	=   cached_download(
       _lowerCAmelCase							,   cache_dir=_lowerCAmelCase							,   force_download=_lowerCAmelCase							,   proxies=_lowerCAmelCase							,   resume_download=_lowerCAmelCase							,   local_files_only=_lowerCAmelCase							,   use_auth_token=_lowerCAmelCase							,   )
   UpperCamelCase	:					Tuple       	=   "git"
   UpperCamelCase	:					List[str]       	=   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(
       _lowerCAmelCase							,   _lowerCAmelCase							,   cache_dir=_lowerCAmelCase							,   force_download=_lowerCAmelCase							,   proxies=_lowerCAmelCase							,   resume_download=_lowerCAmelCase							,   local_files_only=_lowerCAmelCase							,   use_auth_token=_lowerCAmelCase							,   )
   UpperCamelCase	:					List[str]       	=   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	:					List[Any]       	=   check_imports(_lowerCAmelCase     )

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

  UpperCamelCase	:					int       	=   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.
  UpperCamelCase	:					Tuple       	=   submodule_path / commit_hash
  UpperCamelCase	:					Dict       	=   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							,   )						-> Any:
 UpperCamelCase	:					Optional[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"							,   ""     )     )







import functools



def 		A_				(     _lowerCAmelCase							,   _lowerCAmelCase     )						-> int:
 UpperCamelCase	:					Optional[int]       	=   len(_lowerCAmelCase     )
 UpperCamelCase	:					List[str]       	=   len(_lowerCAmelCase     )

 @functools.cache
 def min_distance(_lowerCAmelCase							,   _lowerCAmelCase     ) -> int:
  # if first word index is overflow - delete all from the second word
  if indexa >= len_worda:
   return len_worda - indexa
  # if second word index is overflow - delete all from the first word
  if indexa >= len_worda:
   return len_worda - indexa
  UpperCamelCase	:					Union[str, Any]       	=   int(worda[indexa] != worda[indexa]     )  # current letters not identical
  return min(
      1 + min_distance(indexa + 1							,   _lowerCAmelCase     )							,   1 + min_distance(_lowerCAmelCase							,   indexa + 1     )							,   diff + min_distance(indexa + 1							,   indexa + 1     )							,   )

 return min_distance(0							,   0     )


if __name__ == "__main__":
    import doctest

    doctest.testmod()
