
'''simple docstring'''





import argparse
import os
import re
import zipfile

import torch

from transformers import MegatronBertConfig





def UpperCAmelCase						(					UpperCAmelCase_      ,  UpperCAmelCase_      ,  UpperCAmelCase_=0	):
    # Format the message.
    if name is None:
        _a							:       int       					=			None
    else:
        _a							:       int       					=			""".""" * max(0      ,  spaces - 2	) + """# {:""" + str(50 - spaces	) + """s}"""
        _a							:       Tuple       					=			fmt.format(UpperCAmelCase_	)

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





def UpperCAmelCase						(					UpperCAmelCase_      ,  UpperCAmelCase_      ,  UpperCAmelCase_      ,  UpperCAmelCase_      ,  UpperCAmelCase_	):
    # 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 BERT.
    _a							:       Union[str, Any]       					=			param.size()
    if checkpoint_version == 1.0:
        # version 1.0 stores [num_heads * hidden_size * num_splits, :]
        _a							:       Tuple       					=			(num_heads, hidden_size, num_splits) + input_shape[1:]
        _a							:       Tuple       					=			param.view(*UpperCAmelCase_	)
        _a							:       int       					=			param.transpose(0      ,  2	)
        _a							:       Optional[Any]       					=			param.transpose(1      ,  2	).contiguous()
    elif checkpoint_version >= 2.0:
        # other versions store [num_heads * num_splits * hidden_size, :]
        _a							:       Union[str, Any]       					=			(num_heads, num_splits, hidden_size) + input_shape[1:]
        _a							:       List[Any]       					=			param.view(*UpperCAmelCase_	)
        _a							:       int       					=			param.transpose(0      ,  1	).contiguous()
    _a							:       Any       					=			param.view(*UpperCAmelCase_	)
    return param





def UpperCAmelCase						(					UpperCAmelCase_      ,  UpperCAmelCase_      ,  UpperCAmelCase_	):
    # The converted output model.
    _a							:       Union[str, Any]       					=			{}

    # old versions did not store training args
    _a							:       List[Any]       					=			input_state_dict.get("""args"""      ,  UpperCAmelCase_	)
    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))

        _a							:       List[str]       					=			ds_args.tokenizer_type
        _a							:       Optional[Any]       					=			ds_args.padded_vocab_size
        _a							:       Union[str, Any]       					=			ds_args.max_position_embeddings
        _a							:       Any       					=			ds_args.hidden_size
        _a							:       int       					=			ds_args.num_layers
        _a							:       List[Any]       					=			ds_args.num_attention_heads
        _a							:       List[str]       					=			ds_args.ffn_hidden_size if """ffn_hidden_size""" in ds_args else 4 * ds_args.hidden_size
        # pprint(config)

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

    # The model.
    _a							:       int       					=			input_state_dict["""model"""]
    # The language model.
    _a							:       Tuple       					=			model["""language_model"""]
    # The embeddings.
    _a							:       Any       					=			lm["""embedding"""]

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

    # The position embeddings.
    _a							:       int       					=			embeddings["""position_embeddings"""]["""weight"""]
    assert pos_embeddings.size(0	) == config.max_position_embeddings and pos_embeddings.size(1	) == config.hidden_size
    # Store the position embeddings.
    _a							:       Any       					=			pos_embeddings

    # The token-type embeddings.
    _a							:       Dict       					=			embeddings["""tokentype_embeddings"""]["""weight"""]
    # Store the position embeddings.
    _a							:       Tuple       					=			tokentype_embeddings

    # The transformer.
    _a							:       Dict       					=			lm["""transformer"""] if """transformer""" in lm.keys() else lm["""encoder"""]

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

    # The simple map of names for "automated" rules.
    _a							:       Optional[Any]       					=			{
        """attention.dense""": """.attention.output.dense.""",
        """self_attention.dense""": """.attention.output.dense.""",
        """mlp.dense_h_to_4h""": """.intermediate.dense.""",
        """mlp.dense_4h_to_h""": """.output.dense.""",
    }

    # Keep track of the attention/query/value tensor.
    _a							:       str       					=			None

    # Extract the layers.
    for key, val in transformer.items():
        # Match the name.
        _a							:       Any       					=			layer_re.match(UpperCAmelCase_	)

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

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

        # The name of the layer.
        _a							:       Optional[Any]       					=			F'''bert.encoder.layer.{layer_idx}'''

        # For layernorm(s), simply store the layer norm.
        if op_name.endswith("""layernorm"""	):
            _a							:       str       					=			"""attention.ln""" if op_name.startswith("""input"""	) else """ln"""
            _a							:       Optional[Any]       					=			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":
            # Make sure the QKV pointer is nil.
            assert attention_qkv_weight is None, ""

            _a							:       Tuple       					=			fix_query_key_value_ordering(UpperCAmelCase_      ,  UpperCAmelCase_      ,  3      ,  UpperCAmelCase_      ,  UpperCAmelCase_	)
            # Store the tensor as we need the bias as well to interleave QKV and biases.
            _a							:       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":
            # Make sure we read the weight tensor.
            assert attention_qkv_weight is not None, ""

            # Split the QKV matrix into Q, K and V. Megatron stores Q,K,V interleaved.
            _a							:       Any       					=			attention_qkv_weight[0 * config.hidden_size : 1 * config.hidden_size, :]
            _a							:       str       					=			attention_qkv_weight[1 * config.hidden_size : 2 * config.hidden_size, :]
            _a							:       List[Any]       					=			attention_qkv_weight[2 * config.hidden_size : 3 * config.hidden_size, :]

            _a							:       List[str]       					=			fix_query_key_value_ordering(UpperCAmelCase_      ,  UpperCAmelCase_      ,  3      ,  UpperCAmelCase_      ,  UpperCAmelCase_	)
            # Split the bias.
            _a							:       Tuple       					=			out_val[0 * config.hidden_size : 1 * config.hidden_size]
            _a							:       int       					=			out_val[1 * config.hidden_size : 2 * config.hidden_size]
            _a							:       str       					=			out_val[2 * config.hidden_size : 3 * config.hidden_size]

            # Store.
            _a							:       Tuple       					=			q
            _a							:       Dict       					=			q_bias
            _a							:       List[str]       					=			k
            _a							:       List[str]       					=			k_bias
            _a							:       Union[str, Any]       					=			v
            _a							:       Optional[Any]       					=			v_bias

            # Clear the stored tensor.
            _a							:       str       					=			None

        # Copy weights and biases as is.
        elif weight_or_bias in ["weight", "bias"]:
            _a							:       Tuple       					=			megatron_to_transformers[op_name]
            _a							:       Tuple       					=			val

    # The final layernorm.
    _a							:       int       					=			transformer["""final_layernorm.weight"""]
    _a							:       List[str]       					=			transformer["""final_layernorm.bias"""]

    # The pooler.
    _a							:       List[str]       					=			lm["""pooler"""]

    # Store the matrix and the bias.
    _a							:       List[Any]       					=			pooler["""dense.weight"""]
    _a							:       Dict       					=			pooler["""dense.bias"""]

    # The LM head from Megatron (for RACE).
    _a							:       Dict       					=			model["""lm_head"""]

    # The transform matrix.
    _a							:       int       					=			lm_head["""dense.weight"""]
    _a							:       Any       					=			lm_head["""dense.bias"""]

    # The transform LN.
    _a							:       Union[str, Any]       					=			lm_head["""layernorm.weight"""]
    _a							:       List[Any]       					=			lm_head["""layernorm.bias"""]

    # For the decoder, we replicate the weights.
    _a							:       Any       					=			word_embeddings
    _a							:       Dict       					=			lm_head["""bias"""]

    # The classifier from Megatron (for MLNI).
    _a							:       Optional[int]       					=			model["""binary_head"""]

    # Store the classifier.
    _a							:       Union[str, Any]       					=			binary_head["""weight"""]
    _a							:       Union[str, Any]       					=			binary_head["""bias"""]

    # It should be done!
    return output_state_dict





def UpperCAmelCase						(					):
    # Create the argument parser.
    _a							:       Optional[int]       					=			argparse.ArgumentParser()
    parser.add_argument("""--print-checkpoint-structure"""      ,  action="""store_true"""	)
    parser.add_argument("""path_to_checkpoint"""      ,  type=UpperCAmelCase_      ,  help="""Path to the ZIP file containing the checkpoint"""	)
    parser.add_argument(
        """--config_file"""      ,  default=""""""      ,  type=UpperCAmelCase_      ,  help="""An optional config json file describing the pre-trained model."""      ,  )
    _a							:       Union[str, Any]       					=			parser.parse_args()

    # Extract the basename.
    _a							:       Tuple       					=			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:
                _a							:       Any       					=			torch.load(UpperCAmelCase_      ,  map_location="""cpu"""	)
    else:
        _a							:       Dict       					=			torch.load(args.path_to_checkpoint      ,  map_location="""cpu"""	)

    if args.config_file == "":
        # Default config of megatron-bert 345m
        _a							:       Any       					=			MegatronBertConfig()

        # different megatron-bert-*-345m models have different vocab sizes, so override the default
        # config (which is for megatron-bert-cased-345m) with the actual vocab dimension
        _a							:       List[str]       					=			input_state_dict["""model"""]["""lm_head"""]["""bias"""].numel()
    else:
        _a							:       List[str]       					=			MegatronBertConfig.from_json_file(args.config_file	)

    # Convert.
    print("""Converting"""	)
    _a							:       Optional[int]       					=			convert_megatron_checkpoint(UpperCAmelCase_      ,  UpperCAmelCase_      ,  UpperCAmelCase_	)

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

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

    # Store the state_dict to file.
    _a							:       Union[str, Any]       					=			os.path.join(UpperCAmelCase_      ,  """pytorch_model.bin"""	)
    print(F'''Saving checkpoint to "{output_checkpoint_file}"'''	)
    torch.save(UpperCAmelCase_      ,  UpperCAmelCase_	)


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

if __name__ == "__main__":
  main()

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








'''simple docstring'''





from __future__ import annotations

from enum import Enum




class     UpperCamelCase					(  snake_case__					):
  a    :Optional[int]      					=				2_4
  a    :Dict      					=				2_1
  a    :Dict      					=				1_8
  a    :int      					=				1_5
  a    :List[str]      					=				1_2
  a    :Tuple      					=				9
  a    :Optional[int]      					=				6
  a    :str      					=				3
  a    :Dict      					=				2
  a    :int      					=				1
  a    :List[Any]      					=				-1
  a    :Optional[Any]      					=				-2
  a    :Union[str, Any]      					=				-3
  a    :int      					=				-6
  a    :Union[str, Any]      					=				-9
  a    :Dict      					=				-1_2
  a    :Any      					=				-1_5
  a    :Tuple      					=				-1_8
  a    :Optional[Any]      					=				-2_1
  a    :Union[str, Any]      					=				-2_4






class     UpperCamelCase					(  snake_case__					):
  a    :Optional[Any]      					=				8
  a    :Optional[int]      					=				7
  a    :Optional[int]      					=				6
  a    :List[str]      					=				5
  a    :Union[str, Any]      					=				4
  a    :Union[str, Any]      					=				3
  a    :str      					=				2
  a    :Optional[int]      					=				1






def UpperCAmelCase						(					UpperCAmelCase_      ,  UpperCAmelCase_      ,  UpperCAmelCase_      ,  ):
    if isinstance(UpperCAmelCase_      ,  UpperCAmelCase_	):
        _a							:       int       					=			SIUnit[known_prefix.lower()]
    if isinstance(UpperCAmelCase_      ,  UpperCAmelCase_	):
        _a							:       Optional[Any]       					=			SIUnit[unknown_prefix.lower()]
    _a							:       float       					=			known_amount * (
        10 ** (known_prefix.value - unknown_prefix.value)
    )
    return unknown_amount





def UpperCAmelCase						(					UpperCAmelCase_      ,  UpperCAmelCase_      ,  UpperCAmelCase_      ,  ):
    if isinstance(UpperCAmelCase_      ,  UpperCAmelCase_	):
        _a							:       Optional[Any]       					=			BinaryUnit[known_prefix.lower()]
    if isinstance(UpperCAmelCase_      ,  UpperCAmelCase_	):
        _a							:       Dict       					=			BinaryUnit[unknown_prefix.lower()]
    _a							:       float       					=			known_amount * (
        2 ** ((known_prefix.value - unknown_prefix.value) * 10)
    )
    return unknown_amount


if __name__ == "__main__":
  import doctest

  doctest.testmod()


