





import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


SCREAMING_SNAKE_CASE_   :      int			=   logging.get_logger(__name__)

class 		snake_case_   (							UpperCAmelCase_						):







	'''simple docstring'''
	__UpperCamelCase   =		['''input_features''', '''is_longer''']




	def __init__(						self							:     Dict					,      __lowerCamelCase							:     str=64					,      __lowerCamelCase							:     Any=48_000					,      __lowerCamelCase							:     str=480					,      __lowerCamelCase							:     List[str]=10					,      __lowerCamelCase							:     Tuple=1_024					,      __lowerCamelCase							:     Dict=0.0					,      __lowerCamelCase							:     Dict=False					,      __lowerCamelCase							:     float = 0					,      __lowerCamelCase							:     float = 14_000					,      __lowerCamelCase							:     int = None					,      __lowerCamelCase							:     str = "fusion"					,      __lowerCamelCase							:     str = "repeatpad"					,      **__lowerCamelCase							:     Optional[int]					,      )       -> Dict:

			'''simple docstring'''




			super().__init__(
			    feature_size=__lowerCamelCase					,      sampling_rate=__lowerCamelCase					,      padding_value=__lowerCamelCase					,      return_attention_mask=__lowerCamelCase					,      **__lowerCamelCase					,      )
			__lowercase											=     top_db
			__lowercase											=     truncation
			__lowercase											=     padding
			__lowercase											=     fft_window_size
			__lowercase											=     (fft_window_size >> 1) + 1
			__lowercase											=     hop_length
			__lowercase											=     max_length_s
			__lowercase											=     max_length_s * sampling_rate
			__lowercase											=     sampling_rate
			__lowercase											=     frequency_min
			__lowercase											=     frequency_max
			__lowercase											=     mel_filter_bank(
			    num_frequency_bins=self.nb_frequency_bins					,      num_mel_filters=__lowerCamelCase					,      min_frequency=__lowerCamelCase					,      max_frequency=__lowerCamelCase					,      sampling_rate=__lowerCamelCase					,      norm=__lowerCamelCase					,      mel_scale='htk'					,      )
			__lowercase											=     mel_filter_bank(
			    num_frequency_bins=self.nb_frequency_bins					,      num_mel_filters=__lowerCamelCase					,      min_frequency=__lowerCamelCase					,      max_frequency=__lowerCamelCase					,      sampling_rate=__lowerCamelCase					,      norm='slaney'					,      mel_scale='slaney'					,      )




	def    UpperCAmelCase			(						self							:     Dict   )       -> Dict[str, Any]:

			'''simple docstring'''




			__lowercase											=     copy.deepcopy(self.__dict__   )
			__lowercase											=     self.__class__.__name__
			if "mel_filters" in output:
					del output["mel_filters"]
			if "mel_filters_slaney" in output:
					del output["mel_filters_slaney"]
			return output




	def    UpperCAmelCase			(						self							:     Tuple					,      __lowerCamelCase							:     np.array					,      __lowerCamelCase							:     Optional[np.array] = None   )       -> np.ndarray:

			'''simple docstring'''




			__lowercase											=     spectrogram(
			    __lowerCamelCase					,      window_function(self.fft_window_size					,      'hann'   )					,      frame_length=self.fft_window_size					,      hop_length=self.hop_length					,      power=2.0					,      mel_filters=__lowerCamelCase					,      log_mel='dB'					,      )
			return log_mel_spectrogram.T




	def    UpperCAmelCase			(						self							:     Tuple					,      __lowerCamelCase							:     Any					,      __lowerCamelCase							:     List[str]					,      __lowerCamelCase							:     Dict   )       -> Dict:

			'''simple docstring'''




			__lowercase											=     np.array_split(list(range(0					,      total_frames - chunk_frames + 1   )   )					,      3   )
			if len(ranges[1]   ) == 0:
					# if the audio is too short, we just use the first chunk
					__lowercase											=     [0]
			if len(ranges[2]   ) == 0:
					# if the audio is too short, we just use the first chunk
					__lowercase											=     [0]
			# randomly choose index for each part
			__lowercase											=     np.random.choice(ranges[0]   )
			__lowercase											=     np.random.choice(ranges[1]   )
			__lowercase											=     np.random.choice(ranges[2]   )

			__lowercase											=     mel[idx_front : idx_front + chunk_frames, :]
			__lowercase											=     mel[idx_middle : idx_middle + chunk_frames, :]
			__lowercase											=     mel[idx_back : idx_back + chunk_frames, :]

			__lowercase											=     torch.tensor(mel[None, None, :]   )
			__lowercase											=     torch.nn.functional.interpolate(
			    __lowerCamelCase					,      size=[chunk_frames, 64]					,      mode='bilinear'					,      align_corners=__lowerCamelCase   )
			__lowercase											=     mel_shrink[0][0].numpy()
			__lowercase											=     np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]					,      axis=0   )
			return mel_fusion




	def    UpperCAmelCase			(						self							:     Optional[int]					,      __lowerCamelCase							:     np.array					,      __lowerCamelCase							:     int					,      __lowerCamelCase							:     Tuple					,      __lowerCamelCase							:     str   )       -> np.array:

			'''simple docstring'''




			if waveform.shape[0] > max_length:
					if truncation == "rand_trunc":
							__lowercase											=     True
							# random crop to max_length (for compatibility) -> this should be handled by self.pad
							__lowercase											=     len(__lowerCamelCase   ) - max_length
							__lowercase											=     np.random.randint(0					,      overflow + 1   )
							__lowercase											=     waveform[idx : idx + max_length]
							__lowercase											=     self._np_extract_fbank_features(__lowerCamelCase					,      self.mel_filters_slaney   )[None, :]
					elif truncation == "fusion":
							__lowercase											=     self._np_extract_fbank_features(__lowerCamelCase					,      self.mel_filters   )
							__lowercase											=     max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
							__lowercase											=     mel.shape[0]
							if chunk_frames == total_frames:
									# there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
									# In this case, we just use the whole audio.
									__lowercase											=     np.stack([mel, mel, mel, mel]					,      axis=0   )
									__lowercase											=     False
							else:
									__lowercase											=     self._random_mel_fusion(__lowerCamelCase					,      __lowerCamelCase					,      __lowerCamelCase   )
									__lowercase											=     True
					else:
							raise NotImplementedError(F"data_truncating {truncation} not implemented"   )

			else:
					__lowercase											=     False
					# only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
					if waveform.shape[0] < max_length:
							if padding == "repeat":
									__lowercase											=     int(max_length / len(__lowerCamelCase   )   )
									__lowercase											=     np.stack(np.tile(__lowerCamelCase					,      n_repeat + 1   )   )[:max_length]
							if padding == "repeatpad":
									__lowercase											=     int(max_length / len(__lowerCamelCase   )   )
									__lowercase											=     np.stack(np.tile(__lowerCamelCase					,      __lowerCamelCase   )   )
							__lowercase											=     np.pad(__lowerCamelCase					,      (0, max_length - waveform.shape[0])					,      mode='constant'					,      constant_values=0   )

					if truncation == "fusion":
							__lowercase											=     self._np_extract_fbank_features(__lowerCamelCase					,      self.mel_filters   )
							__lowercase											=     np.stack([input_mel, input_mel, input_mel, input_mel]					,      axis=0   )
					else:
							__lowercase											=     self._np_extract_fbank_features(__lowerCamelCase					,      self.mel_filters_slaney   )[None, :]

			return input_mel, longer






	def __call__(						self							:     int					,      __lowerCamelCase							:     Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]					,      __lowerCamelCase							:     str = None					,      __lowerCamelCase							:     Optional[str] = None					,      __lowerCamelCase							:     Optional[int] = None					,      __lowerCamelCase							:     Optional[int] = None					,      __lowerCamelCase							:     Optional[Union[str, TensorType]] = None					,      **__lowerCamelCase							:     Union[str, Any]					,      )       -> BatchFeature:

			'''simple docstring'''




			__lowercase											=     truncation if truncation is not None else self.truncation
			__lowercase											=     padding if padding else self.padding

			if sampling_rate is not None:
					if sampling_rate != self.sampling_rate:
							raise ValueError(
							    F"The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"
							    F" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"
							    F" was sampled with {self.sampling_rate} and not {sampling_rate}."   )
			else:
					logger.warning(
					    'It is strongly recommended to pass the `sampling_rate` argument to this function. '
					    'Failing to do so can result in silent errors that might be hard to debug.'   )

			__lowercase											=     isinstance(__lowerCamelCase					,      np.ndarray   ) and len(raw_speech.shape   ) > 1
			if is_batched_numpy and len(raw_speech.shape   ) > 2:
					raise ValueError(F"Only mono-channel audio is supported for input to {self}"   )
			__lowercase											=     is_batched_numpy or (
			    isinstance(__lowerCamelCase					,      (list, tuple)   ) and (isinstance(raw_speech[0]					,      (np.ndarray, tuple, list)   ))
			)

			if is_batched:
					__lowercase											=     [np.asarray(__lowerCamelCase					,      dtype=np.floataa   ) for speech in raw_speech]
			elif not is_batched and not isinstance(__lowerCamelCase					,      np.ndarray   ):
					__lowercase											=     np.asarray(__lowerCamelCase					,      dtype=np.floataa   )
			elif isinstance(__lowerCamelCase					,      np.ndarray   ) and raw_speech.dtype is np.dtype(np.floataa   ):
					__lowercase											=     raw_speech.astype(np.floataa   )

			# always return batch
			if not is_batched:
					__lowercase											=     [np.asarray(__lowerCamelCase   )]

			# convert to mel spectrogram, truncate and pad if needed.
			__lowercase											=     [
			    self._get_input_mel(__lowerCamelCase					,      max_length if max_length else self.nb_max_samples					,      __lowerCamelCase					,      __lowerCamelCase   )
			    for waveform in raw_speech
			]

			__lowercase											=     []
			__lowercase											=     []
			for mel, longer in padded_inputs:
					input_mel.append(__lowerCamelCase   )
					is_longer.append(__lowerCamelCase   )

			if truncation == "fusion" and sum(__lowerCamelCase   ) == 0:
					# if no audio is longer than 10s, then randomly select one audio to be longer
					__lowercase											=     np.random.randint(0					,      len(__lowerCamelCase   )   )
					__lowercase											=     True

			if isinstance(input_mel[0]					,      __lowerCamelCase   ):
					__lowercase											=     [np.asarray(__lowerCamelCase					,      dtype=np.floataa   ) for feature in input_mel]

			# is_longer is a list of bool
			__lowercase											=     [[longer] for longer in is_longer]

			__lowercase											=     {'input_features': input_mel, 'is_longer': is_longer}
			__lowercase											=     BatchFeature(__lowerCamelCase   )

			if return_tensors is not None:
					__lowercase											=     input_features.convert_to_tensors(__lowerCamelCase   )

			return input_features











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  SCREAMING_SNAKE_CASE				(      snake_case							)		->    Optional[int]:
		__lowercase											=     model.config

		__lowercase											=     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						,   )
		__lowercase											=     MBartConfig(
		    is_decoder=snake_case						,   is_encoder_decoder=snake_case						,   add_cross_attention=snake_case						,   decoder_layers=original_config.decoder_layer						,   max_position_embeddings=original_config.max_position_embeddings						,   vocab_size=len(
		        model.decoder.tokenizer							)						,   scale_embedding=snake_case						,   add_final_layer_norm=snake_case						,   )

		return encoder_config, decoder_config



def  SCREAMING_SNAKE_CASE				(      snake_case							)		->    Tuple:
		if "encoder.model" in name:
				__lowercase											=     name.replace('encoder.model'						,   'encoder'							)
		if "decoder.model" in name:
				__lowercase											=     name.replace('decoder.model'						,   'decoder'							)
		if "patch_embed.proj" in name:
				__lowercase											=     name.replace('patch_embed.proj'						,   'embeddings.patch_embeddings.projection'							)
		if "patch_embed.norm" in name:
				__lowercase											=     name.replace('patch_embed.norm'						,   'embeddings.norm'							)
		if name.startswith('encoder'							):
				if "layers" in name:
						__lowercase											=     'encoder.' + name
				if "attn.proj" in name:
						__lowercase											=     name.replace('attn.proj'						,   'attention.output.dense'							)
				if "attn" in name and "mask" not in name:
						__lowercase											=     name.replace('attn'						,   'attention.self'							)
				if "norm1" in name:
						__lowercase											=     name.replace('norm1'						,   'layernorm_before'							)
				if "norm2" in name:
						__lowercase											=     name.replace('norm2'						,   'layernorm_after'							)
				if "mlp.fc1" in name:
						__lowercase											=     name.replace('mlp.fc1'						,   'intermediate.dense'							)
				if "mlp.fc2" in name:
						__lowercase											=     name.replace('mlp.fc2'						,   'output.dense'							)

				if name == "encoder.norm.weight":
						__lowercase											=     'encoder.layernorm.weight'
				if name == "encoder.norm.bias":
						__lowercase											=     'encoder.layernorm.bias'

		return name



def  SCREAMING_SNAKE_CASE				(      snake_case						,   snake_case							)		->    List[Any]:
		for key in orig_state_dict.copy().keys():
				__lowercase											=     orig_state_dict.pop(snake_case							)

				if "qkv" in key:
						__lowercase											=     key.split('.'							)
						__lowercase											=     int(key_split[3]							)
						__lowercase											=     int(key_split[5]							)
						__lowercase											=     model.encoder.encoder.layers[layer_num].blocks[block_num].attention.self.all_head_size

						if "weight" in key:
								__lowercase											=     val[:dim, :]
								__lowercase											=     val[dim : dim * 2, :]
								__lowercase											=     val[-dim:, :]
						else:
								__lowercase											=     val[:dim]
								__lowercase											=     val[dim : dim * 2]
								__lowercase											=     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:
						__lowercase											=     val

		return orig_state_dict



def  SCREAMING_SNAKE_CASE				(      snake_case						,   snake_case=None						,   snake_case=False							)		->    Any:
		# load original model
		__lowercase											=     DonutModel.from_pretrained(snake_case							).eval()

		# load HuggingFace model
		__lowercase						,      __lowercase											=     get_configs(snake_case							)
		__lowercase											=     DonutSwinModel(snake_case							)
		__lowercase											=     MBartForCausalLM(snake_case							)
		__lowercase											=     VisionEncoderDecoderModel(encoder=snake_case						,   decoder=snake_case							)
		model.eval()

		__lowercase											=     original_model.state_dict()
		__lowercase											=     convert_state_dict(snake_case						,   snake_case							)
		model.load_state_dict(snake_case							)

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

		__lowercase											=     XLMRobertaTokenizerFast.from_pretrained(snake_case						,   from_slow=snake_case							)
		__lowercase											=     DonutImageProcessor(
		    do_align_long_axis=original_model.config.align_long_axis						,   size=original_model.config.input_size[::-1]							)
		__lowercase											=     DonutProcessor(snake_case						,   snake_case							)
		__lowercase											=     processor(snake_case						,   return_tensors='pt'							).pixel_values

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

		__lowercase											=     original_model.encoder.model.patch_embed(snake_case							)
		__lowercase						,      __lowercase											=     model.encoder.embeddings(snake_case							)
		assert torch.allclose(snake_case						,   snake_case						,   atol=1E-3							)

		# verify encoder hidden states
		__lowercase											=     original_model.encoder(snake_case							)
		__lowercase											=     model.encoder(snake_case							).last_hidden_state
		assert torch.allclose(snake_case						,   snake_case						,   atol=1E-2							)

		# verify decoder hidden states
		__lowercase											=     original_model(snake_case						,   snake_case						,   snake_case							).logits
		__lowercase											=     model(snake_case						,   decoder_input_ids=snake_case							).logits
		assert torch.allclose(snake_case						,   snake_case						,   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(snake_case							)
				processor.save_pretrained(snake_case							)

		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__":
		SCREAMING_SNAKE_CASE_   :      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.''',
		)

		SCREAMING_SNAKE_CASE_   :      Tuple			=   parser.parse_args()
		convert_donut_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)




