



import gc
import random
import unittest

import numpy as np
import torch
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, 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 		UpperCAmelCase__      (      __UpperCamelCase						,unittest.TestCase       ):




			'''simple docstring'''






			UpperCamelCase      						=	KandinskyPipeline
			UpperCamelCase      						=	[
			    """prompt""",
			    """image_embeds""",
			    """negative_image_embeds""",
			]
			UpperCamelCase      						=	["""prompt""", """negative_prompt""", """image_embeds""", """negative_image_embeds"""]
			UpperCamelCase      						=	[
			    """generator""",
			    """height""",
			    """width""",
			    """latents""",
			    """guidance_scale""",
			    """negative_prompt""",
			    """num_inference_steps""",
			    """return_dict""",
			    """guidance_scale""",
			    """num_images_per_prompt""",
			    """output_type""",
			    """return_dict""",
			]
			UpperCamelCase      						=	False




			@property
			def 			snake_case__					(       self							:			Dict			):
					'''simple docstring'''





					return 32




			@property
			def 			snake_case__					(       self							:			Tuple			):
					'''simple docstring'''





					return 32




			@property
			def 			snake_case__					(       self							:			List[str]			):
					'''simple docstring'''





					return self.time_input_dim




			@property
			def 			snake_case__					(       self							:			Union[str, Any]			):
					'''simple docstring'''





					return self.time_input_dim * 4




			@property
			def 			snake_case__					(       self							:			Any			):
					'''simple docstring'''





					return 1_00




			@property
			def 			snake_case__					(       self							:			int			):
					'''simple docstring'''





					__UpperCAmelCase   :					int													=     XLMRobertaTokenizerFast.from_pretrained('''YiYiXu/tiny-random-mclip-base'''			)
					return tokenizer




			@property
			def 			snake_case__					(       self							:			int			):
					'''simple docstring'''





					torch.manual_seed(0			)
					__UpperCAmelCase   :					Tuple													=     MCLIPConfig(
					    numDims=self.cross_attention_dim	,	transformerDimensions=self.text_embedder_hidden_size	,	hidden_size=self.text_embedder_hidden_size	,	intermediate_size=37	,	num_attention_heads=4	,	num_hidden_layers=5	,	vocab_size=10_05	,	)

					__UpperCAmelCase   :					Union[str, Any]													=     MultilingualCLIP(a_			)
					__UpperCAmelCase   :					str													=     text_encoder.eval()

					return text_encoder




			@property
			def 			snake_case__					(       self							:			Dict			):
					'''simple docstring'''





					torch.manual_seed(0			)

					__UpperCAmelCase   :					str													=     {
					    '''in_channels''': 4,
					    # Out channels is double in channels because predicts mean and variance
					    '''out_channels''': 8,
					    '''addition_embed_type''': '''text_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''': '''text_image_proj''',
					    '''cross_attention_dim''': self.cross_attention_dim,
					    '''attention_head_dim''': 4,
					    '''resnet_time_scale_shift''': '''scale_shift''',
					    '''class_embed_type''': None,
					}

					__UpperCAmelCase   :					Any													=     UNetaDConditionModel(**a_			)
					return model




			@property
			def 			snake_case__					(       self							:			str			):
					'''simple docstring'''





					return {
					    "block_out_channels": [32, 64],
					    "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": 12,
					    "out_channels": 3,
					    "up_block_types": [
					        "AttnUpDecoderBlock2D",
					        "UpDecoderBlock2D",
					    ],
					    "vq_embed_dim": 4,
					}




			@property
			def 			snake_case__					(       self							:			str			):
					'''simple docstring'''





					torch.manual_seed(0			)
					__UpperCAmelCase   :					Optional[Any]													=     VQModel(**self.dummy_movq_kwargs			)
					return model




			def 			snake_case__					(       self							:			Optional[int]			):
					'''simple docstring'''





					__UpperCAmelCase   :					Dict													=     self.dummy_text_encoder
					__UpperCAmelCase   :					Dict													=     self.dummy_tokenizer
					__UpperCAmelCase   :					List[str]													=     self.dummy_unet
					__UpperCAmelCase   :					List[Any]													=     self.dummy_movq

					__UpperCAmelCase   :					Optional[Any]													=     DDIMScheduler(
					    num_train_timesteps=10_00	,	beta_schedule='''linear'''	,	beta_start=0.0_0_0_8_5	,	beta_end=0.0_1_2	,	clip_sample=a_	,	set_alpha_to_one=a_	,	steps_offset=1	,	prediction_type='''epsilon'''	,	thresholding=a_	,	)

					__UpperCAmelCase   :					Optional[Any]													=     {
					    '''text_encoder''': text_encoder,
					    '''tokenizer''': tokenizer,
					    '''unet''': unet,
					    '''scheduler''': scheduler,
					    '''movq''': movq,
					}
					return components




			def 			snake_case__					(       self							:			List[str]	,	a_							:			List[Any]	,	a_							:			List[Any]=0			):
					'''simple docstring'''





					__UpperCAmelCase   :					Optional[int]													=     floats_tensor((1, self.cross_attention_dim)	,	rng=random.Random(a_			)			).to(a_			)
					__UpperCAmelCase   :					Optional[Any]													=     floats_tensor((1, self.cross_attention_dim)	,	rng=random.Random(seed + 1			)			).to(a_			)
					if str(a_			).startswith('''mps'''			):
							__UpperCAmelCase   :					Optional[Any]													=     torch.manual_seed(a_			)
					else:
							__UpperCAmelCase   :					int													=     torch.Generator(device=a_			).manual_seed(a_			)
					__UpperCAmelCase   :					Optional[Any]													=     {
					    '''prompt''': '''horse''',
					    '''image_embeds''': image_embeds,
					    '''negative_image_embeds''': negative_image_embeds,
					    '''generator''': generator,
					    '''height''': 64,
					    '''width''': 64,
					    '''guidance_scale''': 4.0,
					    '''num_inference_steps''': 2,
					    '''output_type''': '''np''',
					}
					return inputs




			def 			snake_case__					(       self							:			int			):
					'''simple docstring'''





					__UpperCAmelCase   :					Optional[int]													=     '''cpu'''

					__UpperCAmelCase   :					str													=     self.get_dummy_components()

					__UpperCAmelCase   :					Optional[Any]													=     self.pipeline_class(**a_			)
					__UpperCAmelCase   :					List[Any]													=     pipe.to(a_			)

					pipe.set_progress_bar_config(disable=a_			)

					__UpperCAmelCase   :					Any													=     pipe(**self.get_dummy_inputs(a_			)			)
					__UpperCAmelCase   :					Any													=     output.images

					__UpperCAmelCase   :					int													=     pipe(
					    **self.get_dummy_inputs(a_			)	,	return_dict=a_	,	)[0]

					__UpperCAmelCase   :					Optional[int]													=     image[0, -3:, -3:, -1]
					__UpperCAmelCase   :					List[Any]													=     image_from_tuple[0, -3:, -3:, -1]

					assert image.shape == (1, 64, 64, 3)

					__UpperCAmelCase   :					Dict													=     np.array(
					    [0.3_2_8_6_6_3, 1.0, 0.2_3_2_1_6_8_7_3, 1.0, 0.9_2_7_1_7_5_6_4, 0.4_6_3_9_0_4_6, 0.9_6_8_9_4_7_7_7, 0.3_1_7_1_3_3_7_8, 0.6_2_9_3_9_5_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 		UpperCAmelCase__      (      unittest.TestCase       ):




			'''simple docstring'''






			def 			snake_case__					(       self							:			Dict			):
					'''simple docstring'''





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




			def 			snake_case__					(       self							:			int			):
					'''simple docstring'''





					__UpperCAmelCase   :					List[Any]													=     load_numpy(
					    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
					    '''/kandinsky/kandinsky_text2img_cat_fp16.npy'''			)

					__UpperCAmelCase   :					List[str]													=     KandinskyPriorPipeline.from_pretrained(
					    '''kandinsky-community/kandinsky-2-1-prior'''	,	torch_dtype=torch.floataa			)
					pipe_prior.to(a_			)

					__UpperCAmelCase   :					Dict													=     KandinskyPipeline.from_pretrained('''kandinsky-community/kandinsky-2-1'''	,	torch_dtype=torch.floataa			)
					__UpperCAmelCase   :					str													=     pipeline.to(a_			)
					pipeline.set_progress_bar_config(disable=a_			)

					__UpperCAmelCase   :					str													=     '''red cat, 4k photo'''

					__UpperCAmelCase   :					List[Any]													=     torch.Generator(device='''cuda'''			).manual_seed(0			)
					__UpperCAmelCase							, __UpperCAmelCase   :					Any													=     pipe_prior(
					    a_	,	generator=a_	,	num_inference_steps=5	,	negative_prompt=''''''	,	).to_tuple()

					__UpperCAmelCase   :					Union[str, Any]													=     torch.Generator(device='''cuda'''			).manual_seed(0			)
					__UpperCAmelCase   :					str													=     pipeline(
					    a_	,	image_embeds=a_	,	negative_image_embeds=a_	,	generator=a_	,	num_inference_steps=1_00	,	output_type='''np'''	,	)

					__UpperCAmelCase   :					Union[str, Any]													=     output.images[0]

					assert image.shape == (5_12, 5_12, 3)

					assert_mean_pixel_difference(a_	,	a_			)










import sys
from typing import Tuple

import numpy as np
import torch
from PIL import Image
from torch import nn

from transformers.image_utils import PILImageResampling
from utils import img_tensorize

class 		UpperCAmelCase__      :




			'''simple docstring'''






			def __init__(       self							:			int	,	a_							:			str	,	a_							:			List[str]=sys.maxsize			):
					'''simple docstring'''





					__UpperCAmelCase   :					List[str]													=     '''bilinear'''
					__UpperCAmelCase   :					str													=     max_size
					__UpperCAmelCase   :					Dict													=     short_edge_length




			def __call__(       self							:			str	,	a_							:			Union[str, Any]			):
					'''simple docstring'''





					__UpperCAmelCase   :					List[str]													=     []
					for img in imgs:
							__UpperCAmelCase							, __UpperCAmelCase   :					List[str]													=     img.shape[:2]
							# later: provide list and randomly choose index for resize
							__UpperCAmelCase   :					Optional[Any]													=     np.random.randint(self.short_edge_length[0]	,	self.short_edge_length[1] + 1			)
							if size == 0:
									return img
							__UpperCAmelCase   :					Optional[Any]													=     size * 1.0 / min(a_	,	a_			)
							if h < w:
									__UpperCAmelCase							, __UpperCAmelCase   :					List[str]													=     size, scale * w
							else:
									__UpperCAmelCase							, __UpperCAmelCase   :					Union[str, Any]													=     scale * h, size
							if max(a_	,	a_			) > self.max_size:
									__UpperCAmelCase   :					Union[str, Any]													=     self.max_size * 1.0 / max(a_	,	a_			)
									__UpperCAmelCase   :					int													=     newh * scale
									__UpperCAmelCase   :					Union[str, Any]													=     neww * scale
							__UpperCAmelCase   :					Any													=     int(neww + 0.5			)
							__UpperCAmelCase   :					Optional[int]													=     int(newh + 0.5			)

							if img.dtype == np.uinta:
									__UpperCAmelCase   :					Any													=     Image.fromarray(a_			)
									__UpperCAmelCase   :					Any													=     pil_image.resize((neww, newh)	,	PILImageResampling.BILINEAR			)
									__UpperCAmelCase   :					List[Any]													=     np.asarray(a_			)
							else:
									__UpperCAmelCase   :					Union[str, Any]													=     img.permute(2	,	0	,	1			).unsqueeze(0			)  # 3, 0, 1)  # hw(c) -> nchw
									__UpperCAmelCase   :					Optional[int]													=     nn.functional.interpolate(
									    a_	,	(newh, neww)	,	mode=self.interp_method	,	align_corners=a_			).squeeze(0			)
							img_augs.append(a_			)

					return img_augs

class 		UpperCAmelCase__      :




			'''simple docstring'''






			def __init__(       self							:			Tuple	,	a_							:			List[Any]			):
					'''simple docstring'''





					__UpperCAmelCase   :					Optional[int]													=     ResizeShortestEdge([cfg.INPUT.MIN_SIZE_TEST, cfg.INPUT.MIN_SIZE_TEST]	,	cfg.INPUT.MAX_SIZE_TEST			)
					__UpperCAmelCase   :					List[str]													=     cfg.INPUT.FORMAT
					__UpperCAmelCase   :					int													=     cfg.SIZE_DIVISIBILITY
					__UpperCAmelCase   :					int													=     cfg.PAD_VALUE
					__UpperCAmelCase   :					Any													=     cfg.INPUT.MAX_SIZE_TEST
					__UpperCAmelCase   :					Tuple													=     cfg.MODEL.DEVICE
					__UpperCAmelCase   :					Union[str, Any]													=     torch.tensor(cfg.MODEL.PIXEL_STD			).to(self.device			).view(len(cfg.MODEL.PIXEL_STD			)	,	1	,	1			)
					__UpperCAmelCase   :					List[Any]													=     torch.tensor(cfg.MODEL.PIXEL_MEAN			).to(self.device			).view(len(cfg.MODEL.PIXEL_STD			)	,	1	,	1			)
					__UpperCAmelCase   :					Union[str, Any]													=     lambda a_			: (x - self.pixel_mean) / self.pixel_std




			def 			snake_case__					(       self							:			int	,	a_							:			Tuple			):
					'''simple docstring'''





					__UpperCAmelCase   :					Optional[Any]													=     tuple(max(a_			) for s in zip(*[img.shape for img in images]			)			)
					__UpperCAmelCase   :					Any													=     [im.shape[-2:] for im in images]
					__UpperCAmelCase   :					Union[str, Any]													=     [
					    nn.functional.pad(
					        a_	,	[0, max_size[-1] - size[1], 0, max_size[-2] - size[0]]	,	value=self.pad_value	,	)
					    for size, im in zip(a_	,	a_			)
					]

					return torch.stack(a_			), torch.tensor(a_			)




			def __call__(       self							:			List[str]	,	a_							:			int	,	a_							:			int=False			):
					'''simple docstring'''





					with torch.no_grad():
							if not isinstance(a_	,	a_			):
									__UpperCAmelCase   :					Optional[Any]													=     [images]
							if single_image:
									assert len(a_			) == 1
							for i in range(len(a_			)			):
									if isinstance(images[i]	,	torch.Tensor			):
											images.insert(a_	,	images.pop(a_			).to(self.device			).float()			)
									elif not isinstance(images[i]	,	torch.Tensor			):
											images.insert(
											    a_	,	torch.as_tensor(img_tensorize(images.pop(a_			)	,	input_format=self.input_format			)			)
											    .to(self.device			)
											    .float()	,	)
            # resize smallest edge
							__UpperCAmelCase   :					Optional[Any]													=     torch.tensor([im.shape[:2] for im in images]			)
							__UpperCAmelCase   :					Dict													=     self.aug(a_			)
							# transpose images and convert to torch tensors
							# images = [torch.as_tensor(i.astype("float32")).permute(2, 0, 1).to(self.device) for i in images]
							# now normalize before pad to avoid useless arithmetic
							__UpperCAmelCase   :					Optional[int]													=     [self.normalizer(a_			) for x in images]
							# now pad them to do the following operations
							__UpperCAmelCase							, __UpperCAmelCase   :					Optional[int]													=     self.pad(a_			)
							# Normalize

							if self.size_divisibility > 0:
									raise NotImplementedError()
							# pad
							__UpperCAmelCase   :					str													=     torch.true_divide(a_	,	a_			)
							if single_image:
									return images[0], sizes[0], scales_yx[0]
							else:
									return images, sizes, scales_yx

def 	a						(				_UpperCAmelCase							: str			,       _UpperCAmelCase							: Optional[int]						):







		'''simple docstring'''





		boxes[:, 0::2] *= scale_yx[:, 1]
		boxes[:, 1::2] *= scale_yx[:, 0]
		return boxes

def 	a						(				_UpperCAmelCase							: Any			,       _UpperCAmelCase							: Tuple[int, int]						):







		'''simple docstring'''





		assert torch.isfinite(_UpperCAmelCase						).all(), "Box tensor contains infinite or NaN!"
		__UpperCAmelCase							, __UpperCAmelCase   :					List[str]													=     box_size
		tensor[:, 0].clamp_(min=0			,       max=_UpperCAmelCase						)
		tensor[:, 1].clamp_(min=0			,       max=_UpperCAmelCase						)
		tensor[:, 2].clamp_(min=0			,       max=_UpperCAmelCase						)
		tensor[:, 3].clamp_(min=0			,       max=_UpperCAmelCase						)





