# 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    __lowercase     (						lowerCamelCase					:     Tuple  ,  lowerCamelCase					:     Tuple  ,  lowerCamelCase					:     List[Any]=0	):
 # Format the message.
 if name is None:
  UpperCamelCase_					:						Any =					None
 else:
  UpperCamelCase_					:						Optional[int] =					'.' * max(0  ,  spaces - 2	) + '# {:' + str(50 - spaces	) + 's}'
  UpperCamelCase_					:						Tuple =					fmt.format(lowerCamelCase	)

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





def    __lowercase     (						lowerCamelCase					:     Tuple  ,  lowerCamelCase					:     Union[str, Any]  ,  lowerCamelCase					:     Optional[Any]  ,  lowerCamelCase					:     str  ,  lowerCamelCase					:     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.
 UpperCamelCase_					:						Union[str, Any] =					param.size()
 if checkpoint_version == 1.0:
  # version 1.0 stores [num_heads * hidden_size * num_splits, :]
  UpperCamelCase_					:						Union[str, Any] =					(num_heads, hidden_size, num_splits) + input_shape[1:]
  UpperCamelCase_					:						Dict =					param.view(*lowerCamelCase	)
  UpperCamelCase_					:						str =					param.transpose(0  ,  2	)
  UpperCamelCase_					:						Any =					param.transpose(1  ,  2	).contiguous()
 elif checkpoint_version >= 2.0:
  # other versions store [num_heads * num_splits * hidden_size, :]
  UpperCamelCase_					:						Dict =					(num_heads, num_splits, hidden_size) + input_shape[1:]
  UpperCamelCase_					:						Optional[int] =					param.view(*lowerCamelCase	)
  UpperCamelCase_					:						List[str] =					param.transpose(0  ,  1	).contiguous()
 UpperCamelCase_					:						List[Any] =					param.view(*lowerCamelCase	)
 return param





def    __lowercase     (						lowerCamelCase					:     Dict  ,  lowerCamelCase					:     Union[str, Any]  ,  lowerCamelCase					:     Tuple	):
 # The converted output model.
 UpperCamelCase_					:						Optional[Any] =					{}

 # old versions did not store training args
 UpperCamelCase_					:						Union[str, Any] =					input_state_dict.get('args'  ,  lowerCamelCase	)
 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))

  UpperCamelCase_					:						Optional[Any] =					ds_args.padded_vocab_size
  UpperCamelCase_					:						List[Any] =					ds_args.max_position_embeddings
  UpperCamelCase_					:						Dict =					ds_args.hidden_size
  UpperCamelCase_					:						List[Any] =					ds_args.num_layers
  UpperCamelCase_					:						str =					ds_args.num_attention_heads
  UpperCamelCase_					:						str =					ds_args.ffn_hidden_size
  # pprint(config)

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

 # The model.
 UpperCamelCase_					:						Tuple =					input_state_dict['model']
 # The language model.
 UpperCamelCase_					:						List[Any] =					model['language_model']
 # The embeddings.
 UpperCamelCase_					:						Union[str, Any] =					lm['embedding']

 # The word embeddings.
 UpperCamelCase_					:						Optional[int] =					embeddings['word_embeddings']['weight']
 # Truncate the embedding table to vocab_size rows.
 UpperCamelCase_					:						Union[str, Any] =					word_embeddings[: config.vocab_size, :]
 UpperCamelCase_					:						Dict =					word_embeddings

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

 # The transformer.
 UpperCamelCase_					:						int =					lm['transformer'] if 'transformer' in lm.keys() else lm['encoder']

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

 # The simple map of names for "automated" rules.
 UpperCamelCase_					:						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.
  UpperCamelCase_					:						Union[str, Any] =					layer_re.match(lowerCamelCase	)

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

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

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

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

   # Insert a "dummy" tensor for masked_bias.
   UpperCamelCase_					:						List[Any] =					torch.tensor(-1e4  ,  dtype=torch.floataa	)
   UpperCamelCase_					:						Union[str, Any] =					masked_bias

   UpperCamelCase_					:						List[Any] =					fix_query_key_value_ordering(lowerCamelCase  ,  lowerCamelCase  ,  3  ,  lowerCamelCase  ,  lowerCamelCase	)
   # Megatron stores (3*D) x D but transformers-GPT2 expects D x 3*D.
   UpperCamelCase_					:						Any =					out_val.transpose(0  ,  1	).contiguous()
   # Store.
   UpperCamelCase_					:						List[str] =					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":
   UpperCamelCase_					:						Union[str, Any] =					fix_query_key_value_ordering(lowerCamelCase  ,  lowerCamelCase  ,  3  ,  lowerCamelCase  ,  lowerCamelCase	)
   # Store. No change of shape.
   UpperCamelCase_					:						Optional[int] =					out_val

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

  # Copy the bias.
  elif weight_or_bias == "bias":
   UpperCamelCase_					:						Union[str, Any] =					megatron_to_transformers[op_name]
   UpperCamelCase_					:						Optional[int] =					val

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

 # The final layernorm.
 UpperCamelCase_					:						str =					transformer['final_layernorm.weight']
 UpperCamelCase_					:						str =					transformer['final_layernorm.bias']

 # For LM head, transformers' wants the matrix to weight embeddings.
 UpperCamelCase_					:						Optional[int] =					word_embeddings

 # It should be done!
 return output_state_dict





def    __lowercase     (						):
 # Create the argument parser.
 UpperCamelCase_					:						Dict =					argparse.ArgumentParser()
 parser.add_argument('--print-checkpoint-structure'  ,  action='store_true'	)
 parser.add_argument(
     'path_to_checkpoint'  ,  type=lowerCamelCase  ,  help='Path to the checkpoint file (.zip archive or direct .pt file)'  ,  )
 parser.add_argument(
     '--config_file'  ,  default=''  ,  type=lowerCamelCase  ,  help='An optional config json file describing the pre-trained model.'  ,  )
 UpperCamelCase_					:						List[Any] =					parser.parse_args()

 # Extract the basename.
 UpperCamelCase_					:						Dict =					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:
    UpperCamelCase_					:						List[str] =					torch.load(lowerCamelCase  ,  map_location='cpu'	)
 else:
  UpperCamelCase_					:						str =					torch.load(args.path_to_checkpoint  ,  map_location='cpu'	)

 UpperCamelCase_					:						List[str] =					input_state_dict.get('args'  ,  lowerCamelCase	)

 # 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:
    UpperCamelCase_					:						int =					'gelu_fast'
   elif ds_args.openai_gelu:
    UpperCamelCase_					:						str =					'gelu_new'
   else:
    UpperCamelCase_					:						Optional[int] =					'gelu'
  else:
   # in the very early days this used to be "gelu_new"
   UpperCamelCase_					:						List[str] =					'gelu_new'

  # Spell out all parameters in case the defaults change.
  UpperCamelCase_					:						str =					GPTaConfig(
      vocab_size=50257  ,  n_positions=1024  ,  n_embd=1024  ,  n_layer=24  ,  n_head=16  ,  n_inner=4096  ,  activation_function=lowerCamelCase  ,  resid_pdrop=0.1  ,  embd_pdrop=0.1  ,  attn_pdrop=0.1  ,  layer_norm_epsilon=1e-5  ,  initializer_range=0.0_2  ,  summary_type='cls_index'  ,  summary_use_proj=lowerCamelCase  ,  summary_activation=lowerCamelCase  ,  summary_proj_to_labels=lowerCamelCase  ,  summary_first_dropout=0.1  ,  scale_attn_weights=lowerCamelCase  ,  use_cache=lowerCamelCase  ,  bos_token_id=50256  ,  eos_token_id=50256  ,  )
 else:
  UpperCamelCase_					:						List[str] =					GPTaConfig.from_json_file(args.config_file	)

 UpperCamelCase_					:						Optional[int] =					['GPT2LMHeadModel']

 # Convert.
 print('Converting'	)
 UpperCamelCase_					:						Tuple =					convert_megatron_checkpoint(lowerCamelCase  ,  lowerCamelCase  ,  lowerCamelCase	)

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

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

 UpperCamelCase_					:						Tuple =					AutoTokenizer.from_pretrained(lowerCamelCase	)
 UpperCamelCase_					:						Union[str, Any] =					type(lowerCamelCase	).__name__
 UpperCamelCase_					:						List[str] =					tokenizer_class

 # Store the config to file.
 print('Saving config'	)
 config.save_pretrained(lowerCamelCase	)

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

 # Store the state_dict to file.
 UpperCamelCase_					:						int =					os.path.join(lowerCamelCase  ,  'pytorch_model.bin'	)
 print(F"Saving checkpoint to \"{output_checkpoint_file}\""	)
 torch.save(lowerCamelCase  ,  lowerCamelCase	)


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

if __name__ == "__main__":
   main()

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




from __future__ import annotations





def    __lowercase     (						lowerCamelCase					:     Optional[Any]  ,  lowerCamelCase					:     Dict  ,  lowerCamelCase					:     Union[str, Any]  ,  lowerCamelCase					:     List[str]	):  # noqa: E741
 while r - l > 1:
  UpperCamelCase_					:						Union[str, Any] =					(l + r) // 2
  if v[m] >= key:
   UpperCamelCase_					:						str =					m
  else:
   UpperCamelCase_					:						List[Any] =					m  # noqa: E741
 return r





def    __lowercase     (						lowerCamelCase					:     list[int]	):
 if len(lowerCamelCase	) == 0:
  return 0

 UpperCamelCase_					:						Tuple =					[0] * len(lowerCamelCase	)
 UpperCamelCase_					:						int =					1

 UpperCamelCase_					:						Dict =					v[0]

 for i in range(1  ,  len(lowerCamelCase	)	):
  if v[i] < tail[0]:
   UpperCamelCase_					:						Any =					v[i]
  elif v[i] > tail[length - 1]:
   UpperCamelCase_					:						Dict =					v[i]
   length += 1
  else:
   UpperCamelCase_					:						List[str] =					v[i]

 return length


if __name__ == "__main__":
   import doctest

   doctest.testmod()



