


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






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE				:       Optional[int]						,				_SCREAMING_SNAKE_CASE				:       str						,				_SCREAMING_SNAKE_CASE				:       List[str]       ):
					# The converted output model.
					__a      :	str    =  {}

					# old versions did not store training args
					__a      :	Tuple    =  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))

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

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

					# The model.
					__a      :	int    =  input_state_dict['model']
					# The language model.
					__a      :	Dict    =  model['language_model']
					# The embeddings.
					__a      :	Tuple    =  lm['embedding']

					# The word embeddings.
					__a      :	Optional[int]    =  embeddings['word_embeddings']['weight']
					# Truncate the embedding table to vocab_size rows.
					__a      :	List[str]    =  word_embeddings[: config.vocab_size, :]
					__a      :	int    =  word_embeddings

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

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

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

					# The simple map of names for "automated" rules.
					__a      :	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.
										__a      :	Tuple    =  layer_re.match(_SCREAMING_SNAKE_CASE       )

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

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

										# The name of the layer.
										__a      :	Dict    =  F"""transformer.h.{layer_idx}"""

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

															# Insert a "dummy" tensor for masked_bias.
															__a      :	int    =  torch.tensor(-1e4						,				dtype=torch.floataa       )
															__a      :	Any    =  masked_bias

															__a      :	List[str]    =  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.
															__a      :	int    =  out_val.transpose(0						,				1       ).contiguous()
															# Store.
															__a      :	List[Any]    =  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":
															__a      :	Any    =  fix_query_key_value_ordering(_SCREAMING_SNAKE_CASE						,				_SCREAMING_SNAKE_CASE						,				3						,				_SCREAMING_SNAKE_CASE						,				_SCREAMING_SNAKE_CASE       )
															# Store. No change of shape.
															__a      :	Any    =  out_val

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

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

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

					# The final layernorm.
					__a      :	int    =  transformer['final_layernorm.weight']
					__a      :	int    =  transformer['final_layernorm.bias']

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

					# It should be done!
					return output_state_dict






def   lowerCamelCase							():
					# Create the argument parser.
					__a      :	int    =  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.'						,				)
					__a      :	Tuple    =  parser.parse_args()

					# Extract the basename.
					__a      :	Union[str, 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:
																				__a      :	int    =  torch.load(_SCREAMING_SNAKE_CASE						,				map_location='cpu'       )
					else:
										__a      :	str    =  torch.load(args.path_to_checkpoint						,				map_location='cpu'       )

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

										# Spell out all parameters in case the defaults change.
										__a      :	Tuple    =  GPTaConfig(
										    vocab_size=50_257						,				n_positions=1_024						,				n_embd=1_024						,				n_layer=24						,				n_head=16						,				n_inner=4_096						,				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.0_2						,				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=50_256						,				eos_token_id=50_256						,				)
					else:
										__a      :	Optional[Any]    =  GPTaConfig.from_json_file(args.config_file       )

					__a      :	Dict    =  ['GPT2LMHeadModel']

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

					__a      :	str    =  AutoTokenizer.from_pretrained(_SCREAMING_SNAKE_CASE       )
					__a      :	Union[str, Any]    =  type(_SCREAMING_SNAKE_CASE       ).__name__
					__a      :	str    =  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.
					__a      :	str    =  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 dataclasses import dataclass





@dataclass
class 							__UpperCamelCase    :
	A_           =   42
	A_           =   None
	A_           =   None






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE				:       TreeNode | None       ):

					# Validation
					def is_valid_tree(_SCREAMING_SNAKE_CASE				:       TreeNode | None       ) -> bool:
										if node is None:
															return True

										if not isinstance(_SCREAMING_SNAKE_CASE						,				_SCREAMING_SNAKE_CASE       ):
															return False

										try:
															float(node.data       )
										except (TypeError, ValueError):
															return False

										return is_valid_tree(node.left       ) and is_valid_tree(node.right       )

					if not is_valid_tree(_SCREAMING_SNAKE_CASE       ):
										raise ValueError(
										    'Each node should be type of TreeNode and data should be float.'       )

					def is_binary_search_tree_recursive_check(
					    _SCREAMING_SNAKE_CASE				:       TreeNode | None						,				_SCREAMING_SNAKE_CASE				:       float						,				_SCREAMING_SNAKE_CASE				:       float       ) -> bool:

										if node is None:
															return True

										return (
										    left_bound < node.data < right_bound
										    and is_binary_search_tree_recursive_check(node.left						,				_SCREAMING_SNAKE_CASE						,				node.data       )
										    and is_binary_search_tree_recursive_check(
										        node.right						,				node.data						,				_SCREAMING_SNAKE_CASE       )
										)

					return is_binary_search_tree_recursive_check(_SCREAMING_SNAKE_CASE						,				-float('inf'       )						,				float('inf'       )       )


if __name__ == "__main__":
	import doctest

	doctest.testmod()



