import unittest

import numpy as np
import torch
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyVaaPriorPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()





class 					lowercase__						(   UpperCamelCase_	,							unittest.TestCase):
						UpperCamelCase_     =		KandinskyVaaPriorPipeline
						UpperCamelCase_     =		["""prompt"""]
						UpperCamelCase_     =		["""prompt""", """negative_prompt"""]
						UpperCamelCase_     =		[
						    """num_images_per_prompt""",
						    """generator""",
						    """num_inference_steps""",
						    """latents""",
						    """negative_prompt""",
						    """guidance_scale""",
						    """output_type""",
						    """return_dict""",
						]
						UpperCamelCase_     =		False


						@property
						def        __A       (       self	:				int						):





												'''simple docstring'''





												return 32


						@property
						def        __A       (       self	:				Optional[int]						):





												'''simple docstring'''





												return 32


						@property
						def        __A       (       self	:				Optional[int]						):





												'''simple docstring'''





												return self.time_input_dim


						@property
						def        __A       (       self	:				List[str]						):





												'''simple docstring'''





												return self.time_input_dim * 4


						@property
						def        __A       (       self	:				str						):





												'''simple docstring'''





												return 100


						@property
						def        __A       (       self	:				List[Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''						)
												return tokenizer


						@property
						def        __A       (       self	:				Optional[int]						):





												'''simple docstring'''





												torch.manual_seed(0						)
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							CLIPTextConfig(
												    bos_token_id=0    ,		eos_token_id=2    ,		hidden_size=self.text_embedder_hidden_size    ,		projection_dim=self.text_embedder_hidden_size    ,		intermediate_size=37    ,		layer_norm_eps=1E-05    ,		num_attention_heads=4    ,		num_hidden_layers=5    ,		pad_token_id=1    ,		vocab_size=1000    ,		)
												return CLIPTextModelWithProjection(UpperCamelCase__						)


						@property
						def        __A       (       self	:				Optional[int]						):





												'''simple docstring'''





												torch.manual_seed(0						)

												SCREAMING_SNAKE_CASE							:    List[str]							=							{
												    '''num_attention_heads''': 2,
												    '''attention_head_dim''': 12,
												    '''embedding_dim''': self.text_embedder_hidden_size,
												    '''num_layers''': 1,
												}

												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							PriorTransformer(**UpperCamelCase__						)
												# clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							nn.Parameter(torch.ones(model.clip_std.shape						)						)
												return model


						@property
						def        __A       (       self	:				Union[str, Any]						):





												'''simple docstring'''





												torch.manual_seed(0						)
												SCREAMING_SNAKE_CASE							:    List[Any]							=							CLIPVisionConfig(
												    hidden_size=self.text_embedder_hidden_size    ,		image_size=224    ,		projection_dim=self.text_embedder_hidden_size    ,		intermediate_size=37    ,		num_attention_heads=4    ,		num_channels=3    ,		num_hidden_layers=5    ,		patch_size=14    ,		)

												SCREAMING_SNAKE_CASE							:    Dict							=							CLIPVisionModelWithProjection(UpperCamelCase__						)
												return model


						@property
						def        __A       (       self	:				Dict						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Tuple							=							CLIPImageProcessor(
												    crop_size=224    ,		do_center_crop=UpperCamelCase__    ,		do_normalize=UpperCamelCase__    ,		do_resize=UpperCamelCase__    ,		image_mean=[0.4814_5466, 0.457_8275, 0.4082_1073]    ,		image_std=[0.2686_2954, 0.2613_0258, 0.2757_7711]    ,		resample=3    ,		size=224    ,		)

												return image_processor


						def        __A       (       self	:				List[Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							self.dummy_prior
												SCREAMING_SNAKE_CASE							:    List[Any]							=							self.dummy_image_encoder
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							self.dummy_text_encoder
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							self.dummy_tokenizer
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							self.dummy_image_processor

												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							UnCLIPScheduler(
												    variance_type='''fixed_small_log'''    ,		prediction_type='''sample'''    ,		num_train_timesteps=1000    ,		clip_sample=UpperCamelCase__    ,		clip_sample_range=10.0    ,		)

												SCREAMING_SNAKE_CASE							:    List[Any]							=							{
												    '''prior''': prior,
												    '''image_encoder''': image_encoder,
												    '''text_encoder''': text_encoder,
												    '''tokenizer''': tokenizer,
												    '''scheduler''': scheduler,
												    '''image_processor''': image_processor,
												}

												return components


						def        __A       (       self	:				List[str]    ,		UpperCamelCase__	:				Tuple    ,		UpperCamelCase__	:				List[Any]=0						):





												'''simple docstring'''





												if str(UpperCamelCase__						).startswith('''mps'''						):
																		SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							torch.manual_seed(UpperCamelCase__						)
												else:
																		SCREAMING_SNAKE_CASE							:    Optional[Any]							=							torch.Generator(device=UpperCamelCase__						).manual_seed(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							{
												    '''prompt''': '''horse''',
												    '''generator''': generator,
												    '''guidance_scale''': 4.0,
												    '''num_inference_steps''': 2,
												    '''output_type''': '''np''',
												}
												return inputs


						def        __A       (       self	:				Optional[Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Any							=							'''cpu'''

												SCREAMING_SNAKE_CASE							:    List[Any]							=							self.get_dummy_components()

												SCREAMING_SNAKE_CASE							:    List[str]							=							self.pipeline_class(**UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							pipe.to(UpperCamelCase__						)

												pipe.set_progress_bar_config(disable=UpperCamelCase__						)

												SCREAMING_SNAKE_CASE							:    Any							=							pipe(**self.get_dummy_inputs(UpperCamelCase__						)						)
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							output.image_embeds

												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							pipe(
												    **self.get_dummy_inputs(UpperCamelCase__						)    ,		return_dict=UpperCamelCase__    ,		)[0]

												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							image[0, -10:]
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							image_from_tuple[0, -10:]

												assert image.shape == (1, 32)

												SCREAMING_SNAKE_CASE							:    Tuple							=							np.array(
												    [-0.0532, 1.7120, 0.3656, -1.0852, -0.8946, -1.1756, 0.4348, 0.2482, 0.5146, -0.1156]						)

												assert np.abs(image_slice.flatten() - expected_slice						).max() < 1E-2
												assert np.abs(image_from_tuple_slice.flatten() - expected_slice						).max() < 1E-2


						@skip_mps
						def        __A       (       self	:				Union[str, Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Tuple							=							torch_device == '''cpu'''
												SCREAMING_SNAKE_CASE							:    Any							=							True
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							False

												self._test_inference_batch_single_identical(
												    test_max_difference=UpperCamelCase__    ,		relax_max_difference=UpperCamelCase__    ,		test_mean_pixel_difference=UpperCamelCase__    ,		)



						@skip_mps
						def        __A       (       self	:				Dict						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Optional[int]							=							torch_device == '''cpu'''
												SCREAMING_SNAKE_CASE							:    int							=							False

												self._test_attention_slicing_forward_pass(
												    test_max_difference=UpperCamelCase__    ,		test_mean_pixel_difference=UpperCamelCase__    ,		)






from ...configuration_utils import PretrainedConfig
from ...utils import logging


__UpperCamelCase		:       Any									=     logging.get_logger(__name__)

__UpperCamelCase		:       Optional[int]									=     {
    'facebook/nllb-moe-54B': 'https://huggingface.co/facebook/nllb-moe-54b/resolve/main/config.json',
}





class 					lowercase__						(   UpperCamelCase_):
						UpperCamelCase_     =		"""nllb-moe"""
						UpperCamelCase_     =		["""past_key_values"""]
						UpperCamelCase_     =		{"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}


						def __init__(       self	:				List[str]    ,		UpperCamelCase__	:				List[str]=12_8112    ,		UpperCamelCase__	:				str=1024    ,		UpperCamelCase__	:				Optional[int]=12    ,		UpperCamelCase__	:				Union[str, Any]=4096    ,		UpperCamelCase__	:				Optional[Any]=16    ,		UpperCamelCase__	:				Optional[Any]=12    ,		UpperCamelCase__	:				str=4096    ,		UpperCamelCase__	:				Dict=16    ,		UpperCamelCase__	:				Any=0.05    ,		UpperCamelCase__	:				Any=0.05    ,		UpperCamelCase__	:				Optional[Any]=True    ,		UpperCamelCase__	:				Optional[Any]=True    ,		UpperCamelCase__	:				List[Any]="relu"    ,		UpperCamelCase__	:				Union[str, Any]=1024    ,		UpperCamelCase__	:				Optional[int]=0.1    ,		UpperCamelCase__	:				Any=0.1    ,		UpperCamelCase__	:				Optional[Any]=0.0    ,		UpperCamelCase__	:				str=0.02    ,		UpperCamelCase__	:				Tuple=2    ,		UpperCamelCase__	:				List[str]=True    ,		UpperCamelCase__	:				Dict=False    ,		UpperCamelCase__	:				Any="float32"    ,		UpperCamelCase__	:				List[Any]=False    ,		UpperCamelCase__	:				Tuple=128    ,		UpperCamelCase__	:				Tuple=64    ,		UpperCamelCase__	:				List[str]=4    ,		UpperCamelCase__	:				Union[str, Any]=4    ,		UpperCamelCase__	:				Dict=0.001    ,		UpperCamelCase__	:				Optional[Any]=0.001    ,		UpperCamelCase__	:				Optional[Any]="all"    ,		UpperCamelCase__	:				Tuple=False    ,		UpperCamelCase__	:				List[Any]=False    ,		UpperCamelCase__	:				List[Any]=1.0    ,		UpperCamelCase__	:				str=0.2    ,		UpperCamelCase__	:				List[Any]=1    ,		UpperCamelCase__	:				List[Any]=0    ,		UpperCamelCase__	:				Optional[int]=2    ,		UpperCamelCase__	:				Tuple=False    ,		**UpperCamelCase__	:				Union[str, Any]    ,		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    List[str]							=							vocab_size
												SCREAMING_SNAKE_CASE							:    Dict							=							max_position_embeddings
												SCREAMING_SNAKE_CASE							:    int							=							d_model
												SCREAMING_SNAKE_CASE							:    Any							=							encoder_ffn_dim
												SCREAMING_SNAKE_CASE							:    List[Any]							=							encoder_layers
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							encoder_attention_heads
												SCREAMING_SNAKE_CASE							:    List[str]							=							decoder_ffn_dim
												SCREAMING_SNAKE_CASE							:    Dict							=							decoder_layers
												SCREAMING_SNAKE_CASE							:    List[Any]							=							decoder_attention_heads
												SCREAMING_SNAKE_CASE							:    List[str]							=							dropout
												SCREAMING_SNAKE_CASE							:    Any							=							attention_dropout
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							activation_dropout
												SCREAMING_SNAKE_CASE							:    List[Any]							=							activation_function
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							init_std
												SCREAMING_SNAKE_CASE							:    int							=							encoder_layerdrop
												SCREAMING_SNAKE_CASE							:    List[Any]							=							decoder_layerdrop
												SCREAMING_SNAKE_CASE							:    Any							=							use_cache
												SCREAMING_SNAKE_CASE							:    str							=							encoder_layers
												SCREAMING_SNAKE_CASE							:    List[Any]							=							scale_embedding  # scale factor will be sqrt(d_model) if True
												SCREAMING_SNAKE_CASE							:    List[str]							=							router_z_loss_coef
												SCREAMING_SNAKE_CASE							:    List[str]							=							router_aux_loss_coef
												SCREAMING_SNAKE_CASE							:    int							=							decoder_sparse_step
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							encoder_sparse_step
												SCREAMING_SNAKE_CASE							:    List[str]							=							num_experts
												SCREAMING_SNAKE_CASE							:    int							=							expert_capacity
												SCREAMING_SNAKE_CASE							:    Any							=							router_bias
												if router_dtype not in ["float32", "float16", "bfloat16"]:
																		raise ValueError(f"""`router_dtype` must be one of 'float32', 'float16' or 'bfloat16', got {router_dtype}"""						)
												SCREAMING_SNAKE_CASE							:    str							=							router_dtype

												SCREAMING_SNAKE_CASE							:    List[Any]							=							router_ignore_padding_tokens
												SCREAMING_SNAKE_CASE							:    int							=							batch_prioritized_routing
												SCREAMING_SNAKE_CASE							:    str							=							second_expert_policy
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							normalize_router_prob_before_dropping
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							moe_eval_capacity_token_fraction
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							moe_token_dropout
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							output_router_logits
												super().__init__(
												    pad_token_id=UpperCamelCase__    ,		bos_token_id=UpperCamelCase__    ,		eos_token_id=UpperCamelCase__    ,		is_encoder_decoder=UpperCamelCase__    ,		decoder_start_token_id=UpperCamelCase__    ,		**UpperCamelCase__    ,		)





