




"""simple docstring"""

import unittest

from parameterized import parameterized

from transformers import OpenLlamaConfig, is_torch_available, set_seed
from transformers.testing_utils import require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
		import torch

		from transformers import OpenLlamaForCausalLM, OpenLlamaForSequenceClassification, OpenLlamaModel

class        lowercase      :







				def __init__(							self   :					Tuple      ,     _lowerCamelCase   :					str      ,     _lowerCamelCase   :					Optional[int]=13      ,     _lowerCamelCase   :					Union[str, Any]=7      ,     _lowerCamelCase   :					Optional[Any]=True      ,     _lowerCamelCase   :					Any=True      ,     _lowerCamelCase   :					List[Any]=False      ,     _lowerCamelCase   :					List[Any]=True      ,     _lowerCamelCase   :					Any=99      ,     _lowerCamelCase   :					List[str]=32      ,     _lowerCamelCase   :					int=5      ,     _lowerCamelCase   :					Any=4      ,     _lowerCamelCase   :					List[Any]=37      ,     _lowerCamelCase   :					int="gelu"      ,     _lowerCamelCase   :					Tuple=0.1      ,     _lowerCamelCase   :					Optional[int]=0.1      ,     _lowerCamelCase   :					Dict=5_12      ,     _lowerCamelCase   :					Optional[Any]=16      ,     _lowerCamelCase   :					List[Any]=2      ,     _lowerCamelCase   :					Dict=0.02      ,     _lowerCamelCase   :					Dict=3      ,     _lowerCamelCase   :					Dict=4      ,     _lowerCamelCase   :					int=None      ,     ):



					"""simple docstring"""

					A_  :				Dict	   =		parent
					A_  :				str	   =		batch_size
					A_  :				str	   =		seq_length
					A_  :				List[Any]	   =		is_training
					A_  :				List[Any]	   =		use_input_mask
					A_  :				str	   =		use_token_type_ids
					A_  :				Union[str, Any]	   =		use_labels
					A_  :				List[Any]	   =		vocab_size
					A_  :				Any	   =		hidden_size
					A_  :				Dict	   =		num_hidden_layers
					A_  :				Union[str, Any]	   =		num_attention_heads
					A_  :				Union[str, Any]	   =		intermediate_size
					A_  :				int	   =		hidden_act
					A_  :				int	   =		hidden_dropout_prob
					A_  :				List[str]	   =		attention_probs_dropout_prob
					A_  :				str	   =		max_position_embeddings
					A_  :				List[Any]	   =		type_vocab_size
					A_  :				str	   =		type_sequence_label_size
					A_  :				str	   =		initializer_range
					A_  :				str	   =		num_labels
					A_  :				List[Any]	   =		num_choices
					A_  :				str	   =		scope







				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					A_  :				Optional[int]	   =		ids_tensor([self.batch_size, self.seq_length]      ,     self.vocab_size       )

					A_  :				int	   =		None
					if self.use_input_mask:
						A_  :				Union[str, Any]	   =		random_attention_mask([self.batch_size, self.seq_length]       )

					A_  :				Optional[Any]	   =		None
					if self.use_token_type_ids:
						A_  :				Union[str, Any]	   =		ids_tensor([self.batch_size, self.seq_length]      ,     self.type_vocab_size       )

					A_  :				Optional[int]	   =		None
					A_  :				Union[str, Any]	   =		None
					A_  :				Dict	   =		None
					if self.use_labels:
						A_  :				Tuple	   =		ids_tensor([self.batch_size]      ,     self.type_sequence_label_size       )
						A_  :				int	   =		ids_tensor([self.batch_size, self.seq_length]      ,     self.num_labels       )
						A_  :				int	   =		ids_tensor([self.batch_size]      ,     self.num_choices       )

					A_  :				Any	   =		self.get_config()

					return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels







				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					return OpenLlamaConfig(
					    vocab_size=self.vocab_size      ,     hidden_size=self.hidden_size      ,     num_hidden_layers=self.num_hidden_layers      ,     num_attention_heads=self.num_attention_heads      ,     intermediate_size=self.intermediate_size      ,     hidden_act=self.hidden_act      ,     hidden_dropout_prob=self.hidden_dropout_prob      ,     attention_probs_dropout_prob=self.attention_probs_dropout_prob      ,     max_position_embeddings=self.max_position_embeddings      ,     type_vocab_size=self.type_vocab_size      ,     is_decoder=__A      ,     initializer_range=self.initializer_range      ,     use_stable_embedding=__A      ,     )







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



					"""simple docstring"""

					A_  :				Any	   =		OpenLlamaModel(config=__A       )
					model.to(__A       )
					model.eval()
					A_  :				Optional[int]	   =		model(__A      ,     attention_mask=__A       )
					A_  :				List[Any]	   =		model(__A       )
					self.parent.assertEqual(result.last_hidden_state.shape      ,     (self.batch_size, self.seq_length, self.hidden_size)       )







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



					"""simple docstring"""

					A_  :				List[str]	   =		True
					A_  :				Dict	   =		OpenLlamaModel(__A       )
					model.to(__A       )
					model.eval()
					A_  :				Optional[Any]	   =		model(
					    __A      ,     attention_mask=__A      ,     encoder_hidden_states=__A      ,     encoder_attention_mask=__A      ,     )
					A_  :				int	   =		model(
					    __A      ,     attention_mask=__A      ,     encoder_hidden_states=__A      ,     )
					A_  :				List[str]	   =		model(__A      ,     attention_mask=__A       )
					self.parent.assertEqual(result.last_hidden_state.shape      ,     (self.batch_size, self.seq_length, self.hidden_size)       )







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



					"""simple docstring"""

					A_  :				Dict	   =		OpenLlamaForCausalLM(config=__A       )
					model.to(__A       )
					model.eval()
					A_  :				List[Any]	   =		model(__A      ,     attention_mask=__A      ,     labels=__A       )
					self.parent.assertEqual(result.logits.shape      ,     (self.batch_size, self.seq_length, self.vocab_size)       )







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



					"""simple docstring"""

					A_  :				List[str]	   =		True
					A_  :				Dict	   =		True
					A_  :				Tuple	   =		OpenLlamaForCausalLM(config=__A       )
					model.to(__A       )
					model.eval()

					# first forward pass
					A_  :				Optional[Any]	   =		model(
					    __A      ,     attention_mask=__A      ,     encoder_hidden_states=__A      ,     encoder_attention_mask=__A      ,     use_cache=__A      ,     )
					A_  :				Tuple	   =		outputs.past_key_values

					# create hypothetical multiple next token and extent to next_input_ids
					A_  :				str	   =		ids_tensor((self.batch_size, 3)      ,     config.vocab_size       )
					A_  :				Optional[int]	   =		ids_tensor((self.batch_size, 3)      ,     vocab_size=2       )

					# append to next input_ids and
					A_  :				Union[str, Any]	   =		torch.cat([input_ids, next_tokens]      ,     dim=-1       )
					A_  :				str	   =		torch.cat([input_mask, next_mask]      ,     dim=-1       )

					A_  :				Dict	   =		model(
					    __A      ,     attention_mask=__A      ,     encoder_hidden_states=__A      ,     encoder_attention_mask=__A      ,     output_hidden_states=__A      ,     )['''hidden_states'''][0]
					A_  :				Optional[int]	   =		model(
					    __A      ,     attention_mask=__A      ,     encoder_hidden_states=__A      ,     encoder_attention_mask=__A      ,     past_key_values=__A      ,     output_hidden_states=__A      ,     )['''hidden_states'''][0]

					# select random slice
					A_  :				Tuple	   =		ids_tensor((1,)      ,     output_from_past.shape[-1]       ).item()
					A_  :				Optional[int]	   =		output_from_no_past[:, -3:, random_slice_idx].detach()
					A_  :				int	   =		output_from_past[:, :, random_slice_idx].detach()

					self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]       )

					# test that outputs are equal for slice
					self.parent.assertTrue(torch.allclose(__A      ,     __A      ,     atol=1E-3       )       )







				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					A_  :				Union[str, Any]	   =		self.prepare_config_and_inputs()
					(
					    (
					    A_						
					)   , (
					    A_						
					)   , (
					    A_						
					)   , (
					    A_						
					)   , (
					    A_						
					)   , (
					    A_						
					)   , (
					    A_						
					)   , 						
					)  :				Optional[int]	   =		config_and_inputs
					A_  :				Optional[Any]	   =		{'''input_ids''': input_ids, '''attention_mask''': input_mask}
					return config, inputs_dict

@require_torch
class        lowercase      (    UpperCamelCase__     ,	UpperCamelCase__     ,	UpperCamelCase__     ,	unittest.TestCase):
				__lowerCAmelCase    :    Union[str, Any]         =   (
				    (OpenLlamaModel, OpenLlamaForCausalLM, OpenLlamaForSequenceClassification) if is_torch_available() else ()
				)
				__lowerCAmelCase    :    Any         =   (OpenLlamaForCausalLM,) if is_torch_available() else ()
				__lowerCAmelCase    :    List[str]         =   (
				    {
				        """feature-extraction""": OpenLlamaModel,
				        """text-classification""": OpenLlamaForSequenceClassification,
				        """text-generation""": OpenLlamaForCausalLM,
				        """zero-shot""": OpenLlamaForSequenceClassification,
				    }
				    if is_torch_available()
				    else {}
				)
				__lowerCAmelCase    :    List[Any]         =   False
				__lowerCAmelCase    :    Union[str, Any]         =   False







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



					"""simple docstring"""

					A_  :				Any	   =		OpenLlamaModelTester(self       )
					A_  :				List[str]	   =		ConfigTester(self      ,     config_class=__A      ,     hidden_size=37       )







				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					self.config_tester.run_common_tests()







				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					A_  :				List[Any]	   =		self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*__A       )







				def 					a_  (							self   :					Optional[Any]       ):



					"""simple docstring"""

					A_  :				Dict	   =		self.model_tester.prepare_config_and_inputs()
					for type in ["absolute", "relative_key", "relative_key_query"]:
						A_  :				Dict	   =		type
						self.model_tester.create_and_check_model(*__A       )







				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					A_   , A_  :				Dict	   =		self.model_tester.prepare_config_and_inputs_for_common()
					A_  :				Union[str, Any]	   =		3
					A_  :				str	   =		input_dict['''input_ids''']
					A_  :				int	   =		input_ids.ne(1       ).to(__A       )
					A_  :				Tuple	   =		ids_tensor([self.model_tester.batch_size]      ,     self.model_tester.type_sequence_label_size       )
					A_  :				Tuple	   =		OpenLlamaForSequenceClassification(__A       )
					model.to(__A       )
					model.eval()
					A_  :				List[Any]	   =		model(__A      ,     attention_mask=__A      ,     labels=__A       )
					self.assertEqual(result.logits.shape      ,     (self.model_tester.batch_size, self.model_tester.num_labels)       )







				def 					a_  (							self   :					Dict       ):



					"""simple docstring"""

					A_   , A_  :				Optional[int]	   =		self.model_tester.prepare_config_and_inputs_for_common()
					A_  :				Dict	   =		3
					A_  :				int	   =		'''single_label_classification'''
					A_  :				int	   =		input_dict['''input_ids''']
					A_  :				int	   =		input_ids.ne(1       ).to(__A       )
					A_  :				Optional[int]	   =		ids_tensor([self.model_tester.batch_size]      ,     self.model_tester.type_sequence_label_size       )
					A_  :				Optional[Any]	   =		OpenLlamaForSequenceClassification(__A       )
					model.to(__A       )
					model.eval()
					A_  :				List[Any]	   =		model(__A      ,     attention_mask=__A      ,     labels=__A       )
					self.assertEqual(result.logits.shape      ,     (self.model_tester.batch_size, self.model_tester.num_labels)       )







				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					A_   , A_  :				Union[str, Any]	   =		self.model_tester.prepare_config_and_inputs_for_common()
					A_  :				int	   =		3
					A_  :				Optional[Any]	   =		'''multi_label_classification'''
					A_  :				Optional[Any]	   =		input_dict['''input_ids''']
					A_  :				str	   =		input_ids.ne(1       ).to(__A       )
					A_  :				Any	   =		ids_tensor(
					    [self.model_tester.batch_size, config.num_labels]      ,     self.model_tester.type_sequence_label_size       ).to(torch.float       )
					A_  :				Tuple	   =		OpenLlamaForSequenceClassification(__A       )
					model.to(__A       )
					model.eval()
					A_  :				str	   =		model(__A      ,     attention_mask=__A      ,     labels=__A       )
					self.assertEqual(result.logits.shape      ,     (self.model_tester.batch_size, self.model_tester.num_labels)       )







				@unittest.skip('''Open-Llama buffers include complex numbers, which breaks this test'''       )
				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					pass







				@parameterized.expand([('''linear''',), ('''dynamic''',)]       )
				def 					a_  (							self   :					Tuple      ,     _lowerCamelCase   :					str       ):



					"""simple docstring"""

					A_   , A_  :				List[str]	   =		self.model_tester.prepare_config_and_inputs_for_common()
					A_  :				Union[str, Any]	   =		ids_tensor([1, 10]      ,     config.vocab_size       )
					A_  :				Union[str, Any]	   =		ids_tensor([1, int(config.max_position_embeddings * 1.5       )]      ,     config.vocab_size       )

					set_seed(42       )  # Fixed seed at init time so the two models get the same random weights
					A_  :				Optional[Any]	   =		OpenLlamaModel(__A       )
					original_model.to(__A       )
					original_model.eval()
					A_  :				List[Any]	   =		original_model(__A       ).last_hidden_state
					A_  :				str	   =		original_model(__A       ).last_hidden_state

					set_seed(42       )  # Fixed seed at init time so the two models get the same random weights
					A_  :				List[str]	   =		{'''type''': scaling_type, '''factor''': 10.0}
					A_  :				Dict	   =		OpenLlamaModel(__A       )
					scaled_model.to(__A       )
					scaled_model.eval()
					A_  :				int	   =		scaled_model(__A       ).last_hidden_state
					A_  :				Any	   =		scaled_model(__A       ).last_hidden_state

					# Dynamic scaling does not change the RoPE embeddings until it receives an input longer than the original
					# maximum sequence length, so the outputs for the short input should match.
					if scaling_type == "dynamic":
						self.assertTrue(torch.allclose(__A      ,     __A      ,     atol=1E-5       )       )
					else:
						self.assertFalse(torch.allclose(__A      ,     __A      ,     atol=1E-5       )       )

					# The output should be different for long inputs
					self.assertFalse(torch.allclose(__A      ,     __A      ,     atol=1E-5       )       )









"""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)





