
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 			UpperCamelCase					(    ):
       snake_case       :   Optional[Any]							=      "https://pypi.org/pypi/diffusers/json"
       snake_case       :   List[str]							=      json.loads(request.urlopen(__lowerCamelCase    ).read()    )["releases"].keys()
       return sorted(__lowerCamelCase				,					key=lambda __lowerCamelCase    : version.Version(__lowerCamelCase    )    )


def 			UpperCamelCase					(    ):
       # 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    )
       snake_case       :   int							=      Path(__lowerCamelCase    ) / "__init__.py"
       if not init_path.exists():
              init_path.touch()


def 			UpperCamelCase					(    __lowerCamelCase     : Union[str, os.PathLike]    ):
       init_hf_modules()
       snake_case       :   Union[str, Any]							=      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    )
       snake_case       :   Dict							=      dynamic_module_path / "__init__.py"
       if not init_path.exists():
              init_path.touch()


def 			UpperCamelCase					(    __lowerCamelCase     : Union[str, Any]    ):
       with open(__lowerCamelCase				,					"r"				,					encoding="utf-8"    ) as f:
              snake_case       :   Tuple							=      f.read()

       # Imports of the form `import .xxx`
       snake_case       :   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 			UpperCamelCase					(    __lowerCamelCase     : int    ):
       snake_case       :   Union[str, Any]							=      False
       snake_case       :   str							=      [module_file]
       snake_case       :   str							=      []

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

              snake_case       :   Any							=      Path(__lowerCamelCase    ).parent
              snake_case       :   str							=      [str(module_path / m    ) for m in new_imports]
              snake_case       :   Tuple							=      [f for f in new_import_files if f not in all_relative_imports]
              snake_case       :   Optional[int]							=      [f"""{f}.py""" for f in new_import_files]

              snake_case       :   Optional[Any]							=      len(__lowerCamelCase    ) == 0
              all_relative_imports.extend(__lowerCamelCase    )

       return all_relative_imports


def 			UpperCamelCase					(    __lowerCamelCase     : Union[str, Any]    ):
       with open(__lowerCamelCase				,					"r"				,					encoding="utf-8"    ) as f:
              snake_case       :   int							=      f.read()

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

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

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


def 			UpperCamelCase					(    __lowerCamelCase     : List[str]    ):
       from ..pipelines import DiffusionPipeline

       snake_case       :   List[str]							=      dict(inspect.getmembers(__lowerCamelCase				,					inspect.isclass    )    )

       snake_case       :   str							=      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}."""    )
                     snake_case       :   Dict							=      cls

       return pipeline_class


def 			UpperCamelCase					(    __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				,					):
       snake_case       :   List[str]							=      str(__lowerCamelCase    )

       snake_case       :   Optional[Any]							=      os.path.join(__lowerCamelCase				,					__lowerCamelCase    )

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

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

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

              snake_case       :   Optional[Any]							=      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.
              snake_case       :   Any							=      submodule_path / commit_hash
              snake_case       :   List[Any]							=      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 			UpperCamelCase					(    __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     : Tuple				,					):
       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"				,					""    )    )







from ...configuration_utils import PretrainedConfig
from ...utils import logging


__lowerCamelCase							=       logging.get_logger(__name__)

__lowerCamelCase							=       {
    # See all MEGATRON_BERT models at https://huggingface.co/models?filter=bert
}






class   UpperCAmelCase		(						A_ ):
   A__					:		List[str]						     =     "megatron-bert"



   def __init__(self		:					Optional[int]    ,						snake_case__		:					List[str]=2_90_56    ,						snake_case__		:					List[Any]=10_24    ,						snake_case__		:					str=24    ,						snake_case__		:					Tuple=16    ,						snake_case__		:					Union[str, Any]=40_96    ,						snake_case__		:					str="gelu"    ,						snake_case__		:					str=0.1    ,						snake_case__		:					Optional[int]=0.1    ,						snake_case__		:					Tuple=5_12    ,						snake_case__		:					Union[str, Any]=2    ,						snake_case__		:					Dict=0.02    ,						snake_case__		:					List[Any]=1e-12    ,						snake_case__		:					int=0    ,						snake_case__		:					Tuple="absolute"    ,						snake_case__		:					Any=True    ,						**snake_case__		:					Union[str, Any]    ,						)		->	Optional[Any]:


          '''simple docstring'''


          super().__init__(pad_token_id=snake_case__    ,						**snake_case__ )

          snake_case       :   Tuple							=      vocab_size
          snake_case       :   str							=      hidden_size
          snake_case       :   str							=      num_hidden_layers
          snake_case       :   str							=      num_attention_heads
          snake_case       :   Optional[int]							=      hidden_act
          snake_case       :   int							=      intermediate_size
          snake_case       :   List[str]							=      hidden_dropout_prob
          snake_case       :   Union[str, Any]							=      attention_probs_dropout_prob
          snake_case       :   Dict							=      max_position_embeddings
          snake_case       :   List[str]							=      type_vocab_size
          snake_case       :   List[str]							=      initializer_range
          snake_case       :   Tuple							=      layer_norm_eps
          snake_case       :   int							=      position_embedding_type
          snake_case       :   str							=      use_cache





