
import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class    _UpperCamelCase(   SCREAMING_SNAKE_CASE		,						SCREAMING_SNAKE_CASE		,						SCREAMING_SNAKE_CASE		,						unittest.TestCase				):
						__A:						List[Any]				   =      StableDiffusionImageVariationPipeline
						__A:						Union[str, Any]				   =      IMAGE_VARIATION_PARAMS
						__A:						List[Any]				   =      IMAGE_VARIATION_BATCH_PARAMS
						__A:						Tuple				   =      frozenset([]				)
						# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
						__A:						Any				   =      frozenset([]				)

						def 						a__ (					self		:			Tuple			):
													torch.manual_seed(0			)
													_UpperCAmelCase  :			Any	       =		UNetaDConditionModel(
													    block_out_channels=(32, 64)	,						layers_per_block=2	,						sample_size=32	,						in_channels=4	,						out_channels=4	,						down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")	,						up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")	,						cross_attention_dim=32	,						)
													_UpperCAmelCase  :			List[str]	       =		PNDMScheduler(skip_prk_steps=_lowerCamelCase			)
													torch.manual_seed(0			)
													_UpperCAmelCase  :			int	       =		AutoencoderKL(
													    block_out_channels=[32, 64]	,						in_channels=3	,						out_channels=3	,						down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]	,						up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]	,						latent_channels=4	,						)
													torch.manual_seed(0			)
													_UpperCAmelCase  :			Optional[Any]	       =		CLIPVisionConfig(
													    hidden_size=32	,						projection_dim=32	,						intermediate_size=37	,						layer_norm_eps=1E-05	,						num_attention_heads=4	,						num_hidden_layers=5	,						image_size=32	,						patch_size=4	,						)
													_UpperCAmelCase  :			Tuple	       =		CLIPVisionModelWithProjection(_lowerCamelCase			)
													_UpperCAmelCase  :			str	       =		CLIPImageProcessor(crop_size=32	,						size=32			)

													_UpperCAmelCase  :			Any	       =		{
													    "unet": unet,
													    "scheduler": scheduler,
													    "vae": vae,
													    "image_encoder": image_encoder,
													    "feature_extractor": feature_extractor,
													    "safety_checker": None,
													}
													return components

						def 						a__ (					self		:			List[Any]	,						_lowerCamelCase		:			Tuple	,						_lowerCamelCase		:			List[str]=0			):
													_UpperCAmelCase  :			int	       =		floats_tensor((1, 3, 32, 32)	,						rng=random.Random(_lowerCamelCase			)			)
													_UpperCAmelCase  :			Dict	       =		image.cpu().permute(0	,						2	,						3	,						1			)[0]
													_UpperCAmelCase  :			Any	       =		Image.fromarray(np.uinta(_lowerCamelCase			)			).convert("RGB"			).resize((32, 32)			)
													if str(_lowerCamelCase			).startswith("mps"			):
																				_UpperCAmelCase  :			Optional[int]	       =		torch.manual_seed(_lowerCamelCase			)
													else:
																				_UpperCAmelCase  :			Dict	       =		torch.Generator(device=_lowerCamelCase			).manual_seed(_lowerCamelCase			)
													_UpperCAmelCase  :			Dict	       =		{
													    "image": image,
													    "generator": generator,
													    "num_inference_steps": 2,
													    "guidance_scale": 6.0,
													    "output_type": "numpy",
													}
													return inputs

						def 						a__ (					self		:			Dict			):
													_UpperCAmelCase  :			Any	       =		"cpu"  # ensure determinism for the device-dependent torch.Generator
													_UpperCAmelCase  :			Any	       =		self.get_dummy_components()
													_UpperCAmelCase  :			List[str]	       =		StableDiffusionImageVariationPipeline(**_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		sd_pipe.to(_lowerCamelCase			)
													sd_pipe.set_progress_bar_config(disable=_lowerCamelCase			)

													_UpperCAmelCase  :			Optional[Any]	       =		self.get_dummy_inputs(_lowerCamelCase			)
													_UpperCAmelCase  :			Union[str, Any]	       =		sd_pipe(**_lowerCamelCase			).images
													_UpperCAmelCase  :			Optional[int]	       =		image[0, -3:, -3:, -1]

													assert image.shape == (1, 64, 64, 3)
													_UpperCAmelCase  :			List[Any]	       =		np.array([0.52_39, 0.57_23, 0.47_96, 0.50_49, 0.55_50, 0.46_85, 0.53_29, 0.48_91, 0.49_21]			)

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

						def 						a__ (					self		:			Dict			):
													_UpperCAmelCase  :			Tuple	       =		"cpu"  # ensure determinism for the device-dependent torch.Generator
													_UpperCAmelCase  :			int	       =		self.get_dummy_components()
													_UpperCAmelCase  :			Dict	       =		StableDiffusionImageVariationPipeline(**_lowerCamelCase			)
													_UpperCAmelCase  :			int	       =		sd_pipe.to(_lowerCamelCase			)
													sd_pipe.set_progress_bar_config(disable=_lowerCamelCase			)

													_UpperCAmelCase  :			Optional[int]	       =		self.get_dummy_inputs(_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		2 * [inputs["image"]]
													_UpperCAmelCase  :			List[str]	       =		sd_pipe(**_lowerCamelCase			)

													_UpperCAmelCase  :			Tuple	       =		output.images

													_UpperCAmelCase  :			int	       =		image[-1, -3:, -3:, -1]

													assert image.shape == (2, 64, 64, 3)
													_UpperCAmelCase  :			List[Any]	       =		np.array([0.68_92, 0.56_37, 0.58_36, 0.57_71, 0.62_54, 0.64_09, 0.55_80, 0.55_69, 0.52_89]			)

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






						def 						a__ (					self		:			Tuple			):
													super().test_inference_batch_single_identical(expected_max_diff=3E-3			)







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

						def 						a__ (					self		:			Tuple			):
													super().tearDown()
													gc.collect()
													torch.cuda.empty_cache()

						def 						a__ (					self		:			int	,						_lowerCamelCase		:			Union[str, Any]	,						_lowerCamelCase		:			Optional[Any]="cpu"	,						_lowerCamelCase		:			Union[str, Any]=torch.floataa	,						_lowerCamelCase		:			Tuple=0			):
													_UpperCAmelCase  :			int	       =		torch.Generator(device=_lowerCamelCase			).manual_seed(_lowerCamelCase			)
													_UpperCAmelCase  :			int	       =		load_image(
													    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
													    "/stable_diffusion_imgvar/input_image_vermeer.png"			)
													_UpperCAmelCase  :			str	       =		np.random.RandomState(_lowerCamelCase			).standard_normal((1, 4, 64, 64)			)
													_UpperCAmelCase  :			str	       =		torch.from_numpy(_lowerCamelCase			).to(device=_lowerCamelCase	,						dtype=_lowerCamelCase			)
													_UpperCAmelCase  :			Dict	       =		{
													    "image": init_image,
													    "latents": latents,
													    "generator": generator,
													    "num_inference_steps": 3,
													    "guidance_scale": 7.5,
													    "output_type": "numpy",
													}
													return inputs

						def 						a__ (					self		:			Optional[Any]			):
													_UpperCAmelCase  :			int	       =		StableDiffusionImageVariationPipeline.from_pretrained(
													    "lambdalabs/sd-image-variations-diffusers"	,						safety_checker=_lowerCamelCase			)
													_UpperCAmelCase  :			Any	       =		sd_pipe.to(_lowerCamelCase			)
													sd_pipe.set_progress_bar_config(disable=_lowerCamelCase			)

													_UpperCAmelCase  :			Any	       =		self.get_inputs(_lowerCamelCase			)
													_UpperCAmelCase  :			Dict	       =		sd_pipe(**_lowerCamelCase			).images
													_UpperCAmelCase  :			int	       =		image[0, -3:, -3:, -1].flatten()

													assert image.shape == (1, 5_12, 5_12, 3)
													_UpperCAmelCase  :			Dict	       =		np.array([0.8_44_91, 0.9_07_89, 0.7_57_08, 0.7_87_34, 0.8_34_85, 0.7_00_99, 0.6_69_38, 0.6_87_27, 0.6_13_79]			)
													assert np.abs(image_slice - expected_slice			).max() < 6E-3

						def 						a__ (					self		:			str			):
													_UpperCAmelCase  :			List[Any]	       =		0

													def callback_fn(_lowerCamelCase		:			int	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			torch.FloatTensor			) -> None:
																				_UpperCAmelCase  :			Optional[Any]	       =		True
																				nonlocal number_of_steps
																				number_of_steps += 1
																				if step == 1:
																											_UpperCAmelCase  :			int	       =		latents.detach().cpu().numpy()
																											assert latents.shape == (1, 4, 64, 64)
																											_UpperCAmelCase  :			Union[str, Any]	       =		latents[0, -3:, -3:, -1]
																											_UpperCAmelCase  :			str	       =		np.array(
																											    [-0.16_21, 0.28_37, -0.79_79, -0.12_21, -1.30_57, 0.76_81, -2.11_91, 0.04_64, 1.63_09]			)

																											assert np.abs(latents_slice.flatten() - expected_slice			).max() < 5E-2
																				elif step == 2:
																											_UpperCAmelCase  :			List[str]	       =		latents.detach().cpu().numpy()
																											assert latents.shape == (1, 4, 64, 64)
																											_UpperCAmelCase  :			Tuple	       =		latents[0, -3:, -3:, -1]
																											_UpperCAmelCase  :			str	       =		np.array([0.62_99, 1.75_00, 1.19_92, -2.15_82, -1.89_94, 0.73_34, -0.70_90, 1.01_37, 1.52_73]			)

																											assert np.abs(latents_slice.flatten() - expected_slice			).max() < 5E-2

													_UpperCAmelCase  :			List[Any]	       =		False

													_UpperCAmelCase  :			int	       =		StableDiffusionImageVariationPipeline.from_pretrained(
													    "fusing/sd-image-variations-diffusers"	,						safety_checker=_lowerCamelCase	,						torch_dtype=torch.floataa	,						)
													pipe.to(_lowerCamelCase			)
													pipe.set_progress_bar_config(disable=_lowerCamelCase			)
													pipe.enable_attention_slicing()

													_UpperCAmelCase  :			Optional[Any]	       =		self.get_inputs(_lowerCamelCase	,						dtype=torch.floataa			)
													pipe(**_lowerCamelCase	,						callback=_lowerCamelCase	,						callback_steps=1			)
													assert callback_fn.has_been_called
													assert number_of_steps == inputs["num_inference_steps"]






						def 						a__ (					self		:			str			):
													torch.cuda.empty_cache()
													torch.cuda.reset_max_memory_allocated()
													torch.cuda.reset_peak_memory_stats()

													_UpperCAmelCase  :			int	       =		"fusing/sd-image-variations-diffusers"
													_UpperCAmelCase  :			List[str]	       =		StableDiffusionImageVariationPipeline.from_pretrained(
													    _lowerCamelCase	,						safety_checker=_lowerCamelCase	,						torch_dtype=torch.floataa			)
													_UpperCAmelCase  :			str	       =		pipe.to(_lowerCamelCase			)
													pipe.set_progress_bar_config(disable=_lowerCamelCase			)
													pipe.enable_attention_slicing(1			)
													pipe.enable_sequential_cpu_offload()

													_UpperCAmelCase  :			Any	       =		self.get_inputs(_lowerCamelCase	,						dtype=torch.floataa			)
													_UpperCAmelCase  :			Any	       =		pipe(**_lowerCamelCase			)

													_UpperCAmelCase  :			str	       =		torch.cuda.max_memory_allocated()
													# make sure that less than 2.6 GB is allocated
													assert mem_bytes < 2.6 * 10**9







@nightly
@require_torch_gpu
class    _UpperCamelCase(   unittest.TestCase				):

						def 						a__ (					self		:			str			):
													super().tearDown()
													gc.collect()
													torch.cuda.empty_cache()

						def 						a__ (					self		:			Dict	,						_lowerCamelCase		:			List[Any]	,						_lowerCamelCase		:			int="cpu"	,						_lowerCamelCase		:			Union[str, Any]=torch.floataa	,						_lowerCamelCase		:			List[Any]=0			):
													_UpperCAmelCase  :			Union[str, Any]	       =		torch.Generator(device=_lowerCamelCase			).manual_seed(_lowerCamelCase			)
													_UpperCAmelCase  :			Tuple	       =		load_image(
													    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
													    "/stable_diffusion_imgvar/input_image_vermeer.png"			)
													_UpperCAmelCase  :			str	       =		np.random.RandomState(_lowerCamelCase			).standard_normal((1, 4, 64, 64)			)
													_UpperCAmelCase  :			List[str]	       =		torch.from_numpy(_lowerCamelCase			).to(device=_lowerCamelCase	,						dtype=_lowerCamelCase			)
													_UpperCAmelCase  :			Any	       =		{
													    "image": init_image,
													    "latents": latents,
													    "generator": generator,
													    "num_inference_steps": 50,
													    "guidance_scale": 7.5,
													    "output_type": "numpy",
													}
													return inputs

						def 						a__ (					self		:			Union[str, Any]			):
													_UpperCAmelCase  :			int	       =		StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"			)
													sd_pipe.to(_lowerCamelCase			)
													sd_pipe.set_progress_bar_config(disable=_lowerCamelCase			)

													_UpperCAmelCase  :			Tuple	       =		self.get_inputs(_lowerCamelCase			)
													_UpperCAmelCase  :			int	       =		sd_pipe(**_lowerCamelCase			).images[0]

													_UpperCAmelCase  :			str	       =		load_numpy(
													    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
													    "/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy"			)
													_UpperCAmelCase  :			List[Any]	       =		np.abs(expected_image - image			).max()
													assert max_diff < 1E-3






						def 						a__ (					self		:			Optional[int]			):
													_UpperCAmelCase  :			Tuple	       =		StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"			)
													_UpperCAmelCase  :			Dict	       =		DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config			)
													sd_pipe.to(_lowerCamelCase			)
													sd_pipe.set_progress_bar_config(disable=_lowerCamelCase			)

													_UpperCAmelCase  :			Optional[int]	       =		self.get_inputs(_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		25
													_UpperCAmelCase  :			Optional[int]	       =		sd_pipe(**_lowerCamelCase			).images[0]

													_UpperCAmelCase  :			Union[str, Any]	       =		load_numpy(
													    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
													    "/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy"			)
													_UpperCAmelCase  :			Dict	       =		np.abs(expected_image - image			).max()
													assert max_diff < 1E-3






from __future__ import annotations

from string import ascii_letters




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE = None  )   ->      str:

							"""simple docstring"""






							_UpperCAmelCase  :			List[Any]	       =		alphabet or ascii_letters

							# The final result string
							_UpperCAmelCase  :			Optional[Any]	       =		""

							for character in input_string:
														if character not in alpha:
																					# Append without encryption if character is not in the alphabet
																					result += character
														else:
																					# Get the index of the new key and make sure it isn't too large
																					_UpperCAmelCase  :			Any	       =		(alpha.index(_SCREAMING_SNAKE_CASE  ) + key) % len(_SCREAMING_SNAKE_CASE  )

																					# Append the encoded character to the alphabet
																					result += alpha[new_key]

							return result




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE = None  )   ->      str:

							"""simple docstring"""






							key *= -1

							return encrypt(_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE = None  )   ->      dict[int, str]:

							"""simple docstring"""






							_UpperCAmelCase  :			Any	       =		alphabet or ascii_letters

							# To store data on all the combinations
							_UpperCAmelCase  :			Dict	       =		{}

							# Cycle through each combination
							for key in range(1      ,	len(_SCREAMING_SNAKE_CASE  ) + 1  ):
														# Decrypt the message and store the result in the data
														_UpperCAmelCase  :			Union[str, Any]	       =		decrypt(_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )

							return brute_force_data


if __name__ == "__main__":
				while True:
								print(f'''\n{'-' * 1_0}\n Menu\n{'-' * 1_0}''')
								print(*['1.Encrypt', '2.Decrypt', '3.BruteForce', '4.Quit'], sep='\n')

								# get user input
								__lowerCamelCase								=					input('\nWhat would you like to do?: ').strip() or '4'

								# run functions based on what the user chose
								if choice not in ("1", "2", "3", "4"):
												print('Invalid choice, please enter a valid choice')
								elif choice == "1":
												__lowerCamelCase								=					input('Please enter the string to be encrypted: ')
												__lowerCamelCase								=					int(input('Please enter off-set: ').strip())

												print(encrypt(input_string, key))
								elif choice == "2":
												__lowerCamelCase								=					input('Please enter the string to be decrypted: ')
												__lowerCamelCase								=					int(input('Please enter off-set: ').strip())

												print(decrypt(input_string, key))
								elif choice == "3":
												__lowerCamelCase								=					input('Please enter the string to be decrypted: ')
												__lowerCamelCase								=					brute_force(input_string)

												for key, value in brute_force_data.items():
																print(f'''Key: {key} | Message: {value}''')

								elif choice == "4":
												print('Goodbye.')
												break




