





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							:       Any					 =	logging.get_logger(__name__)

lowerCAmelCase							:       List[str]					 =	[
    ("""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							:       List[str]					 =	[
    """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 		lowerCAmelCase  (	UpperCamelCase__   :   Dict					)      ->    Union[str, Any]:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				str           =   torch.load(UpperCamelCase__       ,			map_location='''cpu'''					)
						return sd






def 		lowerCAmelCase  (	UpperCamelCase__   :   str       ,			UpperCamelCase__   :   Any       ,			UpperCamelCase__   :   Union[str, Any]=rename_keys_prefix					)      ->    str:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				int           =   OrderedDict()
						__SCREAMING_SNAKE_CASE:				Optional[int]           =   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
												__SCREAMING_SNAKE_CASE:				Optional[Any]           =   key
												for name_pair in rename_keys_prefix:
																		__SCREAMING_SNAKE_CASE:				Dict           =   new_key.replace(name_pair[0]       ,			name_pair[1]					)
												__SCREAMING_SNAKE_CASE:				Tuple           =   d[key]
												if key == "bert.cls.predictions.decoder.weight":
																		# Old bert code didn't have `decoder.bias`, but was added separately
																		__SCREAMING_SNAKE_CASE:				Tuple           =   new_d['''cls.predictions.bias''']
						return new_d






@torch.no_grad()
def 		lowerCAmelCase  (	UpperCamelCase__   :   Optional[Any]       ,			UpperCamelCase__   :   Any					)      ->    Tuple:

						"""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:
												__SCREAMING_SNAKE_CASE:				Optional[int]           =   '''pretraining'''
												if "vcr" in checkpoint_path:
																		__SCREAMING_SNAKE_CASE:				str           =   {'''visual_embedding_dim''': 512}
												elif "vqa_advanced" in checkpoint_path:
																		__SCREAMING_SNAKE_CASE:				Tuple           =   {'''visual_embedding_dim''': 2_048}
												elif "vqa" in checkpoint_path:
																		__SCREAMING_SNAKE_CASE:				Any           =   {'''visual_embedding_dim''': 2_048}
												elif "nlvr" in checkpoint_path:
																		__SCREAMING_SNAKE_CASE:				int           =   {'''visual_embedding_dim''': 1_024}
												else:
																		raise NotImplementedError(F"""No implementation found for `{checkpoint_path}`."""					)
						else:
												if "vcr" in checkpoint_path:
																		__SCREAMING_SNAKE_CASE:				Any           =   {'''visual_embedding_dim''': 512}
																		__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   '''multichoice'''
												elif "vqa_advanced" in checkpoint_path:
																		__SCREAMING_SNAKE_CASE:				List[str]           =   {'''visual_embedding_dim''': 2_048}
																		__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   '''vqa_advanced'''
												elif "vqa" in checkpoint_path:
																		__SCREAMING_SNAKE_CASE:				List[str]           =   {'''visual_embedding_dim''': 2_048, '''num_labels''': 3_129}
																		__SCREAMING_SNAKE_CASE:				str           =   '''vqa'''
												elif "nlvr" in checkpoint_path:
																		__SCREAMING_SNAKE_CASE:				str           =   {
																		    '''visual_embedding_dim''': 1_024,
																		    '''num_labels''': 2,
																		}
																		__SCREAMING_SNAKE_CASE:				Any           =   '''nlvr'''

						__SCREAMING_SNAKE_CASE:				str           =   VisualBertConfig(**UpperCamelCase__					)

						# Load State Dict
						__SCREAMING_SNAKE_CASE:				Dict           =   load_state_dict(UpperCamelCase__					)

						__SCREAMING_SNAKE_CASE:				List[Any]           =   get_new_dict(UpperCamelCase__       ,			UpperCamelCase__					)

						if model_type == "pretraining":
												__SCREAMING_SNAKE_CASE:				List[str]           =   VisualBertForPreTraining(UpperCamelCase__					)
						elif model_type == "vqa":
												__SCREAMING_SNAKE_CASE:				List[Any]           =   VisualBertForQuestionAnswering(UpperCamelCase__					)
						elif model_type == "nlvr":
												__SCREAMING_SNAKE_CASE:				Dict           =   VisualBertForVisualReasoning(UpperCamelCase__					)
						elif model_type == "multichoice":
												__SCREAMING_SNAKE_CASE:				Tuple           =   VisualBertForMultipleChoice(UpperCamelCase__					)

						model.load_state_dict(UpperCamelCase__					)
						# Save Checkpoints
						Path(UpperCamelCase__					).mkdir(exist_ok=UpperCamelCase__					)
						model.save_pretrained(UpperCamelCase__					)


if __name__ == "__main__":
							lowerCAmelCase							:       List[Any]					 =	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							:       Dict					 =	parser.parse_args()
							convert_visual_bert_checkpoint(args.orig_checkpoint_path, args.pytorch_dump_folder_path)










import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()





class 				a			(      __lowercase   ,__lowercase   ,__lowercase   ,unittest.TestCase			):
				SCREAMING_SNAKE_CASE__     :   str		      =     StableUnCLIPPipeline
				SCREAMING_SNAKE_CASE__     :   str		      =     TEXT_TO_IMAGE_PARAMS
				SCREAMING_SNAKE_CASE__     :   int		      =     TEXT_TO_IMAGE_BATCH_PARAMS
				SCREAMING_SNAKE_CASE__     :   Dict		      =     TEXT_TO_IMAGE_IMAGE_PARAMS
				SCREAMING_SNAKE_CASE__     :   List[Any]		      =     TEXT_TO_IMAGE_IMAGE_PARAMS

				# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
				SCREAMING_SNAKE_CASE__     :   int		      =     False






				def 			snake_case_    (  self					):

										"""simple docstring"""




										__SCREAMING_SNAKE_CASE:				Tuple           =   32
										__SCREAMING_SNAKE_CASE:				List[Any]           =   embedder_hidden_size

										# prior components

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				Dict           =   CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				int           =   CLIPTextModelWithProjection(
										    CLIPTextConfig(
										        bos_token_id=0			, eos_token_id=2			, hidden_size=_lowerCAmelCase			, projection_dim=_lowerCAmelCase			, intermediate_size=37			, layer_norm_eps=1e-05			, num_attention_heads=4			, num_hidden_layers=5			, pad_token_id=1			, vocab_size=1000			, )					)

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				str           =   PriorTransformer(
										    num_attention_heads=2			, attention_head_dim=12			, embedding_dim=_lowerCAmelCase			, num_layers=1			, )

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				Optional[int]           =   DDPMScheduler(
										    variance_type='''fixed_small_log'''			, prediction_type='''sample'''			, num_train_timesteps=1000			, clip_sample=_lowerCAmelCase			, clip_sample_range=5.0			, beta_schedule='''squaredcos_cap_v2'''			, )

										# regular denoising components

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				List[Any]           =   StableUnCLIPImageNormalizer(embedding_dim=_lowerCAmelCase					)
										__SCREAMING_SNAKE_CASE:				Tuple           =   DDPMScheduler(beta_schedule='''squaredcos_cap_v2'''					)

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				str           =   CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   CLIPTextModel(
										    CLIPTextConfig(
										        bos_token_id=0			, eos_token_id=2			, hidden_size=_lowerCAmelCase			, projection_dim=32			, intermediate_size=37			, layer_norm_eps=1e-05			, num_attention_heads=4			, num_hidden_layers=5			, pad_token_id=1			, vocab_size=1000			, )					)

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				str           =   UNetaDConditionModel(
										    sample_size=32			, in_channels=4			, out_channels=4			, down_block_types=('''CrossAttnDownBlock2D''', '''DownBlock2D''')			, up_block_types=('''UpBlock2D''', '''CrossAttnUpBlock2D''')			, block_out_channels=(32, 64)			, attention_head_dim=(2, 4)			, class_embed_type='''projection'''			, projection_class_embeddings_input_dim=embedder_projection_dim * 2			, cross_attention_dim=_lowerCAmelCase			, layers_per_block=1			, upcast_attention=_lowerCAmelCase			, use_linear_projection=_lowerCAmelCase			, )

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				Optional[int]           =   DDIMScheduler(
										    beta_schedule='''scaled_linear'''			, beta_start=0.00085			, beta_end=0.012			, prediction_type='''v_prediction'''			, set_alpha_to_one=_lowerCAmelCase			, steps_offset=1			, )

										torch.manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				int           =   AutoencoderKL()

										__SCREAMING_SNAKE_CASE:				Dict           =   {
										    # prior components
										    '''prior_tokenizer''': prior_tokenizer,
										    '''prior_text_encoder''': prior_text_encoder,
										    '''prior''': prior,
										    '''prior_scheduler''': prior_scheduler,
										    # image noising components
										    '''image_normalizer''': image_normalizer,
										    '''image_noising_scheduler''': image_noising_scheduler,
										    # regular denoising components
										    '''tokenizer''': tokenizer,
										    '''text_encoder''': text_encoder,
										    '''unet''': unet,
										    '''scheduler''': scheduler,
										    '''vae''': vae,
										}

										return components






				def 			snake_case_    (  self			, _lowerCAmelCase			, _lowerCAmelCase=0					):

										"""simple docstring"""




										if str(_lowerCAmelCase					).startswith('''mps'''					):
																__SCREAMING_SNAKE_CASE:				List[str]           =   torch.manual_seed(_lowerCAmelCase					)
										else:
																__SCREAMING_SNAKE_CASE:				List[str]           =   torch.Generator(device=_lowerCAmelCase					).manual_seed(_lowerCAmelCase					)
										__SCREAMING_SNAKE_CASE:				List[str]           =   {
										    '''prompt''': '''A painting of a squirrel eating a burger''',
										    '''generator''': generator,
										    '''num_inference_steps''': 2,
										    '''prior_num_inference_steps''': 2,
										    '''output_type''': '''numpy''',
										}
										return inputs






				def 			snake_case_    (  self					):

										"""simple docstring"""




										__SCREAMING_SNAKE_CASE:				str           =   torch_device == '''cpu'''

										self._test_attention_slicing_forward_pass(test_max_difference=_lowerCAmelCase					)






				def 			snake_case_    (  self					):

										"""simple docstring"""




										__SCREAMING_SNAKE_CASE:				List[str]           =   torch_device in ['''cpu''', '''mps''']

										self._test_inference_batch_single_identical(test_max_difference=_lowerCAmelCase					)





@slow
@require_torch_gpu
class 				a			(      unittest.TestCase			):






				def 			snake_case_    (  self					):

										"""simple docstring"""




										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()






				def 			snake_case_    (  self					):

										"""simple docstring"""




										__SCREAMING_SNAKE_CASE:				int           =   load_numpy(
										    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy'''					)

										__SCREAMING_SNAKE_CASE:				Optional[int]           =   StableUnCLIPPipeline.from_pretrained('''fusing/stable-unclip-2-1-l'''			, torch_dtype=torch.floataa					)
										pipe.to(_lowerCAmelCase					)
										pipe.set_progress_bar_config(disable=_lowerCAmelCase					)
										# stable unclip will oom when integration tests are run on a V100,
										# so turn on memory savings
										pipe.enable_attention_slicing()
										pipe.enable_sequential_cpu_offload()

										__SCREAMING_SNAKE_CASE:				Tuple           =   torch.Generator(device='''cpu'''					).manual_seed(0					)
										__SCREAMING_SNAKE_CASE:				List[str]           =   pipe('''anime turle'''			, generator=_lowerCAmelCase			, output_type='''np'''					)

										__SCREAMING_SNAKE_CASE:				Optional[int]           =   output.images[0]

										assert image.shape == (768, 768, 3)

										assert_mean_pixel_difference(_lowerCAmelCase			, _lowerCAmelCase					)






				def 			snake_case_    (  self					):

										"""simple docstring"""




										torch.cuda.empty_cache()
										torch.cuda.reset_max_memory_allocated()
										torch.cuda.reset_peak_memory_stats()

										__SCREAMING_SNAKE_CASE:				int           =   StableUnCLIPPipeline.from_pretrained('''fusing/stable-unclip-2-1-l'''			, torch_dtype=torch.floataa					)
										__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   pipe.to(_lowerCAmelCase					)
										pipe.set_progress_bar_config(disable=_lowerCAmelCase					)
										pipe.enable_attention_slicing()
										pipe.enable_sequential_cpu_offload()

										__SCREAMING_SNAKE_CASE:				Dict           =   pipe(
										    '''anime turtle'''			, prior_num_inference_steps=2			, num_inference_steps=2			, output_type='''np'''			, )

										__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   torch.cuda.max_memory_allocated()
										# make sure that less than 7 GB is allocated
										assert mem_bytes < 7 * 10**9



