
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()
_SCREAMING_SNAKE_CASE    :      Union[str, Any]          =			logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE    :      Tuple          =			[
    ('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'),
]

_SCREAMING_SNAKE_CASE    :      Any          =			[
    '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 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Union[str, Any]       ) ->       Optional[Any]:

				"""simple docstring"""







				A__  :	int         =							torch.load(__UpperCamelCase							,						map_location='''cpu'''       )
				return sd

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Tuple							,						__UpperCamelCase :						Union[str, Any]							,						__UpperCamelCase :						Tuple=rename_keys_prefix       ) ->       int:

				"""simple docstring"""







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

@torch.no_grad()
def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						int							,						__UpperCamelCase :						List[str]       ) ->       str:

				"""simple docstring"""







				assert (
				    checkpoint_path.split('''/'''       )[-1] in ACCEPTABLE_CHECKPOINTS
				), F"The checkpoint provided must be in {ACCEPTABLE_CHECKPOINTS}."

				# Get Config
				if "pre" in checkpoint_path:
								A__  :	str         =							'''pretraining'''
								if "vcr" in checkpoint_path:
												A__  :	int         =							{'''visual_embedding_dim''': 5_12}
								elif "vqa_advanced" in checkpoint_path:
												A__  :	List[str]         =							{'''visual_embedding_dim''': 20_48}
								elif "vqa" in checkpoint_path:
												A__  :	Optional[Any]         =							{'''visual_embedding_dim''': 20_48}
								elif "nlvr" in checkpoint_path:
												A__  :	Any         =							{'''visual_embedding_dim''': 10_24}
								else:
												raise NotImplementedError(F"No implementation found for `{checkpoint_path}`."       )
				else:
								if "vcr" in checkpoint_path:
												A__  :	int         =							{'''visual_embedding_dim''': 5_12}
												A__  :	str         =							'''multichoice'''
								elif "vqa_advanced" in checkpoint_path:
												A__  :	Union[str, Any]         =							{'''visual_embedding_dim''': 20_48}
												A__  :	List[Any]         =							'''vqa_advanced'''
								elif "vqa" in checkpoint_path:
												A__  :	Tuple         =							{'''visual_embedding_dim''': 20_48, '''num_labels''': 31_29}
												A__  :	Optional[int]         =							'''vqa'''
								elif "nlvr" in checkpoint_path:
												A__  :	Dict         =							{
												    '''visual_embedding_dim''': 10_24,
												    '''num_labels''': 2,
												}
												A__  :	Optional[int]         =							'''nlvr'''

				A__  :	str         =							VisualBertConfig(**__UpperCamelCase       )

				# Load State Dict
				A__  :	int         =							load_state_dict(__UpperCamelCase       )

				A__  :	Optional[Any]         =							get_new_dict(__UpperCamelCase							,						__UpperCamelCase       )

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

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


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

import inspect
import os
import unittest

import torch

import accelerate
from accelerate import debug_launcher
from accelerate.test_utils import (
    execute_subprocess_async,
    require_cpu,
    require_huggingface_suite,
    require_multi_gpu,
    require_single_gpu,
)
from accelerate.utils import patch_environment







@require_huggingface_suite
class  UpperCamelCase__		(    unittest.TestCase   ):







	'''simple docstring'''





	def   __snake_case				(			self		):
					A__  :	Dict         =							inspect.getfile(accelerate.test_utils		)
					A__  :	Any         =							os.path.sep.join(
					    mod_file.split(os.path.sep		)[:-1] + ['''scripts''', '''external_deps''', '''test_metrics.py''']		)

					from accelerate.test_utils.scripts.external_deps import test_metrics  # noqa: F401

					A__  :	Tuple         =							test_metrics





	@require_cpu
	def   __snake_case				(			self		):
					debug_launcher(self.test_metrics.main				,       num_processes=1		)





	@require_cpu
	def   __snake_case				(			self		):
					debug_launcher(self.test_metrics.main		)





	@require_single_gpu
	def   __snake_case				(			self		):
					self.test_metrics.main()





	@require_multi_gpu
	def   __snake_case				(			self		):
					print(F"Found {torch.cuda.device_count()} devices."		)
					A__  :	int         =							['''torchrun''', F"--nproc_per_node={torch.cuda.device_count()}", self.test_file_path]
					with patch_environment(omp_num_threads=1		):
									execute_subprocess_async(UpperCamelCase__				,       env=os.environ.copy()		)