


'''simple docstring'''
import argparse
import os
import re
import zipfile

import torch

from transformers import AutoTokenizer, GPTaConfig


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


def 			_lowercase       (				UpperCamelCase__       :					Optional[int],				UpperCamelCase__       :					int,				UpperCamelCase__       :					Union[str, Any]			):
	# The converted output model.
	__A  :		Any       =			{}

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

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

	# The model.
	__A  :		List[str]       =			input_state_dict['model']
	# The language model.
	__A  :		Dict       =			model['language_model']
	# The embeddings.
	__A  :		Tuple       =			lm['embedding']

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

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

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

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

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

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

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

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

		# For layernorm(s), simply store the layer norm.
		if op_name.endswith('layernorm'			):
			__A  :		str       =			'ln_1' if op_name.startswith('input'			) else 'ln_2'
			__A  :		List[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":
			# Insert a tensor of 1x1xDxD bias.
			__A  :		str       =			torch.tril(torch.ones((n_positions, n_positions),				dtype=torch.floataa			)			).view(
			    1,				1,				UpperCamelCase__,				UpperCamelCase__			)
			__A  :		Union[str, Any]       =			causal_mask

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

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

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

		# Copy the bias.
		elif weight_or_bias == "bias":
			__A  :		Tuple       =			megatron_to_transformers[op_name]
			__A  :		Union[str, Any]       =			val

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

	# The final layernorm.
	__A  :		List[Any]       =			transformer['final_layernorm.weight']
	__A  :		Any       =			transformer['final_layernorm.bias']

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

	# It should be done!
	return output_state_dict


def 			_lowercase       (				):
	# Create the argument parser.
	__A  :		List[str]       =			argparse.ArgumentParser()
	parser.add_argument('--print-checkpoint-structure',				action='store_true'			)
	parser.add_argument(
	    'path_to_checkpoint',				type=UpperCamelCase__,				help='Path to the checkpoint file (.zip archive or direct .pt file)',				)
	parser.add_argument(
	    '--config_file',				default='',				type=UpperCamelCase__,				help='An optional config json file describing the pre-trained model.',				)
	__A  :		Optional[Any]       =			parser.parse_args()

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

	__A  :		List[Any]       =			input_state_dict.get('args',				UpperCamelCase__			)

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

		# Spell out all parameters in case the defaults change.
		__A  :		str       =			GPTaConfig(
		    vocab_size=50257,				n_positions=1024,				n_embd=1024,				n_layer=24,				n_head=16,				n_inner=4096,				activation_function=UpperCamelCase__,				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=UpperCamelCase__,				summary_activation=UpperCamelCase__,				summary_proj_to_labels=UpperCamelCase__,				summary_first_dropout=0.1,				scale_attn_weights=UpperCamelCase__,				use_cache=UpperCamelCase__,				bos_token_id=50256,				eos_token_id=50256,				)
	else:
		__A  :		str       =			GPTaConfig.from_json_file(args.config_file			)

	__A  :		Any       =			['GPT2LMHeadModel']

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

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

	# 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  :		Any       =			'gpt2'
		elif tokenizer_type == "PretrainedFromHF":
			__A  :		List[Any]       =			ds_args.tokenizer_name_or_path
		else:
			raise ValueError(f"""Unrecognized tokenizer_type {tokenizer_type}"""			)
	else:
		__A  :		Tuple       =			'gpt2'

	__A  :		Union[str, Any]       =			AutoTokenizer.from_pretrained(UpperCamelCase__			)
	__A  :		Optional[Any]       =			type(UpperCamelCase__			).__name__
	__A  :		int       =			tokenizer_class

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

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

	# Store the state_dict to file.
	__A  :		str       =			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'''
import unittest

from transformers import DonutProcessor


UpperCAmelCase_ :     str								=      'naver-clova-ix/donut-base'

class 						_lowerCamelCase					(    unittest.TestCase				):




 '''simple docstring'''


 def  snake_case__   (      self						):

  """simple docstring"""







  __A  :		List[str]       =			DonutProcessor.from_pretrained(__lowercase						)







 def  snake_case__   (      self						):

  """simple docstring"""







  __A  :		List[str]       =			{
      'name': 'John Doe',
      'age': '99',
      'city': 'Atlanta',
      'state': 'GA',
      'zip': '30301',
      'phone': '123-4567',
      'nicknames': [{'nickname': 'Johnny'}, {'nickname': 'JD'}],
  }

  __A  :		Optional[Any]       =			(
      '<s_name>John Doe</s_name><s_age>99</s_age><s_city>Atlanta</s_city>'
      '<s_state>GA</s_state><s_zip>30301</s_zip><s_phone>123-4567</s_phone>'
      '<s_nicknames><s_nickname>Johnny</s_nickname>'
      '<sep/><s_nickname>JD</s_nickname></s_nicknames>'
  )
  __A  :		List[Any]       =			self.processor.tokenajson(__lowercase						)

  self.assertDictEqual(__lowercase      ,      __lowercase						)






