


"""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 					a__			(				SCREAMING_SNAKE_CASE    :    Any				,      SCREAMING_SNAKE_CASE    :    Any				,      SCREAMING_SNAKE_CASE    :    Optional[int]=0				):


  '''simple docstring'''







  if name is None:
    lowerCAmelCase     :     Optional[int]								=							None
  else:
    lowerCAmelCase     :     List[str]								=							"." * max(0				,      spaces - 2				) + "# {:" + str(5_0 - spaces				) + "s}"
    lowerCAmelCase     :     List[str]								=							fmt.format(SCREAMING_SNAKE_CASE				)

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


  '''simple docstring'''







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


  '''simple docstring'''







  lowerCAmelCase     :     Union[str, Any]								=							{}

  # old versions did not store training args
  lowerCAmelCase     :     Dict								=							input_state_dict.get("args"				,      SCREAMING_SNAKE_CASE				)
  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))

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

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

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

  # The word embeddings.
  lowerCAmelCase     :     List[Any]								=							embeddings["word_embeddings"]["weight"]
  # Truncate the embedding table to vocab_size rows.
  lowerCAmelCase     :     Tuple								=							word_embeddings[: config.vocab_size, :]
  lowerCAmelCase     :     Optional[int]								=							word_embeddings

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

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

  # The regex to extract layer names.
  lowerCAmelCase     :     str								=							re.compile(r"layers\.(\d+)\.([a-z0-9_.]+)\.([a-z]+)"				)

  # The simple map of names for "automated" rules.
  lowerCAmelCase     :     Dict								=							{
      "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.
    lowerCAmelCase     :     str								=							layer_re.match(SCREAMING_SNAKE_CASE				)

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

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

    # The name of the layer.
    lowerCAmelCase     :     List[str]								=							f"""transformer.h.{layer_idx}"""

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

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

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

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

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

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

  # The final layernorm.
  lowerCAmelCase     :     Union[str, Any]								=							transformer["final_layernorm.weight"]
  lowerCAmelCase     :     Any								=							transformer["final_layernorm.bias"]

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

  # It should be done!
  return output_state_dict
def 					a__			(				):


  '''simple docstring'''







  lowerCAmelCase     :     Union[str, Any]								=							argparse.ArgumentParser()
  parser.add_argument("--print-checkpoint-structure"				,      action="store_true"				)
  parser.add_argument(
      "path_to_checkpoint"				,      type=SCREAMING_SNAKE_CASE				,      help="Path to the checkpoint file (.zip archive or direct .pt file)"				,      )
  parser.add_argument(
      "--config_file"				,      default=""				,      type=SCREAMING_SNAKE_CASE				,      help="An optional config json file describing the pre-trained model."				,      )
  lowerCAmelCase     :     Optional[Any]								=							parser.parse_args()

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

  lowerCAmelCase     :     List[Any]								=							input_state_dict.get("args"				,      SCREAMING_SNAKE_CASE				)

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

    # Spell out all parameters in case the defaults change.
    lowerCAmelCase     :     Optional[Any]								=							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=SCREAMING_SNAKE_CASE				,      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=SCREAMING_SNAKE_CASE				,      summary_activation=SCREAMING_SNAKE_CASE				,      summary_proj_to_labels=SCREAMING_SNAKE_CASE				,      summary_first_dropout=0.1				,      scale_attn_weights=SCREAMING_SNAKE_CASE				,      use_cache=SCREAMING_SNAKE_CASE				,      bos_token_id=5_0_2_5_6				,      eos_token_id=5_0_2_5_6				,      )
  else:
    lowerCAmelCase     :     int								=							GPTaConfig.from_json_file(args.config_file				)

  lowerCAmelCase     :     Union[str, Any]								=							["GPT2LMHeadModel"]

  # Convert.
  print("Converting"				)
  lowerCAmelCase     :     str								=							convert_megatron_checkpoint(SCREAMING_SNAKE_CASE				,      SCREAMING_SNAKE_CASE				,      SCREAMING_SNAKE_CASE				)

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

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

  lowerCAmelCase     :     Any								=							AutoTokenizer.from_pretrained(SCREAMING_SNAKE_CASE				)
  lowerCAmelCase     :     List[Any]								=							type(SCREAMING_SNAKE_CASE				).__name__
  lowerCAmelCase     :     Dict								=							tokenizer_class

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

  # Save tokenizer based on args
  print(f"""Adding {tokenizer_class} tokenizer files"""				)
  tokenizer.save_pretrained(SCREAMING_SNAKE_CASE				)

  # Store the state_dict to file.
  lowerCAmelCase     :     Tuple								=							os.path.join(SCREAMING_SNAKE_CASE				,      "pytorch_model.bin"				)
  print(f"""Saving checkpoint to \"{output_checkpoint_file}\""""				)
  torch.save(SCREAMING_SNAKE_CASE				,      SCREAMING_SNAKE_CASE				)


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

if __name__ == "__main__":
     main()

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









"""simple docstring"""







from __future__ import annotations

from collections import namedtuple
from dataclasses import dataclass


@dataclass
class       SCREAMING_SNAKE_CASE__				:
       """simple docstring"""


       a  :				int
       a  :				TreeNode | None   =None
       a  :				TreeNode | None   =None

lowerCAmelCase__      				=   namedtuple('''CoinsDistribResult''', '''moves excess''')
def 					a__			(				SCREAMING_SNAKE_CASE    :    TreeNode | None				):


  '''simple docstring'''







  if root is None:
    return 0

  # Validation
  def count_nodes(SCREAMING_SNAKE_CASE    :    TreeNode | None				) -> int:
    if node is None:
      return 0

    return count_nodes(node.left				) + count_nodes(node.right				) + 1

  def count_coins(SCREAMING_SNAKE_CASE    :    TreeNode | None				) -> int:
    if node is None:
      return 0

    return count_coins(node.left				) + count_coins(node.right				) + node.data

  if count_nodes(SCREAMING_SNAKE_CASE				) != count_coins(SCREAMING_SNAKE_CASE				):
    raise ValueError("The nodes number should be same as the number of coins"				)

  # Main calculation
  def get_distrib(SCREAMING_SNAKE_CASE    :    TreeNode | None				) -> CoinsDistribResult:

    if node is None:
      return CoinsDistribResult(0				,      1				)

    lowerCAmelCase						,  lowerCAmelCase     :     List[str]								=							get_distrib(node.left				)
    lowerCAmelCase						,  lowerCAmelCase     :     List[str]								=							get_distrib(node.right				)

    lowerCAmelCase     :     Union[str, Any]								=							1 - left_distrib_excess
    lowerCAmelCase     :     List[Any]								=							1 - right_distrib_excess

    lowerCAmelCase     :     int								=							(
        left_distrib_moves
        + right_distrib_moves
        + abs(SCREAMING_SNAKE_CASE				)
        + abs(SCREAMING_SNAKE_CASE				)
    )
    lowerCAmelCase     :     int								=							node.data - coins_to_left - coins_to_right

    return CoinsDistribResult(SCREAMING_SNAKE_CASE				,      SCREAMING_SNAKE_CASE				)

  return get_distrib(SCREAMING_SNAKE_CASE				)[0]


if __name__ == "__main__":
     import doctest

     doctest.testmod()





