
import argparse
import json
from pathlib import Path

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import (
    BertTokenizer,
    ViltConfig,
    ViltForImageAndTextRetrieval,
    ViltForImagesAndTextClassification,
    ViltForMaskedLM,
    ViltForQuestionAnswering,
    ViltImageProcessor,
    ViltProcessor,
)
from transformers.utils import logging


logging.set_verbosity_info()
lowercase_       :						Union[str, Any]     					=	logging.get_logger(__name__)





def    _lowerCAmelCase    (     lowerCamelCase__							:							int,						lowerCamelCase__							:							Optional[int]=False,						lowerCamelCase__							:							Any=False,						lowerCamelCase__							:							Any=False				)  ->    Dict:
  _SCREAMING_SNAKE_CASE   :				str							    = []
  for i in range(config.num_hidden_layers				):
    # encoder layers: output projection, 2 feedforward neural networks and 2 layernorms
    rename_keys.append((f'''transformer.blocks.{i}.norm1.weight''', f'''vilt.encoder.layer.{i}.layernorm_before.weight''')				)
    rename_keys.append((f'''transformer.blocks.{i}.norm1.bias''', f'''vilt.encoder.layer.{i}.layernorm_before.bias''')				)
    rename_keys.append(
        (f'''transformer.blocks.{i}.attn.proj.weight''', f'''vilt.encoder.layer.{i}.attention.output.dense.weight''')				)
    rename_keys.append(
        (f'''transformer.blocks.{i}.attn.proj.bias''', f'''vilt.encoder.layer.{i}.attention.output.dense.bias''')				)
    rename_keys.append((f'''transformer.blocks.{i}.norm2.weight''', f'''vilt.encoder.layer.{i}.layernorm_after.weight''')				)
    rename_keys.append((f'''transformer.blocks.{i}.norm2.bias''', f'''vilt.encoder.layer.{i}.layernorm_after.bias''')				)
    rename_keys.append(
        (f'''transformer.blocks.{i}.mlp.fc1.weight''', f'''vilt.encoder.layer.{i}.intermediate.dense.weight''')				)
    rename_keys.append((f'''transformer.blocks.{i}.mlp.fc1.bias''', f'''vilt.encoder.layer.{i}.intermediate.dense.bias''')				)
    rename_keys.append((f'''transformer.blocks.{i}.mlp.fc2.weight''', f'''vilt.encoder.layer.{i}.output.dense.weight''')				)
    rename_keys.append((f'''transformer.blocks.{i}.mlp.fc2.bias''', f'''vilt.encoder.layer.{i}.output.dense.bias''')				)

  # embeddings
  rename_keys.extend(
      [
          # text embeddings
          ("text_embeddings.word_embeddings.weight", "vilt.embeddings.text_embeddings.word_embeddings.weight"),
          (
              "text_embeddings.position_embeddings.weight",
              "vilt.embeddings.text_embeddings.position_embeddings.weight",
          ),
          ("text_embeddings.position_ids", "vilt.embeddings.text_embeddings.position_ids"),
          (
              "text_embeddings.token_type_embeddings.weight",
              "vilt.embeddings.text_embeddings.token_type_embeddings.weight",
          ),
          ("text_embeddings.LayerNorm.weight", "vilt.embeddings.text_embeddings.LayerNorm.weight"),
          ("text_embeddings.LayerNorm.bias", "vilt.embeddings.text_embeddings.LayerNorm.bias"),
          # patch embeddings
          ("transformer.cls_token", "vilt.embeddings.cls_token"),
          ("transformer.patch_embed.proj.weight", "vilt.embeddings.patch_embeddings.projection.weight"),
          ("transformer.patch_embed.proj.bias", "vilt.embeddings.patch_embeddings.projection.bias"),
          ("transformer.pos_embed", "vilt.embeddings.position_embeddings"),
          # token type embeddings
          ("token_type_embeddings.weight", "vilt.embeddings.token_type_embeddings.weight"),
      ]				)

  # final layernorm + pooler
  rename_keys.extend(
      [
          ("transformer.norm.weight", "vilt.layernorm.weight"),
          ("transformer.norm.bias", "vilt.layernorm.bias"),
          ("pooler.dense.weight", "vilt.pooler.dense.weight"),
          ("pooler.dense.bias", "vilt.pooler.dense.bias"),
      ]				)

  # classifier head(s)
  if vqa_model:
    # classification head
    rename_keys.extend(
        [
            ("vqa_classifier.0.weight", "classifier.0.weight"),
            ("vqa_classifier.0.bias", "classifier.0.bias"),
            ("vqa_classifier.1.weight", "classifier.1.weight"),
            ("vqa_classifier.1.bias", "classifier.1.bias"),
            ("vqa_classifier.3.weight", "classifier.3.weight"),
            ("vqa_classifier.3.bias", "classifier.3.bias"),
        ]				)
  elif nlvr_model:
    # classification head
    rename_keys.extend(
        [
            ("nlvr2_classifier.0.weight", "classifier.0.weight"),
            ("nlvr2_classifier.0.bias", "classifier.0.bias"),
            ("nlvr2_classifier.1.weight", "classifier.1.weight"),
            ("nlvr2_classifier.1.bias", "classifier.1.bias"),
            ("nlvr2_classifier.3.weight", "classifier.3.weight"),
            ("nlvr2_classifier.3.bias", "classifier.3.bias"),
        ]				)
  else:
    pass

  return rename_keys





def    _lowerCAmelCase    (     lowerCamelCase__							:							Tuple,						lowerCamelCase__							:							Any				)  ->    Tuple:
  for i in range(config.num_hidden_layers				):
    _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = """vilt."""
    # read in weights + bias of input projection layer (in timm, this is a single matrix + bias)
    _SCREAMING_SNAKE_CASE   :				Tuple							    = state_dict.pop(f'''transformer.blocks.{i}.attn.qkv.weight'''				)
    _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = state_dict.pop(f'''transformer.blocks.{i}.attn.qkv.bias'''				)
    # next, add query, keys and values (in that order) to the state dict
    _SCREAMING_SNAKE_CASE   :				List[Any]							    = in_proj_weight[
        : config.hidden_size, :
    ]
    _SCREAMING_SNAKE_CASE   :				List[str]							    = in_proj_bias[: config.hidden_size]
    _SCREAMING_SNAKE_CASE   :				List[str]							    = in_proj_weight[
        config.hidden_size : config.hidden_size * 2, :
    ]
    _SCREAMING_SNAKE_CASE   :				List[Any]							    = in_proj_bias[
        config.hidden_size : config.hidden_size * 2
    ]
    _SCREAMING_SNAKE_CASE   :				Any							    = in_proj_weight[
        -config.hidden_size :, :
    ]
    _SCREAMING_SNAKE_CASE   :				Dict							    = in_proj_bias[-config.hidden_size :]





def    _lowerCAmelCase    (     lowerCamelCase__							:							List[str]				)  ->    List[Any]:
  _SCREAMING_SNAKE_CASE   :				Tuple							    = ["""head.weight""", """head.bias"""]
  for k in ignore_keys:
    state_dict.pop(UpperCAmelCase__,						UpperCAmelCase__				)





def    _lowerCAmelCase    (     lowerCamelCase__							:							Tuple,						lowerCamelCase__							:							List[str],						lowerCamelCase__							:							Union[str, Any]				)  ->    Union[str, Any]:
  _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = dct.pop(UpperCAmelCase__				)
  _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = val





@torch.no_grad()
def    _lowerCAmelCase    (     lowerCamelCase__							:							str,						lowerCamelCase__							:							Optional[int]				)  ->    List[str]:
  _SCREAMING_SNAKE_CASE   :				int							    = ViltConfig(image_size=3_8_4,						patch_size=3_2,						tie_word_embeddings=UpperCAmelCase__				)
  _SCREAMING_SNAKE_CASE   :				str							    = False
  _SCREAMING_SNAKE_CASE   :				str							    = False
  _SCREAMING_SNAKE_CASE   :				str							    = False
  _SCREAMING_SNAKE_CASE   :				Optional[int]							    = False
  if "vqa" in checkpoint_url:
    _SCREAMING_SNAKE_CASE   :				int							    = True
    _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = 3_1_2_9
    _SCREAMING_SNAKE_CASE   :				int							    = """huggingface/label-files"""
    _SCREAMING_SNAKE_CASE   :				Dict							    = """vqa2-id2label.json"""
    _SCREAMING_SNAKE_CASE   :				Dict							    = json.load(open(hf_hub_download(UpperCAmelCase__,						UpperCAmelCase__,						repo_type="dataset"				),						"r"				)				)
    _SCREAMING_SNAKE_CASE   :				Optional[int]							    = {int(UpperCAmelCase__				): v for k, v in idalabel.items()}
    _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = idalabel
    _SCREAMING_SNAKE_CASE   :				Optional[int]							    = {v: k for k, v in idalabel.items()}
    _SCREAMING_SNAKE_CASE   :				List[Any]							    = ViltForQuestionAnswering(UpperCAmelCase__				)
  elif "nlvr" in checkpoint_url:
    _SCREAMING_SNAKE_CASE   :				Any							    = True
    _SCREAMING_SNAKE_CASE   :				Tuple							    = 2
    _SCREAMING_SNAKE_CASE   :				Optional[int]							    = {0: """False""", 1: """True"""}
    _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = {v: k for k, v in config.idalabel.items()}
    _SCREAMING_SNAKE_CASE   :				List[str]							    = 3
    _SCREAMING_SNAKE_CASE   :				Tuple							    = ViltForImagesAndTextClassification(UpperCAmelCase__				)
  elif "irtr" in checkpoint_url:
    _SCREAMING_SNAKE_CASE   :				Optional[int]							    = True
    _SCREAMING_SNAKE_CASE   :				Optional[int]							    = ViltForImageAndTextRetrieval(UpperCAmelCase__				)
  elif "mlm_itm" in checkpoint_url:
    _SCREAMING_SNAKE_CASE   :				Optional[int]							    = True
    _SCREAMING_SNAKE_CASE   :				Tuple							    = ViltForMaskedLM(UpperCAmelCase__				)
  else:
    raise ValueError("Unknown model type"				)

  # load state_dict of original model, remove and rename some keys
  _SCREAMING_SNAKE_CASE   :				Tuple							    = torch.hub.load_state_dict_from_url(UpperCAmelCase__,						map_location="cpu"				)["""state_dict"""]
  _SCREAMING_SNAKE_CASE   :				int							    = create_rename_keys(UpperCAmelCase__,						UpperCAmelCase__,						UpperCAmelCase__,						UpperCAmelCase__				)
  for src, dest in rename_keys:
    rename_key(UpperCAmelCase__,						UpperCAmelCase__,						UpperCAmelCase__				)
  read_in_q_k_v(UpperCAmelCase__,						UpperCAmelCase__				)
  if mlm_model or irtr_model:
    _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = ["""itm_score.fc.weight""", """itm_score.fc.bias"""]
    for k in ignore_keys:
      state_dict.pop(UpperCAmelCase__,						UpperCAmelCase__				)

    # load state dict into HuggingFace model
  model.eval()
  if mlm_model:
    _SCREAMING_SNAKE_CASE   :				Tuple							    = model.load_state_dict(UpperCAmelCase__,						strict=UpperCAmelCase__				)
    assert missing_keys == ["mlm_score.decoder.bias"]
  else:
    model.load_state_dict(UpperCAmelCase__				)

  # Define processor
  _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = ViltImageProcessor(size=3_8_4				)
  _SCREAMING_SNAKE_CASE   :				Optional[int]							    = BertTokenizer.from_pretrained("bert-base-uncased"				)
  _SCREAMING_SNAKE_CASE   :				Any							    = ViltProcessor(UpperCAmelCase__,						UpperCAmelCase__				)

  # Forward pass on example inputs (image + text)
  if nlvr_model:
    _SCREAMING_SNAKE_CASE   :				int							    = Image.open(requests.get("https://lil.nlp.cornell.edu/nlvr/exs/ex0_0.jpg",						stream=UpperCAmelCase__				).raw				)
    _SCREAMING_SNAKE_CASE   :				Optional[int]							    = Image.open(requests.get("https://lil.nlp.cornell.edu/nlvr/exs/ex0_0.jpg",						stream=UpperCAmelCase__				).raw				)
    _SCREAMING_SNAKE_CASE   :				Optional[int]							    = (
        """The left image contains twice the number of dogs as the right image, and at least two dogs in total are"""
        """ standing."""
    )
    _SCREAMING_SNAKE_CASE   :				int							    = processor(UpperCAmelCase__,						UpperCAmelCase__,						return_tensors="pt"				)
    _SCREAMING_SNAKE_CASE   :				Dict							    = processor(UpperCAmelCase__,						UpperCAmelCase__,						return_tensors="pt"				)
    _SCREAMING_SNAKE_CASE   :				Any							    = model(
        input_ids=encoding_a.input_ids,						pixel_values=encoding_a.pixel_values,						pixel_values_a=encoding_a.pixel_values,						)
  else:
    _SCREAMING_SNAKE_CASE   :				Any							    = Image.open(requests.get("http://images.cocodataset.org/val2017/000000039769.jpg",						stream=UpperCAmelCase__				).raw				)
    if mlm_model:
      _SCREAMING_SNAKE_CASE   :				str							    = """a bunch of [MASK] laying on a [MASK]."""
    else:
      _SCREAMING_SNAKE_CASE   :				List[str]							    = """How many cats are there?"""
    _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = processor(UpperCAmelCase__,						UpperCAmelCase__,						return_tensors="pt"				)
    _SCREAMING_SNAKE_CASE   :				Dict							    = model(**UpperCAmelCase__				)

  # Verify outputs
  if mlm_model:
    _SCREAMING_SNAKE_CASE   :				List[str]							    = torch.Size([1, 1_1, 3_0_5_2_2]				)
    _SCREAMING_SNAKE_CASE   :				List[Any]							    = torch.tensor([-12.5061, -12.5123, -12.5174]				)
    assert outputs.logits.shape == expected_shape
    assert torch.allclose(outputs.logits[0, 0, :3],						UpperCAmelCase__,						atol=1E-4				)

    # verify masked token prediction equals "cats"
    _SCREAMING_SNAKE_CASE   :				Any							    = outputs.logits[0, 4, :].argmax(-1				).item()
    assert tokenizer.decode([predicted_id]				) == "cats"
  elif vqa_model:
    _SCREAMING_SNAKE_CASE   :				Any							    = torch.Size([1, 3_1_2_9]				)
    _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = torch.tensor([-15.9495, -18.1472, -10.3041]				)
    assert torch.allclose(outputs.logits[0, :3],						UpperCAmelCase__,						atol=1E-4				)
    assert outputs.logits.shape == expected_shape
    assert torch.allclose(outputs.logits[0, 0, :3],						UpperCAmelCase__,						atol=1E-4				)

    # verify vqa prediction equals "2"
    _SCREAMING_SNAKE_CASE   :				List[Any]							    = outputs.logits.argmax(-1				).item()
    assert model.config.idalabel[predicted_idx] == "2"
  elif nlvr_model:
    _SCREAMING_SNAKE_CASE   :				List[Any]							    = torch.Size([1, 2]				)
    _SCREAMING_SNAKE_CASE   :				List[Any]							    = torch.tensor([-2.8721, 2.1291]				)
    assert torch.allclose(outputs.logits[0, :3],						UpperCAmelCase__,						atol=1E-4				)
    assert outputs.logits.shape == expected_shape

  Path(UpperCAmelCase__				).mkdir(exist_ok=UpperCAmelCase__				)
  print(f'''Saving model and processor to {pytorch_dump_folder_path}'''				)
  model.save_pretrained(UpperCAmelCase__				)
  processor.save_pretrained(UpperCAmelCase__				)


if __name__ == "__main__":
      lowercase_       :						Dict     					=	argparse.ArgumentParser()
      # Required parameters
      parser.add_argument(
          '''--checkpoint_url''',
          default='''https://github.com/dandelin/ViLT/releases/download/200k/vilt_200k_mlm_itm.ckpt''',
          type=str,
          help='''URL of the checkpoint you\'d like to convert.''',
      )
      parser.add_argument(
          '''--pytorch_dump_folder_path''', default=None, type=str, help='''Path to the output PyTorch model directory.'''
      )

      lowercase_       :						int     					=	parser.parse_args()
      convert_vilt_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path)





"""simple docstring"""
from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer
@dataclass
class 			UpperCamelCase      (							__SCREAMING_SNAKE_CASE  ):
     A__      =			42
class 			UpperCamelCase      (							__SCREAMING_SNAKE_CASE      ,       __SCREAMING_SNAKE_CASE  ):
     @register_to_config
     def __init__(     self  ,     snake_case__ = 3  ,     snake_case__ = 3  ,     snake_case__ = ("DownEncoderBlock2D",)  ,     snake_case__ = ("UpDecoderBlock2D",)  ,     snake_case__ = (64,)  ,     snake_case__ = 1  ,     snake_case__ = "silu"  ,     snake_case__ = 3  ,     snake_case__ = 32  ,     snake_case__ = 256  ,     snake_case__ = 32  ,     snake_case__ = None  ,     snake_case__ = 0.18_215  ,     snake_case__ = "group"  ,     ):




       """simple docstring"""


       super().__init__()

       # pass init params to Encoder
       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = Encoder(
           in_channels=snake_case__  ,     out_channels=snake_case__  ,     down_block_types=snake_case__  ,     block_out_channels=snake_case__  ,     layers_per_block=snake_case__  ,     act_fn=snake_case__  ,     norm_num_groups=snake_case__  ,     double_z=snake_case__  ,     )

       _SCREAMING_SNAKE_CASE   :				Dict							    = vq_embed_dim if vq_embed_dim is not None else latent_channels

       _SCREAMING_SNAKE_CASE   :				Tuple							    = nn.Convad(snake_case__  ,     snake_case__  ,     1				)
       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = VectorQuantizer(snake_case__  ,     snake_case__  ,     beta=0.25  ,     remap=snake_case__  ,     sane_index_shape=snake_case__				)
       _SCREAMING_SNAKE_CASE   :				List[str]							    = nn.Convad(snake_case__  ,     snake_case__  ,     1				)

       # pass init params to Decoder
       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = Decoder(
           in_channels=snake_case__  ,     out_channels=snake_case__  ,     up_block_types=snake_case__  ,     block_out_channels=snake_case__  ,     layers_per_block=snake_case__  ,     act_fn=snake_case__  ,     norm_num_groups=snake_case__  ,     norm_type=snake_case__  ,     )
     @apply_forward_hook
     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__  ,     snake_case__ = True				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				List[str]							    = self.encoder(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				Dict							    = self.quant_conv(snake_case__				)

       if not return_dict:
         return (h,)

       return VQEncoderOutput(latents=snake_case__				)
     @apply_forward_hook
     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__  ,     snake_case__ = False  ,     snake_case__ = True				):




       """simple docstring"""


       if not force_not_quantize:
         _SCREAMING_SNAKE_CASE    ,		_SCREAMING_SNAKE_CASE    ,		_SCREAMING_SNAKE_CASE   :				int							    = self.quantize(snake_case__				)
       else:
         _SCREAMING_SNAKE_CASE   :				int							    = h
       _SCREAMING_SNAKE_CASE   :				Dict							    = self.post_quant_conv(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				int							    = self.decoder(snake_case__  ,     quant if self.config.norm_type == "spatial" else None				)

       if not return_dict:
         return (dec,)

       return DecoderOutput(sample=snake_case__				)




     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__  ,     snake_case__ = True				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				Tuple							    = sample
       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = self.encode(snake_case__				).latents
       _SCREAMING_SNAKE_CASE   :				Any							    = self.decode(snake_case__				).sample

       if not return_dict:
         return (dec,)

       return DecoderOutput(sample=snake_case__				)



