

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

      SCREAMING_SNAKE_CASE :      List[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						,    )
      SCREAMING_SNAKE_CASE :      List[Any]       =   MBartConfig(
          is_decoder=_A						,    is_encoder_decoder=_A						,    add_cross_attention=_A						,    decoder_layers=original_config.decoder_layer						,    max_position_embeddings=original_config.max_position_embeddings						,    vocab_size=len(
              model.decoder.tokenizer			)						,    scale_embedding=_A						,    add_final_layer_norm=_A						,    )

      return encoder_config, decoder_config






def 					__lowercase      (     _A			)     ->						Union[str, Any]:







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

            if name == "encoder.norm.weight":
                  SCREAMING_SNAKE_CASE :      str       =   """encoder.layernorm.weight"""
            if name == "encoder.norm.bias":
                  SCREAMING_SNAKE_CASE :      Dict       =   """encoder.layernorm.bias"""

      return name






def 					__lowercase      (     _A						,    _A			)     ->						Optional[Any]:







      for key in orig_state_dict.copy().keys():
            SCREAMING_SNAKE_CASE :      List[str]       =   orig_state_dict.pop(_A			)

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

                  if "weight" in key:
                        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   val[:dim, :]
                        SCREAMING_SNAKE_CASE :      Optional[int]       =   val[dim : dim * 2, :]
                        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   val[-dim:, :]
                  else:
                        SCREAMING_SNAKE_CASE :      List[Any]       =   val[:dim]
                        SCREAMING_SNAKE_CASE :      Optional[Any]       =   val[dim : dim * 2]
                        SCREAMING_SNAKE_CASE :      Dict       =   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:
                  SCREAMING_SNAKE_CASE :      List[str]       =   val

      return orig_state_dict






def 					__lowercase      (     _A						,    _A=None						,    _A=False			)     ->						List[Any]:
      SCREAMING_SNAKE_CASE :      Dict       =   DonutModel.from_pretrained(_A			).eval()

      # load HuggingFace model
      SCREAMING_SNAKE_CASE :      List[str]       =   get_configs(_A			)
      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   DonutSwinModel(_A			)
      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   MBartForCausalLM(_A			)
      SCREAMING_SNAKE_CASE :      Any       =   VisionEncoderDecoderModel(encoder=_A						,    decoder=_A			)
      model.eval()

      SCREAMING_SNAKE_CASE :      List[Any]       =   original_model.state_dict()
      SCREAMING_SNAKE_CASE :      List[Any]       =   convert_state_dict(_A						,    _A			)
      model.load_state_dict(_A			)

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

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

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

      SCREAMING_SNAKE_CASE :      Tuple       =   original_model.encoder.model.patch_embed(_A			)
      SCREAMING_SNAKE_CASE :      str       =   model.encoder.embeddings(_A			)
      assert torch.allclose(_A						,    _A						,    atol=1e-3			)

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

      # verify decoder hidden states
      SCREAMING_SNAKE_CASE :      int       =   original_model(_A						,    _A						,    _A			).logits
      SCREAMING_SNAKE_CASE :      Dict       =   model(_A						,    decoder_input_ids=_A			).logits
      assert torch.allclose(_A						,    _A						,    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(_A			)
            processor.save_pretrained(_A			)

      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__":
       UpperCAmelCase__ :     str				    =    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.""",
       )

       UpperCAmelCase__ :     str				    =    parser.parse_args()
       convert_donut_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)







import importlib
import inspect
import json
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Dict, Optional, Union
from urllib import request

from huggingface_hub import HfFolder, cached_download, hf_hub_download, model_info
from packaging import version

from .. import __version__
from . import DIFFUSERS_DYNAMIC_MODULE_NAME, HF_MODULES_CACHE, logging


UpperCAmelCase__ :     List[str]				    =    (
    """https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py"""
)


UpperCAmelCase__ :     Union[str, Any]				    =    logging.get_logger(__name__)  # pylint: disable=invalid-name






def 					__lowercase      (     )     ->						List[Any]:
      SCREAMING_SNAKE_CASE :      Optional[int]       =   """https://pypi.org/pypi/diffusers/json"""
      SCREAMING_SNAKE_CASE :      Dict       =   json.loads(request.urlopen(_A			).read()			)["""releases"""].keys()
      return sorted(_A						,    key=lambda _A			: version.Version(_A			)			)






def 					__lowercase      (     )     ->						Any:
      # This function has already been executed if HF_MODULES_CACHE already is in the Python path.
      if HF_MODULES_CACHE in sys.path:
            return

      sys.path.append(_A			)
      os.makedirs(_A						,    exist_ok=_A			)
      SCREAMING_SNAKE_CASE :      Dict       =   Path(_A			) / """__init__.py"""
      if not init_path.exists():
            init_path.touch()






def 					__lowercase      (     _A			)     ->						Any:
      init_hf_modules()
      SCREAMING_SNAKE_CASE :      Any       =   Path(_A			) / name
      # If the parent module does not exist yet, recursively create it.
      if not dynamic_module_path.parent.exists():
            create_dynamic_module(dynamic_module_path.parent			)
      os.makedirs(_A						,    exist_ok=_A			)
      SCREAMING_SNAKE_CASE :      str       =   dynamic_module_path / """__init__.py"""
      if not init_path.exists():
            init_path.touch()






def 					__lowercase      (     _A			)     ->						Optional[int]:
      with open(_A						,    """r"""						,    encoding="""utf-8"""			) as f:
            SCREAMING_SNAKE_CASE :      Any       =   f.read()

      # Imports of the form `import .xxx`
      SCREAMING_SNAKE_CASE :      List[str]       =   re.findall("""^\s*import\s+\.(\S+)\s*$"""						,    _A						,    flags=re.MULTILINE			)
      # Imports of the form `from .xxx import yyy`
      relative_imports += re.findall("""^\s*from\s+\.(\S+)\s+import"""						,    _A						,    flags=re.MULTILINE			)
      # Unique-ify
      return list(set(_A			)			)






def 					__lowercase      (     _A			)     ->						Optional[Any]:
      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   False
      SCREAMING_SNAKE_CASE :      List[Any]       =   [module_file]
      SCREAMING_SNAKE_CASE :      Any       =   []

      # Let's recurse through all relative imports
      while not no_change:
            SCREAMING_SNAKE_CASE :      List[str]       =   []
            for f in files_to_check:
                  new_imports.extend(get_relative_imports(_A			)			)

            SCREAMING_SNAKE_CASE :      List[str]       =   Path(_A			).parent
            SCREAMING_SNAKE_CASE :      int       =   [str(module_path / m			) for m in new_imports]
            SCREAMING_SNAKE_CASE :      int       =   [f for f in new_import_files if f not in all_relative_imports]
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   [F"{f}.py" for f in new_import_files]

            SCREAMING_SNAKE_CASE :      List[Any]       =   len(_A			) == 0
            all_relative_imports.extend(_A			)

      return all_relative_imports






def 					__lowercase      (     _A			)     ->						Union[str, Any]:
      with open(_A						,    """r"""						,    encoding="""utf-8"""			) as f:
            SCREAMING_SNAKE_CASE :      List[Any]       =   f.read()

      # Imports of the form `import xxx`
      SCREAMING_SNAKE_CASE :      Any       =   re.findall("""^\s*import\s+(\S+)\s*$"""						,    _A						,    flags=re.MULTILINE			)
      # Imports of the form `from xxx import yyy`
      imports += re.findall("""^\s*from\s+(\S+)\s+import"""						,    _A						,    flags=re.MULTILINE			)
      # Only keep the top-level module
      SCREAMING_SNAKE_CASE :      List[Any]       =   [imp.split("""."""			)[0] for imp in imports if not imp.startswith("""."""			)]

      # Unique-ify and test we got them all
      SCREAMING_SNAKE_CASE :      Dict       =   list(set(_A			)			)
      SCREAMING_SNAKE_CASE :      Dict       =   []
      for imp in imports:
            try:
                  importlib.import_module(_A			)
            except ImportError:
                  missing_packages.append(_A			)

      if len(_A			) > 0:
            raise ImportError(
                """This modeling file requires the following packages that were not found in your environment: """
                F"{', '.join(_A			)}. Run `pip install {' '.join(_A			)}`"			)

      return get_relative_imports(_A			)






def 					__lowercase      (     _A						,    _A			)     ->						List[str]:
      SCREAMING_SNAKE_CASE :      int       =   module_path.replace(os.path.sep						,    """."""			)
      SCREAMING_SNAKE_CASE :      List[str]       =   importlib.import_module(_A			)

      if class_name is None:
            return find_pipeline_class(_A			)
      return getattr(_A						,    _A			)






def 					__lowercase      (     _A			)     ->						Optional[int]:
      from ..pipelines import DiffusionPipeline

      SCREAMING_SNAKE_CASE :      Tuple       =   dict(inspect.getmembers(_A						,    inspect.isclass			)			)

      SCREAMING_SNAKE_CASE :      Dict       =   None
      for cls_name, cls in cls_members.items():
            if (
                cls_name != DiffusionPipeline.__name__
                and issubclass(cls						,    _A			)
                and cls.__module__.split("""."""			)[0] != "diffusers"
            ):
                  if pipeline_class is not None:
                        raise ValueError(
                            F"Multiple classes that inherit from {DiffusionPipeline.__name__} have been found:"
                            F" {pipeline_class.__name__}, and {cls_name}. Please make sure to define only one in"
                            F" {loaded_module}."			)
                  SCREAMING_SNAKE_CASE :      List[str]       =   cls

      return pipeline_class






def 					__lowercase      (     _A						,    _A						,    _A = None						,    _A = False						,    _A = False						,    _A = None						,    _A = None						,    _A = None						,    _A = False						,    )     ->						List[str]:
      SCREAMING_SNAKE_CASE :      List[str]       =   str(_A			)

      SCREAMING_SNAKE_CASE :      List[Any]       =   os.path.join(_A						,    _A			)

      if os.path.isfile(_A			):
            SCREAMING_SNAKE_CASE :      Union[str, Any]       =   module_file_or_url
            SCREAMING_SNAKE_CASE :      Dict       =   """local"""
      elif pretrained_model_name_or_path.count("""/"""			) == 0:
            SCREAMING_SNAKE_CASE :      Tuple       =   get_diffusers_versions()
            # cut ".dev0"
            SCREAMING_SNAKE_CASE :      str       =   """v""" + """.""".join(__version__.split("""."""			)[:3]			)

            # retrieve github version that matches
            if revision is None:
                  SCREAMING_SNAKE_CASE :      Any       =   latest_version if latest_version[1:] in available_versions else """main"""
                  logger.info(F"Defaulting to latest_version: {revision}."			)
            elif revision in available_versions:
                  SCREAMING_SNAKE_CASE :      List[Any]       =   F"v{revision}"
            elif revision == "main":
                  SCREAMING_SNAKE_CASE :      Union[str, Any]       =   revision
            else:
                  raise ValueError(
                      F"`custom_revision`: {revision} does not exist. Please make sure to choose one of"
                      F" {', '.join(available_versions + ['main']			)}."			)

            # community pipeline on GitHub
            SCREAMING_SNAKE_CASE :      Tuple       =   COMMUNITY_PIPELINES_URL.format(revision=_A						,    pipeline=_A			)
            try:
                  SCREAMING_SNAKE_CASE :      str       =   cached_download(
                      _A						,    cache_dir=_A						,    force_download=_A						,    proxies=_A						,    resume_download=_A						,    local_files_only=_A						,    use_auth_token=_A						,    )
                  SCREAMING_SNAKE_CASE :      Tuple       =   """git"""
                  SCREAMING_SNAKE_CASE :      Optional[int]       =   pretrained_model_name_or_path + """.py"""
            except EnvironmentError:
                  logger.error(F"Could not locate the {module_file} inside {pretrained_model_name_or_path}."			)
                  raise
      else:
            try:
                  # Load from URL or cache if already cached
                  SCREAMING_SNAKE_CASE :      int       =   hf_hub_download(
                      _A						,    _A						,    cache_dir=_A						,    force_download=_A						,    proxies=_A						,    resume_download=_A						,    local_files_only=_A						,    use_auth_token=_A						,    )
                  SCREAMING_SNAKE_CASE :      Union[str, Any]       =   os.path.join("""local"""						,    """--""".join(pretrained_model_name_or_path.split("""/"""			)			)			)
            except EnvironmentError:
                  logger.error(F"Could not locate the {module_file} inside {pretrained_model_name_or_path}."			)
                  raise

    # Check we have all the requirements in our environment
      SCREAMING_SNAKE_CASE :      int       =   check_imports(_A			)

      # Now we move the module inside our cached dynamic modules.
      SCREAMING_SNAKE_CASE :      List[Any]       =   DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
      create_dynamic_module(_A			)
      SCREAMING_SNAKE_CASE :      Tuple       =   Path(_A			) / full_submodule
      if submodule == "local" or submodule == "git":
            # We always copy local files (we could hash the file to see if there was a change, and give them the name of
            # that hash, to only copy when there is a modification but it seems overkill for now).
            # The only reason we do the copy is to avoid putting too many folders in sys.path.
            shutil.copy(_A						,    submodule_path / module_file			)
            for module_needed in modules_needed:
                  SCREAMING_SNAKE_CASE :      Optional[int]       =   F"{module_needed}.py"
                  shutil.copy(os.path.join(_A						,    _A			)						,    submodule_path / module_needed			)
      else:
            # Get the commit hash
            # TODO: we will get this info in the etag soon, so retrieve it from there and not here.
            if isinstance(_A						,    _A			):
                  SCREAMING_SNAKE_CASE :      Tuple       =   use_auth_token
            elif use_auth_token is True:
                  SCREAMING_SNAKE_CASE :      str       =   HfFolder.get_token()
            else:
                  SCREAMING_SNAKE_CASE :      Optional[int]       =   None

            SCREAMING_SNAKE_CASE :      List[Any]       =   model_info(_A						,    revision=_A						,    token=_A			).sha

            # The module file will end up being placed in a subfolder with the git hash of the repo. This way we get the
            # benefit of versioning.
            SCREAMING_SNAKE_CASE :      int       =   submodule_path / commit_hash
            SCREAMING_SNAKE_CASE :      Tuple       =   full_submodule + os.path.sep + commit_hash
            create_dynamic_module(_A			)

            if not (submodule_path / module_file).exists():
                  shutil.copy(_A						,    submodule_path / module_file			)
            # Make sure we also have every file with relative
            for module_needed in modules_needed:
                  if not (submodule_path / module_needed).exists():
                        get_cached_module_file(
                            _A						,    F"{module_needed}.py"						,    cache_dir=_A						,    force_download=_A						,    resume_download=_A						,    proxies=_A						,    use_auth_token=_A						,    revision=_A						,    local_files_only=_A						,    )
      return os.path.join(_A						,    _A			)







def 					__lowercase      (     _A						,    _A						,    _A = None						,    _A = None						,    _A = False						,    _A = False						,    _A = None						,    _A = None						,    _A = None						,    _A = False						,    **_A						,    )     ->						Tuple:
      SCREAMING_SNAKE_CASE :      Dict       =   get_cached_module_file(
          _A						,    _A						,    cache_dir=_A						,    force_download=_A						,    resume_download=_A						,    proxies=_A						,    use_auth_token=_A						,    revision=_A						,    local_files_only=_A						,    )
      return get_class_in_module(_A						,    final_module.replace(""".py"""						,    """"""			)			)




