


import argparse
from collections import OrderedDict
from pathlib import Path

import torch

from transformers import (
    VisualBertConfig,
    VisualBertForMultipleChoice,
    VisualBertForPreTraining,
    VisualBertForQuestionAnswering,
    VisualBertForVisualReasoning,
)
from transformers.utils import logging


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

snake_case_    :      int				    =				[
    ('''bert.bert''', '''visual_bert'''),
    ('''bert.cls''', '''cls'''),
    ('''bert.classifier''', '''cls'''),
    ('''token_type_embeddings_visual''', '''visual_token_type_embeddings'''),
    ('''position_embeddings_visual''', '''visual_position_embeddings'''),
    ('''projection''', '''visual_projection'''),
]

snake_case_    :      int				    =				[
    '''nlvr2_coco_pre_trained.th''',
    '''nlvr2_fine_tuned.th''',
    '''nlvr2_pre_trained.th''',
    '''vcr_coco_pre_train.th''',
    '''vcr_fine_tune.th''',
    '''vcr_pre_train.th''',
    '''vqa_coco_pre_trained.th''',
    '''vqa_fine_tuned.th''',
    '''vqa_pre_trained.th''',
]







def     lowerCamelCase( a__):
				_SCREAMING_SNAKE_CASE      	=torch.load(lowercase__						,map_location='''cpu''')
				return sd







def     lowerCamelCase( a__						,a__						,a__=rename_keys_prefix):
				_SCREAMING_SNAKE_CASE      	=OrderedDict()
				_SCREAMING_SNAKE_CASE      	=torch.arange(config.max_position_embeddings).expand((1, -1))
				# detector_d = OrderedDict()
				for key in d:
								if "detector" in key:
												# detector_d[key.replace('detector.','')] = d[key]
												continue
								_SCREAMING_SNAKE_CASE      	=key
								for name_pair in rename_keys_prefix:
												_SCREAMING_SNAKE_CASE      	=new_key.replace(name_pair[0]						,name_pair[1])
								_SCREAMING_SNAKE_CASE      	=d[key]
								if key == "bert.cls.predictions.decoder.weight":
												# Old bert code didn't have `decoder.bias`, but was added separately
												_SCREAMING_SNAKE_CASE      	=new_d["cls.predictions.bias"]
				return new_d







@torch.no_grad()
def     lowerCamelCase( a__						,a__):
				assert (
				    checkpoint_path.split('''/''')[-1] in ACCEPTABLE_CHECKPOINTS
				), f"The checkpoint provided must be in {ACCEPTABLE_CHECKPOINTS}."

				# Get Config
				if "pre" in checkpoint_path:
								_SCREAMING_SNAKE_CASE      	="pretraining"
								if "vcr" in checkpoint_path:
												_SCREAMING_SNAKE_CASE      	={"visual_embedding_dim": 512}
								elif "vqa_advanced" in checkpoint_path:
												_SCREAMING_SNAKE_CASE      	={"visual_embedding_dim": 2048}
								elif "vqa" in checkpoint_path:
												_SCREAMING_SNAKE_CASE      	={"visual_embedding_dim": 2048}
								elif "nlvr" in checkpoint_path:
												_SCREAMING_SNAKE_CASE      	={"visual_embedding_dim": 1024}
								else:
												raise NotImplementedError(f"No implementation found for `{checkpoint_path}`.")
				else:
								if "vcr" in checkpoint_path:
												_SCREAMING_SNAKE_CASE      	={"visual_embedding_dim": 512}
												_SCREAMING_SNAKE_CASE      	="multichoice"
								elif "vqa_advanced" in checkpoint_path:
												_SCREAMING_SNAKE_CASE      	={"visual_embedding_dim": 2048}
												_SCREAMING_SNAKE_CASE      	="vqa_advanced"
								elif "vqa" in checkpoint_path:
												_SCREAMING_SNAKE_CASE      	={"visual_embedding_dim": 2048, "num_labels": 3129}
												_SCREAMING_SNAKE_CASE      	="vqa"
								elif "nlvr" in checkpoint_path:
												_SCREAMING_SNAKE_CASE      	={
												    "visual_embedding_dim": 1024,
												    "num_labels": 2,
												}
												_SCREAMING_SNAKE_CASE      	="nlvr"

				_SCREAMING_SNAKE_CASE      	=VisualBertConfig(**lowercase__)

				# Load State Dict
				_SCREAMING_SNAKE_CASE      	=load_state_dict(lowercase__)

				_SCREAMING_SNAKE_CASE      	=get_new_dict(lowercase__						,lowercase__)

				if model_type == "pretraining":
								_SCREAMING_SNAKE_CASE      	=VisualBertForPreTraining(lowercase__)
				elif model_type == "vqa":
								_SCREAMING_SNAKE_CASE      	=VisualBertForQuestionAnswering(lowercase__)
				elif model_type == "nlvr":
								_SCREAMING_SNAKE_CASE      	=VisualBertForVisualReasoning(lowercase__)
				elif model_type == "multichoice":
								_SCREAMING_SNAKE_CASE      	=VisualBertForMultipleChoice(lowercase__)

				model.load_state_dict(lowercase__)
				# Save Checkpoints
				Path(lowercase__).mkdir(exist_ok=lowercase__)
				model.save_pretrained(lowercase__)


if __name__ == "__main__":
							snake_case_    :      Any				    =				argparse.ArgumentParser()
							# Required parameters
							parser.add_argument('''orig_checkpoint_path''', type=str, help='''A path to .th on local filesystem.''')
							parser.add_argument('''pytorch_dump_folder_path''', type=str, help='''Path to the output PyTorch model.''')
							snake_case_    :      Tuple				    =				parser.parse_args()
							convert_visual_bert_checkpoint(args.orig_checkpoint_path, 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_							(    A				):

 '''simple docstring'''
 lowercase_					:    torch.FloatTensor






class 							UpperCAmelCase_							(    A					,				A				):

 '''simple docstring'''







 @register_to_config
 def __init__(					self		:  List[str]			,				snake_case__		:  int = 3			,				snake_case__		:  int = 3			,				snake_case__		:  Tuple[str] = ("DownEncoderBlock2D",)			,				snake_case__		:  Tuple[str] = ("UpDecoderBlock2D",)			,				snake_case__		:  Tuple[int] = (64,)			,				snake_case__		:  int = 1			,				snake_case__		:  str = "silu"			,				snake_case__		:  int = 3			,				snake_case__		:  int = 32			,				snake_case__		:  int = 2_56			,				snake_case__		:  int = 32			,				snake_case__		:  Optional[int] = None			,				snake_case__		:  float = 0.18215			,				snake_case__		:  str = "group"			,				):



   '''simple docstring'''




   super().__init__()

   # pass init params to Encoder
   UpperCAmelCase__      :					Tuple				       =							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__			,				)

   UpperCAmelCase__      :					str				       =							vq_embed_dim if vq_embed_dim is not None else latent_channels

   UpperCAmelCase__      :					Tuple				       =							nn.Convad(snake_case__			,				snake_case__			,				1 )
   UpperCAmelCase__      :					int				       =							VectorQuantizer(snake_case__			,				snake_case__			,				beta=0.25			,				remap=snake_case__			,				sane_index_shape=snake_case__ )
   UpperCAmelCase__      :					Tuple				       =							nn.Convad(snake_case__			,				snake_case__			,				1 )

   # pass init params to Decoder
   UpperCAmelCase__      :					Union[str, Any]				       =							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 							UpperCamelCase  (					self		:  Union[str, Any]			,				snake_case__		:  torch.FloatTensor			,				snake_case__		:  bool = True ):



   '''simple docstring'''




   UpperCAmelCase__      :					List[Any]				       =							self.encoder(snake_case__ )
   UpperCAmelCase__      :					List[str]				       =							self.quant_conv(snake_case__ )

   if not return_dict:
     return (h,)

   return VQEncoderOutput(latents=snake_case__ )







 @apply_forward_hook
 def 							UpperCamelCase  (					self		:  List[str]			,				snake_case__		:  torch.FloatTensor			,				snake_case__		:  bool = False			,				snake_case__		:  bool = True ):



   '''simple docstring'''




   if not force_not_quantize:
     UpperCAmelCase__       ,			UpperCAmelCase__       ,			UpperCAmelCase__      :					Any				       =							self.quantize(snake_case__ )
   else:
     UpperCAmelCase__      :					Tuple				       =							h
   UpperCAmelCase__      :					int				       =							self.post_quant_conv(snake_case__ )
   UpperCAmelCase__      :					Optional[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 							UpperCamelCase  (					self		:  Any			,				snake_case__		:  torch.FloatTensor			,				snake_case__		:  bool = True ):



   '''simple docstring'''




   UpperCAmelCase__      :					Dict				       =							sample
   UpperCAmelCase__      :					int				       =							self.encode(snake_case__ ).latents
   UpperCAmelCase__      :					Any				       =							self.decode(snake_case__ ).sample

   if not return_dict:
     return (dec,)

   return DecoderOutput(sample=snake_case__ )



