




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 				A__ (snake_case   :		List[Any]				)     ->						Union[str, Any]:
					# 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 				A__ (snake_case   :		Any		,	snake_case   :		Any				)     ->						Dict:
					__UpperCamelCase					:	Union[str, Any]             =      {}

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

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

										__UpperCamelCase					:	Dict             =      value.float()

					for key, value in codebook_state_dict.items():
										__UpperCamelCase					:	List[Any]             =      value

					return upgrade





@torch.no_grad()
def 				A__ (snake_case   :		Any		,	snake_case   :		Optional[int]		,	snake_case   :		Union[str, Any]		,	snake_case   :		Optional[int]=None				)     ->						Optional[int]:
					if config_path is not None:
										__UpperCamelCase					:	str             =      FlavaConfig.from_pretrained(snake_case				)
					else:
										__UpperCamelCase					:	List[str]             =      FlavaConfig()

					__UpperCamelCase					:	Any             =      FlavaForPreTraining(snake_case				).eval()

					__UpperCamelCase					:	str             =      convert_dalle_checkpoint(snake_case		,	snake_case		,	save_checkpoint=snake_case				)

					if os.path.exists(snake_case				):
										__UpperCamelCase					:	Union[str, Any]             =      torch.load(snake_case		,	map_location="""cpu"""				)
					else:
										__UpperCamelCase					:	Any             =      torch.hub.load_state_dict_from_url(snake_case		,	map_location="""cpu"""				)

					__UpperCamelCase					:	str             =      upgrade_state_dict(snake_case		,	snake_case				)
					hf_model.load_state_dict(snake_case				)
					__UpperCamelCase					:	str             =      hf_model.state_dict()
					__UpperCamelCase					:	Any             =      count_parameters(snake_case				)
					__UpperCamelCase					:	Union[str, Any]             =      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__":
	a__		     =     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''')
	a__		     =     parser.parse_args()

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












import importlib
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
import transformers.models.auto
from transformers.models.auto.configuration_auto import CONFIG_MAPPING, AutoConfig
from transformers.models.bert.configuration_bert import BertConfig
from transformers.models.roberta.configuration_roberta import RobertaConfig
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402


a__		     =     get_tests_dir('''fixtures/dummy-config.json''')


class    SCREAMING_SNAKE_CASE_				(       unittest.TestCase   ):



	"""simple docstring"""


	def 					lowerCamelCase__     (					self		:						str	)	->       Dict:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[Any]             =      0






	def 					lowerCamelCase__     (					self		:						Union[str, Any]	)	->       List[str]:







						"""simple docstring"""







						self.assertIsNotNone(transformers.models.auto.__spec__	)
						self.assertIsNotNone(importlib.util.find_spec("""transformers.models.auto"""	)	)






	def 					lowerCamelCase__     (					self		:						Optional[Any]	)	->       Dict:







						"""simple docstring"""







						__UpperCamelCase					:	int             =      AutoConfig.from_pretrained("""bert-base-uncased"""	)
						self.assertIsInstance(lowerCAmelCase					,     lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						List[Any]	)	->       Optional[Any]:







						"""simple docstring"""







						__UpperCamelCase					:	Dict             =      AutoConfig.from_pretrained(lowerCAmelCase	)
						self.assertIsInstance(lowerCAmelCase					,     lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						Any	)	->       Optional[int]:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[int]             =      AutoConfig.from_pretrained(lowerCAmelCase	)
						self.assertIsInstance(lowerCAmelCase					,     lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						Tuple	)	->       Any:







						"""simple docstring"""







						__UpperCamelCase					:	List[Any]             =      AutoConfig.for_model("""roberta"""	)
						self.assertIsInstance(lowerCAmelCase					,     lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						int	)	->       List[Any]:







						"""simple docstring"""







						with tempfile.TemporaryDirectory() as tmp_dir:
											# This model name contains bert and roberta, but roberta ends up being picked.
											__UpperCamelCase					:	Union[str, Any]             =      os.path.join(lowerCAmelCase					,     """fake-roberta"""	)
											os.makedirs(lowerCAmelCase					,     exist_ok=lowerCAmelCase	)
											with open(os.path.join(lowerCAmelCase					,     """config.json"""	)					,     """w"""	) as f:
																f.write(json.dumps({}	)	)
											__UpperCamelCase					:	Dict             =      AutoConfig.from_pretrained(lowerCAmelCase	)
											self.assertEqual(type(lowerCAmelCase	)					,     lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						int	)	->       str:







						"""simple docstring"""







						try:
											AutoConfig.register("""custom"""					,     lowerCAmelCase	)
											# Wrong model type will raise an error
											with self.assertRaises(lowerCAmelCase	):
																AutoConfig.register("""model"""					,     lowerCAmelCase	)
											# Trying to register something existing in the Transformers library will raise an error
											with self.assertRaises(lowerCAmelCase	):
																AutoConfig.register("""bert"""					,     lowerCAmelCase	)

											# Now that the config is registered, it can be used as any other config with the auto-API
											__UpperCamelCase					:	Optional[Any]             =      CustomConfig()
											with tempfile.TemporaryDirectory() as tmp_dir:
																config.save_pretrained(lowerCAmelCase	)
																__UpperCamelCase					:	List[str]             =      AutoConfig.from_pretrained(lowerCAmelCase	)
																self.assertIsInstance(lowerCAmelCase					,     lowerCAmelCase	)

						finally:
											if "custom" in CONFIG_MAPPING._extra_content:
																del CONFIG_MAPPING._extra_content["custom"]






	def 					lowerCamelCase__     (					self		:						int	)	->       Optional[Any]:







						"""simple docstring"""







						with self.assertRaisesRegex(
						    lowerCAmelCase					,     """bert-base is not a local folder and is not a valid model identifier"""	):
											__UpperCamelCase					:	Tuple             =      AutoConfig.from_pretrained("""bert-base"""	)






	def 					lowerCamelCase__     (					self		:						Dict	)	->       Any:







						"""simple docstring"""







						with self.assertRaisesRegex(
						    lowerCAmelCase					,     R"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""	):
											__UpperCamelCase					:	Optional[Any]             =      AutoConfig.from_pretrained(lowerCAmelCase					,     revision="""aaaaaa"""	)






	def 					lowerCamelCase__     (					self		:						str	)	->       List[str]:







						"""simple docstring"""







						with self.assertRaisesRegex(
						    lowerCAmelCase					,     """hf-internal-testing/no-config-test-repo does not appear to have a file named config.json."""					,     ):
											__UpperCamelCase					:	Union[str, Any]             =      AutoConfig.from_pretrained("""hf-internal-testing/no-config-test-repo"""	)






	def 					lowerCamelCase__     (					self		:						Optional[int]	)	->       Optional[int]:







						"""simple docstring"""







						with self.assertRaises(lowerCAmelCase	):
											__UpperCamelCase					:	str             =      AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""	)
						# If remote code is disabled, we can't load this config.
						with self.assertRaises(lowerCAmelCase	):
											__UpperCamelCase					:	Tuple             =      AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""					,     trust_remote_code=lowerCAmelCase	)

						__UpperCamelCase					:	List[str]             =      AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""					,     trust_remote_code=lowerCAmelCase	)
						self.assertEqual(config.__class__.__name__					,     """NewModelConfig"""	)

						# Test config can be reloaded.
						with tempfile.TemporaryDirectory() as tmp_dir:
											config.save_pretrained(lowerCAmelCase	)
											__UpperCamelCase					:	str             =      AutoConfig.from_pretrained(lowerCAmelCase					,     trust_remote_code=lowerCAmelCase	)
						self.assertEqual(reloaded_config.__class__.__name__					,     """NewModelConfig"""	)






	def 					lowerCamelCase__     (					self		:						Optional[int]	)	->       str:







						"""simple docstring"""







						class    SCREAMING_SNAKE_CASE_				(       _UpperCamelCase   ):



							"""simple docstring"""


							__magic_name__   :       int      	=		'new-model'

						try:
											AutoConfig.register("""new-model"""					,     lowerCAmelCase	)
											# If remote code is not set, the default is to use local
											__UpperCamelCase					:	Tuple             =      AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""	)
											self.assertEqual(config.__class__.__name__					,     """NewModelConfigLocal"""	)

											# If remote code is disabled, we load the local one.
											__UpperCamelCase					:	Any             =      AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""					,     trust_remote_code=lowerCAmelCase	)
											self.assertEqual(config.__class__.__name__					,     """NewModelConfigLocal"""	)

											# If remote is enabled, we load from the Hub
											__UpperCamelCase					:	List[Any]             =      AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""					,     trust_remote_code=lowerCAmelCase	)
											self.assertEqual(config.__class__.__name__					,     """NewModelConfig"""	)

						finally:
											if "new-model" in CONFIG_MAPPING._extra_content:
																del CONFIG_MAPPING._extra_content["new-model"]






