


'''simple docstring'''







import argparse
import os
from os.path import isfile

import torch

from transformers import GPTaConfig






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






def    lowerCAmelCase__ (       lowerCamelCase  :    Optional[int]						,lowerCamelCase  :    Any     ):
       _A  :			Optional[Any]											=      config.n_positions
       _A  :			Dict											=      config.n_layer
       _A  :			List[Any]											=      config.vocab_size
       _A  :			List[Any]											=      config.n_head
       _A  :			Tuple											=      config.n_embd // config.n_head

       _A  :			Optional[int]											=      sd_megatron['model.language_model.embedding.word_embeddings.weight'][:vocab_size, :]
       _A  :			str											=      {
           '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  :			Any											=      'model.language_model.encoder.layers.'
       for i in range(lowerCamelCase     ):
              _A  :			Dict											=      torch.tril(torch.ones((n_positions, n_positions)						,dtype=torch.bool     )     )
              _A  :			Any											=      causal_mask.view(1						,1						,lowerCamelCase						,lowerCamelCase     )
              _A  :			Tuple											=      causal_mask
              _A  :			str											=      torch.tensor(-1E4						,dtype=torch.bfloataa     )

              _A  :			Optional[Any]											=      sd_megatron[F'{pf}{i}.input_layernorm.weight']
              _A  :			List[Any]											=      sd_megatron[F'{pf}{i}.input_layernorm.bias']

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

              _A  :			List[Any]											=      sd_megatron[F'{pf}{i}.self_attention.query_key_value.bias']
              _A  :			List[Any]											=      fix_query_key_value_ordering(lowerCamelCase						,3						,lowerCamelCase						,lowerCamelCase     )
              _A  :			Tuple											=      vala

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

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

       return sd_hf






def    lowerCAmelCase__ (       lowerCamelCase  :    Optional[int]						,lowerCamelCase  :    Tuple     ):
       _A  :			List[str]											=      64000
       _A  :			Optional[int]											=      config_megatron['encoder_seq_length']
       _A  :			Dict											=      config_megatron['hidden_size']
       _A  :			Optional[Any]											=      config_megatron['num_layers']
       _A  :			List[Any]											=      config_megatron['num_attention_heads']
       _A  :			Any											=      config_megatron['ffn_hidden_size']
       _A  :			List[Any]											=      'gelu'
       _A  :			Tuple											=      0.1
       _A  :			Optional[int]											=      0.1
       _A  :			Union[str, Any]											=      0.1
       _A  :			List[Any]											=      config_megatron['layernorm_epsilon']  # 1e-5
       _A  :			int											=      config_megatron['init_method_std']  # 0.02
       _A  :			Union[str, Any]											=      config_megatron['apply_query_key_layer_scaling']  # True
       _A  :			Union[str, Any]											=      True
       _A  :			Dict											=      True

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

       return config_hf






def    lowerCAmelCase__ (       lowerCamelCase  :    str     ):
       print(lowerCamelCase     )

       _A  :			Dict											=      args.checkpoint_path
       _A  :			Optional[int]											=      args.save_path
       if isfile(lowerCamelCase     ):
              raise FileNotFoundError(F'ERROR! could not find file {checkpoint_path}'     )

       # Load the model.
       _A  :			Dict											=      torch.load(lowerCamelCase						,map_location='cpu'     )

       # Load the config.
       _A  :			Optional[Any]											=      checkpoint['hyper_parameters']['cfg']
       _A  :			Optional[int]											=      GPTaConfig()
       _A  :			int											=      copy_config(config_hf=lowerCamelCase						,config_megatron=lowerCamelCase     )
       _A  :			int											=      ['GPT2LMHeadModel']

       _A  :			List[Any]											=      checkpoint['state_dict']

       # Convert.
       print('Converting'     )
       _A  :			Optional[Any]											=      convert_megatron_checkpoint(lowerCamelCase						,lowerCamelCase     )

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

       _A  :			int											=      'GPTSw3Tokenizer'

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

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


if __name__ == "__main__":
   A   :  Optional[int]       		=				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''')
   A   :  Optional[int]       		=				parser.parse_args()
   main(_args)










'''simple docstring'''







A   :  Optional[Any]       		=				{
    '''0000''': '''0''',
    '''0001''': '''1''',
    '''0010''': '''2''',
    '''0011''': '''3''',
    '''0100''': '''4''',
    '''0101''': '''5''',
    '''0110''': '''6''',
    '''0111''': '''7''',
    '''1000''': '''8''',
    '''1001''': '''9''',
    '''1010''': '''a''',
    '''1011''': '''b''',
    '''1100''': '''c''',
    '''1101''': '''d''',
    '''1110''': '''e''',
    '''1111''': '''f''',
}






def    lowerCAmelCase__ (       lowerCamelCase  :    str     ):
       _A  :			Union[str, Any]											=      str(lowerCamelCase     ).strip()

       # Exceptions
       if not binary_str:
              raise ValueError('Empty string was passed to the function'     )
       _A  :			Any											=      binary_str[0] == '-'
       _A  :			str											=      binary_str[1:] if is_negative else binary_str
       if not all(char in '01' for char in binary_str     ):
              raise ValueError('Non-binary value was passed to the function'     )

       _A  :			Optional[Any]											=      (
           '0' * (4 * (divmod(len(lowerCamelCase     )						,4     )[0] + 1) - len(lowerCamelCase     )) + binary_str
       )

       _A  :			Optional[int]											=      []
       for x in range(0						,len(lowerCamelCase     )						,4     ):
              hexadecimal.append(BITS_TO_HEX[binary_str[x : x + 4]]     )
       _A  :			int											=      '0x' + ''.join(lowerCamelCase     )

       return "-" + hexadecimal_str if is_negative else hexadecimal_str


if __name__ == "__main__":
   from doctest import testmod

   testmod()






