




"""simple docstring"""

# Lint as: python3
import os
import re
import urllib.parse
from pathlib import Path
from typing import Callable, List, Optional, Union
from zipfile import ZipFile

from ..utils.file_utils import cached_path, hf_github_url
from ..utils.logging import get_logger
from ..utils.version import Version


_lowerCamelCase  :			Optional[int]        =       get_logger(__name__)

class        lowercase      :
				__lowerCAmelCase    :    Any         =   """dummy_data"""
				__lowerCAmelCase    :    Any         =   """datasets"""
				__lowerCAmelCase    :    Dict         =   False







				def __init__(							self   :					Union[str, Any]      ,     _lowerCamelCase   :					str      ,     _lowerCamelCase   :					str      ,     _lowerCamelCase   :					Union[Version, str]      ,     _lowerCamelCase   :					Optional[str] = None      ,     _lowerCamelCase   :					bool = False      ,     _lowerCamelCase   :					bool = True      ,     _lowerCamelCase   :					Optional[List[Callable]] = None      ,     ):



					"""simple docstring"""

					A_  :				Optional[Any]	   =		0
					A_  :				Optional[int]	   =		dataset_name
					A_  :				Dict	   =		cache_dir
					A_  :				str	   =		use_local_dummy_data
					A_  :				Any	   =		config
					# download_callbacks take a single url as input
					A_  :				List[Callable]	   =		download_callbacks or []
					# if False, it doesn't load existing files and it returns the paths of the dummy files relative
					# to the dummy_data zip file root
					A_  :				int	   =		load_existing_dummy_data

					# TODO(PVP, QL) might need to make this more general
					A_  :				Optional[Any]	   =		str(_lowerCamelCase       )
					# to be downloaded
					A_  :				List[str]	   =		None
					A_  :				Union[str, Any]	   =		None







				@property
				def 					a_  (							self   :					Tuple       ):



					"""simple docstring"""

					if self._dummy_file is None:
						A_  :				Dict	   =		self.download_dummy_data()
					return self._dummy_file







				@property
				def 					a_  (							self   :					Union[str, Any]       ):



					"""simple docstring"""

					if self.config is not None:
						# structure is dummy / config_name / version_name
						return os.path.join('''dummy'''      ,     self.config.name      ,     self.version_name       )
					# structure is dummy / version_name
					return os.path.join('''dummy'''      ,     self.version_name       )







				@property
				def 					a_  (							self   :					Optional[Any]       ):



					"""simple docstring"""

					return os.path.join(self.dummy_data_folder      ,     '''dummy_data.zip'''       )







				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					A_  :				Dict	   =		(
					    self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
					)

					A_  :				str	   =		cached_path(
					    _lowerCamelCase      ,     cache_dir=self.cache_dir      ,     extract_compressed_file=_lowerCamelCase      ,     force_extract=_lowerCamelCase       )

					return os.path.join(_lowerCamelCase      ,     self.dummy_file_name       )







				@property
				def 					a_  (							self   :					str       ):



					"""simple docstring"""

					return os.path.join(self.datasets_scripts_dir      ,     self.dataset_name      ,     self.dummy_zip_file       )







				@property
				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					if self._bucket_url is None:
						A_  :				int	   =		hf_github_url(self.dataset_name      ,     self.dummy_zip_file.replace(os.sep      ,     '''/'''       )       )
					return self._bucket_url







				@property
				def 					a_  (							self   :					Optional[Any]       ):



					"""simple docstring"""

					if os.path.isdir(self.dummy_file       ):
						return self.dummy_file
					# else cut off path to file -> example `xsum`.
					return "/".join(self.dummy_file.replace(os.sep      ,     '''/'''       ).split('''/'''       )[:-1]       )







				def 					a_  (							self   :					Tuple      ,     _lowerCamelCase   :					Any      ,     *_lowerCamelCase   :					List[str]       ):



					"""simple docstring"""

					if self.load_existing_dummy_data:
						# dummy data is downloaded and tested
						A_  :				Tuple	   =		self.dummy_file
					else:
						# dummy data cannot be downloaded and only the path to dummy file is returned
						A_  :				Any	   =		self.dummy_file_name

					# special case when data_url is a dict
					if isinstance(_lowerCamelCase      ,     _lowerCamelCase       ):
						return self.create_dummy_data_dict(_lowerCamelCase      ,     _lowerCamelCase       )
					elif isinstance(_lowerCamelCase      ,     (list, tuple)       ):
						return self.create_dummy_data_list(_lowerCamelCase      ,     _lowerCamelCase       )
					else:
						return self.create_dummy_data_single(_lowerCamelCase      ,     _lowerCamelCase       )







				def 					a_  (							self   :					int      ,     _lowerCamelCase   :					Tuple      ,     *_lowerCamelCase   :					Any       ):



					"""simple docstring"""

					return self.download_and_extract(_lowerCamelCase       )







				def 					a_  (							self   :					Tuple      ,     _lowerCamelCase   :					Optional[Any]      ,     _lowerCamelCase   :					Any       ):



					"""simple docstring"""

					return self.download_and_extract(_lowerCamelCase       )







				def 					a_  (							self   :					str      ,     _lowerCamelCase   :					Optional[int]      ,     *_lowerCamelCase   :					int      ,     **_lowerCamelCase   :					Optional[int]       ):



					"""simple docstring"""

					return path







				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					return {}







				def 					a_  (							self   :					List[str]      ,     _lowerCamelCase   :					Dict      ,     _lowerCamelCase   :					List[Any]       ):



					"""simple docstring"""

					A_  :				int	   =		{}
					for key, single_urls in data_url.items():
						for download_callback in self.download_callbacks:
							if isinstance(_lowerCamelCase      ,     _lowerCamelCase       ):
								for single_url in single_urls:
									download_callback(_lowerCamelCase       )
							else:
								A_  :				Tuple	   =		single_urls
								download_callback(_lowerCamelCase       )
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
						if isinstance(_lowerCamelCase      ,     _lowerCamelCase       ):
							A_  :				List[Any]	   =		[os.path.join(_lowerCamelCase      ,     urllib.parse.quote_plus(Path(_lowerCamelCase       ).name       )       ) for x in single_urls]
						else:
							A_  :				List[str]	   =		single_urls
							A_  :				Tuple	   =		os.path.join(_lowerCamelCase      ,     urllib.parse.quote_plus(Path(_lowerCamelCase       ).name       )       )
						A_  :				int	   =		value

					# make sure that values are unique
					if all(isinstance(_lowerCamelCase      ,     _lowerCamelCase       ) for i in dummy_data_dict.values()       ) and len(set(dummy_data_dict.values()       )       ) < len(
					    dummy_data_dict.values()       ):
						# append key to value to make its name unique
						A_  :				Optional[Any]	   =		{key: value + key for key, value in dummy_data_dict.items()}

					return dummy_data_dict







				def 					a_  (							self   :					Union[str, Any]      ,     _lowerCamelCase   :					Optional[int]      ,     _lowerCamelCase   :					Optional[Any]       ):



					"""simple docstring"""

					A_  :				Optional[int]	   =		[]
					# trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
					A_  :				List[str]	   =		all(bool(re.findall('''[0-9]{3,}-of-[0-9]{3,}'''      ,     _lowerCamelCase       )       ) for url in data_url       )
					A_  :				Optional[int]	   =		all(
					    url.startswith('''https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/pubmed'''       ) for url in data_url       )
					if data_url and (is_tf_records or is_pubmed_records):
						A_  :				List[Any]	   =		[data_url[0]] * len(_lowerCamelCase       )
					for single_url in data_url:
						for download_callback in self.download_callbacks:
							download_callback(_lowerCamelCase       )
						# we force the name of each key to be the last file / folder name of the url path
						# if the url has arguments, we need to encode them with urllib.parse.quote_plus
						A_  :				List[Any]	   =		os.path.join(_lowerCamelCase      ,     urllib.parse.quote_plus(single_url.split('''/'''       )[-1]       )       )
						dummy_data_list.append(_lowerCamelCase       )
					return dummy_data_list







				def 					a_  (							self   :					Any      ,     _lowerCamelCase   :					str      ,     _lowerCamelCase   :					List[Any]       ):



					"""simple docstring"""

					for download_callback in self.download_callbacks:
						download_callback(_lowerCamelCase       )
					# we force the name of each key to be the last file / folder name of the url path
					# if the url has arguments, we need to encode them with urllib.parse.quote_plus
					A_  :				Optional[int]	   =		os.path.join(_lowerCamelCase      ,     urllib.parse.quote_plus(data_url.split('''/'''       )[-1]       )       )
					if os.path.exists(_lowerCamelCase       ) or not self.load_existing_dummy_data:
						return value
					else:
						# Backward compatibility, maybe deprecate at one point.
						# For many datasets with single url calls to dl_manager.download_and_extract,
						# the dummy_data.zip file is actually the zipped downloaded file
						# while now we expected the dummy_data.zip file to be a directory containing
						# the downloaded file.
						return path_to_dummy_data







				def 					a_  (							self   :					Tuple       ):



					"""simple docstring"""

					pass







				def 					a_  (							self   :					Optional[Any]       ):



					"""simple docstring"""

					pass







				def 					a_  (							self   :					Optional[Any]      ,     _lowerCamelCase   :					Any       ):



					"""simple docstring"""

					def _iter_archive_members(_lowerCamelCase   :					List[Any]       ):
						# this preserves the order of the members inside the ZIP archive
						A_  :				List[str]	   =		Path(self.dummy_file       ).parent
						A_  :				List[str]	   =		path.relative_to(_lowerCamelCase       )
						with ZipFile(self.local_path_to_dummy_data       ) as zip_file:
							A_  :				List[str]	   =		zip_file.namelist()
						for member in members:
							if member.startswith(relative_path.as_posix()       ):
								yield dummy_parent_path.joinpath(_lowerCamelCase       )

					A_  :				Optional[Any]	   =		Path(_lowerCamelCase       )
					A_  :				List[Any]	   =		_iter_archive_members(_lowerCamelCase       ) if self.use_local_dummy_data else path.rglob('''*'''       )
					for file_path in file_paths:
						if file_path.is_file() and not file_path.name.startswith(('''.''', '''__''')       ):
							yield file_path.relative_to(_lowerCamelCase       ).as_posix(), file_path.open('''rb'''       )







				def 					a_  (							self   :					Dict      ,     _lowerCamelCase   :					Optional[int]       ):



					"""simple docstring"""

					if not isinstance(_lowerCamelCase      ,     _lowerCamelCase       ):
						A_  :				int	   =		[paths]
					for path in paths:
						if os.path.isfile(_lowerCamelCase       ):
							if os.path.basename(_lowerCamelCase       ).startswith(('''.''', '''__''')       ):
								return
							yield path
						else:
							for dirpath, dirnames, filenames in os.walk(_lowerCamelCase       ):
								if os.path.basename(_lowerCamelCase       ).startswith(('''.''', '''__''')       ):
									continue
								dirnames.sort()
								for filename in sorted(_lowerCamelCase       ):
									if filename.startswith(('''.''', '''__''')       ):
										continue
									yield os.path.join(_lowerCamelCase      ,     _lowerCamelCase       )











"""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()
_lowerCamelCase  :			Optional[Any]        =       logging.get_logger(__name__)

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

_lowerCamelCase  :			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 		lowercase_      (		_UpperCAmelCase	):




	"""simple docstring"""






	A_  :				Union[str, Any]	   =		torch.load(_UpperCAmelCase  ,      map_location='''cpu'''	)
	return sd

def 		lowercase_      (		_UpperCAmelCase  ,      _UpperCAmelCase  ,      _UpperCAmelCase=rename_keys_prefix	):




	"""simple docstring"""






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

@torch.no_grad()
def 		lowercase_      (		_UpperCAmelCase  ,      _UpperCAmelCase	):




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

	A_  :				Dict	   =		VisualBertConfig(**_UpperCAmelCase	)

	# Load State Dict
	A_  :				List[str]	   =		load_state_dict(_UpperCAmelCase	)

	A_  :				List[str]	   =		get_new_dict(_UpperCAmelCase  ,      _UpperCAmelCase	)

	if model_type == "pretraining":
		A_  :				List[Any]	   =		VisualBertForPreTraining(_UpperCAmelCase	)
	elif model_type == "vqa":
		A_  :				Optional[Any]	   =		VisualBertForQuestionAnswering(_UpperCAmelCase	)
	elif model_type == "nlvr":
		A_  :				Any	   =		VisualBertForVisualReasoning(_UpperCAmelCase	)
	elif model_type == "multichoice":
		A_  :				Optional[int]	   =		VisualBertForMultipleChoice(_UpperCAmelCase	)

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


if __name__ == "__main__":
		_lowerCamelCase  :			Dict        =       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.')
		_lowerCamelCase  :			Optional[int]        =       parser.parse_args()
		convert_visual_bert_checkpoint(args.orig_checkpoint_path, args.pytorch_dump_folder_path)





