





'''simple docstring'''






# 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        UpperCamelCase__		(						__magic_name__  :   Optional[Any]		,       __magic_name__  :   Tuple		,       __magic_name__  :   Union[str, Any]=0      )				->			int:
 '''simple docstring'''






 if name is None:
  snake_case__     :					Optional[Any]    =  None
 else:
  snake_case__     :					int    =  "." * max(0		,       spaces - 2      ) + "# {:" + str(50 - spaces      ) + "s}"
  snake_case__     :					int    =  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        UpperCamelCase__		(						__magic_name__  :   Optional[Any]		,       __magic_name__  :   Union[str, Any]		,       __magic_name__  :   Dict		,       __magic_name__  :   str		,       __magic_name__  :   List[Any]      )				->			Union[str, Any]:
 '''simple docstring'''






 snake_case__     :					str    =  param.size()
 if checkpoint_version == 1.0:
  # version 1.0 stores [num_heads * hidden_size * num_splits, :]
  snake_case__     :					List[str]    =  (num_heads, hidden_size, num_splits) + input_shape[1:]
  snake_case__     :					List[str]    =  param.view(*_lowerCAmelCase      )
  snake_case__     :					int    =  param.transpose(0		,       2      )
  snake_case__     :					Optional[int]    =  param.transpose(1		,       2      ).contiguous()
 elif checkpoint_version >= 2.0:
  # other versions store [num_heads * num_splits * hidden_size, :]
  snake_case__     :					Optional[Any]    =  (num_heads, num_splits, hidden_size) + input_shape[1:]
  snake_case__     :					List[str]    =  param.view(*_lowerCAmelCase      )
  snake_case__     :					List[str]    =  param.transpose(0		,       1      ).contiguous()
 snake_case__     :					Union[str, Any]    =  param.view(*_lowerCAmelCase      )
 return param



def        UpperCamelCase__		(						__magic_name__  :   str		,       __magic_name__  :   str		,       __magic_name__  :   str      )				->			Any:
 '''simple docstring'''






 snake_case__     :					List[str]    =  {}

 # old versions did not store training args
 snake_case__     :					List[str]    =  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))

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

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

 # The model.
 snake_case__     :					Optional[int]    =  input_state_dict["model"]
 # The language model.
 snake_case__     :					List[str]    =  model["language_model"]
 # The embeddings.
 snake_case__     :					Union[str, Any]    =  lm["embedding"]

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

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

 # The transformer.
 snake_case__     :					List[Any]    =  lm["transformer"] if "transformer" in lm.keys() else lm["encoder"]

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

 # The simple map of names for "automated" rules.
 snake_case__     :					str    =  {
     "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.
  snake_case__     :					Optional[Any]    =  layer_re.match(_lowerCAmelCase      )

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

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

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

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

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

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

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

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

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

 # The final layernorm.
 snake_case__     :					Any    =  transformer["final_layernorm.weight"]
 snake_case__     :					Dict    =  transformer["final_layernorm.bias"]

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

 # It should be done!
 return output_state_dict



def        UpperCamelCase__		(						)				->			Tuple:
 '''simple docstring'''






 snake_case__     :					str    =  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."""		,       )
 snake_case__     :					Dict    =  parser.parse_args()

 # Extract the basename.
 snake_case__     :					int    =  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:
    snake_case__     :					str    =  torch.load(_lowerCAmelCase		,       map_location="""cpu"""      )
 else:
  snake_case__     :					Optional[int]    =  torch.load(args.path_to_checkpoint		,       map_location="""cpu"""      )

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

  # Spell out all parameters in case the defaults change.
  snake_case__     :					int    =  GPTaConfig(
      vocab_size=5_02_57		,       n_positions=10_24		,       n_embd=10_24		,       n_layer=24		,       n_head=16		,       n_inner=40_96		,       activation_function=_lowerCAmelCase		,       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=_lowerCAmelCase		,       summary_activation=_lowerCAmelCase		,       summary_proj_to_labels=_lowerCAmelCase		,       summary_first_dropout=0.1		,       scale_attn_weights=_lowerCAmelCase		,       use_cache=_lowerCAmelCase		,       bos_token_id=5_02_56		,       eos_token_id=5_02_56		,       )
 else:
  snake_case__     :					Any    =  GPTaConfig.from_json_file(args.config_file      )

 snake_case__     :					Dict    =  ["GPT2LMHeadModel"]

 # Convert.
 print("""Converting"""      )
 snake_case__     :					Union[str, Any]    =  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:
  snake_case__     :					Union[str, Any]    =  ds_args.tokenizer_type
  if tokenizer_type == "GPT2BPETokenizer":
   snake_case__     :					List[Any]    =  "gpt2"
  elif tokenizer_type == "PretrainedFromHF":
   snake_case__     :					List[Any]    =  ds_args.tokenizer_name_or_path
  else:
   raise ValueError(f"Unrecognized tokenizer_type {tokenizer_type}"      )
 else:
  snake_case__     :					Dict    =  "gpt2"

 snake_case__     :					Dict    =  AutoTokenizer.from_pretrained(_lowerCAmelCase      )
 snake_case__     :					Dict    =  type(_lowerCAmelCase      ).__name__
 snake_case__     :					Optional[Any]    =  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.
 snake_case__     :					Any    =  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

from cmath import sqrt

def   __snake_case    (       _lowerCAmelCase			:			int					,					_lowerCAmelCase			:			int					,					_lowerCAmelCase			:			int						)			->    tuple[complex, complex]:

   if a == 0:
      raise ValueError("Coefficient 'a' must not be zero."						)
   A_       :					Tuple        =      b * b - 4 * a * c

   A_       :					Any        =      (-b + sqrt(_lowerCAmelCase						)) / (2 * a)
   A_       :					Union[str, Any]        =      (-b - sqrt(_lowerCAmelCase						)) / (2 * a)

   return (
       root_a.real if not root_a.imag else root_a,
       root_a.real if not root_a.imag else root_a,
   )

def   __snake_case    (       )			->    List[Any]:
   A_   ,    A_       :					Optional[Any]        =      quadratic_roots(a=5					,					b=6					,					c=1						)
   print(f"The solutions are: {solutiona} and {solutiona}"						)


if __name__ == "__main__":
 main()
