
import json
import os
import shutil
import warnings
from argparse import ArgumentParser, Namespace
from pathlib import Path
from typing import List

from ..utils import logging
from . import BaseTransformersCLICommand


try:
      from cookiecutter.main import cookiecutter

      lowercase__       =True
except ImportError:
      lowercase__       =False

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




def    __UpperCamelCase     (				lowerCAmelCase__							:				Namespace  ):
 return AddNewModelCommand(args.testing  ,   args.testing_file  ,   path=args.path  )



class 		UpperCamelCase__  (       __lowercase    ):
   @staticmethod
   def       lowerCAmelCase			(snake_case_ :					ArgumentParser   ):
    __a :  List[Any]						=  parser.add_parser('''add-new-model'''   )
    add_new_model_parser.add_argument('''--testing'''					,				action='''store_true'''					,				help='''If in testing mode.'''   )
    add_new_model_parser.add_argument('''--testing_file'''					,				type=snake_case_					,				help='''Configuration file on which to run.'''   )
    add_new_model_parser.add_argument(
        '''--path'''					,				type=snake_case_					,				help='''Path to cookiecutter. Should only be used for testing purposes.'''   )
    add_new_model_parser.set_defaults(func=snake_case_   )
   def __init__(self :					Dict					,				snake_case_ :					bool					,				snake_case_ :					str					,				snake_case_ :					Dict=None					,				*snake_case_ :					Optional[Any]   ):
    __a :  Union[str, Any]						=  testing
    __a :  List[Any]						=  testing_file
    __a :  Any						=  path







   def       lowerCAmelCase			(self :					int   ):
    warnings.warn(
        '''The command `transformers-cli add-new-model` is deprecated and will be removed in v5 of Transformers. '''
        '''It is not actively maintained anymore, so might give a result that won\'t pass all tests and quality '''
        '''checks, you should use `transformers-cli add-new-model-like` instead.'''   )
    if not _has_cookiecutter:
     raise ImportError(
         '''Model creation dependencies are required to use the `add_new_model` command. Install them by running '''
         '''the following at the root of your `transformers` clone:\n\n\t$ pip install -e .[modelcreation]\n'''   )
    # Ensure that there is no other `cookiecutter-template-xxx` directory in the current working directory
    __a :  Union[str, Any]						=  [directory for directory in os.listdir() if '''cookiecutter-template-''' == directory[:2_2]]
    if len(snake_case_   ) > 0:
     raise ValueError(
         '''Several directories starting with `cookiecutter-template-` in current working directory. '''
         '''Please clean your directory by removing all folders starting with `cookiecutter-template-` or '''
         '''change your working directory.'''   )

    __a :  Union[str, Any]						=  (
        Path(snake_case_   ).parent.parent.parent.parent if self._path is None else Path(self._path   ).parent.parent
    )
    __a :  Union[str, Any]						=  path_to_transformer_root / '''templates''' / '''adding_a_new_model'''

    # Execute cookiecutter
    if not self._testing:
     cookiecutter(str(snake_case_   )   )
    else:
     with open(self._testing_file					,				'''r'''   ) as configuration_file:
      __a :  List[Any]						=  json.load(snake_case_   )

     cookiecutter(
         str(path_to_cookiecutter if self._path is None else self._path   )					,				no_input=snake_case_					,				extra_context=snake_case_					,				)

    __a :  List[str]						=  [directory for directory in os.listdir() if '''cookiecutter-template-''' in directory[:2_2]][0]

    # Retrieve configuration
    with open(directory + '''/configuration.json'''					,				'''r'''   ) as configuration_file:
     __a :  Optional[Any]						=  json.load(snake_case_   )

    __a :  str						=  configuration['''lowercase_modelname''']
    __a :  int						=  configuration['''generate_tensorflow_pytorch_and_flax''']
    os.remove(f"{directory}/configuration.json"   )

    __a :  Any						=  '''PyTorch''' in generate_tensorflow_pytorch_and_flax
    __a :  Dict						=  '''TensorFlow''' in generate_tensorflow_pytorch_and_flax
    __a :  Optional[int]						=  '''Flax''' in generate_tensorflow_pytorch_and_flax

    __a :  Dict						=  f"{path_to_transformer_root}/src/transformers/models/{lowercase_model_name}"
    os.makedirs(snake_case_					,				exist_ok=snake_case_   )
    os.makedirs(f"{path_to_transformer_root}/tests/models/{lowercase_model_name}"					,				exist_ok=snake_case_   )

    # Tests require submodules as they have parent imports
    with open(f"{path_to_transformer_root}/tests/models/{lowercase_model_name}/__init__.py"					,				'''w'''   ):
     pass

    shutil.move(
        f"{directory}/__init__.py"					,				f"{model_dir}/__init__.py"					,				)
    shutil.move(
        f"{directory}/configuration_{lowercase_model_name}.py"					,				f"{model_dir}/configuration_{lowercase_model_name}.py"					,				)

    def remove_copy_lines(snake_case_ :					Union[str, Any]   ):
     with open(snake_case_					,				'''r'''   ) as f:
      __a :  Union[str, Any]						=  f.readlines()
     with open(snake_case_					,				'''w'''   ) as f:
      for line in lines:
       if "# Copied from transformers." not in line:
        f.write(snake_case_   )

    if output_pytorch:
     if not self._testing:
      remove_copy_lines(f"{directory}/modeling_{lowercase_model_name}.py"   )

     shutil.move(
         f"{directory}/modeling_{lowercase_model_name}.py"					,				f"{model_dir}/modeling_{lowercase_model_name}.py"					,				)

     shutil.move(
         f"{directory}/test_modeling_{lowercase_model_name}.py"					,				f"{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_{lowercase_model_name}.py"					,				)
    else:
     os.remove(f"{directory}/modeling_{lowercase_model_name}.py"   )
     os.remove(f"{directory}/test_modeling_{lowercase_model_name}.py"   )

    if output_tensorflow:
     if not self._testing:
      remove_copy_lines(f"{directory}/modeling_tf_{lowercase_model_name}.py"   )

     shutil.move(
         f"{directory}/modeling_tf_{lowercase_model_name}.py"					,				f"{model_dir}/modeling_tf_{lowercase_model_name}.py"					,				)

     shutil.move(
         f"{directory}/test_modeling_tf_{lowercase_model_name}.py"					,				f"{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_tf_{lowercase_model_name}.py"					,				)
    else:
     os.remove(f"{directory}/modeling_tf_{lowercase_model_name}.py"   )
     os.remove(f"{directory}/test_modeling_tf_{lowercase_model_name}.py"   )

    if output_flax:
     if not self._testing:
      remove_copy_lines(f"{directory}/modeling_flax_{lowercase_model_name}.py"   )

     shutil.move(
         f"{directory}/modeling_flax_{lowercase_model_name}.py"					,				f"{model_dir}/modeling_flax_{lowercase_model_name}.py"					,				)

     shutil.move(
         f"{directory}/test_modeling_flax_{lowercase_model_name}.py"					,				f"{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_flax_{lowercase_model_name}.py"					,				)
    else:
     os.remove(f"{directory}/modeling_flax_{lowercase_model_name}.py"   )
     os.remove(f"{directory}/test_modeling_flax_{lowercase_model_name}.py"   )

    shutil.move(
        f"{directory}/{lowercase_model_name}.md"					,				f"{path_to_transformer_root}/docs/source/en/model_doc/{lowercase_model_name}.md"					,				)

    shutil.move(
        f"{directory}/tokenization_{lowercase_model_name}.py"					,				f"{model_dir}/tokenization_{lowercase_model_name}.py"					,				)

    shutil.move(
        f"{directory}/tokenization_fast_{lowercase_model_name}.py"					,				f"{model_dir}/tokenization_{lowercase_model_name}_fast.py"					,				)

    from os import fdopen, remove
    from shutil import copymode, move
    from tempfile import mkstemp

    def replace(snake_case_ :					str					,				snake_case_ :					str					,				snake_case_ :					List[str]   ):
     # Create temp file
     __a				,       __a :  Tuple						=  mkstemp()
     __a :  Optional[Any]						=  False
     with fdopen(snake_case_					,				'''w'''   ) as new_file:
      with open(snake_case_   ) as old_file:
       for line in old_file:
        new_file.write(snake_case_   )
        if line_to_copy_below in line:
         __a :  Tuple						=  True
         for line_to_copy in lines_to_copy:
          new_file.write(snake_case_   )

     if not line_found:
      raise ValueError(f"Line {line_to_copy_below} was not found in file."   )

     # Copy the file permissions from the old file to the new file
     copymode(snake_case_					,				snake_case_   )
     # Remove original file
     remove(snake_case_   )
     # Move new file
     move(snake_case_					,				snake_case_   )

    def skip_units(snake_case_ :					Any   ):
     return (
         ("generating PyTorch" in line and not output_pytorch)
         or ("generating TensorFlow" in line and not output_tensorflow)
         or ("generating Flax" in line and not output_flax)
     )

    def replace_in_files(snake_case_ :					int   ):
     with open(snake_case_   ) as datafile:
      __a :  List[Any]						=  []
      __a :  int						=  False
      __a :  Tuple						=  False
      for line in datafile:
       if "# To replace in: " in line and "##" not in line:
        __a :  Optional[Any]						=  line.split('''"'''   )[1]
        __a :  Dict						=  skip_units(snake_case_   )
       elif "# Below: " in line and "##" not in line:
        __a :  str						=  line.split('''"'''   )[1]
        __a :  Any						=  skip_units(snake_case_   )
       elif "# End." in line and "##" not in line:
        if not skip_file and not skip_snippet:
         replace(snake_case_					,				snake_case_					,				snake_case_   )

        __a :  str						=  []
       elif "# Replace with" in line and "##" not in line:
        __a :  Optional[int]						=  []
       elif "##" not in line:
        lines_to_copy.append(snake_case_   )

     remove(snake_case_   )

    replace_in_files(f"{directory}/to_replace_{lowercase_model_name}.py"   )
    os.rmdir(snake_case_   )


import argparse
import os
import re

import tensorflow as tf
import torch

from transformers import BertConfig, BertModel
from transformers.utils import logging


logging.set_verbosity_info()
lowercase__       =logging.get_logger(__name__)




def    __UpperCamelCase     (				lowerCAmelCase__							:				Union[str, Any]  ,   lowerCAmelCase__							:				int  ,   lowerCAmelCase__							:				Dict  ):
 __a :  List[str]						=  os.path.abspath(lowerCAmelCase__  )
 logger.info(f"Converting TensorFlow checkpoint from {tf_path}"  )
 # Load weights from TF model
 __a :  Tuple						=  tf.train.list_variables(lowerCAmelCase__  )
 __a :  Optional[Any]						=  []
 __a :  Union[str, Any]						=  []
 __a :  str						=  []
 for full_name, shape in init_vars:
  # logger.info(f"Loading TF weight {name} with shape {shape}")
  __a :  Any						=  full_name.split('''/'''  )
  if full_name == "_CHECKPOINTABLE_OBJECT_GRAPH" or name[0] in ["global_step", "save_counter"]:
   logger.info(f"Skipping non-model layer {full_name}"  )
   continue
  if "optimizer" in full_name:
   logger.info(f"Skipping optimization layer {full_name}"  )
   continue
  if name[0] == "model":
   # ignore initial 'model'
   __a :  Any						=  name[1:]
  # figure out how many levels deep the name is
  __a :  List[Any]						=  0
  for _name in name:
   if _name.startswith('''layer_with_weights'''  ):
    depth += 1
   else:
    break
  layer_depth.append(lowerCAmelCase__  )
  # read data
  __a :  Tuple						=  tf.train.load_variable(lowerCAmelCase__  ,   lowerCAmelCase__  )
  names.append('''/'''.join(lowerCAmelCase__  )  )
  arrays.append(lowerCAmelCase__  )
 logger.info(f"Read a total of {len(lowerCAmelCase__  ):,} layers"  )

 # Sanity check
 if len(set(lowerCAmelCase__  )  ) != 1:
  raise ValueError(f"Found layer names with different depths (layer depth {list(set(lowerCAmelCase__  )  )})"  )
 __a :  int						=  list(set(lowerCAmelCase__  )  )[0]
 if layer_depth != 1:
  raise ValueError(
      '''The model contains more than just the embedding/encoder layers. This script does not handle MLM/NSP'''
      ''' heads.'''  )

 # convert layers
 logger.info('''Converting weights...'''  )
 for full_name, array in zip(lowerCAmelCase__  ,   lowerCAmelCase__  ):
  __a :  int						=  full_name.split('''/'''  )
  __a :  Tuple						=  model
  __a :  Dict						=  []
  for i, m_name in enumerate(lowerCAmelCase__  ):
   if m_name == ".ATTRIBUTES":
    # variable names end with .ATTRIBUTES/VARIABLE_VALUE
    break
   if m_name.startswith('''layer_with_weights'''  ):
    __a :  Union[str, Any]						=  int(m_name.split('''-'''  )[-1]  )
    if layer_num <= 2:
     # embedding layers
     # layer_num 0: word_embeddings
     # layer_num 1: position_embeddings
     # layer_num 2: token_type_embeddings
     continue
    elif layer_num == 3:
     # embedding LayerNorm
     trace.extend(['''embeddings''', '''LayerNorm''']  )
     __a :  Union[str, Any]						=  getattr(lowerCAmelCase__  ,   '''embeddings'''  )
     __a :  List[str]						=  getattr(lowerCAmelCase__  ,   '''LayerNorm'''  )
    elif layer_num > 3 and layer_num < config.num_hidden_layers + 4:
     # encoder layers
     trace.extend(['''encoder''', '''layer''', str(layer_num - 4  )]  )
     __a :  Dict						=  getattr(lowerCAmelCase__  ,   '''encoder'''  )
     __a :  Union[str, Any]						=  getattr(lowerCAmelCase__  ,   '''layer'''  )
     __a :  Any						=  pointer[layer_num - 4]
    elif layer_num == config.num_hidden_layers + 4:
     # pooler layer
     trace.extend(['''pooler''', '''dense''']  )
     __a :  Any						=  getattr(lowerCAmelCase__  ,   '''pooler'''  )
     __a :  Optional[int]						=  getattr(lowerCAmelCase__  ,   '''dense'''  )
   elif m_name == "embeddings":
    trace.append('''embeddings'''  )
    __a :  int						=  getattr(lowerCAmelCase__  ,   '''embeddings'''  )
    if layer_num == 0:
     trace.append('''word_embeddings'''  )
     __a :  Optional[int]						=  getattr(lowerCAmelCase__  ,   '''word_embeddings'''  )
    elif layer_num == 1:
     trace.append('''position_embeddings'''  )
     __a :  List[str]						=  getattr(lowerCAmelCase__  ,   '''position_embeddings'''  )
    elif layer_num == 2:
     trace.append('''token_type_embeddings'''  )
     __a :  Optional[Any]						=  getattr(lowerCAmelCase__  ,   '''token_type_embeddings'''  )
    else:
     raise ValueError(f"Unknown embedding layer with name {full_name}"  )
    trace.append('''weight'''  )
    __a :  Tuple						=  getattr(lowerCAmelCase__  ,   '''weight'''  )
   elif m_name == "_attention_layer":
    # self-attention layer
    trace.extend(['''attention''', '''self''']  )
    __a :  Optional[Any]						=  getattr(lowerCAmelCase__  ,   '''attention'''  )
    __a :  Union[str, Any]						=  getattr(lowerCAmelCase__  ,   '''self'''  )
   elif m_name == "_attention_layer_norm":
    # output attention norm
    trace.extend(['''attention''', '''output''', '''LayerNorm''']  )
    __a :  int						=  getattr(lowerCAmelCase__  ,   '''attention'''  )
    __a :  List[Any]						=  getattr(lowerCAmelCase__  ,   '''output'''  )
    __a :  List[Any]						=  getattr(lowerCAmelCase__  ,   '''LayerNorm'''  )
   elif m_name == "_attention_output_dense":
    # output attention dense
    trace.extend(['''attention''', '''output''', '''dense''']  )
    __a :  Optional[int]						=  getattr(lowerCAmelCase__  ,   '''attention'''  )
    __a :  Optional[Any]						=  getattr(lowerCAmelCase__  ,   '''output'''  )
    __a :  Any						=  getattr(lowerCAmelCase__  ,   '''dense'''  )
   elif m_name == "_output_dense":
    # output dense
    trace.extend(['''output''', '''dense''']  )
    __a :  Tuple						=  getattr(lowerCAmelCase__  ,   '''output'''  )
    __a :  str						=  getattr(lowerCAmelCase__  ,   '''dense'''  )
   elif m_name == "_output_layer_norm":
    # output dense
    trace.extend(['''output''', '''LayerNorm''']  )
    __a :  int						=  getattr(lowerCAmelCase__  ,   '''output'''  )
    __a :  str						=  getattr(lowerCAmelCase__  ,   '''LayerNorm'''  )
   elif m_name == "_key_dense":
    # attention key
    trace.append('''key'''  )
    __a :  Union[str, Any]						=  getattr(lowerCAmelCase__  ,   '''key'''  )
   elif m_name == "_query_dense":
    # attention query
    trace.append('''query'''  )
    __a :  Union[str, Any]						=  getattr(lowerCAmelCase__  ,   '''query'''  )
   elif m_name == "_value_dense":
    # attention value
    trace.append('''value'''  )
    __a :  Union[str, Any]						=  getattr(lowerCAmelCase__  ,   '''value'''  )
   elif m_name == "_intermediate_dense":
    # attention intermediate dense
    trace.extend(['''intermediate''', '''dense''']  )
    __a :  Optional[Any]						=  getattr(lowerCAmelCase__  ,   '''intermediate'''  )
    __a :  Optional[int]						=  getattr(lowerCAmelCase__  ,   '''dense'''  )
   elif m_name == "_output_layer_norm":
    # output layer norm
    trace.append('''output'''  )
    __a :  int						=  getattr(lowerCAmelCase__  ,   '''output'''  )
   # weights & biases
   elif m_name in ["bias", "beta"]:
    trace.append('''bias'''  )
    __a :  Dict						=  getattr(lowerCAmelCase__  ,   '''bias'''  )
   elif m_name in ["kernel", "gamma"]:
    trace.append('''weight'''  )
    __a :  List[Any]						=  getattr(lowerCAmelCase__  ,   '''weight'''  )
   else:
    logger.warning(f"Ignored {m_name}"  )
        # for certain layers reshape is necessary
  __a :  List[str]						=  '''.'''.join(lowerCAmelCase__  )
  if re.match(R'''(\S+)\.attention\.self\.(key|value|query)\.(bias|weight)'''  ,   lowerCAmelCase__  ) or re.match(
      R'''(\S+)\.attention\.output\.dense\.weight'''  ,   lowerCAmelCase__  ):
   __a :  str						=  array.reshape(pointer.data.shape  )
  if "kernel" in full_name:
   __a :  Optional[Any]						=  array.transpose()
  if pointer.shape == array.shape:
   __a :  str						=  torch.from_numpy(lowerCAmelCase__  )
  else:
   raise ValueError(
       f"Shape mismatch in layer {full_name}: Model expects shape {pointer.shape} but layer contains shape:"
       f" {array.shape}"  )
  logger.info(f"Successfully set variable {full_name} to PyTorch layer {trace}"  )
 return model




def    __UpperCamelCase     (				lowerCAmelCase__							:				Optional[Any]  ,   lowerCAmelCase__							:				Union[str, Any]  ,   lowerCAmelCase__							:				Optional[Any]  ):
 # Instantiate model
 logger.info(f"Loading model based on config from {config_path}..."  )
 __a :  Dict						=  BertConfig.from_json_file(lowerCAmelCase__  )
 __a :  int						=  BertModel(lowerCAmelCase__  )

 # Load weights from checkpoint
 logger.info(f"Loading weights from checkpoint {tf_checkpoint_path}..."  )
 load_tfa_weights_in_bert(lowerCAmelCase__  ,   lowerCAmelCase__  ,   lowerCAmelCase__  )

 # Save pytorch-model
 logger.info(f"Saving PyTorch model to {pytorch_dump_path}..."  )
 torch.save(model.state_dict()  ,   lowerCAmelCase__  )


if __name__ == "__main__":
      lowercase__       =argparse.ArgumentParser()
      parser.add_argument(
          '--tf_checkpoint_path', type=str, required=True, help='Path to the TensorFlow 2.x checkpoint path.'
      )
      parser.add_argument(
          '--bert_config_file',
          type=str,
          required=True,
          help='The config json file corresponding to the BERT model. This specifies the model architecture.',
      )
      parser.add_argument(
          '--pytorch_dump_path',
          type=str,
          required=True,
          help='Path to the output PyTorch model (must include filename).',
      )
      lowercase__       =parser.parse_args()
      convert_tfa_checkpoint_to_pytorch(args.tf_checkpoint_path, args.bert_config_file, args.pytorch_dump_path)
