


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							:   str						=						logging.get_logger(__name__)





def 	_lowerCAmelCase					( UpperCamelCase__:  List[Any]				,       UpperCamelCase__:  Dict=False				,       UpperCamelCase__:  Optional[int]=False				,       UpperCamelCase__:  str=False    )			-> Optional[Any]:





	"""simple docstring"""







	A             =					[]
	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					( UpperCamelCase__:  int				,       UpperCamelCase__:  int    )			-> Any:





	"""simple docstring"""







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





def 	_lowerCAmelCase					( UpperCamelCase__:  Tuple    )			-> str:





	"""simple docstring"""







	A             =					["""head.weight""", """head.bias"""]
	for k in ignore_keys:
		state_dict.pop(_snake_case				,       _snake_case    )





def 	_lowerCAmelCase					( UpperCamelCase__:  Dict				,       UpperCamelCase__:  str				,       UpperCamelCase__:  Tuple    )			-> str:





	"""simple docstring"""







	A             =					dct.pop(_snake_case    )
	A             =					val





@torch.no_grad()
def 	_lowerCAmelCase					( UpperCamelCase__:  List[str]				,       UpperCamelCase__:  Dict    )			-> List[str]:





	"""simple docstring"""







	A             =					ViltConfig(image_size=3_84				,       patch_size=32				,       tie_word_embeddings=_snake_case    )
	A             =					False
	A             =					False
	A             =					False
	A             =					False
	if "vqa" in checkpoint_url:
		A             =					True
		A             =					31_29
		A             =					"""huggingface/label-files"""
		A             =					"""vqa2-id2label.json"""
		A             =					json.load(open(hf_hub_download(_snake_case				,       _snake_case				,       repo_type="""dataset"""    )				,       """r"""    )    )
		A             =					{int(_snake_case    ): v for k, v in idalabel.items()}
		A             =					idalabel
		A             =					{v: k for k, v in idalabel.items()}
		A             =					ViltForQuestionAnswering(_snake_case    )
	elif "nlvr" in checkpoint_url:
		A             =					True
		A             =					2
		A             =					{0: """False""", 1: """True"""}
		A             =					{v: k for k, v in config.idalabel.items()}
		A             =					3
		A             =					ViltForImagesAndTextClassification(_snake_case    )
	elif "irtr" in checkpoint_url:
		A             =					True
		A             =					ViltForImageAndTextRetrieval(_snake_case    )
	elif "mlm_itm" in checkpoint_url:
		A             =					True
		A             =					ViltForMaskedLM(_snake_case    )
	else:
		raise ValueError("""Unknown model type"""    )

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

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

	# Define processor
	A             =					ViltImageProcessor(size=3_84    )
	A             =					BertTokenizer.from_pretrained("""bert-base-uncased"""    )
	A             =					ViltProcessor(_snake_case				,       _snake_case    )

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

	# Verify outputs
	if mlm_model:
		A             =					torch.Size([1, 11, 3_05_22]    )
		A             =					torch.tensor([-12.50_61, -12.51_23, -12.51_74]    )
		assert outputs.logits.shape == expected_shape
		assert torch.allclose(outputs.logits[0, 0, :3]				,       _snake_case				,       atol=1e-4    )

		# verify masked token prediction equals "cats"
		A             =					outputs.logits[0, 4, :].argmax(-1    ).item()
		assert tokenizer.decode([predicted_id]    ) == "cats"
	elif vqa_model:
		A             =					torch.Size([1, 31_29]    )
		A             =					torch.tensor([-15.94_95, -18.14_72, -10.30_41]    )
		assert torch.allclose(outputs.logits[0, :3]				,       _snake_case				,       atol=1e-4    )
		assert outputs.logits.shape == expected_shape
		assert torch.allclose(outputs.logits[0, 0, :3]				,       _snake_case				,       atol=1e-4    )

		# verify vqa prediction equals "2"
		A             =					outputs.logits.argmax(-1    ).item()
		assert model.config.idalabel[predicted_idx] == "2"
	elif nlvr_model:
		A             =					torch.Size([1, 2]    )
		A             =					torch.tensor([-2.87_21, 2.12_91]    )
		assert torch.allclose(outputs.logits[0, :3]				,       _snake_case				,       atol=1e-4    )
		assert outputs.logits.shape == expected_shape

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


if __name__ == "__main__":
				_lowercase							:   List[str]						=						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							:   List[str]						=						parser.parse_args()
				convert_vilt_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path)









"""simple docstring"""

import os
import tempfile
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
       import torch
       from torch import nn

       from transformers import (
           Adafactor,
           AdamW,
           get_constant_schedule,
           get_constant_schedule_with_warmup,
           get_cosine_schedule_with_warmup,
           get_cosine_with_hard_restarts_schedule_with_warmup,
           get_inverse_sqrt_schedule,
           get_linear_schedule_with_warmup,
           get_polynomial_decay_schedule_with_warmup,
       )






def 		_snake_case	(   _snake_case				:   Optional[int]							,    _snake_case				:   Optional[Any]=10				)       ->   Optional[int]:






      '''simple docstring'''


      _A     =  []
      for _ in range(_snake_case				):
            lrs.append(scheduler.get_lr()[0]				)
            scheduler.step()
      return lrs






def 		_snake_case	(   _snake_case				:   Optional[Any]							,    _snake_case				:   Union[str, Any]=10				)       ->   List[str]:






      '''simple docstring'''


      _A     =  []
      for step in range(_snake_case				):
            lrs.append(scheduler.get_lr()[0]				)
            scheduler.step()
            if step == num_steps // 2:
                  with tempfile.TemporaryDirectory() as tmpdirname:
                        _A     =  os.path.join(_snake_case							,    'schedule.bin'				)
                        torch.save(scheduler.state_dict()							,    _snake_case				)

                        _A     =  torch.load(_snake_case				)
                        scheduler.load_state_dict(_snake_case				)
      return lrs





@require_torch
class 							lowercase_			(	unittest.TestCase				):



 '''simple docstring'''

 def      lowerCAmelCase_				(					self			:			List[Any]       ,				_UpperCAmelCase			:			List[Any]       ,				_UpperCAmelCase			:			Optional[Any]       ,				_UpperCAmelCase			:			Tuple						):
       self.assertEqual(len(_UpperCAmelCase						)       ,				len(_UpperCAmelCase						)						)
       for a, b in zip(_UpperCAmelCase       ,				_UpperCAmelCase						):
             self.assertAlmostEqual(_UpperCAmelCase       ,				_UpperCAmelCase       ,				delta=_UpperCAmelCase						)

 def      lowerCAmelCase_				(					self			:			Any						):
       _A     =  torch.tensor([0.1, -0.2, -0.1]       ,				requires_grad=_UpperCAmelCase						)
       _A     =  torch.tensor([0.4, 0.2, -0.5]						)
       _A     =  nn.MSELoss()
       # No warmup, constant schedule, no gradient clipping
       _A     =  AdamW(params=[w]       ,				lr=2E-1       ,				weight_decay=0.0						)
       for _ in range(100						):
             _A     =  criterion(_UpperCAmelCase       ,				_UpperCAmelCase						)
             loss.backward()
             optimizer.step()
             w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
             w.grad.zero_()
       self.assertListAlmostEqual(w.tolist()       ,				[0.4, 0.2, -0.5]       ,				tol=1E-2						)



 def      lowerCAmelCase_				(					self			:			int						):
       _A     =  torch.tensor([0.1, -0.2, -0.1]       ,				requires_grad=_UpperCAmelCase						)
       _A     =  torch.tensor([0.4, 0.2, -0.5]						)
       _A     =  nn.MSELoss()
       # No warmup, constant schedule, no gradient clipping
       _A     =  Adafactor(
           params=[w]       ,				lr=1E-2       ,				eps=(1E-3_0, 1E-3)       ,				clip_threshold=1.0       ,				decay_rate=-0.8       ,				betaa=_UpperCAmelCase       ,				weight_decay=0.0       ,				relative_step=_UpperCAmelCase       ,				scale_parameter=_UpperCAmelCase       ,				warmup_init=_UpperCAmelCase       ,				)
       for _ in range(1_000						):
             _A     =  criterion(_UpperCAmelCase       ,				_UpperCAmelCase						)
             loss.backward()
             optimizer.step()
             w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
             w.grad.zero_()
       self.assertListAlmostEqual(w.tolist()       ,				[0.4, 0.2, -0.5]       ,				tol=1E-2						)





@require_torch
class 							lowercase_			(	unittest.TestCase				):



 '''simple docstring'''
 UpperCAmelCase       :	List[str]     =      nn.Linear(50     ,      50				) if is_torch_available() else None
 UpperCAmelCase       :	Tuple     =      AdamW(m.parameters()     ,      lr=10.0				) if is_torch_available() else None
 UpperCAmelCase       :	Dict     =      10

 def      lowerCAmelCase_				(					self			:			Any       ,				_UpperCAmelCase			:			Optional[Any]       ,				_UpperCAmelCase			:			str       ,				_UpperCAmelCase			:			List[Any]       ,				_UpperCAmelCase			:			List[Any]=None						):
       self.assertEqual(len(_UpperCAmelCase						)       ,				len(_UpperCAmelCase						)						)
       for a, b in zip(_UpperCAmelCase       ,				_UpperCAmelCase						):
             self.assertAlmostEqual(_UpperCAmelCase       ,				_UpperCAmelCase       ,				delta=_UpperCAmelCase       ,				msg=_UpperCAmelCase						)



 def      lowerCAmelCase_				(					self			:			List[Any]						):
       _A     =  {'num_warmup_steps': 2, 'num_training_steps': 10}
       # schedulers doct format
       # function: (sched_args_dict, expected_learning_rates)
       _A     =  {
           get_constant_schedule: ({}, [10.0] * self.num_steps),
           get_constant_schedule_with_warmup: (
               {'num_warmup_steps': 4},
               [0.0, 2.5, 5.0, 7.5, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0],
           ),
           get_linear_schedule_with_warmup: (
               {**common_kwargs},
               [0.0, 5.0, 10.0, 8.75, 7.5, 6.25, 5.0, 3.75, 2.5, 1.25],
           ),
           get_cosine_schedule_with_warmup: (
               {**common_kwargs},
               [0.0, 5.0, 10.0, 9.61, 8.53, 6.91, 5.0, 3.08, 1.46, 0.38],
           ),
           get_cosine_with_hard_restarts_schedule_with_warmup: (
               {**common_kwargs, 'num_cycles': 2},
               [0.0, 5.0, 10.0, 8.53, 5.0, 1.46, 10.0, 8.53, 5.0, 1.46],
           ),
           get_polynomial_decay_schedule_with_warmup: (
               {**common_kwargs, 'power': 2.0, 'lr_end': 1E-7},
               [0.0, 5.0, 10.0, 7.656, 5.625, 3.906, 2.5, 1.406, 0.625, 0.156],
           ),
           get_inverse_sqrt_schedule: (
               {'num_warmup_steps': 2},
               [0.0, 5.0, 10.0, 8.165, 7.071, 6.325, 5.774, 5.345, 5.0, 4.714],
           ),
       }

       for scheduler_func, data in scheds.items():
             _A  ,       _A     =  data

             _A     =  scheduler_func(self.optimizer       ,				**_UpperCAmelCase						)
             self.assertEqual(len([scheduler.get_lr()[0]]						)       ,				1						)
             _A     =  unwrap_schedule(_UpperCAmelCase       ,				self.num_steps						)
             self.assertListAlmostEqual(
                 _UpperCAmelCase       ,				_UpperCAmelCase       ,				tol=1E-2       ,				msg=F'''failed for {scheduler_func} in normal scheduler'''       ,				)

             _A     =  scheduler_func(self.optimizer       ,				**_UpperCAmelCase						)
             if scheduler_func.__name__ != "get_constant_schedule":
                   LambdaScheduleWrapper.wrap_scheduler(_UpperCAmelCase						)  # wrap to test picklability of the schedule
             _A     =  unwrap_and_save_reload_schedule(_UpperCAmelCase       ,				self.num_steps						)
             self.assertListEqual(_UpperCAmelCase       ,				_UpperCAmelCase       ,				msg=F'''failed for {scheduler_func} in save and reload'''						)





class 							lowercase_			:



 '''simple docstring'''

 def __init__(					self			:			Union[str, Any]       ,				_UpperCAmelCase			:			Optional[int]						):
       _A     =  fn

 def __call__(					self			:			Tuple       ,				*_UpperCAmelCase			:			List[str]       ,				**_UpperCAmelCase			:			List[str]						):
       return self.fn(*_UpperCAmelCase       ,				**_UpperCAmelCase						)



 @classmethod
 def      lowerCAmelCase_				(					self			:			Union[str, Any]       ,				_UpperCAmelCase			:			Any						):
       _A     =  list(map(self       ,				scheduler.lr_lambdas						)						)

