
"""simple docstring"""






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      lowerCAmelCase__			(		__magic_name__			,   __magic_name__			,   __magic_name__      )						->str:
			__lowercase       =       os.path.abspath(__magic_name__      )
			logger.info(F'''Converting TensorFlow checkpoint from {tf_path}'''      )
			# Load weights from TF model
			__lowercase       =       tf.train.list_variables(__magic_name__      )
			__lowercase       =       []
			__lowercase       =       []
			__lowercase       =       []
			for full_name, shape in init_vars:
						# logger.info(f"Loading TF weight {name} with shape {shape}")
						__lowercase       =       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'
									__lowercase       =       name[1:]
						# figure out how many levels deep the name is
						__lowercase       =       0
						for _name in name:
									if _name.startswith("layer_with_weights"      ):
												depth += 1
									else:
												break
						layer_depth.append(__magic_name__      )
						# read data
						__lowercase       =       tf.train.load_variable(__magic_name__			,   __magic_name__      )
						names.append("/".join(__magic_name__      )      )
						arrays.append(__magic_name__      )
			logger.info(F'''Read a total of {len(__magic_name__      ):,} layers'''      )

			# Sanity check
			if len(set(__magic_name__      )      ) != 1:
						raise ValueError(F'''Found layer names with different depths (layer depth {list(set(__magic_name__      )      )})'''      )
			__lowercase       =       list(set(__magic_name__      )      )[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(__magic_name__			,   __magic_name__      ):
						__lowercase       =       full_name.split("/"      )
						__lowercase       =       model
						__lowercase       =       []
						for i, m_name in enumerate(__magic_name__      ):
									if m_name == ".ATTRIBUTES":
												# variable names end with .ATTRIBUTES/VARIABLE_VALUE
												break
									if m_name.startswith("layer_with_weights"      ):
												__lowercase       =       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"]      )
															__lowercase       =       getattr(__magic_name__			,   "embeddings"      )
															__lowercase       =       getattr(__magic_name__			,   "LayerNorm"      )
												elif layer_num > 3 and layer_num < config.num_hidden_layers + 4:
															# encoder layers
															trace.extend(["encoder", "layer", str(layer_num - 4      )]      )
															__lowercase       =       getattr(__magic_name__			,   "encoder"      )
															__lowercase       =       getattr(__magic_name__			,   "layer"      )
															__lowercase       =       pointer[layer_num - 4]
												elif layer_num == config.num_hidden_layers + 4:
															# pooler layer
															trace.extend(["pooler", "dense"]      )
															__lowercase       =       getattr(__magic_name__			,   "pooler"      )
															__lowercase       =       getattr(__magic_name__			,   "dense"      )
									elif m_name == "embeddings":
												trace.append("embeddings"      )
												__lowercase       =       getattr(__magic_name__			,   "embeddings"      )
												if layer_num == 0:
															trace.append("word_embeddings"      )
															__lowercase       =       getattr(__magic_name__			,   "word_embeddings"      )
												elif layer_num == 1:
															trace.append("position_embeddings"      )
															__lowercase       =       getattr(__magic_name__			,   "position_embeddings"      )
												elif layer_num == 2:
															trace.append("token_type_embeddings"      )
															__lowercase       =       getattr(__magic_name__			,   "token_type_embeddings"      )
												else:
															raise ValueError(F'''Unknown embedding layer with name {full_name}'''      )
												trace.append("weight"      )
												__lowercase       =       getattr(__magic_name__			,   "weight"      )
									elif m_name == "_attention_layer":
												# self-attention layer
												trace.extend(["attention", "self"]      )
												__lowercase       =       getattr(__magic_name__			,   "attention"      )
												__lowercase       =       getattr(__magic_name__			,   "self"      )
									elif m_name == "_attention_layer_norm":
												# output attention norm
												trace.extend(["attention", "output", "LayerNorm"]      )
												__lowercase       =       getattr(__magic_name__			,   "attention"      )
												__lowercase       =       getattr(__magic_name__			,   "output"      )
												__lowercase       =       getattr(__magic_name__			,   "LayerNorm"      )
									elif m_name == "_attention_output_dense":
												# output attention dense
												trace.extend(["attention", "output", "dense"]      )
												__lowercase       =       getattr(__magic_name__			,   "attention"      )
												__lowercase       =       getattr(__magic_name__			,   "output"      )
												__lowercase       =       getattr(__magic_name__			,   "dense"      )
									elif m_name == "_output_dense":
												# output dense
												trace.extend(["output", "dense"]      )
												__lowercase       =       getattr(__magic_name__			,   "output"      )
												__lowercase       =       getattr(__magic_name__			,   "dense"      )
									elif m_name == "_output_layer_norm":
												# output dense
												trace.extend(["output", "LayerNorm"]      )
												__lowercase       =       getattr(__magic_name__			,   "output"      )
												__lowercase       =       getattr(__magic_name__			,   "LayerNorm"      )
									elif m_name == "_key_dense":
												# attention key
												trace.append("key"      )
												__lowercase       =       getattr(__magic_name__			,   "key"      )
									elif m_name == "_query_dense":
												# attention query
												trace.append("query"      )
												__lowercase       =       getattr(__magic_name__			,   "query"      )
									elif m_name == "_value_dense":
												# attention value
												trace.append("value"      )
												__lowercase       =       getattr(__magic_name__			,   "value"      )
									elif m_name == "_intermediate_dense":
												# attention intermediate dense
												trace.extend(["intermediate", "dense"]      )
												__lowercase       =       getattr(__magic_name__			,   "intermediate"      )
												__lowercase       =       getattr(__magic_name__			,   "dense"      )
									elif m_name == "_output_layer_norm":
												# output layer norm
												trace.append("output"      )
												__lowercase       =       getattr(__magic_name__			,   "output"      )
									# weights & biases
									elif m_name in ["bias", "beta"]:
												trace.append("bias"      )
												__lowercase       =       getattr(__magic_name__			,   "bias"      )
									elif m_name in ["kernel", "gamma"]:
												trace.append("weight"      )
												__lowercase       =       getattr(__magic_name__			,   "weight"      )
									else:
												logger.warning(F'''Ignored {m_name}'''      )
        # for certain layers reshape is necessary
						__lowercase       =       ".".join(__magic_name__      )
						if re.match(R"(\S+)\.attention\.self\.(key|value|query)\.(bias|weight)"			,   __magic_name__      ) or re.match(
						    R"(\S+)\.attention\.output\.dense\.weight"			,   __magic_name__      ):
									__lowercase       =       array.reshape(pointer.data.shape      )
						if "kernel" in full_name:
									__lowercase       =       array.transpose()
						if pointer.shape == array.shape:
									__lowercase       =       torch.from_numpy(__magic_name__      )
						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      lowerCAmelCase__			(		__magic_name__			,   __magic_name__			,   __magic_name__      )						->str:
			# Instantiate model
			logger.info(F'''Loading model based on config from {config_path}...'''      )
			__lowercase       =       BertConfig.from_json_file(__magic_name__      )
			__lowercase       =       BertModel(__magic_name__      )

			# Load weights from checkpoint
			logger.info(F'''Loading weights from checkpoint {tf_checkpoint_path}...'''      )
			load_tfa_weights_in_bert(__magic_name__			,   __magic_name__			,   __magic_name__      )

			# Save pytorch-model
			logger.info(F'''Saving PyTorch model to {pytorch_dump_path}...'''      )
			torch.save(model.state_dict()			,   __magic_name__      )


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)







"""simple docstring"""






import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / '''utils'''))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
					from test_module.custom_tokenization_fast import CustomTokenizerFast

class 					__a     ( unittest.TestCase       ):



							'''simple docstring'''







							def   SCREAMING_SNAKE_CASE						(     self       )			->							int:




										'''simple docstring'''
										# A mock response for an HTTP head request to emulate server down
										__lowercase       =       mock.Mock()
										__lowercase       =       500
										__lowercase       =       {}
										__lowercase       =       HTTPError
										__lowercase       =       {}

										# Download this model to make sure it's in the cache.
										__lowercase       =       BertTokenizer.from_pretrained("hf-internal-testing/tiny-random-bert"       )

										# Under the mock environment we get a 500 error when trying to reach the tokenizer.
										with mock.patch("requests.Session.request"  ,    return_value=_lowerCamelCase       ) as mock_head:
													__lowercase       =       BertTokenizer.from_pretrained("hf-internal-testing/tiny-random-bert"       )
													# This check we did call the fake head request
													mock_head.assert_called()
							@require_tokenizers
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										# A mock response for an HTTP head request to emulate server down
										__lowercase       =       mock.Mock()
										__lowercase       =       500
										__lowercase       =       {}
										__lowercase       =       HTTPError
										__lowercase       =       {}

										# Download this model to make sure it's in the cache.
										__lowercase       =       GPTaTokenizerFast.from_pretrained("gpt2"       )

										# Under the mock environment we get a 500 error when trying to reach the tokenizer.
										with mock.patch("requests.Session.request"  ,    return_value=_lowerCamelCase       ) as mock_head:
													__lowercase       =       GPTaTokenizerFast.from_pretrained("gpt2"       )
													# This check we did call the fake head request
													mock_head.assert_called()
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										# This test is for deprecated behavior and can be removed in v5
										try:
													__lowercase       =       tempfile.mktemp()
													with open(_lowerCamelCase  ,    "wb"       ) as f:
																http_get("https://huggingface.co/albert-base-v1/resolve/main/spiece.model"  ,    _lowerCamelCase       )

													__lowercase       =       AlbertTokenizer.from_pretrained(_lowerCamelCase       )
										finally:
													os.remove(_lowerCamelCase       )

										# Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
										# the current folder and have the right name.
										if os.path.isfile("tokenizer.json"       ):
													# We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
													return
										try:
													with open("tokenizer.json"  ,    "wb"       ) as f:
																http_get("https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json"  ,    _lowerCamelCase       )
													__lowercase       =       AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2"       )
													# The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
													self.assertEqual(tokenizer.vocab_size  ,    1_000       )
													# Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

										finally:
													os.remove("tokenizer.json"       )





							def   SCREAMING_SNAKE_CASE						(     self       )			->							int:




										'''simple docstring'''
										# This test is for deprecated behavior and can be removed in v5
										__lowercase       =       AlbertTokenizer.from_pretrained("https://huggingface.co/albert-base-v1/resolve/main/spiece.model"       )





@is_staging_test
class 					__a     ( unittest.TestCase       ):



							'''simple docstring'''







							_lowerCamelCase					:			str          =   ["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]
							@classmethod
							def   SCREAMING_SNAKE_CASE						(     cls       )			->							List[Any]:




										'''simple docstring'''
										__lowercase       =       TOKEN
										HfFolder.save_token(_lowerCamelCase       )
							@classmethod
							def   SCREAMING_SNAKE_CASE						(     cls       )			->							List[str]:




										'''simple docstring'''
										try:
													delete_repo(token=cls._token  ,    repo_id="test-tokenizer"       )
										except HTTPError:
													pass

										try:
													delete_repo(token=cls._token  ,    repo_id="valid_org/test-tokenizer-org"       )
										except HTTPError:
													pass

										try:
													delete_repo(token=cls._token  ,    repo_id="test-dynamic-tokenizer"       )
										except HTTPError:
													pass
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										with tempfile.TemporaryDirectory() as tmp_dir:
													__lowercase       =       os.path.join(_lowerCamelCase  ,    "vocab.txt"       )
													with open(_lowerCamelCase  ,    "w"  ,    encoding="utf-8"       ) as vocab_writer:
																vocab_writer.write("".join([x + "\n" for x in self.vocab_tokens]       )       )
													__lowercase       =       BertTokenizer(_lowerCamelCase       )

										tokenizer.push_to_hub("test-tokenizer"  ,    use_auth_token=self._token       )
										__lowercase       =       BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''       )
										self.assertDictEqual(new_tokenizer.vocab  ,    tokenizer.vocab       )

										# Reset repo
										delete_repo(token=self._token  ,    repo_id="test-tokenizer"       )

										# Push to hub via save_pretrained
										with tempfile.TemporaryDirectory() as tmp_dir:
													tokenizer.save_pretrained(_lowerCamelCase  ,    repo_id="test-tokenizer"  ,    push_to_hub=_lowerCamelCase  ,    use_auth_token=self._token       )

										__lowercase       =       BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''       )
										self.assertDictEqual(new_tokenizer.vocab  ,    tokenizer.vocab       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										with tempfile.TemporaryDirectory() as tmp_dir:
													__lowercase       =       os.path.join(_lowerCamelCase  ,    "vocab.txt"       )
													with open(_lowerCamelCase  ,    "w"  ,    encoding="utf-8"       ) as vocab_writer:
																vocab_writer.write("".join([x + "\n" for x in self.vocab_tokens]       )       )
													__lowercase       =       BertTokenizer(_lowerCamelCase       )

										tokenizer.push_to_hub("valid_org/test-tokenizer-org"  ,    use_auth_token=self._token       )
										__lowercase       =       BertTokenizer.from_pretrained("valid_org/test-tokenizer-org"       )
										self.assertDictEqual(new_tokenizer.vocab  ,    tokenizer.vocab       )

										# Reset repo
										delete_repo(token=self._token  ,    repo_id="valid_org/test-tokenizer-org"       )

										# Push to hub via save_pretrained
										with tempfile.TemporaryDirectory() as tmp_dir:
													tokenizer.save_pretrained(
													    _lowerCamelCase  ,    repo_id="valid_org/test-tokenizer-org"  ,    push_to_hub=_lowerCamelCase  ,    use_auth_token=self._token       )

										__lowercase       =       BertTokenizer.from_pretrained("valid_org/test-tokenizer-org"       )
										self.assertDictEqual(new_tokenizer.vocab  ,    tokenizer.vocab       )





							@require_tokenizers
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Optional[int]:




										'''simple docstring'''
										CustomTokenizer.register_for_auto_class()
										with tempfile.TemporaryDirectory() as tmp_dir:
													__lowercase       =       os.path.join(_lowerCamelCase  ,    "vocab.txt"       )
													with open(_lowerCamelCase  ,    "w"  ,    encoding="utf-8"       ) as vocab_writer:
																vocab_writer.write("".join([x + "\n" for x in self.vocab_tokens]       )       )
													__lowercase       =       CustomTokenizer(_lowerCamelCase       )

										# No fast custom tokenizer
										tokenizer.push_to_hub("test-dynamic-tokenizer"  ,    use_auth_token=self._token       )

										__lowercase       =       AutoTokenizer.from_pretrained(f'''{USER}/test-dynamic-tokenizer'''  ,    trust_remote_code=_lowerCamelCase       )
										# Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
										self.assertEqual(tokenizer.__class__.__name__  ,    "CustomTokenizer"       )

										# Fast and slow custom tokenizer
										CustomTokenizerFast.register_for_auto_class()
										with tempfile.TemporaryDirectory() as tmp_dir:
													__lowercase       =       os.path.join(_lowerCamelCase  ,    "vocab.txt"       )
													with open(_lowerCamelCase  ,    "w"  ,    encoding="utf-8"       ) as vocab_writer:
																vocab_writer.write("".join([x + "\n" for x in self.vocab_tokens]       )       )

													__lowercase       =       BertTokenizerFast.from_pretrained(_lowerCamelCase       )
													bert_tokenizer.save_pretrained(_lowerCamelCase       )
													__lowercase       =       CustomTokenizerFast.from_pretrained(_lowerCamelCase       )

										tokenizer.push_to_hub("test-dynamic-tokenizer"  ,    use_auth_token=self._token       )

										__lowercase       =       AutoTokenizer.from_pretrained(f'''{USER}/test-dynamic-tokenizer'''  ,    trust_remote_code=_lowerCamelCase       )
										# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
										self.assertEqual(tokenizer.__class__.__name__  ,    "CustomTokenizerFast"       )
										__lowercase       =       AutoTokenizer.from_pretrained(
										    f'''{USER}/test-dynamic-tokenizer'''  ,    use_fast=_lowerCamelCase  ,    trust_remote_code=_lowerCamelCase       )
										# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
										self.assertEqual(tokenizer.__class__.__name__  ,    "CustomTokenizer"       )





class 					__a     ( unittest.TestCase       ):



							'''simple docstring'''







							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("Hello 友達"       )
										self.assertEqual(trie.data  ,    {"H": {"e": {"l": {"l": {"o": {" ": {"友": {"達": {"": 1}}}}}}}}}       )
										trie.add("Hello"       )
										trie.data
										self.assertEqual(trie.data  ,    {"H": {"e": {"l": {"l": {"o": {"": 1, " ": {"友": {"達": {"": 1}}}}}}}}}       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										__lowercase       =       Trie()
										self.assertEqual(trie.split("[CLS] This is a extra_id_100"       )  ,    ["[CLS] This is a extra_id_100"]       )
										trie.add("[CLS]"       )
										trie.add("extra_id_1"       )
										trie.add("extra_id_100"       )
										self.assertEqual(trie.split("[CLS] This is a extra_id_100"       )  ,    ["[CLS]", " This is a ", "extra_id_100"]       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("A"       )
										self.assertEqual(trie.split("ABC"       )  ,    ["A", "BC"]       )
										self.assertEqual(trie.split("BCA"       )  ,    ["BC", "A"]       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							str:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("TOKEN]"       )
										trie.add("[SPECIAL_TOKEN]"       )
										self.assertEqual(trie.split("This is something [SPECIAL_TOKEN]"       )  ,    ["This is something ", "[SPECIAL_TOKEN]"]       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Union[str, Any]:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("A"       )
										trie.add("P"       )
										trie.add("[SPECIAL_TOKEN]"       )
										self.assertEqual(trie.split("This is something [SPECIAL_TOKEN]"       )  ,    ["This is something ", "[SPECIAL_TOKEN]"]       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("AB"       )
										trie.add("B"       )
										trie.add("C"       )
										self.assertEqual(trie.split("ABC"       )  ,    ["AB", "C"]       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("ABC"       )
										trie.add("B"       )
										trie.add("CD"       )
										self.assertEqual(trie.split("ABCD"       )  ,    ["ABC", "D"]       )





							def   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										# Even if the offsets are wrong, we necessarily output correct string
										# parts.
										__lowercase       =       Trie()
										__lowercase       =       trie.cut_text("ABC"  ,    [0, 0, 2, 1, 2, 3]       )
										self.assertEqual(_lowerCamelCase  ,    ["AB", "C"]       )





