






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


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







def 			A__							(      ):





 '''simple docstring'''







 snake_case__  :    List[Any]							       ="""https://pypi.org/pypi/diffusers/json"""
 snake_case__  :    Dict							       =json.loads(request.urlopen(_a	).read()	)["""releases"""].keys()
 return sorted(_a		,  key=lambda _a	: version.Version(_a	)	)







def 			A__							(      ):





 '''simple docstring'''







 if HF_MODULES_CACHE in sys.path:
  return

 sys.path.append(_a	)
 os.makedirs(_a		,  exist_ok=_a	)
 snake_case__  :    str							       =Path(_a	) / """__init__.py"""
 if not init_path.exists():
  init_path.touch()







def 			A__							(      _a  :					Union[str, os.PathLike]	):





 '''simple docstring'''







 init_hf_modules()
 snake_case__  :    List[Any]							       =Path(_a	) / 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(_a		,  exist_ok=_a	)
 snake_case__  :    Union[str, Any]							       =dynamic_module_path / """__init__.py"""
 if not init_path.exists():
  init_path.touch()







def 			A__							(      _a  :					List[Any]	):





 '''simple docstring'''







 with open(_a		,  """r"""		,  encoding="""utf-8"""	) as f:
  snake_case__  :    Union[str, Any]							       =f.read()

 # Imports of the form `import .xxx`
 snake_case__  :    List[Any]							       =re.findall("""^\s*import\s+\.(\S+)\s*$"""		,  _a		,  flags=re.MULTILINE	)
 # Imports of the form `from .xxx import yyy`
 relative_imports += re.findall("""^\s*from\s+\.(\S+)\s+import"""		,  _a		,  flags=re.MULTILINE	)
 # Unique-ify
 return list(set(_a	)	)







def 			A__							(      _a  :					str	):





 '''simple docstring'''







 snake_case__  :    int							       =False
 snake_case__  :    List[str]							       =[module_file]
 snake_case__  :    Tuple							       =[]

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

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

  snake_case__  :    List[Any]							       =len(_a	) == 0
  all_relative_imports.extend(_a	)

 return all_relative_imports







def 			A__							(      _a  :					Optional[Any]	):





 '''simple docstring'''







 with open(_a		,  """r"""		,  encoding="""utf-8"""	) as f:
  snake_case__  :    Union[str, Any]							       =f.read()

 # Imports of the form `import xxx`
 snake_case__  :    str							       =re.findall("""^\s*import\s+(\S+)\s*$"""		,  _a		,  flags=re.MULTILINE	)
 # Imports of the form `from xxx import yyy`
 imports += re.findall("""^\s*from\s+(\S+)\s+import"""		,  _a		,  flags=re.MULTILINE	)
 # Only keep the top-level module
 snake_case__  :    Tuple							       =[imp.split("""."""	)[0] for imp in imports if not imp.startswith("""."""	)]

 # Unique-ify and test we got them all
 snake_case__  :    Tuple							       =list(set(_a	)	)
 snake_case__  :    str							       =[]
 for imp in imports:
  try:
   importlib.import_module(_a	)
  except ImportError:
   missing_packages.append(_a	)

 if len(_a	) > 0:
  raise ImportError(
      """This modeling file requires the following packages that were not found in your environment: """
      f"{', '.join(_a	)}. Run `pip install {' '.join(_a	)}`"	)

 return get_relative_imports(_a	)







def 			A__							(      _a  :					Optional[Any]		,  _a  :					Dict	):





 '''simple docstring'''







 snake_case__  :    Any							       =module_path.replace(os.path.sep		,  """."""	)
 snake_case__  :    Union[str, Any]							       =importlib.import_module(_a	)

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







def 			A__							(      _a  :					str	):





 '''simple docstring'''







 from ..pipelines import DiffusionPipeline

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

 snake_case__  :    str							       =None
 for cls_name, cls in cls_members.items():
  if (
      cls_name != DiffusionPipeline.__name__
      and issubclass(cls		,  _a	)
      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__  :    Optional[Any]							       =cls

 return pipeline_class







def 			A__							(      _a  :					Union[str, os.PathLike]		,  _a  :					str		,  _a  :					Optional[Union[str, os.PathLike]] = None		,  _a  :					bool = False		,  _a  :					bool = False		,  _a  :					Optional[Dict[str, str]] = None		,  _a  :					Optional[Union[bool, str]] = None		,  _a  :					Optional[str] = None		,  _a  :					bool = False		,  ):





 '''simple docstring'''







 snake_case__  :    List[str]							       =str(_a	)

 snake_case__  :    Any							       =os.path.join(_a		,  _a	)

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

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

 # Now we move the module inside our cached dynamic modules.
 snake_case__  :    str							       =DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
 create_dynamic_module(_a	)
 snake_case__  :    List[str]							       =Path(_a	) / 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(_a		,  submodule_path / module_file	)
  for module_needed in modules_needed:
   snake_case__  :    int							       =f"{module_needed}.py"
   shutil.copy(os.path.join(_a		,  _a	)		,  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(_a		,  _a	):
   snake_case__  :    List[str]							       =use_auth_token
  elif use_auth_token is True:
   snake_case__  :    Optional[Any]							       =HfFolder.get_token()
  else:
   snake_case__  :    List[str]							       =None

  snake_case__  :    Optional[int]							       =model_info(_a		,  revision=_a		,  token=_a	).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__  :    Tuple							       =submodule_path / commit_hash
  snake_case__  :    Optional[Any]							       =full_submodule + os.path.sep + commit_hash
  create_dynamic_module(_a	)

  if not (submodule_path / module_file).exists():
   shutil.copy(_a		,  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(
        _a		,  f"{module_needed}.py"		,  cache_dir=_a		,  force_download=_a		,  resume_download=_a		,  proxies=_a		,  use_auth_token=_a		,  revision=_a		,  local_files_only=_a		,  )
 return os.path.join(_a		,  _a	)







def 			A__							(      _a  :					Union[str, os.PathLike]		,  _a  :					str		,  _a  :					Optional[str] = None		,  _a  :					Optional[Union[str, os.PathLike]] = None		,  _a  :					bool = False		,  _a  :					bool = False		,  _a  :					Optional[Dict[str, str]] = None		,  _a  :					Optional[Union[bool, str]] = None		,  _a  :					Optional[str] = None		,  _a  :					bool = False		,  **_a  :					Optional[int]		,  ):





 '''simple docstring'''







 snake_case__  :    Union[str, Any]							       =get_cached_module_file(
     _a		,  _a		,  cache_dir=_a		,  force_download=_a		,  resume_download=_a		,  proxies=_a		,  use_auth_token=_a		,  revision=_a		,  local_files_only=_a		,  )
 return get_class_in_module(_a		,  final_module.replace(""".py"""		,  """"""	)	)













import argparse

from transformers import BigBirdConfig, BigBirdForPreTraining, BigBirdForQuestionAnswering, load_tf_weights_in_big_bird
from transformers.utils import logging


logging.set_verbosity_info()







def 			A__							(      _a  :					int		,  _a  :					Any		,  _a  :					Union[str, Any]		,  _a  :					Tuple	):





 '''simple docstring'''







 snake_case__  :    Any							       =BigBirdConfig.from_json_file(_a	)
 print(f"Building PyTorch model from configuration: {config}"	)

 if is_trivia_qa:
  snake_case__  :    str							       =BigBirdForQuestionAnswering(_a	)
 else:
  snake_case__  :    Optional[int]							       =BigBirdForPreTraining(_a	)

 # Load weights from tf checkpoint
 load_tf_weights_in_big_bird(_a		,  _a		,  is_trivia_qa=_a	)

 # Save pytorch-model
 print(f"Save PyTorch model to {pytorch_dump_path}"	)
 model.save_pretrained(_a	)


if __name__ == "__main__":
       __lowerCamelCase	:		Union[str, Any]										=   argparse.ArgumentParser()
       # Required parameters
       parser.add_argument(
           """--tf_checkpoint_path""", default=None, type=str, required=True, help="""Path to the TensorFlow checkpoint path."""
       )
       parser.add_argument(
           """--big_bird_config_file""",
           default=None,
           type=str,
           required=True,
           help=(
               """The config json file corresponding to the pre-trained BERT model. \n"""
               """This specifies the model architecture."""
           ),
       )
       parser.add_argument(
           """--pytorch_dump_path""", default=None, type=str, required=True, help="""Path to the output PyTorch model."""
       )
       parser.add_argument(
           """--is_trivia_qa""", action="""store_true""", help="""Whether to convert a model with a trivia_qa head."""
       )
       __lowerCamelCase	:		Tuple										=   parser.parse_args()
       convert_tf_checkpoint_to_pytorch(
           args.tf_checkpoint_path, args.big_bird_config_file, args.pytorch_dump_path, args.is_trivia_qa
       )





