






# Copyright (c) 2021-, NVIDIA CORPORATION.  All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

####################################################################################################

#
# Note: If when running this conversion script you're getting an exception:
#     ModuleNotFoundError: No module named 'megatron.model.enums'
# you need to tell python where to find the clone of Megatron-LM, e.g.:
#
# cd /tmp
# git clone https://github.com/NVIDIA/Megatron-LM
# PYTHONPATH=/tmp/Megatron-LM python src/transformers/models/megatron_gpt2/convert_megatron_gpt2_checkpoint.py ...
#
# if you already have it cloned elsewhere, simply adjust the path to the existing path
#
# If the training was done using a Megatron-LM fork, e.g.,
# https://github.com/microsoft/Megatron-DeepSpeed/ then chances are that you need to have that one
# in your path, i.e., /path/to/Megatron-DeepSpeed/
#

import argparse
import os
import re
import zipfile

import torch

from transformers import AutoTokenizer, GPTaConfig







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							Any					,  snake_case_ :							Any					,  snake_case_ :							Optional[int]=0       ):
	# Format the message.
	if name is None:
		snake_case__			:				Any       =  None
	else:
		snake_case__			:				Tuple       =  "." * max(0					,  spaces - 2       ) + "# {:" + str(50 - spaces       ) + "s}"
		snake_case__			:				Dict       =  fmt.format(snake_case_       )

	# Print and recurse (if needed).
	if isinstance(snake_case_					,  snake_case_       ):
		if msg is not None:
			print(snake_case_       )
		for k in val.keys():
			recursive_print(snake_case_					,  val[k]					,  spaces + 2       )
	elif isinstance(snake_case_					,  torch.Tensor       ):
		print(snake_case_					,  ":"					,  val.size()       )
	else:
		print(snake_case_					,  ":"					,  snake_case_       )







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							int					,  snake_case_ :							Optional[int]					,  snake_case_ :							List[Any]					,  snake_case_ :							Optional[int]					,  snake_case_ :							List[Any]       ):
	# Permutes layout of param tensor to [num_splits * num_heads * hidden_size, :]
	# for compatibility with later versions of NVIDIA Megatron-LM.
	# The inverse operation is performed inside Megatron-LM to read checkpoints:
	# https://github.com/NVIDIA/Megatron-LM/blob/v2.4/megatron/checkpointing.py#L209
	# If param is the weight tensor of the self-attention block, the returned tensor
	# will have to be transposed one more time to be read by HuggingFace GPT2.
	snake_case__			:				Optional[Any]       =  param.size()
	if checkpoint_version == 1.0:
		# version 1.0 stores [num_heads * hidden_size * num_splits, :]
		snake_case__			:				Optional[int]       =  (num_heads, hidden_size, num_splits) + input_shape[1:]
		snake_case__			:				List[Any]       =  param.view(*snake_case_       )
		snake_case__			:				List[Any]       =  param.transpose(0					,  2       )
		snake_case__			:				List[str]       =  param.transpose(1					,  2       ).contiguous()
	elif checkpoint_version >= 2.0:
		# other versions store [num_heads * num_splits * hidden_size, :]
		snake_case__			:				Union[str, Any]       =  (num_heads, num_splits, hidden_size) + input_shape[1:]
		snake_case__			:				Any       =  param.view(*snake_case_       )
		snake_case__			:				Tuple       =  param.transpose(0					,  1       ).contiguous()
	snake_case__			:				Any       =  param.view(*snake_case_       )
	return param







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							int					,  snake_case_ :							int					,  snake_case_ :							Union[str, Any]       ):
	# The converted output model.
	snake_case__			:				Optional[int]       =  {}

	# old versions did not store training args
	snake_case__			:				Any       =  input_state_dict.get("args"					,  snake_case_       )
	if ds_args is not None:
		# do not make the user write a config file when the exact dimensions/sizes are already in the checkpoint
		# from pprint import pprint
		# pprint(vars(ds_args))

		snake_case__			:				Union[str, Any]       =  ds_args.padded_vocab_size
		snake_case__			:				Union[str, Any]       =  ds_args.max_position_embeddings
		snake_case__			:				int       =  ds_args.hidden_size
		snake_case__			:				Optional[int]       =  ds_args.num_layers
		snake_case__			:				Optional[Any]       =  ds_args.num_attention_heads
		snake_case__			:				Dict       =  ds_args.ffn_hidden_size
		# pprint(config)

	# The number of heads.
	snake_case__			:				Union[str, Any]       =  config.n_head
	# The hidden_size per head.
	snake_case__			:				str       =  config.n_embd // config.n_head
	# Megatron-LM checkpoint version
	if "checkpoint_version" in input_state_dict.keys():
		snake_case__			:				Union[str, Any]       =  input_state_dict["checkpoint_version"]
	else:
		snake_case__			:				int       =  0.0

	# The model.
	snake_case__			:				Dict       =  input_state_dict["model"]
	# The language model.
	snake_case__			:				Optional[int]       =  model["language_model"]
	# The embeddings.
	snake_case__			:				Union[str, Any]       =  lm["embedding"]

	# The word embeddings.
	snake_case__			:				Optional[Any]       =  embeddings["word_embeddings"]["weight"]
	# Truncate the embedding table to vocab_size rows.
	snake_case__			:				Tuple       =  word_embeddings[: config.vocab_size, :]
	snake_case__			:				Any       =  word_embeddings

	# The position embeddings.
	snake_case__			:				Union[str, Any]       =  embeddings["position_embeddings"]["weight"]
	# Read the causal mask dimension (seqlen). [max_sequence_length, hidden_size]
	snake_case__			:				str       =  pos_embeddings.size(0       )
	if n_positions != config.n_positions:
		raise ValueError(
		    F'''pos_embeddings.max_sequence_length={n_positions} and config.n_positions={config.n_positions} don\'t match'''       )
	# Store the position embeddings.
	snake_case__			:				str       =  pos_embeddings

	# The transformer.
	snake_case__			:				Any       =  lm["transformer"] if "transformer" in lm.keys() else lm["encoder"]

	# The regex to extract layer names.
	snake_case__			:				Optional[Any]       =  re.compile(R"layers\.(\d+)\.([a-z0-9_.]+)\.([a-z]+)"       )

	# The simple map of names for "automated" rules.
	snake_case__			:				List[Any]       =  {
	    "attention.dense": ".attn.c_proj.",
	    "self_attention.dense": ".attn.c_proj.",
	    "mlp.dense_h_to_4h": ".mlp.c_fc.",
	    "mlp.dense_4h_to_h": ".mlp.c_proj.",
	}

	# Extract the layers.
	for key, val in transformer.items():
		# Match the name.
		snake_case__			:				Union[str, Any]       =  layer_re.match(snake_case_       )

		# Stop if that's not a layer
		if m is None:
			break

		# The index of the layer.
		snake_case__			:				Dict       =  int(m.group(1       )       )
		# The name of the operation.
		snake_case__			:				int       =  m.group(2       )
		# Is it a weight or a bias?
		snake_case__			:				str       =  m.group(3       )

		# The name of the layer.
		snake_case__			:				Dict       =  F'''transformer.h.{layer_idx}'''

		# For layernorm(s), simply store the layer norm.
		if op_name.endswith("layernorm"       ):
			snake_case__			:				Dict       =  "ln_1" if op_name.startswith("input"       ) else "ln_2"
			snake_case__			:				Tuple       =  val

		# Transpose the QKV matrix.
		elif (
		    op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
		) and weight_or_bias == "weight":
			# Insert a tensor of 1x1xDxD bias.
			snake_case__			:				Optional[Any]       =  torch.tril(torch.ones((n_positions, n_positions)					,  dtype=torch.floataa       )       ).view(
			    1					,  1					,  snake_case_					,  snake_case_       )
			snake_case__			:				Optional[Any]       =  causal_mask

			# Insert a "dummy" tensor for masked_bias.
			snake_case__			:				Any       =  torch.tensor(-1E4					,  dtype=torch.floataa       )
			snake_case__			:				List[Any]       =  masked_bias

			snake_case__			:				Optional[int]       =  fix_query_key_value_ordering(snake_case_					,  snake_case_					,  3					,  snake_case_					,  snake_case_       )
			# Megatron stores (3*D) x D but transformers-GPT2 expects D x 3*D.
			snake_case__			:				int       =  out_val.transpose(0					,  1       ).contiguous()
			# Store.
			snake_case__			:				int       =  out_val

		# Transpose the bias.
		elif (
		    op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
		) and weight_or_bias == "bias":
			snake_case__			:				Any       =  fix_query_key_value_ordering(snake_case_					,  snake_case_					,  3					,  snake_case_					,  snake_case_       )
			# Store. No change of shape.
			snake_case__			:				Tuple       =  out_val

		# Transpose the weights.
		elif weight_or_bias == "weight":
			snake_case__			:				int       =  megatron_to_transformers[op_name]
			snake_case__			:				Dict       =  val.transpose(0					,  1       )

		# Copy the bias.
		elif weight_or_bias == "bias":
			snake_case__			:				Tuple       =  megatron_to_transformers[op_name]
			snake_case__			:				List[Any]       =  val

    # DEBUG.
	assert config.n_layer == layer_idx + 1

	# The final layernorm.
	snake_case__			:				Tuple       =  transformer["final_layernorm.weight"]
	snake_case__			:				Tuple       =  transformer["final_layernorm.bias"]

	# For LM head, transformers' wants the matrix to weight embeddings.
	snake_case__			:				int       =  word_embeddings

	# It should be done!
	return output_state_dict







def 					SCREAMING_SNAKE_CASE    (			):
	# Create the argument parser.
	snake_case__			:				Optional[int]       =  argparse.ArgumentParser()
	parser.add_argument("--print-checkpoint-structure"					,  action="store_true"       )
	parser.add_argument(
	    "path_to_checkpoint"					,  type=snake_case_					,  help="Path to the checkpoint file (.zip archive or direct .pt file)"					,  )
	parser.add_argument(
	    "--config_file"					,  default=""					,  type=snake_case_					,  help="An optional config json file describing the pre-trained model."					,  )
	snake_case__			:				Union[str, Any]       =  parser.parse_args()

	# Extract the basename.
	snake_case__			:				str       =  os.path.dirname(args.path_to_checkpoint       )

	# Load the model.
	# the .zip is very optional, let's keep it for backward compatibility
	print(F'''Extracting PyTorch state dictionary from {args.path_to_checkpoint}'''       )
	if args.path_to_checkpoint.endswith(".zip"       ):
		with zipfile.ZipFile(args.path_to_checkpoint					,  "r"       ) as checkpoint:
			with checkpoint.open("release/mp_rank_00/model_optim_rng.pt"       ) as pytorch_dict:
				snake_case__			:				int       =  torch.load(snake_case_					,  map_location="cpu"       )
	else:
		snake_case__			:				str       =  torch.load(args.path_to_checkpoint					,  map_location="cpu"       )

	snake_case__			:				Optional[int]       =  input_state_dict.get("args"					,  snake_case_       )

	# Read the config, or default to the model released by NVIDIA.
	if args.config_file == "":
		if ds_args is not None:
			if ds_args.bias_gelu_fusion:
				snake_case__			:				Any       =  "gelu_fast"
			elif ds_args.openai_gelu:
				snake_case__			:				Any       =  "gelu_new"
			else:
				snake_case__			:				Optional[Any]       =  "gelu"
		else:
			# in the very early days this used to be "gelu_new"
			snake_case__			:				Tuple       =  "gelu_new"

		# Spell out all parameters in case the defaults change.
		snake_case__			:				Optional[int]       =  GPTaConfig(
		    vocab_size=50257					,  n_positions=1024					,  n_embd=1024					,  n_layer=24					,  n_head=16					,  n_inner=4096					,  activation_function=snake_case_					,  resid_pdrop=0.1					,  embd_pdrop=0.1					,  attn_pdrop=0.1					,  layer_norm_epsilon=1E-5					,  initializer_range=0.02					,  summary_type="cls_index"					,  summary_use_proj=snake_case_					,  summary_activation=snake_case_					,  summary_proj_to_labels=snake_case_					,  summary_first_dropout=0.1					,  scale_attn_weights=snake_case_					,  use_cache=snake_case_					,  bos_token_id=50256					,  eos_token_id=50256					,  )
	else:
		snake_case__			:				Any       =  GPTaConfig.from_json_file(args.config_file       )

	snake_case__			:				List[str]       =  ["GPT2LMHeadModel"]

	# Convert.
	print("Converting"       )
	snake_case__			:				int       =  convert_megatron_checkpoint(snake_case_					,  snake_case_					,  snake_case_       )

	# Print the structure of converted state dict.
	if args.print_checkpoint_structure:
		recursive_print(snake_case_					,  snake_case_       )

	# Add tokenizer class info to config
	# see https://github.com/huggingface/transformers/issues/13906)
	if ds_args is not None:
		snake_case__			:				List[str]       =  ds_args.tokenizer_type
		if tokenizer_type == "GPT2BPETokenizer":
			snake_case__			:				Tuple       =  "gpt2"
		elif tokenizer_type == "PretrainedFromHF":
			snake_case__			:				Optional[int]       =  ds_args.tokenizer_name_or_path
		else:
			raise ValueError(F'''Unrecognized tokenizer_type {tokenizer_type}'''       )
	else:
		snake_case__			:				int       =  "gpt2"

	snake_case__			:				Union[str, Any]       =  AutoTokenizer.from_pretrained(snake_case_       )
	snake_case__			:				int       =  type(snake_case_       ).__name__
	snake_case__			:				Tuple       =  tokenizer_class

	# Store the config to file.
	print("Saving config"       )
	config.save_pretrained(snake_case_       )

	# Save tokenizer based on args
	print(F'''Adding {tokenizer_class} tokenizer files'''       )
	tokenizer.save_pretrained(snake_case_       )

	# Store the state_dict to file.
	snake_case__			:				Optional[Any]       =  os.path.join(snake_case_					,  "pytorch_model.bin"       )
	print(F'''Saving checkpoint to "{output_checkpoint_file}"'''       )
	torch.save(snake_case_					,  snake_case_       )


####################################################################################################

if __name__ == "__main__":
	main()

####################################################################################################











from __future__ import annotations

import os
import tempfile
import unittest

from transformers import ConvBertConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
	import tensorflow as tf

	from transformers import (
	    TFConvBertForMaskedLM,
	    TFConvBertForMultipleChoice,
	    TFConvBertForQuestionAnswering,
	    TFConvBertForSequenceClassification,
	    TFConvBertForTokenClassification,
	    TFConvBertModel,
	)



class 							SCREAMING_SNAKE_CASE__						:





					"""simple docstring"""






					def __init__( self					:  Union[str, Any]							,		__A					:  str							,		__A					:  Optional[Any]=1_3							,		__A					:  Dict=7							,		__A					:  List[str]=True							,		__A					:  Any=True							,		__A					:  str=True							,		__A					:  Optional[Any]=True							,		__A					:  List[str]=9_9							,		__A					:  Dict=3_2							,		__A					:  Tuple=2							,		__A					:  Tuple=4							,		__A					:  Dict=3_7							,		__A					:  Tuple="gelu"							,		__A					:  Any=0.1							,		__A					:  str=0.1							,		__A					:  int=5_1_2							,		__A					:  Union[str, Any]=1_6							,		__A					:  Optional[int]=2							,		__A					:  Union[str, Any]=0.0_2							,		__A					:  Tuple=3							,		__A					:  Union[str, Any]=4							,		__A					:  Optional[int]=None							,		):
						snake_case__			:				Optional[int]       =  parent
						snake_case__			:				Optional[Any]       =  1_3
						snake_case__			:				int       =  7
						snake_case__			:				Optional[int]       =  True
						snake_case__			:				Optional[Any]       =  True
						snake_case__			:				List[str]       =  True
						snake_case__			:				int       =  True
						snake_case__			:				Optional[int]       =  9_9
						snake_case__			:				Union[str, Any]       =  3_8_4
						snake_case__			:				Optional[Any]       =  2
						snake_case__			:				Union[str, Any]       =  4
						snake_case__			:				Any       =  3_7
						snake_case__			:				Any       =  "gelu"
						snake_case__			:				str       =  0.1
						snake_case__			:				Optional[Any]       =  0.1
						snake_case__			:				Union[str, Any]       =  5_1_2
						snake_case__			:				Optional[Any]       =  1_6
						snake_case__			:				List[Any]       =  2
						snake_case__			:				Optional[int]       =  0.0_2
						snake_case__			:				Dict       =  3
						snake_case__			:				Any       =  4
						snake_case__			:				int       =  1_2_8
						snake_case__			:				Dict       =  2
						snake_case__			:				Any       =  9
						snake_case__			:				List[str]       =  1
						snake_case__			:				List[Any]       =  None



					def 	_lowercase					( self					:  List[str]	):
						snake_case__			:				Optional[int]       =  ids_tensor([self.batch_size, self.seq_length]							,		self.vocab_size	)

						snake_case__			:				str       =  None
						if self.use_input_mask:
							snake_case__			:				str       =  random_attention_mask([self.batch_size, self.seq_length]	)

						snake_case__			:				Union[str, Any]       =  None
						if self.use_token_type_ids:
							snake_case__			:				Tuple       =  ids_tensor([self.batch_size, self.seq_length]							,		self.type_vocab_size	)

						snake_case__			:				Optional[Any]       =  None
						snake_case__			:				Any       =  None
						snake_case__			:				Tuple       =  None
						if self.use_labels:
							snake_case__			:				List[str]       =  ids_tensor([self.batch_size]							,		self.type_sequence_label_size	)
							snake_case__			:				str       =  ids_tensor([self.batch_size, self.seq_length]							,		self.num_labels	)
							snake_case__			:				int       =  ids_tensor([self.batch_size]							,		self.num_choices	)

						snake_case__			:				int       =  ConvBertConfig(
						    vocab_size=self.vocab_size							,		hidden_size=self.hidden_size							,		num_hidden_layers=self.num_hidden_layers							,		num_attention_heads=self.num_attention_heads							,		intermediate_size=self.intermediate_size							,		hidden_act=self.hidden_act							,		hidden_dropout_prob=self.hidden_dropout_prob							,		attention_probs_dropout_prob=self.attention_probs_dropout_prob							,		max_position_embeddings=self.max_position_embeddings							,		type_vocab_size=self.type_vocab_size							,		initializer_range=self.initializer_range							,		return_dict=__A							,		)

						return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels



					def 	_lowercase					( self					:  Dict							,		__A					:  Dict							,		__A					:  Dict							,		__A					:  Union[str, Any]							,		__A					:  Optional[int]							,		__A					:  Any							,		__A					:  Union[str, Any]							,		__A					:  Tuple	):
						snake_case__			:				Optional[int]       =  TFConvBertModel(config=__A	)
						snake_case__			:				Optional[int]       =  {"input_ids": input_ids, "attention_mask": input_mask, "token_type_ids": token_type_ids}

						snake_case__			:				List[str]       =  [input_ids, input_mask]
						snake_case__			:				Union[str, Any]       =  model(__A	)

						snake_case__			:				str       =  model(__A	)

						self.parent.assertEqual(result.last_hidden_state.shape							,		(self.batch_size, self.seq_length, self.hidden_size)	)



					def 	_lowercase					( self					:  Union[str, Any]							,		__A					:  List[Any]							,		__A					:  Any							,		__A					:  Union[str, Any]							,		__A					:  int							,		__A					:  Optional[Any]							,		__A					:  Dict							,		__A					:  Optional[int]	):
						snake_case__			:				List[str]       =  TFConvBertForMaskedLM(config=__A	)
						snake_case__			:				Any       =  {
						    "input_ids": input_ids,
						    "attention_mask": input_mask,
						    "token_type_ids": token_type_ids,
						}
						snake_case__			:				int       =  model(__A	)
						self.parent.assertEqual(result.logits.shape							,		(self.batch_size, self.seq_length, self.vocab_size)	)



					def 	_lowercase					( self					:  Tuple							,		__A					:  Union[str, Any]							,		__A					:  List[Any]							,		__A					:  Any							,		__A					:  List[Any]							,		__A					:  List[Any]							,		__A					:  Optional[int]							,		__A					:  List[str]	):
						snake_case__			:				Any       =  self.num_labels
						snake_case__			:				List[Any]       =  TFConvBertForSequenceClassification(config=__A	)
						snake_case__			:				Any       =  {
						    "input_ids": input_ids,
						    "attention_mask": input_mask,
						    "token_type_ids": token_type_ids,
						}

						snake_case__			:				Optional[int]       =  model(__A	)
						self.parent.assertEqual(result.logits.shape							,		(self.batch_size, self.num_labels)	)



					def 	_lowercase					( self					:  int							,		__A					:  List[Any]							,		__A					:  Union[str, Any]							,		__A					:  Optional[Any]							,		__A					:  List[Any]							,		__A					:  Union[str, Any]							,		__A					:  Union[str, Any]							,		__A					:  Optional[int]	):
						snake_case__			:				Optional[Any]       =  self.num_choices
						snake_case__			:				Any       =  TFConvBertForMultipleChoice(config=__A	)
						snake_case__			:				Optional[int]       =  tf.tile(tf.expand_dims(__A							,		1	)							,		(1, self.num_choices, 1)	)
						snake_case__			:				Optional[Any]       =  tf.tile(tf.expand_dims(__A							,		1	)							,		(1, self.num_choices, 1)	)
						snake_case__			:				Optional[int]       =  tf.tile(tf.expand_dims(__A							,		1	)							,		(1, self.num_choices, 1)	)
						snake_case__			:				int       =  {
						    "input_ids": multiple_choice_inputs_ids,
						    "attention_mask": multiple_choice_input_mask,
						    "token_type_ids": multiple_choice_token_type_ids,
						}
						snake_case__			:				Optional[Any]       =  model(__A	)
						self.parent.assertEqual(result.logits.shape							,		(self.batch_size, self.num_choices)	)



					def 	_lowercase					( self					:  List[str]							,		__A					:  Tuple							,		__A					:  str							,		__A					:  Union[str, Any]							,		__A					:  Union[str, Any]							,		__A					:  Any							,		__A					:  int							,		__A					:  Tuple	):
						snake_case__			:				Dict       =  self.num_labels
						snake_case__			:				str       =  TFConvBertForTokenClassification(config=__A	)
						snake_case__			:				Optional[Any]       =  {
						    "input_ids": input_ids,
						    "attention_mask": input_mask,
						    "token_type_ids": token_type_ids,
						}
						snake_case__			:				List[str]       =  model(__A	)
						self.parent.assertEqual(result.logits.shape							,		(self.batch_size, self.seq_length, self.num_labels)	)



					def 	_lowercase					( self					:  Optional[int]							,		__A					:  Union[str, Any]							,		__A					:  List[Any]							,		__A					:  List[str]							,		__A					:  Any							,		__A					:  Any							,		__A					:  Optional[int]							,		__A					:  Optional[Any]	):
						snake_case__			:				Any       =  TFConvBertForQuestionAnswering(config=__A	)
						snake_case__			:				List[str]       =  {
						    "input_ids": input_ids,
						    "attention_mask": input_mask,
						    "token_type_ids": token_type_ids,
						}

						snake_case__			:				int       =  model(__A	)
						self.parent.assertEqual(result.start_logits.shape							,		(self.batch_size, self.seq_length)	)
						self.parent.assertEqual(result.end_logits.shape							,		(self.batch_size, self.seq_length)	)






					def 	_lowercase					( self					:  Any	):
						snake_case__			:				List[Any]       =  self.prepare_config_and_inputs()
						(
						    (
						    snake_case__
						),						(
						    snake_case__
						),						(
						    snake_case__
						),						(
						    snake_case__
						),						(
						    snake_case__
						),						(
						    snake_case__
						),						(
						    snake_case__
						),						
						)			:				List[str]       =  config_and_inputs
						snake_case__			:				Tuple       =  {"input_ids": input_ids, "token_type_ids": token_type_ids, "attention_mask": input_mask}
						return config, inputs_dict



@require_tf
class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_			,       UpperCamelCase_			,       unittest.TestCase						):





					"""simple docstring"""






					a_     =		(
					    (
					        TFConvBertModel,
					        TFConvBertForMaskedLM,
					        TFConvBertForQuestionAnswering,
					        TFConvBertForSequenceClassification,
					        TFConvBertForTokenClassification,
					        TFConvBertForMultipleChoice,
					    )
					    if is_tf_available()
					    else ()
					)
					a_     =		(
					    {
					        "feature-extraction": TFConvBertModel,
					        "fill-mask": TFConvBertForMaskedLM,
					        "question-answering": TFConvBertForQuestionAnswering,
					        "text-classification": TFConvBertForSequenceClassification,
					        "token-classification": TFConvBertForTokenClassification,
					        "zero-shot": TFConvBertForSequenceClassification,
					    }
					    if is_tf_available()
					    else {}
					)
					a_     =		False
					a_     =		False
					a_     =		False



					def 	_lowercase					( self					:  int	):
						snake_case__			:				Optional[Any]       =  TFConvBertModelTester(self	)
						snake_case__			:				List[str]       =  ConfigTester(self							,		config_class=__A							,		hidden_size=3_7	)



					def 	_lowercase					( self					:  List[Any]	):
						self.config_tester.run_common_tests()



					def 	_lowercase					( self					:  Any	):
						snake_case__			:				List[str]       =  self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_model(*__A	)



					def 	_lowercase					( self					:  Union[str, Any]	):
						snake_case__			:				Any       =  self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_masked_lm(*__A	)



					def 	_lowercase					( self					:  Dict	):
						snake_case__			:				str       =  self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_multiple_choice(*__A	)



					def 	_lowercase					( self					:  Optional[Any]	):
						snake_case__			:				List[str]       =  self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_question_answering(*__A	)



					def 	_lowercase					( self					:  Optional[int]	):
						snake_case__			:				int       =  self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_sequence_classification(*__A	)



					def 	_lowercase					( self					:  Dict	):
						snake_case__			:				Any       =  self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_token_classification(*__A	)



					@slow
					def 	_lowercase					( self					:  Dict	):
						snake_case__,						snake_case__			:				Tuple       =  self.model_tester.prepare_config_and_inputs_for_common()
						snake_case__			:				int       =  True
						snake_case__			:				int       =  True

						if hasattr(__A							,		"use_cache"	):
							snake_case__			:				Optional[Any]       =  True

						snake_case__			:				Dict       =  getattr(self.model_tester							,		"encoder_seq_length"							,		self.model_tester.seq_length	)
						snake_case__			:				List[str]       =  getattr(self.model_tester							,		"key_length"							,		__A	)

						for model_class in self.all_model_classes:
							snake_case__			:				Tuple       =  self._prepare_for_class(__A							,		__A	)
							snake_case__			:				List[str]       =  model_class(__A	)
							snake_case__			:				List[Any]       =  len(model(__A	)	)

							with tempfile.TemporaryDirectory() as tmpdirname:
								model.save_pretrained(__A							,		saved_model=__A	)
								snake_case__			:				str       =  os.path.join(__A							,		"saved_model"							,		"1"	)
								snake_case__			:				str       =  tf.keras.models.load_model(__A	)
								snake_case__			:				Optional[Any]       =  model(__A	)

								if self.is_encoder_decoder:
									snake_case__			:				Tuple       =  outputs["encoder_hidden_states"]
									snake_case__			:				str       =  outputs["encoder_attentions"]
								else:
									snake_case__			:				Dict       =  outputs["hidden_states"]
									snake_case__			:				Tuple       =  outputs["attentions"]

								self.assertEqual(len(__A	)							,		__A	)

								snake_case__			:				int       =  getattr(
								    self.model_tester							,		"expected_num_hidden_layers"							,		self.model_tester.num_hidden_layers + 1	)

								self.assertEqual(len(__A	)							,		__A	)
								self.assertListEqual(
								    list(output_hidden_states[0].shape[-2:]	)							,		[self.model_tester.seq_length, self.model_tester.hidden_size]							,		)

								self.assertEqual(len(__A	)							,		self.model_tester.num_hidden_layers	)
								self.assertListEqual(
								    list(output_attentions[0].shape[-3:]	)							,		[self.model_tester.num_attention_heads / 2, encoder_seq_length, encoder_key_length]							,		)



					@slow
					def 	_lowercase					( self					:  Tuple	):
						snake_case__			:				Optional[Any]       =  TFConvBertModel.from_pretrained("YituTech/conv-bert-base"	)
						self.assertIsNotNone(__A	)






					def 	_lowercase					( self					:  List[str]	):
						snake_case__,						snake_case__			:				Optional[Any]       =  self.model_tester.prepare_config_and_inputs_for_common()
						snake_case__			:				Optional[Any]       =  True
						snake_case__			:				List[Any]       =  getattr(self.model_tester							,		"decoder_seq_length"							,		self.model_tester.seq_length	)
						snake_case__			:				int       =  getattr(self.model_tester							,		"encoder_seq_length"							,		self.model_tester.seq_length	)
						snake_case__			:				Any       =  getattr(self.model_tester							,		"key_length"							,		__A	)
						snake_case__			:				List[Any]       =  getattr(self.model_tester							,		"key_length"							,		__A	)

						def check_decoder_attentions_output(__A					:  Optional[int]	):
							snake_case__			:				Optional[Any]       =  len(__A	)
							self.assertEqual(out_len % 2							,		0	)
							snake_case__			:				Optional[int]       =  outputs.decoder_attentions
							self.assertEqual(len(__A	)							,		self.model_tester.num_hidden_layers	)
							self.assertListEqual(
							    list(decoder_attentions[0].shape[-3:]	)							,		[self.model_tester.num_attention_heads / 2, decoder_seq_length, decoder_key_length]							,		)

						def check_encoder_attentions_output(__A					:  Any	):
							snake_case__			:				List[Any]       =  [
							    t.numpy() for t in (outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions)
							]
							self.assertEqual(len(__A	)							,		self.model_tester.num_hidden_layers	)
							self.assertListEqual(
							    list(attentions[0].shape[-3:]	)							,		[self.model_tester.num_attention_heads / 2, encoder_seq_length, encoder_key_length]							,		)

						for model_class in self.all_model_classes:
							snake_case__			:				Optional[int]       =  True
							snake_case__			:				Any       =  False
							snake_case__			:				Dict       =  model_class(__A	)
							snake_case__			:				List[Any]       =  model(self._prepare_for_class(__A							,		__A	)	)
							snake_case__			:				Dict       =  len(__A	)
							self.assertEqual(config.output_hidden_states							,		__A	)
							check_encoder_attentions_output(__A	)

							if self.is_encoder_decoder:
								snake_case__			:				str       =  model_class(__A	)
								snake_case__			:				List[Any]       =  model(self._prepare_for_class(__A							,		__A	)	)
								self.assertEqual(config.output_hidden_states							,		__A	)
								check_decoder_attentions_output(__A	)

							# Check that output attentions can also be changed via the config
							del inputs_dict["output_attentions"]
							snake_case__			:				Optional[int]       =  True
							snake_case__			:				Optional[Any]       =  model_class(__A	)
							snake_case__			:				Union[str, Any]       =  model(self._prepare_for_class(__A							,		__A	)	)
							self.assertEqual(config.output_hidden_states							,		__A	)
							check_encoder_attentions_output(__A	)

							# Check attention is always last and order is fine
							snake_case__			:				Optional[int]       =  True
							snake_case__			:				List[Any]       =  True
							snake_case__			:				Any       =  model_class(__A	)
							snake_case__			:				str       =  model(self._prepare_for_class(__A							,		__A	)	)

							self.assertEqual(out_len + (2 if self.is_encoder_decoder else 1)							,		len(__A	)	)
							self.assertEqual(model.config.output_hidden_states							,		__A	)
							check_encoder_attentions_output(__A	)



@require_tf
class 							SCREAMING_SNAKE_CASE__						(    unittest.TestCase						):





					"""simple docstring"""






					@slow
					def 	_lowercase					( self					:  int	):
						snake_case__			:				int       =  TFConvBertModel.from_pretrained("YituTech/conv-bert-base"	)
						snake_case__			:				int       =  tf.constant([[0, 1, 2, 3, 4, 5]]	)
						snake_case__			:				str       =  model(__A	)[0]

						snake_case__			:				int       =  [1, 6, 7_6_8]
						self.assertEqual(output.shape							,		__A	)

						snake_case__			:				List[Any]       =  tf.constant(
						    [
						        [
						            [-0.0_3_4_7_5_4_9_3, -0.4_6_8_6_0_3_4, -0.3_0_6_3_8_8_3_2],
						            [0.2_2_6_3_7_2_4_8, -0.2_6_9_8_8_6_4_6, -0.7_4_2_3_4_2_4],
						            [0.1_0_3_2_4_8_6_8, -0.4_5_0_1_3_5_0_8, -0.5_8_2_8_0_7_8_4],
						        ]
						    ]	)
						tf.debugging.assert_near(output[:, :3, :3]							,		__A							,		atol=1e-4	)



