
"""simple docstring"""






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
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 			lowercase					( lowercase__   ,unittest.TestCase		):
			lowercase        =     KandinskyImgaImgPipeline
			lowercase        =     ['''prompt''', '''image_embeds''', '''negative_image_embeds''', '''image''']
			lowercase        =     [
			    '''prompt''',
			    '''negative_prompt''',
			    '''image_embeds''',
			    '''negative_image_embeds''',
			    '''image''',
			]
			lowercase        =     [
			    '''generator''',
			    '''height''',
			    '''width''',
			    '''strength''',
			    '''guidance_scale''',
			    '''negative_prompt''',
			    '''num_inference_steps''',
			    '''return_dict''',
			    '''guidance_scale''',
			    '''num_images_per_prompt''',
			    '''output_type''',
			    '''return_dict''',
			]
			lowercase        =     False






			@property
			def 					UpperCAmelCase							(self  :	int			)       ->	Union[str, Any]:





				"""simple docstring"""


				return 32






			@property
			def 					UpperCAmelCase							(self  :	List[str]			)       ->	Optional[Any]:





				"""simple docstring"""


				return 32






			@property
			def 					UpperCAmelCase							(self  :	Union[str, Any]			)       ->	Union[str, Any]:





				"""simple docstring"""


				return self.time_input_dim






			@property
			def 					UpperCAmelCase							(self  :	List[Any]			)       ->	int:





				"""simple docstring"""


				return self.time_input_dim * 4






			@property
			def 					UpperCAmelCase							(self  :	Optional[int]			)       ->	Any:





				"""simple docstring"""


				return 100






			@property
			def 					UpperCAmelCase							(self  :	Dict			)       ->	int:





				"""simple docstring"""


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






			@property
			def 					UpperCAmelCase							(self  :	List[str]			)       ->	List[str]:





				"""simple docstring"""


				torch.manual_seed(0			)
				lowerCAmelCase												=  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=1_005      ,)

				lowerCAmelCase												=  MultilingualCLIP(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  text_encoder.eval()

				return text_encoder






			@property
			def 					UpperCAmelCase							(self  :	str			)       ->	Dict:





				"""simple docstring"""


				torch.manual_seed(0			)

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

				lowerCAmelCase												=  UNetaDConditionModel(**SCREAMING_SNAKE_CASE_			)
				return model






			@property
			def 					UpperCAmelCase							(self  :	Tuple			)       ->	Union[str, Any]:





				"""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 					UpperCAmelCase							(self  :	Tuple			)       ->	Any:





				"""simple docstring"""


				torch.manual_seed(0			)
				lowerCAmelCase												=  VQModel(**self.dummy_movq_kwargs			)
				return model






			def 					UpperCAmelCase							(self  :	Optional[Any]			)       ->	List[str]:





				"""simple docstring"""


				lowerCAmelCase												=  self.dummy_text_encoder
				lowerCAmelCase												=  self.dummy_tokenizer
				lowerCAmelCase												=  self.dummy_unet
				lowerCAmelCase												=  self.dummy_movq

				lowerCAmelCase												=  {
				    '''num_train_timesteps''': 1_000,
				    '''beta_schedule''': '''linear''',
				    '''beta_start''': 0.0_00_85,
				    '''beta_end''': 0.0_12,
				    '''clip_sample''': False,
				    '''set_alpha_to_one''': False,
				    '''steps_offset''': 0,
				    '''prediction_type''': '''epsilon''',
				    '''thresholding''': False,
				}

				lowerCAmelCase												=  DDIMScheduler(**SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  {
				    '''text_encoder''': text_encoder,
				    '''tokenizer''': tokenizer,
				    '''unet''': unet,
				    '''scheduler''': scheduler,
				    '''movq''': movq,
				}

				return components






			def 					UpperCAmelCase							(self  :	Dict      ,SCREAMING_SNAKE_CASE_  :	int      ,SCREAMING_SNAKE_CASE_  :	Optional[int]=0			)       ->	int:





				"""simple docstring"""


				lowerCAmelCase												=  floats_tensor((1, self.cross_attention_dim)      ,rng=random.Random(SCREAMING_SNAKE_CASE_			)			).to(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  floats_tensor((1, self.cross_attention_dim)      ,rng=random.Random(seed + 1			)			).to(SCREAMING_SNAKE_CASE_			)
				# create init_image
				lowerCAmelCase												=  floats_tensor((1, 3, 64, 64)      ,rng=random.Random(SCREAMING_SNAKE_CASE_			)			).to(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  image.cpu().permute(0      ,2      ,3      ,1			)[0]
				lowerCAmelCase												=  Image.fromarray(np.uinta(SCREAMING_SNAKE_CASE_			)			).convert('''RGB'''			).resize((256, 256)			)

				if str(SCREAMING_SNAKE_CASE_			).startswith('''mps'''			):
					lowerCAmelCase												=  torch.manual_seed(SCREAMING_SNAKE_CASE_			)
				else:
					lowerCAmelCase												=  torch.Generator(device=SCREAMING_SNAKE_CASE_			).manual_seed(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  {
				    '''prompt''': '''horse''',
				    '''image''': init_image,
				    '''image_embeds''': image_embeds,
				    '''negative_image_embeds''': negative_image_embeds,
				    '''generator''': generator,
				    '''height''': 64,
				    '''width''': 64,
				    '''num_inference_steps''': 10,
				    '''guidance_scale''': 7.0,
				    '''strength''': 0.2,
				    '''output_type''': '''np''',
				}
				return inputs






			def 					UpperCAmelCase							(self  :	Any			)       ->	List[str]:





				"""simple docstring"""


				lowerCAmelCase												=  '''cpu'''

				lowerCAmelCase												=  self.get_dummy_components()

				lowerCAmelCase												=  self.pipeline_class(**SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  pipe.to(SCREAMING_SNAKE_CASE_			)

				pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  pipe(**self.get_dummy_inputs(SCREAMING_SNAKE_CASE_			)			)
				lowerCAmelCase												=  output.images

				lowerCAmelCase												=  pipe(
				    **self.get_dummy_inputs(SCREAMING_SNAKE_CASE_			)      ,return_dict=SCREAMING_SNAKE_CASE_      ,)[0]

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

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

				lowerCAmelCase												=  np.array(
				    [0.61_47_49_43, 0.6_07_35_39, 0.43_30_85_44, 0.5_92_82_69, 0.47_49_35_95, 0.46_75_59_73, 0.4_61_38_38, 0.45_36_87_97, 0.50_11_92_33]			)
				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 			lowercase					( unittest.TestCase		):






			def 					UpperCAmelCase							(self  :	Optional[Any]			)       ->	Optional[Any]:





				"""simple docstring"""


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






			def 					UpperCAmelCase							(self  :	Dict			)       ->	List[Any]:





				"""simple docstring"""


				lowerCAmelCase												=  load_numpy(
				    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
				    '''/kandinsky/kandinsky_img2img_frog.npy'''			)

				lowerCAmelCase												=  load_image(
				    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main''' '''/kandinsky/cat.png'''			)
				lowerCAmelCase												=  '''A red cartoon frog, 4k'''

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

				lowerCAmelCase												=  KandinskyImgaImgPipeline.from_pretrained(
				    '''kandinsky-community/kandinsky-2-1'''      ,torch_dtype=torch.floataa			)
				lowerCAmelCase												=  pipeline.to(SCREAMING_SNAKE_CASE_			)

				pipeline.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  torch.Generator(device='''cpu'''			).manual_seed(0			)
				lowerCAmelCase							,       lowerCAmelCase												=  pipe_prior(
				    SCREAMING_SNAKE_CASE_      ,generator=SCREAMING_SNAKE_CASE_      ,num_inference_steps=5      ,negative_prompt=''''''      ,).to_tuple()

				lowerCAmelCase												=  pipeline(
				    SCREAMING_SNAKE_CASE_      ,image=SCREAMING_SNAKE_CASE_      ,image_embeds=SCREAMING_SNAKE_CASE_      ,negative_image_embeds=SCREAMING_SNAKE_CASE_      ,generator=SCREAMING_SNAKE_CASE_      ,num_inference_steps=100      ,height=768      ,width=768      ,strength=0.2      ,output_type='''np'''      ,)

				lowerCAmelCase												=  output.images[0]

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

				assert_mean_pixel_difference(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)




"""simple docstring"""






from __future__ import annotations



class 			lowercase					:






			def __init__(self  :	List[Any]      ,SCREAMING_SNAKE_CASE_  :	int = 0			)       ->	List[Any]:





				"""simple docstring"""


				lowerCAmelCase												=  key






			def 					UpperCAmelCase							(self  :	Tuple      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int			)       ->	list[str]:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  key or self.__key or 1

				# make sure key is an appropriate size
				key %= 255

				return [chr(ord(SCREAMING_SNAKE_CASE_			) ^ key			) for ch in content]






			def 					UpperCAmelCase							(self  :	int      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int			)       ->	list[str]:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  key or self.__key or 1

				# make sure key is an appropriate size
				key %= 255

				return [chr(ord(SCREAMING_SNAKE_CASE_			) ^ key			) for ch in content]






			def 					UpperCAmelCase							(self  :	Union[str, Any]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int = 0			)       ->	str:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  key or self.__key or 1

				# make sure key can be any size
				while key > 255:
					key -= 255

				# This will be returned
				lowerCAmelCase												=  ''''''

				for ch in content:
					ans += chr(ord(SCREAMING_SNAKE_CASE_			) ^ key			)

				return ans






			def 					UpperCAmelCase							(self  :	Union[str, Any]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int = 0			)       ->	str:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  key or self.__key or 1

				# make sure key can be any size
				while key > 255:
					key -= 255

				# This will be returned
				lowerCAmelCase												=  ''''''

				for ch in content:
					ans += chr(ord(SCREAMING_SNAKE_CASE_			) ^ key			)

				return ans






			def 					UpperCAmelCase							(self  :	List[str]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int = 0			)       ->	bool:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				try:
					with open(SCREAMING_SNAKE_CASE_			) as fin, open('''encrypt.out'''      ,'''w+'''			) as fout:
						# actual encrypt-process
						for line in fin:
							fout.write(self.encrypt_string(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)			)

				except OSError:
					return False

				return True






			def 					UpperCAmelCase							(self  :	List[Any]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int			)       ->	bool:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				try:
					with open(SCREAMING_SNAKE_CASE_			) as fin, open('''decrypt.out'''      ,'''w+'''			) as fout:
						# actual encrypt-process
						for line in fin:
							fout.write(self.decrypt_string(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)			)

				except OSError:
					return False

				return True


# Tests
# crypt = XORCipher()
# key = 67

# # test encrypt
# print(crypt.encrypt("hallo welt",key))
# # test decrypt
# print(crypt.decrypt(crypt.encrypt("hallo welt",key), key))

# # test encrypt_string
# print(crypt.encrypt_string("hallo welt",key))

# # test decrypt_string
# print(crypt.decrypt_string(crypt.encrypt_string("hallo welt",key),key))

# if (crypt.encrypt_file("test.txt",key)):
#       print("encrypt successful")
# else:
#       print("encrypt unsuccessful")

# if (crypt.decrypt_file("encrypt.out",key)):
#       print("decrypt successful")
# else:
#       print("decrypt unsuccessful")


