





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


_snake_case    :       Optional[Any]		       =    (
    'https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py'
)


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

def     a_       (					):
   __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     a_       (					):
   # 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_      )
   __lowerCAmelCase				      =      Path(lowerCAmelCase_      ) / '__init__.py'
   if not init_path.exists():
      init_path.touch()

def     a_       (					lowerCAmelCase_ :     Union[str, os.PathLike]      ):
   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     a_       (					lowerCAmelCase_ :     List[Any]      ):
   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     a_       (					lowerCAmelCase_ :     List[str]      ):
   __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     a_       (					lowerCAmelCase_ :     List[str]      ):
   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     a_       (					lowerCAmelCase_ :     List[Any],      lowerCAmelCase_ :     Dict      ):
   __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     a_       (					lowerCAmelCase_ :     Optional[int]      ):
   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     a_       (					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,      ):
   __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     a_       (					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_ :     List[Any],      ):
   __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',      ''      )      )



"""simple docstring"""






from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


__SCREAMING_SNAKE_CASE							=     {
    'configuration_whisper': ['WHISPER_PRETRAINED_CONFIG_ARCHIVE_MAP', 'WhisperConfig', 'WhisperOnnxConfig'],
    'feature_extraction_whisper': ['WhisperFeatureExtractor'],
    'processing_whisper': ['WhisperProcessor'],
    'tokenization_whisper': ['WhisperTokenizer'],
}

try:
 if not is_tokenizers_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 __SCREAMING_SNAKE_CASE							=     ['WhisperTokenizerFast']

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 __SCREAMING_SNAKE_CASE							=     [
     'WHISPER_PRETRAINED_MODEL_ARCHIVE_LIST',
     'WhisperForConditionalGeneration',
     'WhisperModel',
     'WhisperPreTrainedModel',
     'WhisperForAudioClassification',
 ]

try:
 if not is_tf_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 __SCREAMING_SNAKE_CASE							=     [
     'TF_WHISPER_PRETRAINED_MODEL_ARCHIVE_LIST',
     'TFWhisperForConditionalGeneration',
     'TFWhisperModel',
     'TFWhisperPreTrainedModel',
 ]

try:
 if not is_flax_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 __SCREAMING_SNAKE_CASE							=     [
     'FlaxWhisperForConditionalGeneration',
     'FlaxWhisperModel',
     'FlaxWhisperPreTrainedModel',
     'FlaxWhisperForAudioClassification',
 ]


if TYPE_CHECKING:
 from .configuration_whisper import WHISPER_PRETRAINED_CONFIG_ARCHIVE_MAP, WhisperConfig, WhisperOnnxConfig
 from .feature_extraction_whisper import WhisperFeatureExtractor
 from .processing_whisper import WhisperProcessor
 from .tokenization_whisper import WhisperTokenizer

 try:
  if not is_tokenizers_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .tokenization_whisper_fast import WhisperTokenizerFast

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_whisper import (
      WHISPER_PRETRAINED_MODEL_ARCHIVE_LIST,
      WhisperForAudioClassification,
      WhisperForConditionalGeneration,
      WhisperModel,
      WhisperPreTrainedModel,
  )

 try:
  if not is_tf_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_tf_whisper import (
      TF_WHISPER_PRETRAINED_MODEL_ARCHIVE_LIST,
      TFWhisperForConditionalGeneration,
      TFWhisperModel,
      TFWhisperPreTrainedModel,
  )

 try:
  if not is_flax_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_flax_whisper import (
      FlaxWhisperForAudioClassification,
      FlaxWhisperForConditionalGeneration,
      FlaxWhisperModel,
      FlaxWhisperPreTrainedModel,
  )

else:
 import sys

 __SCREAMING_SNAKE_CASE							=     _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)
