



import argparse

import torch
from datasets import load_dataset
from donut import DonutModel

from transformers import (
    DonutImageProcessor,
    DonutProcessor,
    DonutSwinConfig,
    DonutSwinModel,
    MBartConfig,
    MBartForCausalLM,
    VisionEncoderDecoderModel,
    XLMRobertaTokenizerFast,
)






def 					snake_case__       (		__lowerCamelCase    :   Tuple     ):
			"""simple docstring"""
			lowerCamelCase__		:		Optional[Any]		     =model.config

			lowerCamelCase__		:		Optional[Any]		     =DonutSwinConfig(
			    image_size=original_config.input_size							,		patch_size=4							,		depths=original_config.encoder_layer							,		num_heads=[4, 8, 16, 32]							,		window_size=original_config.window_size							,		embed_dim=128							,		)
			lowerCamelCase__		:		int		     =MBartConfig(
			    is_decoder=__lowerCamelCase							,		is_encoder_decoder=__lowerCamelCase							,		add_cross_attention=__lowerCamelCase							,		decoder_layers=original_config.decoder_layer							,		max_position_embeddings=original_config.max_position_embeddings							,		vocab_size=len(
			        model.decoder.tokenizer     )							,		scale_embedding=__lowerCamelCase							,		add_final_layer_norm=__lowerCamelCase							,		)

			return encoder_config, decoder_config






def 					snake_case__       (		__lowerCamelCase    :   Dict     ):
			"""simple docstring"""
			if "encoder.model" in name:
						lowerCamelCase__		:		List[Any]		     =name.replace('''encoder.model'''							,		'''encoder'''     )
			if "decoder.model" in name:
						lowerCamelCase__		:		Union[str, Any]		     =name.replace('''decoder.model'''							,		'''decoder'''     )
			if "patch_embed.proj" in name:
						lowerCamelCase__		:		str		     =name.replace('''patch_embed.proj'''							,		'''embeddings.patch_embeddings.projection'''     )
			if "patch_embed.norm" in name:
						lowerCamelCase__		:		Dict		     =name.replace('''patch_embed.norm'''							,		'''embeddings.norm'''     )
			if name.startswith('''encoder'''     ):
						if "layers" in name:
									lowerCamelCase__		:		Optional[int]		     ='''encoder.''' + name
						if "attn.proj" in name:
									lowerCamelCase__		:		List[str]		     =name.replace('''attn.proj'''							,		'''attention.output.dense'''     )
						if "attn" in name and "mask" not in name:
									lowerCamelCase__		:		Optional[int]		     =name.replace('''attn'''							,		'''attention.self'''     )
						if "norm1" in name:
									lowerCamelCase__		:		Any		     =name.replace('''norm1'''							,		'''layernorm_before'''     )
						if "norm2" in name:
									lowerCamelCase__		:		Optional[int]		     =name.replace('''norm2'''							,		'''layernorm_after'''     )
						if "mlp.fc1" in name:
									lowerCamelCase__		:		Dict		     =name.replace('''mlp.fc1'''							,		'''intermediate.dense'''     )
						if "mlp.fc2" in name:
									lowerCamelCase__		:		List[str]		     =name.replace('''mlp.fc2'''							,		'''output.dense'''     )

						if name == "encoder.norm.weight":
									lowerCamelCase__		:		Tuple		     ='''encoder.layernorm.weight'''
						if name == "encoder.norm.bias":
									lowerCamelCase__		:		Optional[int]		     ='''encoder.layernorm.bias'''

			return name






def 					snake_case__       (		__lowerCamelCase    :   List[str]							,		__lowerCamelCase    :   str     ):
			"""simple docstring"""
			for key in orig_state_dict.copy().keys():
						lowerCamelCase__		:		Union[str, Any]		     =orig_state_dict.pop(__lowerCamelCase     )

						if "qkv" in key:
									lowerCamelCase__		:		List[str]		     =key.split('''.'''     )
									lowerCamelCase__		:		Optional[Any]		     =int(key_split[3]     )
									lowerCamelCase__		:		Optional[Any]		     =int(key_split[5]     )
									lowerCamelCase__		:		Union[str, Any]		     =model.encoder.encoder.layers[layer_num].blocks[block_num].attention.self.all_head_size

									if "weight" in key:
												lowerCamelCase__		:		Dict		     =val[:dim, :]
												lowerCamelCase__		:		List[str]		     =val[dim : dim * 2, :]
												lowerCamelCase__		:		Optional[Any]		     =val[-dim:, :]
									else:
												lowerCamelCase__		:		List[Any]		     =val[:dim]
												lowerCamelCase__		:		List[str]		     =val[dim : dim * 2]
												lowerCamelCase__		:		Union[str, Any]		     =val[-dim:]
						elif "attn_mask" in key or key in ["encoder.model.norm.weight", "encoder.model.norm.bias"]:
									# HuggingFace implementation doesn't use attn_mask buffer
									# and model doesn't use final LayerNorms for the encoder
									pass
						else:
									lowerCamelCase__		:		Union[str, Any]		     =val

			return orig_state_dict






def 					snake_case__       (		__lowerCamelCase    :   Tuple							,		__lowerCamelCase    :   List[Any]=None							,		__lowerCamelCase    :   str=False     ):
			"""simple docstring"""
			lowerCamelCase__		:		int		     =DonutModel.from_pretrained(__lowerCamelCase     ).eval()

			# load HuggingFace model
			lowerCamelCase__		:		List[str]		     =get_configs(__lowerCamelCase     )
			lowerCamelCase__		:		Dict		     =DonutSwinModel(__lowerCamelCase     )
			lowerCamelCase__		:		List[str]		     =MBartForCausalLM(__lowerCamelCase     )
			lowerCamelCase__		:		Tuple		     =VisionEncoderDecoderModel(encoder=__lowerCamelCase							,		decoder=__lowerCamelCase     )
			model.eval()

			lowerCamelCase__		:		List[str]		     =original_model.state_dict()
			lowerCamelCase__		:		Optional[int]		     =convert_state_dict(__lowerCamelCase							,		__lowerCamelCase     )
			model.load_state_dict(__lowerCamelCase     )

			# verify results on scanned document
			lowerCamelCase__		:		int		     =load_dataset('''hf-internal-testing/example-documents'''     )
			lowerCamelCase__		:		List[str]		     =dataset['''test'''][0]['''image'''].convert('''RGB'''     )

			lowerCamelCase__		:		Optional[int]		     =XLMRobertaTokenizerFast.from_pretrained(__lowerCamelCase							,		from_slow=__lowerCamelCase     )
			lowerCamelCase__		:		int		     =DonutImageProcessor(
			    do_align_long_axis=original_model.config.align_long_axis							,		size=original_model.config.input_size[::-1]     )
			lowerCamelCase__		:		Any		     =DonutProcessor(__lowerCamelCase							,		__lowerCamelCase     )
			lowerCamelCase__		:		List[Any]		     =processor(__lowerCamelCase							,		return_tensors='''pt'''     ).pixel_values

			if model_name == "naver-clova-ix/donut-base-finetuned-docvqa":
						lowerCamelCase__		:		Optional[Any]		     ='''<s_docvqa><s_question>{user_input}</s_question><s_answer>'''
						lowerCamelCase__		:		Optional[Any]		     ='''When is the coffee break?'''
						lowerCamelCase__		:		List[str]		     =task_prompt.replace('''{user_input}'''							,		__lowerCamelCase     )
			elif model_name == "naver-clova-ix/donut-base-finetuned-rvlcdip":
						lowerCamelCase__		:		Any		     ='''<s_rvlcdip>'''
			elif model_name in [
			    "naver-clova-ix/donut-base-finetuned-cord-v1",
			    "naver-clova-ix/donut-base-finetuned-cord-v1-2560",
			]:
						lowerCamelCase__		:		Any		     ='''<s_cord>'''
			elif model_name == "naver-clova-ix/donut-base-finetuned-cord-v2":
						lowerCamelCase__		:		Dict		     ='''s_cord-v2>'''
			elif model_name == "naver-clova-ix/donut-base-finetuned-zhtrainticket":
						lowerCamelCase__		:		List[Any]		     ='''<s_zhtrainticket>'''
			elif model_name in ["naver-clova-ix/donut-proto", "naver-clova-ix/donut-base"]:
						# use a random prompt
						lowerCamelCase__		:		int		     ='''hello world'''
			else:
						raise ValueError('''Model name not supported'''     )
			lowerCamelCase__		:		str		     =original_model.decoder.tokenizer(__lowerCamelCase							,		add_special_tokens=__lowerCamelCase							,		return_tensors='''pt'''     )[
			    '''input_ids'''
			]

			lowerCamelCase__		:		Optional[Any]		     =original_model.encoder.model.patch_embed(__lowerCamelCase     )
			lowerCamelCase__		:		Dict		     =model.encoder.embeddings(__lowerCamelCase     )
			assert torch.allclose(__lowerCamelCase							,		__lowerCamelCase							,		atol=1e-3     )

			# verify encoder hidden states
			lowerCamelCase__		:		Optional[int]		     =original_model.encoder(__lowerCamelCase     )
			lowerCamelCase__		:		Tuple		     =model.encoder(__lowerCamelCase     ).last_hidden_state
			assert torch.allclose(__lowerCamelCase							,		__lowerCamelCase							,		atol=1e-2     )

			# verify decoder hidden states
			lowerCamelCase__		:		str		     =original_model(__lowerCamelCase							,		__lowerCamelCase							,		__lowerCamelCase     ).logits
			lowerCamelCase__		:		Any		     =model(__lowerCamelCase							,		decoder_input_ids=__lowerCamelCase     ).logits
			assert torch.allclose(__lowerCamelCase							,		__lowerCamelCase							,		atol=1e-3     )
			print('''Looks ok!'''     )

			if pytorch_dump_folder_path is not None:
						print(f'''Saving model and processor to {pytorch_dump_folder_path}'''     )
						model.save_pretrained(__lowerCamelCase     )
						processor.save_pretrained(__lowerCamelCase     )

			if push_to_hub:
						model.push_to_hub('''nielsr/''' + model_name.split('''/'''     )[-1]							,		commit_message='''Update model'''     )
						processor.push_to_hub('''nielsr/''' + model_name.split('''/'''     )[-1]							,		commit_message='''Update model'''     )


if __name__ == "__main__":
		_lowercase			:			Dict										=						argparse.ArgumentParser()
		# Required parameters
		parser.add_argument(
		    "--model_name",
		    default="naver-clova-ix/donut-base-finetuned-docvqa",
		    required=False,
		    type=str,
		    help="Name of the original model you'd like to convert.",
		)
		parser.add_argument(
		    "--pytorch_dump_folder_path",
		    default=None,
		    required=False,
		    type=str,
		    help="Path to the output PyTorch model directory.",
		)
		parser.add_argument(
		    "--push_to_hub",
		    action="store_true",
		    help="Whether or not to push the converted model and processor to the 🤗 hub.",
		)

		_lowercase			:			int										=						parser.parse_args()
		convert_donut_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)









"""simple docstring"""






import numpy as np
from PIL import Image






def 					snake_case__       (		__lowerCamelCase    :   np.ndarray							,		__lowerCamelCase    :   int							,		__lowerCamelCase    :   int     ):
			"""simple docstring"""
			lowerCamelCase__		:		List[Any]		     =np.array(__lowerCamelCase     )
			if arr.shape[0] != arr.shape[1]:
						raise ValueError('''The input array is not a square matrix'''     )
			lowerCamelCase__		:		int		     =0
			lowerCamelCase__		:		int		     =0
			lowerCamelCase__		:		Optional[int]		     =0
			lowerCamelCase__		:		List[Any]		     =0

			# compute the shape of the output matrix
			lowerCamelCase__		:		Union[str, Any]		     =(arr.shape[0] - size) // stride + 1
			# initialize the output matrix with zeros of shape maxpool_shape
			lowerCamelCase__		:		Union[str, Any]		     =np.zeros((maxpool_shape, maxpool_shape)     )

			while i < arr.shape[0]:
						if i + size > arr.shape[0]:
									# if the end of the matrix is reached, break
									break
						while j < arr.shape[1]:
									# if the end of the matrix is reached, break
									if j + size > arr.shape[1]:
												break
									# compute the maximum of the pooling matrix
									lowerCamelCase__		:		str		     =np.max(arr[i : i + size, j : j + size]     )
									# shift the pooling matrix by stride of column pixels
									j += stride
									mat_j += 1

						# shift the pooling matrix by stride of row pixels
						i += stride
						mat_i += 1

						# reset the column index to 0
						lowerCamelCase__		:		Optional[int]		     =0
						lowerCamelCase__		:		Optional[int]		     =0

			return updated_arr






def 					snake_case__       (		__lowerCamelCase    :   np.ndarray							,		__lowerCamelCase    :   int							,		__lowerCamelCase    :   int     ):
			"""simple docstring"""
			lowerCamelCase__		:		List[Any]		     =np.array(__lowerCamelCase     )
			if arr.shape[0] != arr.shape[1]:
						raise ValueError('''The input array is not a square matrix'''     )
			lowerCamelCase__		:		str		     =0
			lowerCamelCase__		:		List[Any]		     =0
			lowerCamelCase__		:		Optional[int]		     =0
			lowerCamelCase__		:		List[Any]		     =0

			# compute the shape of the output matrix
			lowerCamelCase__		:		Dict		     =(arr.shape[0] - size) // stride + 1
			# initialize the output matrix with zeros of shape avgpool_shape
			lowerCamelCase__		:		Optional[Any]		     =np.zeros((avgpool_shape, avgpool_shape)     )

			while i < arr.shape[0]:
						# if the end of the matrix is reached, break
						if i + size > arr.shape[0]:
									break
						while j < arr.shape[1]:
									# if the end of the matrix is reached, break
									if j + size > arr.shape[1]:
												break
									# compute the average of the pooling matrix
									lowerCamelCase__		:		Optional[int]		     =int(np.average(arr[i : i + size, j : j + size]     )     )
									# shift the pooling matrix by stride of column pixels
									j += stride
									mat_j += 1

						# shift the pooling matrix by stride of row pixels
						i += stride
						mat_i += 1
						# reset the column index to 0
						lowerCamelCase__		:		Optional[Any]		     =0
						lowerCamelCase__		:		int		     =0

			return updated_arr


# Main Function
if __name__ == "__main__":
		from doctest import testmod

		testmod(name="avgpooling", verbose=True)

		# Loading the image
		_lowercase			:			int										=						Image.open("path_to_image")

		# Converting the image to numpy array and maxpooling, displaying the result
		# Ensure that the image is a square matrix

		Image.fromarray(maxpooling(np.array(image), size=3, stride=2)).show()

		# Converting the image to numpy array and averagepooling, displaying the result
		# Ensure that the image is a square matrix

		Image.fromarray(avgpooling(np.array(image), size=3, stride=2)).show()




