




import math
from typing import Any, Callable, List, Optional, Tuple, Union

import numpy as np
import torch

from ...models import TaFilmDecoder
from ...schedulers import DDPMScheduler
from ...utils import is_onnx_available, logging, randn_tensor


if is_onnx_available():
      from ..onnx_utils import OnnxRuntimeModel

from ..pipeline_utils import AudioPipelineOutput, DiffusionPipeline
from .continous_encoder import SpectrogramContEncoder
from .notes_encoder import SpectrogramNotesEncoder


__magic_name__  :  Dict					=       logging.get_logger(__name__)  # pylint: disable=invalid-name

__magic_name__  :  Optional[int]					=       2_56





class     SCREAMING_SNAKE_CASE__		(_a			):
     lowercase_ :      List[str]				= ["melgan"]
     def __init__(						self	:		int       ,						__lowerCamelCase	:		SpectrogramNotesEncoder       ,						__lowerCamelCase	:		SpectrogramContEncoder       ,						__lowerCamelCase	:		TaFilmDecoder       ,						__lowerCamelCase	:		DDPMScheduler       ,						__lowerCamelCase	:		OnnxRuntimeModel if is_onnx_available() else Any       ,						):



       """simple docstring"""
       super().__init__()

       # From MELGAN
       lowerCAmelCase__										=      math.log(1e-5						)  # Matches MelGAN training.
       lowerCAmelCase__										=      4.0  # Largest value for most examples
       lowerCAmelCase__										=      1_28

       self.register_modules(
           notes_encoder=__lowerCamelCase       ,						continuous_encoder=__lowerCamelCase       ,						decoder=__lowerCamelCase       ,						scheduler=__lowerCamelCase       ,						melgan=__lowerCamelCase       ,						)
     def        A__      (						self	:		str       ,						__lowerCamelCase	:		Tuple       ,						__lowerCamelCase	:		int=(-1.0, 1.0)       ,						__lowerCamelCase	:		List[Any]=False						):



       """simple docstring"""
       lowerCAmelCase__				,      lowerCAmelCase__										=      output_range
       if clip:
         lowerCAmelCase__										=      torch.clip(__lowerCamelCase       ,						self.min_value       ,						self.max_value						)
       # Scale to [0, 1].
       lowerCAmelCase__										=      (features - self.min_value) / (self.max_value - self.min_value)
       # Scale to [min_out, max_out].
       return zero_one * (max_out - min_out) + min_out
     def        A__      (						self	:		Optional[Any]       ,						__lowerCamelCase	:		List[Any]       ,						__lowerCamelCase	:		List[str]=(-1.0, 1.0)       ,						__lowerCamelCase	:		List[Any]=False						):



       """simple docstring"""
       lowerCAmelCase__				,      lowerCAmelCase__										=      input_range
       lowerCAmelCase__										=      torch.clip(__lowerCamelCase       ,						__lowerCamelCase       ,						__lowerCamelCase						) if clip else outputs
       # Scale to [0, 1].
       lowerCAmelCase__										=      (outputs - min_out) / (max_out - min_out)
       # Scale to [self.min_value, self.max_value].
       return zero_one * (self.max_value - self.min_value) + self.min_value
     def        A__      (						self	:		List[Any]       ,						__lowerCamelCase	:		Optional[Any]       ,						__lowerCamelCase	:		Optional[int]       ,						__lowerCamelCase	:		List[str]						):



       """simple docstring"""
       lowerCAmelCase__										=      input_tokens > 0
       lowerCAmelCase__				,      lowerCAmelCase__										=      self.notes_encoder(
           encoder_input_tokens=__lowerCamelCase       ,						encoder_inputs_mask=__lowerCamelCase						)

       lowerCAmelCase__				,      lowerCAmelCase__										=      self.continuous_encoder(
           encoder_inputs=__lowerCamelCase       ,						encoder_inputs_mask=__lowerCamelCase						)

       return [(tokens_encoded, tokens_mask), (continuous_encoded, continuous_mask)]
     def        A__      (						self	:		Dict       ,						__lowerCamelCase	:		int       ,						__lowerCamelCase	:		Dict       ,						__lowerCamelCase	:		Union[str, Any]						):



       """simple docstring"""
       lowerCAmelCase__										=      noise_time
       if not torch.is_tensor(__lowerCamelCase						):
         lowerCAmelCase__										=      torch.tensor([timesteps]       ,						dtype=torch.long       ,						device=input_tokens.device						)
       elif torch.is_tensor(__lowerCamelCase						) and len(timesteps.shape						) == 0:
         lowerCAmelCase__										=      timesteps[None].to(input_tokens.device						)

       # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
       lowerCAmelCase__										=      timesteps * torch.ones(input_tokens.shape[0]       ,						dtype=timesteps.dtype       ,						device=timesteps.device						)

       lowerCAmelCase__										=      self.decoder(
           encodings_and_masks=__lowerCamelCase       ,						decoder_input_tokens=__lowerCamelCase       ,						decoder_noise_time=__lowerCamelCase						)
       return logits



     @torch.no_grad()
     def __call__(						self	:		Tuple       ,						__lowerCamelCase	:		List[List[int]]       ,						__lowerCamelCase	:		Optional[torch.Generator] = None       ,						__lowerCamelCase	:		int = 1_00       ,						__lowerCamelCase	:		bool = True       ,						__lowerCamelCase	:		str = "numpy"       ,						__lowerCamelCase	:		Optional[Callable[[int, int, torch.FloatTensor], None]] = None       ,						__lowerCamelCase	:		int = 1       ,						):



       """simple docstring"""
       if (callback_steps is None) or (
           callback_steps is not None and (not isinstance(__lowerCamelCase       ,						__lowerCamelCase						) or callback_steps <= 0)
       ):
         raise ValueError(
             F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
             F""" {type(__lowerCamelCase						)}."""						)

       lowerCAmelCase__										=      np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]       ,						dtype=np.floataa						)
       lowerCAmelCase__										=      np.zeros([1, 0, self.n_dims]       ,						np.floataa						)
       lowerCAmelCase__										=      torch.ones((1, TARGET_FEATURE_LENGTH)       ,						dtype=__lowerCamelCase       ,						device=self.device						)

       for i, encoder_input_tokens in enumerate(__lowerCamelCase						):
         if i == 0:
           lowerCAmelCase__										=      torch.from_numpy(pred_mel[:1].copy()						).to(
               device=self.device       ,						dtype=self.decoder.dtype						)
           # The first chunk has no previous context.
           lowerCAmelCase__										=      torch.zeros((1, TARGET_FEATURE_LENGTH)       ,						dtype=__lowerCamelCase       ,						device=self.device						)
         else:
           # The full song pipeline does not feed in a context feature, so the mask
           # will be all 0s after the feature converter. Because we know we're
           # feeding in a full context chunk from the previous prediction, set it
           # to all 1s.
           lowerCAmelCase__										=      ones

         lowerCAmelCase__										=      self.scale_features(
             __lowerCamelCase       ,						output_range=[-1.0, 1.0]       ,						clip=__lowerCamelCase						)

         lowerCAmelCase__										=      self.encode(
             input_tokens=torch.IntTensor([encoder_input_tokens]						).to(device=self.device						)       ,						continuous_inputs=__lowerCamelCase       ,						continuous_mask=__lowerCamelCase       ,						)

         # Sample encoder_continuous_inputs shaped gaussian noise to begin loop
         lowerCAmelCase__										=      randn_tensor(
             shape=encoder_continuous_inputs.shape       ,						generator=__lowerCamelCase       ,						device=self.device       ,						dtype=self.decoder.dtype       ,						)

         # set step values
         self.scheduler.set_timesteps(__lowerCamelCase						)

         # Denoising diffusion loop
         for j, t in enumerate(self.progress_bar(self.scheduler.timesteps						)						):
           lowerCAmelCase__										=      self.decode(
               encodings_and_masks=__lowerCamelCase       ,						input_tokens=__lowerCamelCase       ,						noise_time=t / self.scheduler.config.num_train_timesteps       ,						)

           # Compute previous output: x_t -> x_t-1
           lowerCAmelCase__										=      self.scheduler.step(__lowerCamelCase       ,						__lowerCamelCase       ,						__lowerCamelCase       ,						generator=__lowerCamelCase						).prev_sample

         lowerCAmelCase__										=      self.scale_to_features(__lowerCamelCase       ,						input_range=[-1.0, 1.0]						)
         lowerCAmelCase__										=      mel[:1]
         lowerCAmelCase__										=      mel.cpu().float().numpy()

         lowerCAmelCase__										=      np.concatenate([full_pred_mel, pred_mel[:1]]       ,						axis=1						)

         # call the callback, if provided
         if callback is not None and i % callback_steps == 0:
           callback(__lowerCamelCase       ,						__lowerCamelCase						)

         logger.info('''Generated segment'''       ,						__lowerCamelCase						)

       if output_type == "numpy" and not is_onnx_available():
         raise ValueError(
             '''Cannot return output in \'np\' format if ONNX is not available. Make sure to have ONNX installed or set \'output_type\' to \'mel\'.'''						)
       elif output_type == "numpy" and self.melgan is None:
         raise ValueError(
             '''Cannot return output in \'np\' format if melgan component is not defined. Make sure to define `self.melgan` or set \'output_type\' to \'mel\'.'''						)

       if output_type == "numpy":
         lowerCAmelCase__										=      self.melgan(input_features=full_pred_mel.astype(np.floataa						)						)
       else:
         lowerCAmelCase__										=      full_pred_mel

       if not return_dict:
         return (output,)

       return AudioPipelineOutput(audios=__lowerCamelCase						)







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 		a_							( __lowerCAmelCase				):
  lowerCAmelCase__										=      model.config

  lowerCAmelCase__										=      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=1_28   , )
  lowerCAmelCase__										=      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 		a_							( __lowerCAmelCase				):
  if "encoder.model" in name:
    lowerCAmelCase__										=      name.replace('''encoder.model'''   , '''encoder'''				)
  if "decoder.model" in name:
    lowerCAmelCase__										=      name.replace('''decoder.model'''   , '''decoder'''				)
  if "patch_embed.proj" in name:
    lowerCAmelCase__										=      name.replace('''patch_embed.proj'''   , '''embeddings.patch_embeddings.projection'''				)
  if "patch_embed.norm" in name:
    lowerCAmelCase__										=      name.replace('''patch_embed.norm'''   , '''embeddings.norm'''				)
  if name.startswith('''encoder'''				):
    if "layers" in name:
      lowerCAmelCase__										=      '''encoder.''' + name
    if "attn.proj" in name:
      lowerCAmelCase__										=      name.replace('''attn.proj'''   , '''attention.output.dense'''				)
    if "attn" in name and "mask" not in name:
      lowerCAmelCase__										=      name.replace('''attn'''   , '''attention.self'''				)
    if "norm1" in name:
      lowerCAmelCase__										=      name.replace('''norm1'''   , '''layernorm_before'''				)
    if "norm2" in name:
      lowerCAmelCase__										=      name.replace('''norm2'''   , '''layernorm_after'''				)
    if "mlp.fc1" in name:
      lowerCAmelCase__										=      name.replace('''mlp.fc1'''   , '''intermediate.dense'''				)
    if "mlp.fc2" in name:
      lowerCAmelCase__										=      name.replace('''mlp.fc2'''   , '''output.dense'''				)

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

  return name
def 		a_							( __lowerCAmelCase   , __lowerCAmelCase				):
  for key in orig_state_dict.copy().keys():
    lowerCAmelCase__										=      orig_state_dict.pop(__lowerCAmelCase				)

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

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

  return orig_state_dict
def 		a_							( __lowerCAmelCase   , __lowerCAmelCase=None   , __lowerCAmelCase=False				):
  # load original model
  lowerCAmelCase__										=      DonutModel.from_pretrained(__lowerCAmelCase				).eval()

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

  lowerCAmelCase__										=      original_model.state_dict()
  lowerCAmelCase__										=      convert_state_dict(__lowerCAmelCase   , __lowerCAmelCase				)
  model.load_state_dict(__lowerCAmelCase				)

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

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

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

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

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

  # verify decoder hidden states
  lowerCAmelCase__										=      original_model(__lowerCAmelCase   , __lowerCAmelCase   , __lowerCAmelCase				).logits
  lowerCAmelCase__										=      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__":
      __magic_name__  :  Union[str, Any]					=       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.""",
      )

      __magic_name__  :  Any					=       parser.parse_args()
      convert_donut_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)

