

'''simple docstring'''

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

A_       						=							[
    ("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"),
]

A_       						=							[
    "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    A_  ( snake_case ):
  SCREAMING_SNAKE_CASE:int						      =					torch.load(snake_case      ,       map_location="cpu" )
  return sd





def    A_  ( snake_case      ,       snake_case      ,       snake_case=rename_keys_prefix ):
  SCREAMING_SNAKE_CASE:Dict						      =					OrderedDict()
  SCREAMING_SNAKE_CASE:Tuple						      =					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:Optional[Any]						      =					key
    for name_pair in rename_keys_prefix:
      SCREAMING_SNAKE_CASE:Optional[Any]						      =					new_key.replace(name_pair[0]      ,       name_pair[1] )
    SCREAMING_SNAKE_CASE:int						      =					d[key]
    if key == "bert.cls.predictions.decoder.weight":
      # Old bert code didn't have `decoder.bias`, but was added separately
      SCREAMING_SNAKE_CASE:List[Any]						      =					new_d['cls.predictions.bias']
  return new_d





@torch.no_grad()
def    A_  ( snake_case      ,       snake_case ):
  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:Any						      =					'pretraining'
    if "vcr" in checkpoint_path:
      SCREAMING_SNAKE_CASE:Dict						      =					{'visual_embedding_dim': 512}
    elif "vqa_advanced" in checkpoint_path:
      SCREAMING_SNAKE_CASE:Tuple						      =					{'visual_embedding_dim': 2048}
    elif "vqa" in checkpoint_path:
      SCREAMING_SNAKE_CASE:List[Any]						      =					{'visual_embedding_dim': 2048}
    elif "nlvr" in checkpoint_path:
      SCREAMING_SNAKE_CASE:Optional[int]						      =					{'visual_embedding_dim': 1024}
    else:
      raise NotImplementedError(F'''No implementation found for `{checkpoint_path}`.''' )
  else:
    if "vcr" in checkpoint_path:
      SCREAMING_SNAKE_CASE:Tuple						      =					{'visual_embedding_dim': 512}
      SCREAMING_SNAKE_CASE:List[Any]						      =					'multichoice'
    elif "vqa_advanced" in checkpoint_path:
      SCREAMING_SNAKE_CASE:Tuple						      =					{'visual_embedding_dim': 2048}
      SCREAMING_SNAKE_CASE:str						      =					'vqa_advanced'
    elif "vqa" in checkpoint_path:
      SCREAMING_SNAKE_CASE:List[str]						      =					{'visual_embedding_dim': 2048, 'num_labels': 3129}
      SCREAMING_SNAKE_CASE:Optional[Any]						      =					'vqa'
    elif "nlvr" in checkpoint_path:
      SCREAMING_SNAKE_CASE:int						      =					{
          'visual_embedding_dim': 1024,
          'num_labels': 2,
      }
      SCREAMING_SNAKE_CASE:List[Any]						      =					'nlvr'

  SCREAMING_SNAKE_CASE:Any						      =					VisualBertConfig(**snake_case )

  # Load State Dict
  SCREAMING_SNAKE_CASE:Any						      =					load_state_dict(snake_case )

  SCREAMING_SNAKE_CASE:Optional[Any]						      =					get_new_dict(snake_case      ,       snake_case )

  if model_type == "pretraining":
    SCREAMING_SNAKE_CASE:Union[str, Any]						      =					VisualBertForPreTraining(snake_case )
  elif model_type == "vqa":
    SCREAMING_SNAKE_CASE:int						      =					VisualBertForQuestionAnswering(snake_case )
  elif model_type == "nlvr":
    SCREAMING_SNAKE_CASE:Union[str, Any]						      =					VisualBertForVisualReasoning(snake_case )
  elif model_type == "multichoice":
    SCREAMING_SNAKE_CASE:List[Any]						      =					VisualBertForMultipleChoice(snake_case )

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


if __name__ == "__main__":
  A_       						=							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.")
  A_       						=							parser.parse_args()
  convert_visual_bert_checkpoint(args.orig_checkpoint_path, args.pytorch_dump_folder_path)









'''simple docstring'''

from importlib import import_module

from .logging import get_logger


A_       						=							get_logger(__name__)





class 							_snake_case :
   def __init__( self			:      Dict     ,SCREAMING_SNAKE_CASE__			:      Any     ,SCREAMING_SNAKE_CASE__			:      Dict=None							):
     SCREAMING_SNAKE_CASE:Any						      =					attrs or []
     if module is not None:
       for key in module.__dict__:
         if key in attrs or not key.startswith("__"							):
           setattr(self     ,SCREAMING_SNAKE_CASE__     ,getattr(SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__							)							)
     SCREAMING_SNAKE_CASE:Any						      =					module._original_module if isinstance(SCREAMING_SNAKE_CASE__     ,_PatchedModuleObj							) else module





class 							_snake_case :
   _A       :     Any       =					[]
   def __init__( self			:      Any     ,SCREAMING_SNAKE_CASE__			:      List[str]     ,SCREAMING_SNAKE_CASE__			:      str     ,SCREAMING_SNAKE_CASE__			:      str     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]=None							):
     SCREAMING_SNAKE_CASE:Any						      =					obj
     SCREAMING_SNAKE_CASE:Any						      =					target
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					new
     SCREAMING_SNAKE_CASE:Any						      =					target.split("."							)[0]
     SCREAMING_SNAKE_CASE:str						      =					{}
     SCREAMING_SNAKE_CASE:List[Any]						      =					attrs or []
   def __enter__( self			:      Tuple							):
     *SCREAMING_SNAKE_CASE      ,    SCREAMING_SNAKE_CASE:int						      =					self.target.split("."							)

     # Patch modules:
     # it's used to patch attributes of submodules like "os.path.join";
     # in this case we need to patch "os" and "os.path"

     for i in range(len(SCREAMING_SNAKE_CASE__							)							):
       try:
         SCREAMING_SNAKE_CASE:int						      =					import_module(".".join(submodules[: i + 1]							)							)
       except ModuleNotFoundError:
         continue
       # We iterate over all the globals in self.obj in case we find "os" or "os.path"
       for attr in self.obj.__dir__():
         SCREAMING_SNAKE_CASE:str						      =					getattr(self.obj     ,SCREAMING_SNAKE_CASE__							)
         # We don't check for the name of the global, but rather if its value *is* "os" or "os.path".
         # This allows to patch renamed modules like "from os import path as ospath".
         if obj_attr is submodule or (
             (isinstance(SCREAMING_SNAKE_CASE__     ,_PatchedModuleObj							) and obj_attr._original_module is submodule)
         ):
           SCREAMING_SNAKE_CASE:Union[str, Any]						      =					obj_attr
           # patch at top level
           setattr(self.obj     ,SCREAMING_SNAKE_CASE__     ,_PatchedModuleObj(SCREAMING_SNAKE_CASE__     ,attrs=self.attrs							)							)
           SCREAMING_SNAKE_CASE:int						      =					getattr(self.obj     ,SCREAMING_SNAKE_CASE__							)
           # construct lower levels patches
           for key in submodules[i + 1 :]:
             setattr(SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__     ,_PatchedModuleObj(getattr(SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__							)     ,attrs=self.attrs							)							)
             SCREAMING_SNAKE_CASE:Dict						      =					getattr(SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__							)
           # finally set the target attribute
           setattr(SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__     ,self.new							)

        # Patch attribute itself:
        # it's used for builtins like "open",
        # and also to patch "os.path.join" we may also need to patch "join"
        # itself if it was imported as "from os.path import join".

     if submodules:  # if it's an attribute of a submodule like "os.path.join"
       try:
         SCREAMING_SNAKE_CASE:Any						      =					getattr(import_module(".".join(SCREAMING_SNAKE_CASE__							)							)     ,SCREAMING_SNAKE_CASE__							)
       except (AttributeError, ModuleNotFoundError):
         return
       # We iterate over all the globals in self.obj in case we find "os.path.join"
       for attr in self.obj.__dir__():
         # We don't check for the name of the global, but rather if its value *is* "os.path.join".
         # This allows to patch renamed attributes like "from os.path import join as pjoin".
         if getattr(self.obj     ,SCREAMING_SNAKE_CASE__							) is attr_value:
           SCREAMING_SNAKE_CASE:Union[str, Any]						      =					getattr(self.obj     ,SCREAMING_SNAKE_CASE__							)
           setattr(self.obj     ,SCREAMING_SNAKE_CASE__     ,self.new							)
     elif target_attr in globals()["__builtins__"]:  # if it'a s builtin like "open"
       SCREAMING_SNAKE_CASE:int						      =					globals()["__builtins__"][target_attr]
       setattr(self.obj     ,SCREAMING_SNAKE_CASE__     ,self.new							)
     else:
       raise RuntimeError(F'''Tried to patch attribute {target_attr} instead of a submodule.'''							)
   def __exit__( self			:      List[Any]     ,*SCREAMING_SNAKE_CASE__			:      Union[str, Any]							):
     for attr in list(self.original							):
       setattr(self.obj     ,SCREAMING_SNAKE_CASE__     ,self.original.pop(SCREAMING_SNAKE_CASE__							)							)
   def     __UpperCamelCase						( self			:      Optional[int]							):
     self.__enter__()
     self._active_patches.append(self							)







   def     __UpperCamelCase						( self			:      Dict							):
     try:
       self._active_patches.remove(self							)
     except ValueError:
       # If the patch hasn't been started this will fail
       return None

     return self.__exit__()






