


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



def        UpperCamelCase_(   lowerCamelCase_							,				lowerCamelCase_=False							,				lowerCamelCase_=False							,				lowerCamelCase_=False				)					->		Union[str, Any]:
    _lowercase :							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        UpperCamelCase_(   lowerCamelCase_							,				lowerCamelCase_				)					->		List[Any]:







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



def        UpperCamelCase_(   lowerCamelCase_				)					->		Tuple:
    _lowercase :							Dict   	=	['head.weight', 'head.bias']
    for k in ignore_keys:
        state_dict.pop(lowerCAmelCase__							,				lowerCAmelCase__				)



def        UpperCamelCase_(   lowerCamelCase_							,				lowerCamelCase_							,				lowerCamelCase_				)					->		Optional[Any]:
    _lowercase :							Optional[int]   	=	dct.pop(lowerCAmelCase__				)
    _lowercase :							Tuple   	=	val



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

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

    # load state dict into HuggingFace model
    model.eval()
    if mlm_model:
        _lowercase :							Union[str, Any]   	=	model.load_state_dict(lowerCAmelCase__							,				strict=lowerCAmelCase__				)
        assert missing_keys == ["mlm_score.decoder.bias"]
    else:
        model.load_state_dict(lowerCAmelCase__				)

    # Define processor
    _lowercase :							Tuple   	=	ViltImageProcessor(size=384				)
    _lowercase :							Optional[int]   	=	BertTokenizer.from_pretrained('bert-base-uncased'				)
    _lowercase :							int   	=	ViltProcessor(lowerCAmelCase__							,				lowerCAmelCase__				)

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

    # Verify outputs
    if mlm_model:
        _lowercase :							List[Any]   	=	torch.Size([1, 11, 3_0522]				)
        _lowercase :							Union[str, Any]   	=	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]							,				lowerCAmelCase__							,				atol=1e-4				)

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

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

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


if __name__ == "__main__":
      SCREAMING_SNAKE_CASE :   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."
      )

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






import os

import pytest
import yaml

from datasets.features.features import Features, Value
from datasets.info import DatasetInfo, DatasetInfosDict



@pytest.mark.parametrize(
    'files'							,				[
        ['full:README.md', 'dataset_infos.json'],
        ['empty:README.md', 'dataset_infos.json'],
        ['dataset_infos.json'],
        ['full:README.md'],
    ]							,				)
def        UpperCamelCase_(   lowerCamelCase_							,				lowerCamelCase_				)					->		List[str]:
    _lowercase :							Any   	=	tmp_path_factory.mktemp('dset_infos_dir'				)
    if "full:README.md" in files:
        with open(dataset_infos_dir / 'README.md'							,				'w'				) as f:
            f.write('---\ndataset_info:\n  dataset_size: 42\n---'				)
    if "empty:README.md" in files:
        with open(dataset_infos_dir / 'README.md'							,				'w'				) as f:
            f.write(''				)
    # we want to support dataset_infos.json for backward compatibility
    if "dataset_infos.json" in files:
        with open(dataset_infos_dir / 'dataset_infos.json'							,				'w'				) as f:
            f.write('{"default": {"dataset_size": 42}}'				)
    _lowercase :							Any   	=	DatasetInfosDict.from_directory(lowerCamelCase_				)
    assert dataset_infos
    assert dataset_infos["default"].dataset_size == 42



@pytest.mark.parametrize(
    'dataset_info'							,				[
        DatasetInfo(),
        DatasetInfo(
            description='foo'							,				features=Features({'a': Value('int32'				)}				)							,				builder_name='builder'							,				config_name='config'							,				version='1.0.0'							,				splits=[{'name': 'train'}]							,				download_size=42							,				),
    ]							,				)
def        UpperCamelCase_(   lowerCamelCase_							,				lowerCamelCase_				)					->		Optional[int]:
    _lowercase :							Union[str, Any]   	=	str(lowerCamelCase_				)
    dataset_info.write_to_directory(lowerCamelCase_				)
    _lowercase :							List[str]   	=	DatasetInfo.from_directory(lowerCamelCase_				)
    assert dataset_info == reloaded
    assert os.path.exists(os.path.join(lowerCamelCase_							,				'dataset_info.json'				)				)



def        UpperCamelCase_(   )					->		int:
    _lowercase :							Tuple   	=	DatasetInfo(
        description='foo'							,				citation='bar'							,				homepage='https://foo.bar'							,				license='CC0'							,				features=Features({'a': Value('int32'				)}				)							,				post_processed={}							,				supervised_keys=()							,				task_templates=[]							,				builder_name='builder'							,				config_name='config'							,				version='1.0.0'							,				splits=[{'name': 'train', 'num_examples': 42}]							,				download_checksums={}							,				download_size=1337							,				post_processing_size=442							,				dataset_size=1234							,				size_in_bytes=1337 + 442 + 1234							,				)
    _lowercase :							Optional[int]   	=	dataset_info._to_yaml_dict()
    assert sorted(lowerCamelCase_				) == sorted(DatasetInfo._INCLUDED_INFO_IN_YAML				)
    for key in DatasetInfo._INCLUDED_INFO_IN_YAML:
        assert key in dataset_info_yaml_dict
        assert isinstance(dataset_info_yaml_dict[key]							,				(list, dict, int, str)				)
    _lowercase :							str   	=	yaml.safe_dump(lowerCamelCase_				)
    _lowercase :							str   	=	yaml.safe_load(lowerCamelCase_				)
    assert dataset_info_yaml_dict == reloaded



def        UpperCamelCase_(   )					->		int:
    _lowercase :							Tuple   	=	DatasetInfo()
    _lowercase :							Tuple   	=	dataset_info._to_yaml_dict()
    assert dataset_info_yaml_dict == {}



@pytest.mark.parametrize(
    'dataset_infos_dict'							,				[
        DatasetInfosDict(),
        DatasetInfosDict({'default': DatasetInfo()}				),
        DatasetInfosDict({'my_config_name': DatasetInfo()}				),
        DatasetInfosDict(
            {
                'default': DatasetInfo(
                    description='foo'							,				features=Features({'a': Value('int32'				)}				)							,				builder_name='builder'							,				config_name='config'							,				version='1.0.0'							,				splits=[{'name': 'train'}]							,				download_size=42							,				)
            }				),
        DatasetInfosDict(
            {
                'v1': DatasetInfo(dataset_size=42				),
                'v2': DatasetInfo(dataset_size=1337				),
            }				),
    ]							,				)
def        UpperCamelCase_(   lowerCamelCase_							,				lowerCamelCase_				)					->		Dict:
    _lowercase :							Tuple   	=	str(lowerCamelCase_				)
    dataset_infos_dict.write_to_directory(lowerCamelCase_				)
    _lowercase :							Tuple   	=	DatasetInfosDict.from_directory(lowerCamelCase_				)

    # the config_name of the dataset_infos_dict take over the attribute
    for config_name, dataset_info in dataset_infos_dict.items():
        _lowercase :							Any   	=	config_name
        # the yaml representation doesn't include fields like description or citation
        # so we just test that we can recover what we can from the yaml
        _lowercase :							str   	=	DatasetInfo._from_yaml_dict(dataset_info._to_yaml_dict()				)
    assert dataset_infos_dict == reloaded

    if dataset_infos_dict:
        assert os.path.exists(os.path.join(lowerCamelCase_							,				'README.md'				)				)


