






"""simple docstring"""
# 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 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__=0					):
	# Format the message.
	if name is None:
		A_		:						Optional[Any]	    =   None
	else:
		A_		:						Dict	    =   """.""" * max(0   ,   spaces - 2					) + """# {:""" + str(50 - spaces					) + """s}"""
		A_		:						Any	    =   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 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__					):
	# 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.
	A_		:						Optional[Any]	    =   param.size()
	if checkpoint_version == 1.0:
		# version 1.0 stores [num_heads * hidden_size * num_splits, :]
		A_		:						Union[str, Any]	    =   (num_heads, hidden_size, num_splits) + input_shape[1:]
		A_		:						List[Any]	    =   param.view(*snake_case__					)
		A_		:						str	    =   param.transpose(0   ,   2					)
		A_		:						Dict	    =   param.transpose(1   ,   2					).contiguous()
	elif checkpoint_version >= 2.0:
		# other versions store [num_heads * num_splits * hidden_size, :]
		A_		:						str	    =   (num_heads, num_splits, hidden_size) + input_shape[1:]
		A_		:						Dict	    =   param.view(*snake_case__					)
		A_		:						Any	    =   param.transpose(0   ,   1					).contiguous()
	A_		:						List[Any]	    =   param.view(*snake_case__					)
	return param






def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__					):
	# The converted output model.
	A_		:						List[Any]	    =   {}

	# old versions did not store training args
	A_		:						Optional[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))

		A_		:						Any	    =   ds_args.padded_vocab_size
		A_		:						Tuple	    =   ds_args.max_position_embeddings
		A_		:						Union[str, Any]	    =   ds_args.hidden_size
		A_		:						List[Any]	    =   ds_args.num_layers
		A_		:						Dict	    =   ds_args.num_attention_heads
		A_		:						List[Any]	    =   ds_args.ffn_hidden_size
		# pprint(config)

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

	# The model.
	A_		:						Tuple	    =   input_state_dict["""model"""]
	# The language model.
	A_		:						List[str]	    =   model["""language_model"""]
	# The embeddings.
	A_		:						Dict	    =   lm["""embedding"""]

	# The word embeddings.
	A_		:						Optional[int]	    =   embeddings["""word_embeddings"""]["""weight"""]
	# Truncate the embedding table to vocab_size rows.
	A_		:						int	    =   word_embeddings[: config.vocab_size, :]
	A_		:						str	    =   word_embeddings

	# The position embeddings.
	A_		:						List[Any]	    =   embeddings["""position_embeddings"""]["""weight"""]
	# Read the causal mask dimension (seqlen). [max_sequence_length, hidden_size]
	A_		:						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.
	A_		:						Optional[int]	    =   pos_embeddings

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

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

	# The simple map of names for "automated" rules.
	A_		:						Optional[int]	    =   {
	    """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.
		A_		:						Any	    =   layer_re.match(snake_case__					)

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

		# The index of the layer.
		A_		:						str	    =   int(m.group(1					)					)
		# The name of the operation.
		A_		:						Optional[Any]	    =   m.group(2					)
		# Is it a weight or a bias?
		A_		:						Optional[int]	    =   m.group(3					)

		# The name of the layer.
		A_		:						str	    =   F"""transformer.h.{layer_idx}"""

		# For layernorm(s), simply store the layer norm.
		if op_name.endswith("""layernorm"""					):
			A_		:						Tuple	    =   """ln_1""" if op_name.startswith("""input"""					) else """ln_2"""
			A_		:						int	    =   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.
			A_		:						List[str]	    =   torch.tril(torch.ones((n_positions, n_positions)   ,   dtype=torch.floataa					)					).view(
			    1   ,   1   ,   snake_case__   ,   snake_case__					)
			A_		:						Dict	    =   causal_mask

			# Insert a "dummy" tensor for masked_bias.
			A_		:						Dict	    =   torch.tensor(-1E4   ,   dtype=torch.floataa					)
			A_		:						str	    =   masked_bias

			A_		:						Dict	    =   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.
			A_		:						Dict	    =   out_val.transpose(0   ,   1					).contiguous()
			# Store.
			A_		:						Optional[Any]	    =   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":
			A_		:						Union[str, Any]	    =   fix_query_key_value_ordering(snake_case__   ,   snake_case__   ,   3   ,   snake_case__   ,   snake_case__					)
			# Store. No change of shape.
			A_		:						List[str]	    =   out_val

		# Transpose the weights.
		elif weight_or_bias == "weight":
			A_		:						Any	    =   megatron_to_transformers[op_name]
			A_		:						str	    =   val.transpose(0   ,   1					)

		# Copy the bias.
		elif weight_or_bias == "bias":
			A_		:						int	    =   megatron_to_transformers[op_name]
			A_		:						Dict	    =   val

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

	# The final layernorm.
	A_		:						List[Any]	    =   transformer["""final_layernorm.weight"""]
	A_		:						Optional[Any]	    =   transformer["""final_layernorm.bias"""]

	# For LM head, transformers' wants the matrix to weight embeddings.
	A_		:						Union[str, Any]	    =   word_embeddings

	# It should be done!
	return output_state_dict






def 			__UpperCamelCase	(		):
	# Create the argument parser.
	A_		:						str	    =   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."""   ,   )
	A_		:						List[str]	    =   parser.parse_args()

	# Extract the basename.
	A_		:						int	    =   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:
				A_		:						Dict	    =   torch.load(snake_case__   ,   map_location="""cpu"""					)
	else:
		A_		:						Any	    =   torch.load(args.path_to_checkpoint   ,   map_location="""cpu"""					)

	A_		:						Dict	    =   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:
				A_		:						Union[str, Any]	    =   """gelu_fast"""
			elif ds_args.openai_gelu:
				A_		:						Tuple	    =   """gelu_new"""
			else:
				A_		:						int	    =   """gelu"""
		else:
			# in the very early days this used to be "gelu_new"
			A_		:						List[Any]	    =   """gelu_new"""

		# Spell out all parameters in case the defaults change.
		A_		:						Optional[Any]	    =   GPTaConfig(
		    vocab_size=50_257   ,   n_positions=1_024   ,   n_embd=1_024   ,   n_layer=24   ,   n_head=16   ,   n_inner=4_096   ,   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=50_256   ,   eos_token_id=50_256   ,   )
	else:
		A_		:						Tuple	    =   GPTaConfig.from_json_file(args.config_file					)

	A_		:						str	    =   ["""GPT2LMHeadModel"""]

	# Convert.
	print("""Converting"""					)
	A_		:						Optional[Any]	    =   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:
		A_		:						str	    =   ds_args.tokenizer_type
		if tokenizer_type == "GPT2BPETokenizer":
			A_		:						Optional[int]	    =   """gpt2"""
		elif tokenizer_type == "PretrainedFromHF":
			A_		:						Tuple	    =   ds_args.tokenizer_name_or_path
		else:
			raise ValueError(F"""Unrecognized tokenizer_type {tokenizer_type}"""					)
	else:
		A_		:						List[Any]	    =   """gpt2"""

	A_		:						Dict	    =   AutoTokenizer.from_pretrained(snake_case__					)
	A_		:						Optional[int]	    =   type(snake_case__					).__name__
	A_		:						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.
	A_		:						int	    =   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()

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









"""simple docstring"""
import unittest

from transformers import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING, is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
							from PIL import Image
else:



							class      SCREAMING_SNAKE_CASE  :





								"""simple docstring"""




								@staticmethod
								def    lowerCamelCase(*lowerCAmelCase_     ,     **lowerCAmelCase_							):
									pass



@is_pipeline_test
@require_torch
@require_vision
class      SCREAMING_SNAKE_CASE  (				unittest.TestCase						):





	"""simple docstring"""




	_A       :				List[Any]    =				MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING





	def    lowerCamelCase(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_							):
		A_		:						Union[str, Any]	    =   pipeline("""visual-question-answering"""     ,     model="""hf-internal-testing/tiny-vilt-random-vqa"""							)
		A_		:						Union[str, Any]	    =   [
		    {
		        """image""": Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""							),
		        """question""": """How many cats are there?""",
		    },
		    {
		        """image""": """./tests/fixtures/tests_samples/COCO/000000039769.png""",
		        """question""": """How many cats are there?""",
		    },
		]
		return vqa_pipeline, examples





	def    lowerCamelCase(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_							):
		A_		:						str	    =   vqa_pipeline(lowerCAmelCase_     ,     top_k=1							)
		self.assertEqual(
		    lowerCAmelCase_     ,     [
		        [{"""score""": ANY(lowerCAmelCase_							), """answer""": ANY(lowerCAmelCase_							)}],
		        [{"""score""": ANY(lowerCAmelCase_							), """answer""": ANY(lowerCAmelCase_							)}],
		    ]     ,     )





	@require_torch
	def    lowerCamelCase(self							):
		A_		:						Tuple	    =   pipeline("""visual-question-answering"""     ,     model="""hf-internal-testing/tiny-vilt-random-vqa"""							)
		A_		:						Any	    =   """./tests/fixtures/tests_samples/COCO/000000039769.png"""
		A_		:						Any	    =   """How many cats are there?"""

		A_		:						Optional[int]	    =   vqa_pipeline(image=lowerCAmelCase_     ,     question="""How many cats are there?"""     ,     top_k=2							)
		self.assertEqual(
		    lowerCAmelCase_     ,     [{"""score""": ANY(lowerCAmelCase_							), """answer""": ANY(lowerCAmelCase_							)}, {"""score""": ANY(lowerCAmelCase_							), """answer""": ANY(lowerCAmelCase_							)}]							)

		A_		:						Any	    =   vqa_pipeline({"""image""": image, """question""": question}     ,     top_k=2							)
		self.assertEqual(
		    lowerCAmelCase_     ,     [{"""score""": ANY(lowerCAmelCase_							), """answer""": ANY(lowerCAmelCase_							)}, {"""score""": ANY(lowerCAmelCase_							), """answer""": ANY(lowerCAmelCase_							)}]							)





	@slow
	@require_torch
	def    lowerCamelCase(self							):
		A_		:						int	    =   pipeline("""visual-question-answering"""     ,     model="""dandelin/vilt-b32-finetuned-vqa"""							)
		A_		:						Tuple	    =   """./tests/fixtures/tests_samples/COCO/000000039769.png"""
		A_		:						int	    =   """How many cats are there?"""

		A_		:						List[Any]	    =   vqa_pipeline(image=lowerCAmelCase_     ,     question=lowerCAmelCase_     ,     top_k=2							)
		self.assertEqual(
		    nested_simplify(lowerCAmelCase_     ,     decimals=4							)     ,     [{"""score""": 0.8799, """answer""": """2"""}, {"""score""": 0.296, """answer""": """1"""}]							)

		A_		:						Any	    =   vqa_pipeline({"""image""": image, """question""": question}     ,     top_k=2							)
		self.assertEqual(
		    nested_simplify(lowerCAmelCase_     ,     decimals=4							)     ,     [{"""score""": 0.8799, """answer""": """2"""}, {"""score""": 0.296, """answer""": """1"""}]							)

		A_		:						Tuple	    =   vqa_pipeline(
		    [{"""image""": image, """question""": question}, {"""image""": image, """question""": question}]     ,     top_k=2							)
		self.assertEqual(
		    nested_simplify(lowerCAmelCase_     ,     decimals=4							)     ,     [[{"""score""": 0.8799, """answer""": """2"""}, {"""score""": 0.296, """answer""": """1"""}]] * 2     ,     )





	@require_tf
	@unittest.skip("""Visual question answering not implemented in TF"""							)
	def    lowerCamelCase(self							):
		pass

