






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaImgaImgPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()



class 							SCREAMING_SNAKE_CASE__						(    lowerCamelCase_			,       unittest.TestCase						):





					"""simple docstring"""






					a_     =		KandinskyVaaImgaImgPipeline
					a_     =		['''image_embeds''', '''negative_image_embeds''', '''image''']
					a_     =		[
					    '''image_embeds''',
					    '''negative_image_embeds''',
					    '''image''',
					]
					a_     =		[
					    '''generator''',
					    '''height''',
					    '''width''',
					    '''strength''',
					    '''guidance_scale''',
					    '''num_inference_steps''',
					    '''return_dict''',
					    '''guidance_scale''',
					    '''num_images_per_prompt''',
					    '''output_type''',
					    '''return_dict''',
					]
					a_     =		False



					@property
					def 	_lowercase					( self					:  Tuple	):
						return 3_2



					@property
					def 	_lowercase					( self					:  str	):
						return 3_2



					@property
					def 	_lowercase					( self					:  List[Any]	):
						return self.time_input_dim



					@property
					def 	_lowercase					( self					:  int	):
						return self.time_input_dim * 4



					@property
					def 	_lowercase					( self					:  Any	):
						return 1_0_0



					@property
					def 	_lowercase					( self					:  List[Any]	):
						torch.manual_seed(0	)

						snake_case__			:				Optional[int]       =  {
						    "in_channels": 4,
						    # Out channels is double in channels because predicts mean and variance
						    "out_channels": 8,
						    "addition_embed_type": "image",
						    "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
						    "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
						    "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
						    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
						    "layers_per_block": 1,
						    "encoder_hid_dim": self.text_embedder_hidden_size,
						    "encoder_hid_dim_type": "image_proj",
						    "cross_attention_dim": self.cross_attention_dim,
						    "attention_head_dim": 4,
						    "resnet_time_scale_shift": "scale_shift",
						    "class_embed_type": None,
						}

						snake_case__			:				Dict       =  UNetaDConditionModel(**__snake_case	)
						return model



					@property
					def 	_lowercase					( self					:  str	):
						return {
						    "block_out_channels": [3_2, 6_4],
						    "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
						    "in_channels": 3,
						    "latent_channels": 4,
						    "layers_per_block": 1,
						    "norm_num_groups": 8,
						    "norm_type": "spatial",
						    "num_vq_embeddings": 1_2,
						    "out_channels": 3,
						    "up_block_types": [
						        "AttnUpDecoderBlock2D",
						        "UpDecoderBlock2D",
						    ],
						    "vq_embed_dim": 4,
						}



					@property
					def 	_lowercase					( self					:  str	):
						torch.manual_seed(0	)
						snake_case__			:				Any       =  VQModel(**self.dummy_movq_kwargs	)
						return model



					def 	_lowercase					( self					:  Optional[int]	):
						snake_case__			:				Optional[Any]       =  self.dummy_unet
						snake_case__			:				Optional[int]       =  self.dummy_movq

						snake_case__			:				List[Any]       =  {
						    "num_train_timesteps": 1_0_0_0,
						    "beta_schedule": "linear",
						    "beta_start": 0.0_0_0_8_5,
						    "beta_end": 0.0_1_2,
						    "clip_sample": False,
						    "set_alpha_to_one": False,
						    "steps_offset": 0,
						    "prediction_type": "epsilon",
						    "thresholding": False,
						}

						snake_case__			:				Union[str, Any]       =  DDIMScheduler(**__snake_case	)

						snake_case__			:				List[str]       =  {
						    "unet": unet,
						    "scheduler": scheduler,
						    "movq": movq,
						}

						return components



					def 	_lowercase					( self					:  int							,		__A					:  Union[str, Any]							,		__A					:  Dict=0	):
						snake_case__			:				Tuple       =  floats_tensor((1, self.text_embedder_hidden_size)							,		rng=random.Random(__snake_case	)	).to(__snake_case	)
						snake_case__			:				Dict       =  floats_tensor((1, self.text_embedder_hidden_size)							,		rng=random.Random(seed + 1	)	).to(
						    __snake_case	)
						# create init_image
						snake_case__			:				Optional[int]       =  floats_tensor((1, 3, 6_4, 6_4)							,		rng=random.Random(__snake_case	)	).to(__snake_case	)
						snake_case__			:				Union[str, Any]       =  image.cpu().permute(0							,		2							,		3							,		1	)[0]
						snake_case__			:				Optional[Any]       =  Image.fromarray(np.uinta(__snake_case	)	).convert("RGB"	).resize((2_5_6, 2_5_6)	)

						if str(__snake_case	).startswith("mps"	):
							snake_case__			:				List[Any]       =  torch.manual_seed(__snake_case	)
						else:
							snake_case__			:				Tuple       =  torch.Generator(device=__snake_case	).manual_seed(__snake_case	)
						snake_case__			:				Optional[int]       =  {
						    "image": init_image,
						    "image_embeds": image_embeds,
						    "negative_image_embeds": negative_image_embeds,
						    "generator": generator,
						    "height": 6_4,
						    "width": 6_4,
						    "num_inference_steps": 1_0,
						    "guidance_scale": 7.0,
						    "strength": 0.2,
						    "output_type": "np",
						}
						return inputs






					def 	_lowercase					( self					:  List[Any]	):
						snake_case__			:				Tuple       =  "cpu"

						snake_case__			:				Tuple       =  self.get_dummy_components()

						snake_case__			:				Optional[int]       =  self.pipeline_class(**__snake_case	)
						snake_case__			:				Tuple       =  pipe.to(__snake_case	)

						pipe.set_progress_bar_config(disable=__snake_case	)

						snake_case__			:				List[Any]       =  pipe(**self.get_dummy_inputs(__snake_case	)	)
						snake_case__			:				Union[str, Any]       =  output.images

						snake_case__			:				int       =  pipe(
						    **self.get_dummy_inputs(__snake_case	)							,		return_dict=__snake_case							,		)[0]

						snake_case__			:				str       =  image[0, -3:, -3:, -1]
						snake_case__			:				Union[str, Any]       =  image_from_tuple[0, -3:, -3:, -1]

						assert image.shape == (1, 6_4, 6_4, 3)

						snake_case__			:				Optional[int]       =  np.array(
						    [0.6_1_9_9_7_7_8, 0.6_3_9_8_4_4_0_6, 0.4_6_1_4_5_7_8_5, 0.6_2_9_4_4_9_8_4, 0.5_6_2_2_2_1_5, 0.4_7_3_0_6_1_3_2, 0.4_7_4_4_1_4_5_6, 0.4_6_0_7_6_0_6, 0.4_8_7_1_9_2_6_3]	)
						assert (
						    np.abs(image_slice.flatten() - expected_slice	).max() < 1e-2
						), f''' expected_slice {expected_slice}, but got {image_slice.flatten()}'''
						assert (
						    np.abs(image_from_tuple_slice.flatten() - expected_slice	).max() < 1e-2
						), f''' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'''



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





					"""simple docstring"""






					def 	_lowercase					( self					:  Union[str, Any]	):
						# clean up the VRAM after each test
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()






					def 	_lowercase					( self					:  Optional[int]	):
						snake_case__			:				Any       =  load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/kandinskyv22/kandinskyv22_img2img_frog.npy"	)

						snake_case__			:				List[Any]       =  load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"	)
						snake_case__			:				List[Any]       =  "A red cartoon frog, 4k"

						snake_case__			:				Dict       =  KandinskyVaaPriorPipeline.from_pretrained(
						    "kandinsky-community/kandinsky-2-2-prior"							,		torch_dtype=torch.floataa	)
						pipe_prior.to(__snake_case	)

						snake_case__			:				List[Any]       =  KandinskyVaaImgaImgPipeline.from_pretrained(
						    "kandinsky-community/kandinsky-2-2-decoder"							,		torch_dtype=torch.floataa	)
						snake_case__			:				Dict       =  pipeline.to(__snake_case	)

						pipeline.set_progress_bar_config(disable=__snake_case	)

						snake_case__			:				Optional[Any]       =  torch.Generator(device="cpu"	).manual_seed(0	)
						snake_case__,						snake_case__			:				Union[str, Any]       =  pipe_prior(
						    __snake_case							,		generator=__snake_case							,		num_inference_steps=5							,		negative_prompt=""							,		).to_tuple()

						snake_case__			:				Optional[int]       =  pipeline(
						    image=__snake_case							,		image_embeds=__snake_case							,		negative_image_embeds=__snake_case							,		generator=__snake_case							,		num_inference_steps=1_0_0							,		height=7_6_8							,		width=7_6_8							,		strength=0.2							,		output_type="np"							,		)

						snake_case__			:				Optional[Any]       =  output.images[0]

						assert image.shape == (7_6_8, 7_6_8, 3)

						assert_mean_pixel_difference(__snake_case							,		__snake_case	)











from collections import OrderedDict
from typing import Mapping

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


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

__lowerCamelCase :      List[Any]										=				{
    """facebook/xmod-base""": """https://huggingface.co/facebook/xmod-base/resolve/main/config.json""",
    """facebook/xmod-large-prenorm""": """https://huggingface.co/facebook/xmod-large-prenorm/resolve/main/config.json""",
    """facebook/xmod-base-13-125k""": """https://huggingface.co/facebook/xmod-base-13-125k/resolve/main/config.json""",
    """facebook/xmod-base-30-125k""": """https://huggingface.co/facebook/xmod-base-30-125k/resolve/main/config.json""",
    """facebook/xmod-base-30-195k""": """https://huggingface.co/facebook/xmod-base-30-195k/resolve/main/config.json""",
    """facebook/xmod-base-60-125k""": """https://huggingface.co/facebook/xmod-base-60-125k/resolve/main/config.json""",
    """facebook/xmod-base-60-265k""": """https://huggingface.co/facebook/xmod-base-60-265k/resolve/main/config.json""",
    """facebook/xmod-base-75-125k""": """https://huggingface.co/facebook/xmod-base-75-125k/resolve/main/config.json""",
    """facebook/xmod-base-75-269k""": """https://huggingface.co/facebook/xmod-base-75-269k/resolve/main/config.json""",
}



class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					a_     =		"xmod"



					def __init__( self					:  List[str]							,		__A					:  List[str]=3_0_5_2_2							,		__A					:  Tuple=7_6_8							,		__A					:  str=1_2							,		__A					:  List[Any]=1_2							,		__A					:  List[str]=3_0_7_2							,		__A					:  List[str]="gelu"							,		__A					:  List[Any]=0.1							,		__A					:  Tuple=0.1							,		__A					:  str=5_1_2							,		__A					:  Union[str, Any]=2							,		__A					:  List[Any]=0.0_2							,		__A					:  List[str]=1e-1_2							,		__A					:  Tuple=1							,		__A					:  List[Any]=0							,		__A					:  Optional[Any]=2							,		__A					:  Optional[int]="absolute"							,		__A					:  Optional[int]=True							,		__A					:  Dict=None							,		__A					:  Optional[int]=False							,		__A					:  Dict=2							,		__A					:  List[str]=False							,		__A					:  Dict=True							,		__A					:  Union[str, Any]=True							,		__A					:  Tuple=("en_XX",)							,		__A					:  Optional[Any]=None							,		**__A					:  Tuple							,		):
						super().__init__(pad_token_id=__A							,		bos_token_id=__A							,		eos_token_id=__A							,		**__A	)

						snake_case__			:				Tuple       =  vocab_size
						snake_case__			:				Union[str, Any]       =  hidden_size
						snake_case__			:				int       =  num_hidden_layers
						snake_case__			:				str       =  num_attention_heads
						snake_case__			:				Tuple       =  hidden_act
						snake_case__			:				Optional[Any]       =  intermediate_size
						snake_case__			:				List[str]       =  hidden_dropout_prob
						snake_case__			:				Optional[int]       =  attention_probs_dropout_prob
						snake_case__			:				Union[str, Any]       =  max_position_embeddings
						snake_case__			:				str       =  type_vocab_size
						snake_case__			:				List[str]       =  initializer_range
						snake_case__			:				Dict       =  layer_norm_eps
						snake_case__			:				str       =  position_embedding_type
						snake_case__			:				List[str]       =  use_cache
						snake_case__			:				Tuple       =  classifier_dropout
						snake_case__			:				Any       =  pre_norm
						snake_case__			:				List[str]       =  adapter_reduction_factor
						snake_case__			:				List[Any]       =  adapter_layer_norm
						snake_case__			:				str       =  adapter_reuse_layer_norm
						snake_case__			:				Union[str, Any]       =  ln_before_adapter
						snake_case__			:				Tuple       =  list(__A	)
						snake_case__			:				int       =  default_language



class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					@property
					def 	_lowercase					( self					:  int	):
						if self.task == "multiple-choice":
							snake_case__			:				Any       =  {0: "batch", 1: "choice", 2: "sequence"}
						else:
							snake_case__			:				Any       =  {0: "batch", 1: "sequence"}
						return OrderedDict(
						    [
						        ("input_ids", dynamic_axis),
						        ("attention_mask", dynamic_axis),
						    ]	)



