





import argparse
import os
from os.path import isfile

import torch

from transformers import GPTaConfig

def      __snake_case						(					_lowerCAmelCase : Any				,				_lowerCAmelCase : str				,				_lowerCAmelCase : List[str]=0	) ->						List[str]:
   # Format the message.
   if name is None:
      A_					:				Tuple          =			None
   else:
      A_					:				Tuple          =			"." * max(0				,				spaces - 2	) + "# {:" + str(50 - spaces	) + "s}"
      A_					:				Dict          =			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						(					_lowerCAmelCase : List[Any]				,				_lowerCAmelCase : str				,				_lowerCAmelCase : Dict				,				_lowerCAmelCase : List[Any]	) ->						Dict:
   # 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[int]          =			param.size()
   # other versions store [num_heads * num_splits * hidden_size, :]
   A_					:				List[str]          =			(num_heads, num_splits, hidden_size) + input_shape[1:]
   A_					:				Optional[Any]          =			param.view(*a_	)
   A_					:				str          =			param.transpose(0				,				1	).contiguous()
   A_					:				Optional[Any]          =			param.view(*a_	)
   return param

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

   A_					:				int          =			sd_megatron["model.language_model.embedding.word_embeddings.weight"][:vocab_size, :]
   A_					:				Optional[Any]          =			{
       "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"],
   }

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

      A_					:				Union[str, Any]          =			sd_megatron[f"{pf}{i}.input_layernorm.weight"]
      A_					:				Tuple          =			sd_megatron[f"{pf}{i}.input_layernorm.bias"]

      A_					:				Optional[int]          =			sd_megatron[f"{pf}{i}.self_attention.query_key_value.weight"]
      A_					:				Dict          =			fix_query_key_value_ordering(a_				,				3				,				a_				,				a_	)
      A_					:				Optional[Any]          =			vala.transpose(0				,				1	).contiguous()

      A_					:				Tuple          =			sd_megatron[f"{pf}{i}.self_attention.query_key_value.bias"]
      A_					:				str          =			fix_query_key_value_ordering(a_				,				3				,				a_				,				a_	)
      A_					:				Tuple          =			vala

      A_					:				List[Any]          =			sd_megatron[f"{pf}{i}.self_attention.dense.weight"].transpose(
          0				,				1	)
      A_					:				Optional[Any]          =			sd_megatron[f"{pf}{i}.self_attention.dense.bias"]
      A_					:				Union[str, Any]          =			sd_megatron[f"{pf}{i}.post_attention_layernorm.weight"]
      A_					:				str          =			sd_megatron[f"{pf}{i}.post_attention_layernorm.bias"]
      A_					:				Tuple          =			sd_megatron[f"{pf}{i}.mlp.dense_h_to_4h.weight"].transpose(0				,				1	)
      A_					:				Any          =			sd_megatron[f"{pf}{i}.mlp.dense_h_to_4h.bias"]
      A_					:				Optional[Any]          =			sd_megatron[f"{pf}{i}.mlp.dense_4h_to_h.weight"].transpose(
          0				,				1	)
      A_					:				Optional[Any]          =			sd_megatron[f"{pf}{i}.mlp.dense_4h_to_h.bias"]

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

   return sd_hf

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

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

   return config_hf

def      __snake_case						(					_lowerCAmelCase : Optional[Any]	) ->						List[Any]:
   print(a_	)

   A_					:				Dict          =			args.checkpoint_path
   A_					:				Dict          =			args.save_path
   if isfile(a_	):
      raise FileNotFoundError(f"ERROR! could not find file {checkpoint_path}"	)

   # Load the model.
   A_					:				Tuple          =			torch.load(a_				,				map_location="cpu"	)

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

   A_					:				Optional[Any]          =			checkpoint["state_dict"]

   # Convert.
   print("Converting"	)
   A_					:				int          =			convert_megatron_checkpoint(a_				,				a_	)

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

   A_					:				Dict          =			"GPTSw3Tokenizer"

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

   # Store the state_dict to file.
   A_					:				List[str]          =			os.path.join(a_				,				"pytorch_model.bin"	)
   print(f"Saving checkpoint to \"{output_checkpoint_file}\""	)
   torch.save(a_				,				a_	)


if __name__ == "__main__":
 _lowerCAmelCase							:      Dict						=	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							:      int						=	parser.parse_args()
 main(_args)








from __future__ import annotations

def      __snake_case						(					_lowerCAmelCase : int	) ->						list[int]:
   assert (
       isinstance(_lowerCAmelCase				,				_lowerCAmelCase	) and mask > 0
   ), f"mask needs to be positive integer, your input {mask}"
   A_					:				Optional[Any]          =			[]
   A_					:				Optional[int]          =			mask

   while submask:
      all_submasks.append(_lowerCAmelCase	)
      A_					:				List[str]          =			(submask - 1) & mask

   return all_submasks


if __name__ == "__main__":
 import doctest

 doctest.testmod()
