




from __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any


class 							__SCREAMING_SNAKE_CASE				:
					_lowerCamelCase						    =					("""_front""", """_back""", """_len""")


					@dataclass
					class 							__SCREAMING_SNAKE_CASE				:
										_lowerCamelCase						    =					None
										_lowerCamelCase						    =					None
										_lowerCamelCase						    =					None


					class 							__SCREAMING_SNAKE_CASE				:
										_lowerCamelCase						    =					("""_cur""",)

										def __init__(							self					,			__lowercase   ):







											'''simple docstring'''




											lowerCamelCase_				:						Optional[int]				  =				cur

										def __iter__(							self   ):







											'''simple docstring'''




											return self





										def       lowercase_				(							self   ):







											'''simple docstring'''




											if self._cur is None:
												# finished iterating
												raise StopIteration
											lowerCamelCase_				:						Any				  =				self._cur.val
											lowerCamelCase_				:						Dict				  =				self._cur.next_node

											return val

					def __init__(							self					,			__lowercase = None   ):







						'''simple docstring'''




						lowerCamelCase_				:						Any				  =				None
						lowerCamelCase_				:						Any				  =				None
						lowerCamelCase_				:						int				  =				0

						if iterable is not None:
							# append every value to the deque
							for val in iterable:
								self.append(__lowercase   )

					def       lowercase_				(							self					,			__lowercase   ):







						'''simple docstring'''




						lowerCamelCase_				:						Dict				  =				self._Node(__lowercase					,			__lowercase					,			__lowercase   )
						if self.is_empty():
							# front = back
							lowerCamelCase_				:						Any				  =				node
							lowerCamelCase_				:						int				  =				1
						else:
							# connect nodes
							lowerCamelCase_				:						Optional[int]				  =				node
							lowerCamelCase_				:						Optional[Any]				  =				self._back
							lowerCamelCase_				:						str				  =				node  # assign new back to the new node

							self._len += 1

							# make sure there were no errors
							assert not self.is_empty(), "Error on appending value."

					def       lowercase_				(							self					,			__lowercase   ):







						'''simple docstring'''




						lowerCamelCase_				:						List[str]				  =				self._Node(__lowercase					,			__lowercase					,			__lowercase   )
						if self.is_empty():
							# front = back
							lowerCamelCase_				:						Optional[int]				  =				node
							lowerCamelCase_				:						Dict				  =				1
						else:
							# connect nodes
							lowerCamelCase_				:						Dict				  =				self._front
							lowerCamelCase_				:						Union[str, Any]				  =				node
							lowerCamelCase_				:						Tuple				  =				node  # assign new front to the new node

							self._len += 1

							# make sure there were no errors
							assert not self.is_empty(), "Error on appending value."

					def       lowercase_				(							self					,			__lowercase   ):







						'''simple docstring'''




						for val in iterable:
							self.append(__lowercase   )

					def       lowercase_				(							self					,			__lowercase   ):







						'''simple docstring'''




						for val in iterable:
							self.appendleft(__lowercase   )

					def       lowercase_				(							self   ):







						'''simple docstring'''




						assert not self.is_empty(), "Deque is empty."

						lowerCamelCase_				:						Tuple				  =				self._back
						lowerCamelCase_				:						Union[str, Any]				  =				self._back.prev_node  # set new back
						# drop the last node - python will deallocate memory automatically
						lowerCamelCase_				:						Tuple				  =				None

						self._len -= 1

						return topop.val

					def       lowercase_				(							self   ):







						'''simple docstring'''




						assert not self.is_empty(), "Deque is empty."

						lowerCamelCase_				:						Union[str, Any]				  =				self._front
						lowerCamelCase_				:						Tuple				  =				self._front.next_node  # set new front and drop the first node
						lowerCamelCase_				:						int				  =				None

						self._len -= 1

						return topop.val

					def       lowercase_				(							self   ):







						'''simple docstring'''




						return self._front is None

					def __len__(							self   ):







						'''simple docstring'''




						return self._len

					def __eq__(							self					,			__lowercase   ):







						'''simple docstring'''




						if not isinstance(__lowercase					,			__lowercase   ):
							return NotImplemented

						lowerCamelCase_				:						Any				  =				self._front
						lowerCamelCase_				:						Tuple				  =				other._front

						# if the length of the dequeues are not the same, they are not equal
						if len(self   ) != len(__lowercase   ):
							return False

						while me is not None and oth is not None:
							# compare every value
							if me.val != oth.val:
								return False
							lowerCamelCase_				:						Tuple				  =				me.next_node
							lowerCamelCase_				:						Any				  =				oth.next_node

						return True

					def __iter__(							self   ):







						'''simple docstring'''




						return Deque._Iterator(self._front   )





					def __repr__(							self   ):







						'''simple docstring'''




						lowerCamelCase_				:						Optional[int]				  =				[]
						lowerCamelCase_				:						str				  =				self._front
						while aux is not None:
							# append the values in a list to display
							values_list.append(aux.val   )
							lowerCamelCase_				:						Optional[int]				  =				aux.next_node

						return f"[{', '.join(repr(__lowercase   ) for val in values_list   )}]"


if __name__ == "__main__":
							import doctest

							doctest.testmod()

"""simple docstring"""




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__(      __A		,    unittest.TestCase			):
						lowerCAmelCase__	:    Tuple    				=		KandinskyPipeline
						lowerCAmelCase__	:    int    				=		[
						    'prompt',
						    'image_embeds',
						    'negative_image_embeds',
						]
						lowerCAmelCase__	:    str    				=		['prompt', 'negative_prompt', 'image_embeds', 'negative_image_embeds']
						lowerCAmelCase__	:    Any    				=		[
						    'generator',
						    'height',
						    'width',
						    'latents',
						    'guidance_scale',
						    'negative_prompt',
						    'num_inference_steps',
						    'return_dict',
						    'guidance_scale',
						    'num_images_per_prompt',
						    'output_type',
						    'return_dict',
						]
						lowerCAmelCase__	:    Dict    				=		False





						@property
						def 					snake_case__						(							self  )       ->					List[Any]:
													return 32





						@property
						def 					snake_case__						(							self  )       ->					Any:
													return 32





						@property
						def 					snake_case__						(							self  )       ->					Dict:
													return self.time_input_dim





						@property
						def 					snake_case__						(							self  )       ->					Optional[Any]:
													return self.time_input_dim * 4





						@property
						def 					snake_case__						(							self  )       ->					List[str]:
													return 1_00





						@property
						def 					snake_case__						(							self  )       ->					Any:
													A__						  =				XLMRobertaTokenizerFast.from_pretrained('YiYiXu/tiny-random-mclip-base'  )
													return tokenizer





						@property
						def 					snake_case__						(							self  )       ->					Tuple:
													torch.manual_seed(0  )
													A__						  =				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		,)

													A__						  =				MultilingualCLIP(__UpperCAmelCase  )
													A__						  =				text_encoder.eval()

													return text_encoder





						@property
						def 					snake_case__						(							self  )       ->					Optional[Any]:
													torch.manual_seed(0  )

													A__						  =				{
													    '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,
													}

													A__						  =				UNetaDConditionModel(**__UpperCAmelCase  )
													return model





						@property
						def 					snake_case__						(							self  )       ->					Tuple:
													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  )       ->					Any:
													torch.manual_seed(0  )
													A__						  =				VQModel(**self.dummy_movq_kwargs  )
													return model





						def 					snake_case__						(							self  )       ->					Optional[int]:
													A__						  =				self.dummy_text_encoder
													A__						  =				self.dummy_tokenizer
													A__						  =				self.dummy_unet
													A__						  =				self.dummy_movq

													A__						  =				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=__UpperCAmelCase		,set_alpha_to_one=__UpperCAmelCase		,steps_offset=1		,prediction_type='epsilon'		,thresholding=__UpperCAmelCase		,)

													A__						  =				{
													    'text_encoder': text_encoder,
													    'tokenizer': tokenizer,
													    'unet': unet,
													    'scheduler': scheduler,
													    'movq': movq,
													}
													return components





						def 					snake_case__						(							self		,__UpperCAmelCase		,__UpperCAmelCase=0  )       ->					Optional[Any]:
													A__						  =				floats_tensor((1, self.cross_attention_dim)		,rng=random.Random(__UpperCAmelCase  )  ).to(__UpperCAmelCase  )
													A__						  =				floats_tensor((1, self.cross_attention_dim)		,rng=random.Random(seed + 1  )  ).to(__UpperCAmelCase  )
													if str(__UpperCAmelCase  ).startswith('mps'  ):
																				A__						  =				torch.manual_seed(__UpperCAmelCase  )
													else:
																				A__						  =				torch.Generator(device=__UpperCAmelCase  ).manual_seed(__UpperCAmelCase  )
													A__						  =				{
													    '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  )       ->					Optional[Any]:
													A__						  =				'cpu'

													A__						  =				self.get_dummy_components()

													A__						  =				self.pipeline_class(**__UpperCAmelCase  )
													A__						  =				pipe.to(__UpperCAmelCase  )

													pipe.set_progress_bar_config(disable=__UpperCAmelCase  )

													A__						  =				pipe(**self.get_dummy_inputs(__UpperCAmelCase  )  )
													A__						  =				output.images

													A__						  =				pipe(
													    **self.get_dummy_inputs(__UpperCAmelCase  )		,return_dict=__UpperCAmelCase		,)[0]

													A__						  =				image[0, -3:, -3:, -1]
													A__						  =				image_from_tuple[0, -3:, -3:, -1]

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

													A__						  =				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			):





						def 					snake_case__						(							self  )       ->					Tuple:
													# clean up the VRAM after each test
													super().tearDown()
													gc.collect()
													torch.cuda.empty_cache()





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

													A__						  =				KandinskyPriorPipeline.from_pretrained(
													    'kandinsky-community/kandinsky-2-1-prior'		,torch_dtype=torch.floataa  )
													pipe_prior.to(__UpperCAmelCase  )

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

													A__						  =				'red cat, 4k photo'

													A__						  =				torch.Generator(device='cuda'  ).manual_seed(0  )
													A__  ,	A__						  =				pipe_prior(
													    __UpperCAmelCase		,generator=__UpperCAmelCase		,num_inference_steps=5		,negative_prompt=''		,).to_tuple()

													A__						  =				torch.Generator(device='cuda'  ).manual_seed(0  )
													A__						  =				pipeline(
													    __UpperCAmelCase		,image_embeds=__UpperCAmelCase		,negative_image_embeds=__UpperCAmelCase		,generator=__UpperCAmelCase		,num_inference_steps=1_00		,output_type='np'		,)

													A__						  =				output.images[0]

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

													assert_mean_pixel_difference(__UpperCAmelCase		,__UpperCAmelCase  )





