






import argparse
import os
import re
import zipfile

import torch

from transformers import AutoTokenizer, GPTaConfig

def   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__=0  )	->    Union[str, Any]:
 # Format the message.
 if name is None:
  UpperCamelCase_: Tuple   				=    None
 else:
  UpperCamelCase_: Dict   				=    '.' * max(0     ,   spaces - 2  ) + '# {:' + str(5_0 - spaces  ) + 's}'
  UpperCamelCase_: 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 (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__  )	->    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.
 UpperCamelCase_: str   				=    param.size()
 if checkpoint_version == 1.0:
  # version 1.0 stores [num_heads * hidden_size * num_splits, :]
  UpperCamelCase_: List[Any]   				=    (num_heads, hidden_size, num_splits) + input_shape[1:]
  UpperCamelCase_: int   				=    param.view(*_A  )
  UpperCamelCase_: List[str]   				=    param.transpose(0     ,   2  )
  UpperCamelCase_: Union[str, Any]   				=    param.transpose(1     ,   2  ).contiguous()
 elif checkpoint_version >= 2.0:
  # other versions store [num_heads * num_splits * hidden_size, :]
  UpperCamelCase_: Union[str, Any]   				=    (num_heads, num_splits, hidden_size) + input_shape[1:]
  UpperCamelCase_: List[str]   				=    param.view(*_A  )
  UpperCamelCase_: Union[str, Any]   				=    param.transpose(0     ,   1  ).contiguous()
 UpperCamelCase_: List[Any]   				=    param.view(*_A  )
 return param

def   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__  )	->    int:
 # The converted output model.
 UpperCamelCase_: Optional[Any]   				=    {}

 # old versions did not store training args
 UpperCamelCase_: Dict   				=    input_state_dict.get('args'     ,   _A  )
 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_: Union[str, Any]   				=    ds_args.padded_vocab_size
  UpperCamelCase_: str   				=    ds_args.max_position_embeddings
  UpperCamelCase_: Dict   				=    ds_args.hidden_size
  UpperCamelCase_: Union[str, Any]   				=    ds_args.num_layers
  UpperCamelCase_: Dict   				=    ds_args.num_attention_heads
  UpperCamelCase_: int   				=    ds_args.ffn_hidden_size
  # pprint(config)

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

 # The model.
 UpperCamelCase_: Optional[int]   				=    input_state_dict['model']
 # The language model.
 UpperCamelCase_: Optional[Any]   				=    model['language_model']
 # The embeddings.
 UpperCamelCase_: List[str]   				=    lm['embedding']

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

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

 # The transformer.
 UpperCamelCase_: Union[str, Any]   				=    lm['transformer'] if 'transformer' in lm.keys() else lm['encoder']

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

 # The simple map of names for "automated" rules.
 UpperCamelCase_: Optional[Any]   				=    {
     '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_: Tuple   				=    layer_re.match(_A  )

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

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

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

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

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

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

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

  # Copy the bias.
  elif weight_or_bias == "bias":
   UpperCamelCase_: Dict   				=    megatron_to_transformers[op_name]
   UpperCamelCase_: Optional[Any]   				=    val

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

 # The final layernorm.
 UpperCamelCase_: str   				=    transformer['final_layernorm.weight']
 UpperCamelCase_: List[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   snake_case ()	->    List[str]:
 # Create the argument parser.
 UpperCamelCase_: Dict   				=    argparse.ArgumentParser()
 parser.add_argument('--print-checkpoint-structure'     ,   action='store_true'  )
 parser.add_argument(
     'path_to_checkpoint'     ,   type=_A     ,   help='Path to the checkpoint file (.zip archive or direct .pt file)'     ,   )
 parser.add_argument(
     '--config_file'     ,   default=''     ,   type=_A     ,   help='An optional config json file describing the pre-trained model.'     ,   )
 UpperCamelCase_: Union[str, Any]   				=    parser.parse_args()

 # Extract the basename.
 UpperCamelCase_: Optional[Any]   				=    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_: Union[str, Any]   				=    torch.load(_A     ,   map_location='cpu'  )
 else:
  UpperCamelCase_: Any   				=    torch.load(args.path_to_checkpoint     ,   map_location='cpu'  )

 UpperCamelCase_: List[Any]   				=    input_state_dict.get('args'     ,   _A  )

 # 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_: Dict   				=    'gelu_new'
   else:
    UpperCamelCase_: Any   				=    'gelu'
  else:
   # in the very early days this used to be "gelu_new"
   UpperCamelCase_: Any   				=    'gelu_new'

  # Spell out all parameters in case the defaults change.
  UpperCamelCase_: Tuple   				=    GPTaConfig(
      vocab_size=5_0_2_5_7     ,   n_positions=1_0_2_4     ,   n_embd=1_0_2_4     ,   n_layer=2_4     ,   n_head=1_6     ,   n_inner=4_0_9_6     ,   activation_function=_A     ,   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=_A     ,   summary_activation=_A     ,   summary_proj_to_labels=_A     ,   summary_first_dropout=0.1     ,   scale_attn_weights=_A     ,   use_cache=_A     ,   bos_token_id=5_0_2_5_6     ,   eos_token_id=5_0_2_5_6     ,   )
 else:
  UpperCamelCase_: str   				=    GPTaConfig.from_json_file(args.config_file  )

 UpperCamelCase_: Any   				=    ['GPT2LMHeadModel']

 # Convert.
 print('Converting'  )
 UpperCamelCase_: Union[str, Any]   				=    convert_megatron_checkpoint(_A     ,   _A     ,   _A  )

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

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

 UpperCamelCase_: Tuple   				=    AutoTokenizer.from_pretrained(_A  )
 UpperCamelCase_: str   				=    type(_A  ).__name__
 UpperCamelCase_: List[str]   				=    tokenizer_class

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

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

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


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

if __name__ == "__main__":
   main()

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







from abc import ABC, abstractmethod
from argparse import ArgumentParser




class     _lowerCAmelCase(	UpperCAmelCase_ ):





       """simple docstring"""







       @staticmethod
       @abstractmethod
       def       _a   ( _lowerCamelCase							):
         raise NotImplementedError()





       @abstractmethod
       def       _a   ( self							):
         raise NotImplementedError()