

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 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							Optional[Any]      )			->      Dict:




				'''simple docstring'''





				return sum(param.float().sum() if "encoder.embeddings" not in key else 0 for key, param in state_dict.items()      )



def 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							Any	,       SCREAMING_SNAKE_CASE_:							Optional[int]      )			->      int:




				'''simple docstring'''





				A__        =			{}

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

								A__        =			key.replace("heads.cmd.mim_head.cls.predictions"	,       "mmm_image_head"      )
								A__        =			key.replace("heads.cmd.mlm_head.cls.predictions"	,       "mmm_text_head"      )
								A__        =			key.replace("heads.cmd.itm_head.cls"	,       "itm_head"      )
								A__        =			key.replace("heads.cmd.itm_head.pooler"	,       "itm_head.pooler"      )
								A__        =			key.replace("heads.cmd.clip_head.logit_scale"	,       "flava.logit_scale"      )
								A__        =			key.replace("heads.fairseq_mlm.cls.predictions"	,       "mlm_head"      )
								A__        =			key.replace("heads.imagenet.mim_head.cls.predictions"	,       "mim_head"      )
								A__        =			key.replace("mm_text_projection"	,       "flava.text_to_mm_projection"      )
								A__        =			key.replace("mm_image_projection"	,       "flava.image_to_mm_projection"      )
								A__        =			key.replace("image_encoder.module"	,       "flava.image_model"      )
								A__        =			key.replace("text_encoder.module"	,       "flava.text_model"      )
								A__        =			key.replace("mm_encoder.module.encoder.cls_token"	,       "flava.multimodal_model.cls_token"      )
								A__        =			key.replace("mm_encoder.module"	,       "flava.multimodal_model"      )
								A__        =			key.replace("text_projection"	,       "flava.text_projection"      )
								A__        =			key.replace("image_projection"	,       "flava.image_projection"      )

								A__        =			value.float()

				for key, value in codebook_state_dict.items():
								A__        =			value

				return upgrade



@torch.no_grad()
def 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							int	,       SCREAMING_SNAKE_CASE_:							List[str]	,       SCREAMING_SNAKE_CASE_:							List[str]	,       SCREAMING_SNAKE_CASE_:							int=None      )			->      Any:




				'''simple docstring'''





				if config_path is not None:
								A__        =			FlavaConfig.from_pretrained(SCREAMING_SNAKE_CASE_      )
				else:
								A__        =			FlavaConfig()

				A__        =			FlavaForPreTraining(SCREAMING_SNAKE_CASE_      ).eval()

				A__        =			convert_dalle_checkpoint(SCREAMING_SNAKE_CASE_	,       SCREAMING_SNAKE_CASE_	,       save_checkpoint=SCREAMING_SNAKE_CASE_      )

				if os.path.exists(SCREAMING_SNAKE_CASE_      ):
								A__        =			torch.load(SCREAMING_SNAKE_CASE_	,       map_location="cpu"      )
				else:
								A__        =			torch.hub.load_state_dict_from_url(SCREAMING_SNAKE_CASE_	,       map_location="cpu"      )

				A__        =			upgrade_state_dict(SCREAMING_SNAKE_CASE_	,       SCREAMING_SNAKE_CASE_      )
				hf_model.load_state_dict(SCREAMING_SNAKE_CASE_      )
				A__        =			hf_model.state_dict()
				A__        =			count_parameters(SCREAMING_SNAKE_CASE_      )
				A__        =			count_parameters(SCREAMING_SNAKE_CASE_      ) + count_parameters(SCREAMING_SNAKE_CASE_      )

				assert torch.allclose(SCREAMING_SNAKE_CASE_	,       SCREAMING_SNAKE_CASE_	,       atol=1e-3      )

				hf_model.save_pretrained(SCREAMING_SNAKE_CASE_      )


if __name__ == "__main__":
	lowerCAmelCase__	   = 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""")
	lowerCAmelCase__	   = parser.parse_args()

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




lowerCAmelCase__	   = """ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"""



def 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							bytes      )			->      bytes:




				'''simple docstring'''





				if not isinstance(SCREAMING_SNAKE_CASE_	,       SCREAMING_SNAKE_CASE_      ):
								A__        =			F'a bytes-like object is required, not \'{data.__class__.__name__}\''
								raise TypeError(SCREAMING_SNAKE_CASE_      )

				A__        =			"".join(bin(SCREAMING_SNAKE_CASE_      )[2:].zfill(8      ) for byte in data      )

				A__        =			len(SCREAMING_SNAKE_CASE_      ) % 6 != 0

				if padding_needed:
								# The padding that will be added later
								A__        =			b"=" * ((6 - len(SCREAMING_SNAKE_CASE_      ) % 6) // 2)

								# Append binary_stream with arbitrary binary digits (0's by default) to make its
								# length a multiple of 6.
								binary_stream += "0" * (6 - len(SCREAMING_SNAKE_CASE_      ) % 6)
				else:
								A__        =			b""

				# Encode every 6 binary digits to their corresponding Base64 character
				return (
				    "".join(
				        B64_CHARSET[int(binary_stream[index : index + 6]	,       2      )]
				        for index in range(0	,       len(SCREAMING_SNAKE_CASE_      )	,       6      )      ).encode()
				    + padding
				)



def 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							str      )			->      bytes:




				'''simple docstring'''





				if not isinstance(SCREAMING_SNAKE_CASE_	,       SCREAMING_SNAKE_CASE_      ) and not isinstance(SCREAMING_SNAKE_CASE_	,       SCREAMING_SNAKE_CASE_      ):
								A__        =			(
								    "argument should be a bytes-like object or ASCII string, "
								    F'not \'{encoded_data.__class__.__name__}\''
								)
								raise TypeError(SCREAMING_SNAKE_CASE_      )

				# In case encoded_data is a bytes-like object, make sure it contains only
				# ASCII characters so we convert it to a string object
				if isinstance(SCREAMING_SNAKE_CASE_	,       SCREAMING_SNAKE_CASE_      ):
								try:
												A__        =			encoded_data.decode("utf-8"      )
								except UnicodeDecodeError:
												raise ValueError("base64 encoded data should only contain ASCII characters"      )

				A__        =			encoded_data.count("="      )

				# Check if the encoded string contains non base64 characters
				if padding:
								assert all(
								    char in B64_CHARSET for char in encoded_data[:-padding]      ), "Invalid base64 character(s) found."
				else:
								assert all(
								    char in B64_CHARSET for char in encoded_data      ), "Invalid base64 character(s) found."

				# Check the padding
				assert len(SCREAMING_SNAKE_CASE_      ) % 4 == 0 and padding < 3, "Incorrect padding"

				if padding:
								# Remove padding if there is one
								A__        =			encoded_data[:-padding]

								A__        =			"".join(
								    bin(B64_CHARSET.index(SCREAMING_SNAKE_CASE_      )      )[2:].zfill(6      ) for char in encoded_data      )[: -padding * 2]
				else:
								A__        =			"".join(
								    bin(B64_CHARSET.index(SCREAMING_SNAKE_CASE_      )      )[2:].zfill(6      ) for char in encoded_data      )

				A__        =			[
				    int(binary_stream[index : index + 8]	,       2      )
				    for index in range(0	,       len(SCREAMING_SNAKE_CASE_      )	,       8      )
				]

				return bytes(SCREAMING_SNAKE_CASE_      )


if __name__ == "__main__":
	import doctest

	doctest.testmod()

