
"""simple docstring"""




import unittest
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory

from transformers import BertConfig, BertTokenizerFast, FeatureExtractionPipeline
from transformers.convert_graph_to_onnx import (
    convert,
    ensure_valid_input,
    generate_identified_filename,
    infer_shapes,
    quantize,
)
from transformers.testing_utils import require_tf, require_tokenizers, require_torch, slow



class 	lowerCAmelCase_						:


				"""simple docstring"""



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )					->		Optional[int]:



											"""simple docstring"""




											return None






class 	lowerCAmelCase_						:


				"""simple docstring"""



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )					->		Optional[int]:



											"""simple docstring"""




											return None






class 	lowerCAmelCase_						(unittest.TestCase  ):


				"""simple docstring"""



				__UpperCamelCase      :						Any					       =   [
				    # (model_name, model_kwargs)
				    ('''bert-base-cased''', {}),
				    ('''gpt2''', {'''use_cache''': False}),  # We don't support exporting GPT2 past keys anymore
				]



				@require_tf
				@slow
				def 	__magic_name__    (self    )					->		int:



											"""simple docstring"""




											for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
																		self._test_export(SCREAMING_SNAKE_CASE__   ,       """tf"""   ,       12   ,       **SCREAMING_SNAKE_CASE__    )



				@require_torch
				@slow
				def 	__magic_name__    (self    )					->		Optional[Any]:



											"""simple docstring"""




											for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
																		self._test_export(SCREAMING_SNAKE_CASE__   ,       """pt"""   ,       12   ,       **SCREAMING_SNAKE_CASE__    )



				@require_torch
				@slow
				def 	__magic_name__    (self    )					->		Union[str, Any]:



											"""simple docstring"""




											from transformers import BertModel

											SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   ["""[UNK]""", """[SEP]""", """[CLS]""", """[PAD]""", """[MASK]""", """some""", """other""", """words"""]
											with NamedTemporaryFile(mode="""w+t"""    ) as vocab_file:
																		vocab_file.write("""\n""".join(SCREAMING_SNAKE_CASE__    )    )
																		vocab_file.flush()
																		SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   BertTokenizerFast(vocab_file.name    )

											with TemporaryDirectory() as bert_save_dir:
																		SCREAMING_SNAKE_CASE__						:       int           =   BertModel(BertConfig(vocab_size=len(SCREAMING_SNAKE_CASE__    )    )    )
																		model.save_pretrained(SCREAMING_SNAKE_CASE__    )
																		self._test_export(SCREAMING_SNAKE_CASE__   ,       """pt"""   ,       12   ,       SCREAMING_SNAKE_CASE__    )



				@require_tf
				@slow
				def 	__magic_name__    (self    )					->		List[str]:



											"""simple docstring"""




											for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
																		SCREAMING_SNAKE_CASE__						:       Any           =   self._test_export(SCREAMING_SNAKE_CASE__   ,       """tf"""   ,       12   ,       **SCREAMING_SNAKE_CASE__    )
																		SCREAMING_SNAKE_CASE__						:       Optional[int]           =   quantize(Path(SCREAMING_SNAKE_CASE__    )    )

																		# Ensure the actual quantized model is not bigger than the original one
																		if quantized_path.stat().st_size >= Path(SCREAMING_SNAKE_CASE__    ).stat().st_size:
																									self.fail("""Quantized model is bigger than initial ONNX model"""    )



				@require_torch
				@slow
				def 	__magic_name__    (self    )					->		Optional[Any]:



											"""simple docstring"""




											for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
																		SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   self._test_export(SCREAMING_SNAKE_CASE__   ,       """pt"""   ,       12   ,       **SCREAMING_SNAKE_CASE__    )
																		SCREAMING_SNAKE_CASE__						:       List[Any]           =   quantize(SCREAMING_SNAKE_CASE__    )

																		# Ensure the actual quantized model is not bigger than the original one
																		if quantized_path.stat().st_size >= Path(SCREAMING_SNAKE_CASE__    ).stat().st_size:
																									self.fail("""Quantized model is bigger than initial ONNX model"""    )



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__=None   ,       **SCREAMING_SNAKE_CASE__    )					->		List[Any]:



											"""simple docstring"""




											try:
																		# Compute path
																		with TemporaryDirectory() as tempdir:
																									SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   Path(SCREAMING_SNAKE_CASE__    ).joinpath("""model.onnx"""    )

																		# Remove folder if exists
																		if path.parent.exists():
																									path.parent.rmdir()

																		# Export
																		convert(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       **SCREAMING_SNAKE_CASE__    )

																		return path
											except Exception as e:
																		self.fail(SCREAMING_SNAKE_CASE__    )



				@require_torch
				@require_tokenizers
				@slow
				def 	__magic_name__    (self    )					->		Any:



											"""simple docstring"""




											from transformers import BertModel

											SCREAMING_SNAKE_CASE__						:       Dict           =   BertModel(BertConfig.from_pretrained("""lysandre/tiny-bert-random"""    )    )
											SCREAMING_SNAKE_CASE__						:       Any           =   BertTokenizerFast.from_pretrained("""lysandre/tiny-bert-random"""    )
											self._test_infer_dynamic_axis(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       """pt"""    )



				@require_tf
				@require_tokenizers
				@slow
				def 	__magic_name__    (self    )					->		List[Any]:



											"""simple docstring"""




											from transformers import TFBertModel

											SCREAMING_SNAKE_CASE__						:       Dict           =   TFBertModel(BertConfig.from_pretrained("""lysandre/tiny-bert-random"""    )    )
											SCREAMING_SNAKE_CASE__						:       List[Any]           =   BertTokenizerFast.from_pretrained("""lysandre/tiny-bert-random"""    )
											self._test_infer_dynamic_axis(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       """tf"""    )



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )					->		Optional[int]:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__						:       int           =   FeatureExtractionPipeline(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )

											SCREAMING_SNAKE_CASE__						:       Dict           =   ["""input_ids""", """token_type_ids""", """attention_mask""", """output_0""", """output_1"""]
											SCREAMING_SNAKE_CASE__      ,      SCREAMING_SNAKE_CASE__      ,      SCREAMING_SNAKE_CASE__      ,      SCREAMING_SNAKE_CASE__						:       Any           =   infer_shapes(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )

											# Assert all variables are present
											self.assertEqual(len(SCREAMING_SNAKE_CASE__    )   ,       len(SCREAMING_SNAKE_CASE__    )    )
											self.assertTrue(all(var_name in shapes for var_name in variable_names    )    )
											self.assertSequenceEqual(variable_names[:3]   ,       SCREAMING_SNAKE_CASE__    )
											self.assertSequenceEqual(variable_names[3:]   ,       SCREAMING_SNAKE_CASE__    )

											# Assert inputs are {0: batch, 1: sequence}
											for var_name in ["input_ids", "token_type_ids", "attention_mask"]:
																		self.assertDictEqual(shapes[var_name]   ,       {0: """batch""", 1: """sequence"""}    )

											# Assert outputs are {0: batch, 1: sequence} and {0: batch}
											self.assertDictEqual(shapes["""output_0"""]   ,       {0: """batch""", 1: """sequence"""}    )
											self.assertDictEqual(shapes["""output_1"""]   ,       {0: """batch"""}    )



				def 	__magic_name__    (self    )					->		Optional[Any]:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__						:       Union[str, Any]           =   ["""input_ids""", """attention_mask""", """token_type_ids"""]
											SCREAMING_SNAKE_CASE__						:       List[str]           =   {"""input_ids""": [1, 2, 3, 4], """attention_mask""": [0, 0, 0, 0], """token_type_ids""": [1, 1, 1, 1]}
											SCREAMING_SNAKE_CASE__      ,      SCREAMING_SNAKE_CASE__						:       int           =   ensure_valid_input(FuncContiguousArgs()   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )

											# Should have exactly the same number of args (all are valid)
											self.assertEqual(len(SCREAMING_SNAKE_CASE__    )   ,       3    )

											# Should have exactly the same input names
											self.assertEqual(set(SCREAMING_SNAKE_CASE__    )   ,       set(SCREAMING_SNAKE_CASE__    )    )

											# Parameter should be reordered according to their respective place in the function:
											# (input_ids, token_type_ids, attention_mask)
											self.assertEqual(SCREAMING_SNAKE_CASE__   ,       (tokens["""input_ids"""], tokens["""token_type_ids"""], tokens["""attention_mask"""])    )

											# Generated args are interleaved with another args (for instance parameter "past" in GPT2)
											SCREAMING_SNAKE_CASE__      ,      SCREAMING_SNAKE_CASE__						:       str           =   ensure_valid_input(FuncNonContiguousArgs()   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )

											# Should have exactly the one arg (all before the one not provided "some_other_args")
											self.assertEqual(len(SCREAMING_SNAKE_CASE__    )   ,       1    )
											self.assertEqual(len(SCREAMING_SNAKE_CASE__    )   ,       1    )

											# Should have only "input_ids"
											self.assertEqual(inputs_args[0]   ,       tokens["""input_ids"""]    )
											self.assertEqual(ordered_input_names[0]   ,       """input_ids"""    )



				def 	__magic_name__    (self    )					->		List[Any]:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__						:       List[Any]           =   generate_identified_filename(Path("""/home/something/my_fake_model.onnx"""    )   ,       """-test"""    )
											self.assertEqual("""/home/something/my_fake_model-test.onnx"""   ,       generated.as_posix()    )






"""simple docstring"""




import argparse
import os

import torch

from transformers import FlavaConfig, FlavaForPreTraining
from transformers.models.flava.convert_dalle_to_flava_codebook import convert_dalle_checkpoint

def lowercase_  (   _snake_case						):
							# encoder.embeddings are double copied in original FLAVA
							return sum(param.float().sum() if """encoder.embeddings""" not in key else 0 for key, param in state_dict.items()						)

def lowercase_  (   _snake_case   ,_snake_case						):
							SCREAMING_SNAKE_CASE__						:       Any           =   {}

							for key, value in state_dict.items():
														if "text_encoder.embeddings" in key or "image_encoder.embeddings" in key:
																					continue

														SCREAMING_SNAKE_CASE__						:       Optional[int]           =   key.replace("""heads.cmd.mim_head.cls.predictions"""   ,"""mmm_image_head"""						)
														SCREAMING_SNAKE_CASE__						:       Dict           =   key.replace("""heads.cmd.mlm_head.cls.predictions"""   ,"""mmm_text_head"""						)
														SCREAMING_SNAKE_CASE__						:       List[Any]           =   key.replace("""heads.cmd.itm_head.cls"""   ,"""itm_head"""						)
														SCREAMING_SNAKE_CASE__						:       Tuple           =   key.replace("""heads.cmd.itm_head.pooler"""   ,"""itm_head.pooler"""						)
														SCREAMING_SNAKE_CASE__						:       int           =   key.replace("""heads.cmd.clip_head.logit_scale"""   ,"""flava.logit_scale"""						)
														SCREAMING_SNAKE_CASE__						:       Tuple           =   key.replace("""heads.fairseq_mlm.cls.predictions"""   ,"""mlm_head"""						)
														SCREAMING_SNAKE_CASE__						:       str           =   key.replace("""heads.imagenet.mim_head.cls.predictions"""   ,"""mim_head"""						)
														SCREAMING_SNAKE_CASE__						:       List[str]           =   key.replace("""mm_text_projection"""   ,"""flava.text_to_mm_projection"""						)
														SCREAMING_SNAKE_CASE__						:       Dict           =   key.replace("""mm_image_projection"""   ,"""flava.image_to_mm_projection"""						)
														SCREAMING_SNAKE_CASE__						:       str           =   key.replace("""image_encoder.module"""   ,"""flava.image_model"""						)
														SCREAMING_SNAKE_CASE__						:       Tuple           =   key.replace("""text_encoder.module"""   ,"""flava.text_model"""						)
														SCREAMING_SNAKE_CASE__						:       int           =   key.replace("""mm_encoder.module.encoder.cls_token"""   ,"""flava.multimodal_model.cls_token"""						)
														SCREAMING_SNAKE_CASE__						:       Dict           =   key.replace("""mm_encoder.module"""   ,"""flava.multimodal_model"""						)
														SCREAMING_SNAKE_CASE__						:       Any           =   key.replace("""text_projection"""   ,"""flava.text_projection"""						)
														SCREAMING_SNAKE_CASE__						:       List[Any]           =   key.replace("""image_projection"""   ,"""flava.image_projection"""						)

														SCREAMING_SNAKE_CASE__						:       Tuple           =   value.float()

							for key, value in codebook_state_dict.items():
														SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   value

							return upgrade

@torch.no_grad()
def lowercase_  (   _snake_case   ,_snake_case   ,_snake_case   ,_snake_case=None						):
							if config_path is not None:
														SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   FlavaConfig.from_pretrained(_snake_case						)
							else:
														SCREAMING_SNAKE_CASE__						:       List[str]           =   FlavaConfig()

							SCREAMING_SNAKE_CASE__						:       Optional[int]           =   FlavaForPreTraining(_snake_case						).eval()

							SCREAMING_SNAKE_CASE__						:       List[Any]           =   convert_dalle_checkpoint(_snake_case   ,_snake_case   ,save_checkpoint=_snake_case						)

							if os.path.exists(_snake_case						):
														SCREAMING_SNAKE_CASE__						:       List[str]           =   torch.load(_snake_case   ,map_location="""cpu"""						)
							else:
														SCREAMING_SNAKE_CASE__						:       Tuple           =   torch.hub.load_state_dict_from_url(_snake_case   ,map_location="""cpu"""						)

							SCREAMING_SNAKE_CASE__						:       Dict           =   upgrade_state_dict(_snake_case   ,_snake_case						)
							hf_model.load_state_dict(_snake_case						)
							SCREAMING_SNAKE_CASE__						:       Any           =   hf_model.state_dict()
							SCREAMING_SNAKE_CASE__						:       Any           =   count_parameters(_snake_case						)
							SCREAMING_SNAKE_CASE__						:       str           =   count_parameters(_snake_case						) + count_parameters(_snake_case						)

							assert torch.allclose(_snake_case   ,_snake_case   ,atol=1E-3						)

							hf_model.save_pretrained(_snake_case						)


if __name__ == "__main__":
					UpperCAmelCase__      :    List[Any]    							= argparse.ArgumentParser()
					parser.add_argument('--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model.')
					parser.add_argument('--checkpoint_path', default=None, type=str, help='Path to flava checkpoint')
					parser.add_argument('--codebook_path', default=None, type=str, help='Path to flava codebook checkpoint')
					parser.add_argument('--config_path', default=None, type=str, help='Path to hf config.json of model to convert')
					UpperCAmelCase__      :    Optional[int]    							= parser.parse_args()

					convert_flava_checkpoint(args.checkpoint_path, args.codebook_path, args.pytorch_dump_folder_path, args.config_path)




