
"""simple docstring"""




import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


UpperCAmelCase__      :    List[str]    							= logging.get_logger(__name__)

def lowercase_  (   ):
							# Get the sagemaker specific mp parameters from smp_options variable.
							SCREAMING_SNAKE_CASE__						:       Dict           =   os.getenv("""SM_HP_MP_PARAMETERS"""   ,"""{}"""						)
							try:
														# Parse it and check the field "partitions" is included, it is required for model parallel.
														SCREAMING_SNAKE_CASE__						:       List[str]           =   json.loads(_snake_case						)
														if "partitions" not in smp_options:
																					return False
							except json.JSONDecodeError:
														return False

							# Get the sagemaker specific framework parameters from mpi_options variable.
							SCREAMING_SNAKE_CASE__						:       int           =   os.getenv("""SM_FRAMEWORK_PARAMS"""   ,"""{}"""						)
							try:
														# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
														SCREAMING_SNAKE_CASE__						:       Union[str, Any]           =   json.loads(_snake_case						)
														if not mpi_options.get("""sagemaker_mpi_enabled"""   ,_snake_case						):
																					return False
							except json.JSONDecodeError:
														return False
							# Lastly, check if the `smdistributed` module is present.
							return importlib.util.find_spec("""smdistributed"""						) is not None


if is_sagemaker_model_parallel_available():
					import smdistributed.modelparallel.torch as smp

					smp.init()



@dataclass
class 	lowerCAmelCase_						(a__  ):


				"""simple docstring"""



				__UpperCamelCase      :						str					       =   field(
				    default='''''' ,					metadata={'''help''': '''Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer'''} ,					)



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



											"""simple docstring"""




											super().__post_init__()
											warnings.warn(
											    """`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use """
											    """`TrainingArguments` instead."""   ,       SCREAMING_SNAKE_CASE__   ,       )



				@cached_property
				def 	__magic_name__    (self    )					->		"torch.device":



											"""simple docstring"""




											logger.info("""PyTorch: setting up devices"""    )
											if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
																		logger.warning(
																		    """torch.distributed process group is initialized, but local_rank == -1. """
																		    """In order to use Torch DDP, launch your script with `python -m torch.distributed.launch"""    )
											if self.no_cuda:
																		SCREAMING_SNAKE_CASE__						:       Optional[int]           =   torch.device("""cpu"""    )
																		SCREAMING_SNAKE_CASE__						:       int           =   0
											elif is_sagemaker_model_parallel_available():
																		SCREAMING_SNAKE_CASE__						:       Optional[int]           =   smp.local_rank()
																		SCREAMING_SNAKE_CASE__						:       Any           =   torch.device("""cuda"""   ,       SCREAMING_SNAKE_CASE__    )
																		SCREAMING_SNAKE_CASE__						:       List[Any]           =   1
											elif is_sagemaker_dp_enabled():
																		import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

																		torch.distributed.init_process_group(backend="""smddp"""   ,       timeout=self.ddp_timeout_delta    )
																		SCREAMING_SNAKE_CASE__						:       str           =   int(os.getenv("""SMDATAPARALLEL_LOCAL_RANK"""    )    )
																		SCREAMING_SNAKE_CASE__						:       Dict           =   torch.device("""cuda"""   ,       self.local_rank    )
																		SCREAMING_SNAKE_CASE__						:       Optional[int]           =   1
											elif self.local_rank == -1:
																		# if n_gpu is > 1 we'll use nn.DataParallel.
																		# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
																		# Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
																		# trigger an error that a device index is missing. Index 0 takes into account the
																		# GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
																		# will use the first GPU in that env, i.e. GPU#1
																		SCREAMING_SNAKE_CASE__						:       List[str]           =   torch.device("""cuda:0""" if torch.cuda.is_available() else """cpu"""    )
																		# Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
																		# the default value.
																		SCREAMING_SNAKE_CASE__						:       List[str]           =   torch.cuda.device_count()
											else:
																		# Here, we'll use torch.distributed.
																		# Initializes the distributed backend which will take care of synchronizing nodes/GPUs
																		if not torch.distributed.is_initialized():
																									torch.distributed.init_process_group(backend="""nccl"""   ,       timeout=self.ddp_timeout_delta    )
																		SCREAMING_SNAKE_CASE__						:       List[Any]           =   torch.device("""cuda"""   ,       self.local_rank    )
																		SCREAMING_SNAKE_CASE__						:       str           =   1

											if device.type == "cuda":
																		torch.cuda.set_device(SCREAMING_SNAKE_CASE__    )

											return device



				@property
				def 	__magic_name__    (self    )					->		str:



											"""simple docstring"""




											if is_sagemaker_model_parallel_available():
																		return smp.dp_size()

											return super().world_size



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



											"""simple docstring"""




											return not is_sagemaker_model_parallel_available()



				@property
				def 	__magic_name__    (self    )					->		Tuple:



											"""simple docstring"""




											return False






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




