





"""simple docstring"""
import argparse
import os
from os.path import isfile

import torch

from transformers import GPTaConfig




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

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




def 			__snake_case						(      __A ,__A ,__A ,__A    )		->				Optional[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.
   lowercase    :			Union[str, Any]  =       param.size()
   # other versions store [num_heads * num_splits * hidden_size, :]
   lowercase    :			int  =       (num_heads, num_splits, hidden_size) + input_shape[1:]
   lowercase    :			Tuple  =       param.view(*__A    )
   lowercase    :			Optional[Any]  =       param.transpose(0 ,1    ).contiguous()
   lowercase    :			Tuple  =       param.view(*__A    )
   return param




def 			__snake_case						(      __A ,__A    )		->				int:
   lowercase    :			Dict  =       config.n_positions
   lowercase    :			str  =       config.n_layer
   lowercase    :			Optional[int]  =       config.vocab_size
   lowercase    :			Any  =       config.n_head
   lowercase    :			int  =       config.n_embd // config.n_head

   lowercase    :			Dict  =       sd_megatron["""model.language_model.embedding.word_embeddings.weight"""][:vocab_size, :]
   lowercase    :			Optional[int]  =       {
       """transformer.wte.weight""": word_embeddings,
       """transformer.wpe.weight""": sd_megatron["""model.language_model.embedding.position_embeddings.weight"""],
       """transformer.ln_f.weight""": sd_megatron["""model.language_model.encoder.final_layernorm.weight"""],
       """transformer.ln_f.bias""": sd_megatron["""model.language_model.encoder.final_layernorm.bias"""],
   }

   lowercase    :			str  =       """model.language_model.encoder.layers."""
   for i in range(__A    ):
      lowercase    :			Union[str, Any]  =       torch.tril(torch.ones((n_positions, n_positions) ,dtype=torch.bool    )    )
      lowercase    :			Optional[Any]  =       causal_mask.view(1 ,1 ,__A ,__A    )
      lowercase    :			List[Any]  =       causal_mask
      lowercase    :			Optional[Any]  =       torch.tensor(-1E4 ,dtype=torch.bfloataa    )

      lowercase    :			str  =       sd_megatron[F'''{pf}{i}.input_layernorm.weight''']
      lowercase    :			List[str]  =       sd_megatron[F'''{pf}{i}.input_layernorm.bias''']

      lowercase    :			Optional[Any]  =       sd_megatron[F'''{pf}{i}.self_attention.query_key_value.weight''']
      lowercase    :			List[str]  =       fix_query_key_value_ordering(__A ,3 ,__A ,__A    )
      lowercase    :			Any  =       vala.transpose(0 ,1    ).contiguous()

      lowercase    :			Optional[Any]  =       sd_megatron[F'''{pf}{i}.self_attention.query_key_value.bias''']
      lowercase    :			int  =       fix_query_key_value_ordering(__A ,3 ,__A ,__A    )
      lowercase    :			Optional[Any]  =       vala

      lowercase    :			Any  =       sd_megatron[F'''{pf}{i}.self_attention.dense.weight'''].transpose(
          0 ,1    )
      lowercase    :			Any  =       sd_megatron[F'''{pf}{i}.self_attention.dense.bias''']
      lowercase    :			Tuple  =       sd_megatron[F'''{pf}{i}.post_attention_layernorm.weight''']
      lowercase    :			List[Any]  =       sd_megatron[F'''{pf}{i}.post_attention_layernorm.bias''']
      lowercase    :			Optional[Any]  =       sd_megatron[F'''{pf}{i}.mlp.dense_h_to_4h.weight'''].transpose(0 ,1    )
      lowercase    :			int  =       sd_megatron[F'''{pf}{i}.mlp.dense_h_to_4h.bias''']
      lowercase    :			Tuple  =       sd_megatron[F'''{pf}{i}.mlp.dense_4h_to_h.weight'''].transpose(
          0 ,1    )
      lowercase    :			List[Any]  =       sd_megatron[F'''{pf}{i}.mlp.dense_4h_to_h.bias''']

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

   return sd_hf




def 			__snake_case						(      __A ,__A    )		->				Optional[Any]:
   lowercase    :			str  =       64000
   lowercase    :			List[Any]  =       config_megatron["""encoder_seq_length"""]
   lowercase    :			str  =       config_megatron["""hidden_size"""]
   lowercase    :			Optional[Any]  =       config_megatron["""num_layers"""]
   lowercase    :			int  =       config_megatron["""num_attention_heads"""]
   lowercase    :			Optional[Any]  =       config_megatron["""ffn_hidden_size"""]
   lowercase    :			str  =       """gelu"""
   lowercase    :			List[Any]  =       0.1
   lowercase    :			Optional[Any]  =       0.1
   lowercase    :			str  =       0.1
   lowercase    :			str  =       config_megatron["""layernorm_epsilon"""]  # 1e-5
   lowercase    :			Dict  =       config_megatron["""init_method_std"""]  # 0.02
   lowercase    :			Optional[int]  =       config_megatron["""apply_query_key_layer_scaling"""]  # True
   lowercase    :			List[Any]  =       True
   lowercase    :			Optional[Any]  =       True

   # This identifies the 6.7B (7B) model which uses a different tokenizer
   if config_megatron["hidden_size"] == 4096:
      lowercase    :			Dict  =       1  # <|endoftext|>
      lowercase    :			Union[str, Any]  =       1  # <|endoftext|>
      lowercase    :			Any  =       0  # <unk>
   else:
      lowercase    :			List[str]  =       2  # <s>
      lowercase    :			Tuple  =       3  # <|endoftext|>
      lowercase    :			List[Any]  =       0  # <pad>

   return config_hf




def 			__snake_case						(      __A    )		->				Dict:
   print(__A    )

   lowercase    :			Dict  =       args.checkpoint_path
   lowercase    :			int  =       args.save_path
   if isfile(__A    ):
      raise FileNotFoundError(F'''ERROR! could not find file {checkpoint_path}'''    )

   # Load the model.
   lowercase    :			List[str]  =       torch.load(__A ,map_location="""cpu"""    )

   # Load the config.
   lowercase    :			List[str]  =       checkpoint["""hyper_parameters"""]["""cfg"""]
   lowercase    :			List[Any]  =       GPTaConfig()
   lowercase    :			Tuple  =       copy_config(config_hf=__A ,config_megatron=__A    )
   lowercase    :			Dict  =       ["""GPT2LMHeadModel"""]

   lowercase    :			Union[str, Any]  =       checkpoint["""state_dict"""]

   # Convert.
   print("""Converting"""    )
   lowercase    :			Any  =       convert_megatron_checkpoint(__A ,__A    )

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

   lowercase    :			List[str]  =       """GPTSw3Tokenizer"""

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

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


if __name__ == "__main__":
   lowerCAmelCase:	List[Any]							      =argparse.ArgumentParser()
   parser.add_argument(
       "--checkpoint_path",
       type=str,
       required=True,
       help="e.g. megatron_gpt--val_loss=2.42-step=38000-consumed_samples=54720000",
   )
   parser.add_argument("--save_path", type=str, required=True, help="e.g. /home/user/gpt-sw3/hf")
   parser.add_argument("--print-checkpoint-structure", action="store_true")
   lowerCAmelCase:	Any							      =parser.parse_args()
   main(_args)









"""simple docstring"""
import torch.nn as nn

from .imports import is_fpa_available


if is_fpa_available():
   import transformer_engine.pytorch as te




def 			__snake_case						(      __A ,__A=True ,__A=True ,__A=True    )		->				Union[str, Any]:
   if not is_fpa_available():
      raise ImportError("""Using `convert_model` requires transformer_engine to be installed."""    )
   for name, module in model.named_children():
      if isinstance(__A ,nn.Linear    ) and to_transformer_engine and _convert_linear:
         # Return early if the linear layer weights are not multiples of 16
         if any(p % 16 != 0 for p in module.weight.shape    ):
            return
         lowercase    :			Optional[int]  =       module.bias is not None
         lowercase    :			Union[str, Any]  =       te.Linear(
             module.in_features ,module.out_features ,bias=__A ,params_dtype=module.weight.dtype    )
         lowercase    :			Union[str, Any]  =       module.weight.data.clone()
         if has_bias:
            lowercase    :			Dict  =       module.bias.data.clone()

         setattr(__A ,__A ,__A    )
      elif isinstance(__A ,nn.LayerNorm    ) and to_transformer_engine and _convert_ln:
         lowercase    :			List[str]  =       te.LayerNorm(module.normalized_shape[0] ,eps=module.eps ,params_dtype=module.weight.dtype    )
         lowercase    :			Optional[Any]  =       module.weight.data.clone()
         lowercase    :			Any  =       module.bias.data.clone()

         setattr(__A ,__A ,__A    )
      elif isinstance(__A ,te.Linear    ) and not to_transformer_engine and _convert_linear:
         lowercase    :			Any  =       module.bias is not None
         lowercase    :			List[Any]  =       nn.Linear(
             module.in_features ,module.out_features ,bias=__A ,params_dtype=module.weight.dtype    )
         lowercase    :			Union[str, Any]  =       module.weight.data.clone()
         if has_bias:
            lowercase    :			Optional[Any]  =       module.bias.data.clone()

         setattr(__A ,__A ,__A    )
      elif isinstance(__A ,te.LayerNorm    ) and not to_transformer_engine and _convert_ln:
         lowercase    :			Any  =       nn.LayerNorm(module.normalized_shape[0] ,eps=module.eps ,params_dtype=module.weight.dtype    )
         lowercase    :			str  =       module.weight.data.clone()
         lowercase    :			List[str]  =       module.bias.data.clone()

         setattr(__A ,__A ,__A    )
      else:
         convert_model(
             __A ,to_transformer_engine=__A ,_convert_linear=_convert_linear ,_convert_ln=_convert_ln ,)







def 			__snake_case						(      __A    )		->				Union[str, Any]:
   if not is_fpa_available():
      raise ImportError("""Using `has_transformer_engine_layers` requires transformer_engine to be installed."""    )
   for m in model.modules():
      if isinstance(__A ,(te.LayerNorm, te.Linear)    ):
         return True
   return False


