



'''simple docstring'''






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 	_lowerCAmelCase     (    _lowerCAmelCase     ,		_lowerCAmelCase     ,		_lowerCAmelCase     ,		unittest.TestCase       ):






		'''simple docstring'''
		snake_case_							= StableDiffusionImageVariationPipeline
		snake_case_							= IMAGE_VARIATION_PARAMS
		snake_case_							= IMAGE_VARIATION_BATCH_PARAMS
		snake_case_							= frozenset([]       )
		# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
		snake_case_							= frozenset([]       )
		def 						__lowercase      (						self							:       Optional[int]		)      ->			List[Any]:







									'''simple docstring'''





									torch.manual_seed(0		)
									_lowercase							:		Optional[int]				=   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      ,		)
									_lowercase							:		List[Any]				=   PNDMScheduler(skip_prk_steps=_UpperCAmelCase		)
									torch.manual_seed(0		)
									_lowercase							:		Union[str, Any]				=   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		)
									_lowercase							:		Tuple				=   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      ,		)
									_lowercase							:		int				=   CLIPVisionModelWithProjection(_UpperCAmelCase		)
									_lowercase							:		Tuple				=   CLIPImageProcessor(crop_size=32      ,		size=32		)

									_lowercase							:		Tuple				=   {
									    'unet': unet,
									    'scheduler': scheduler,
									    'vae': vae,
									    'image_encoder': image_encoder,
									    'feature_extractor': feature_extractor,
									    'safety_checker': None,
									}
									return components
		def 						__lowercase      (						self							:       Any      ,		UpperCamelCase_							:       List[str]      ,		UpperCamelCase_							:       Tuple=0		)      ->			List[Any]:







									'''simple docstring'''





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







									'''simple docstring'''





									_lowercase							:		Optional[Any]				=   'cpu'  # ensure determinism for the device-dependent torch.Generator
									_lowercase							:		Tuple				=   self.get_dummy_components()
									_lowercase							:		Optional[int]				=   StableDiffusionImageVariationPipeline(**_UpperCAmelCase		)
									_lowercase							:		Tuple				=   sd_pipe.to(_UpperCAmelCase		)
									sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

									_lowercase							:		List[Any]				=   self.get_dummy_inputs(_UpperCAmelCase		)
									_lowercase							:		Tuple				=   sd_pipe(**_UpperCAmelCase		).images
									_lowercase							:		Optional[int]				=   image[0, -3:, -3:, -1]

									assert image.shape == (1, 64, 64, 3)
									_lowercase							:		List[str]				=   np.array([0.5_239, 0.5_723, 0.4_796, 0.5_049, 0.5_550, 0.4_685, 0.5_329, 0.4_891, 0.4_921]		)

									assert np.abs(image_slice.flatten() - expected_slice		).max() < 1e-3
		def 						__lowercase      (						self							:       List[str]		)      ->			Optional[Any]:







									'''simple docstring'''





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

									_lowercase							:		Optional[int]				=   self.get_dummy_inputs(_UpperCAmelCase		)
									_lowercase							:		Dict				=   2 * [inputs['image']]
									_lowercase							:		Tuple				=   sd_pipe(**_UpperCAmelCase		)

									_lowercase							:		List[Any]				=   output.images

									_lowercase							:		Tuple				=   image[-1, -3:, -3:, -1]

									assert image.shape == (2, 64, 64, 3)
									_lowercase							:		int				=   np.array([0.6_892, 0.5_637, 0.5_836, 0.5_771, 0.6_254, 0.6_409, 0.5_580, 0.5_569, 0.5_289]		)

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





		def 						__lowercase      (						self							:       Tuple		)      ->			Optional[int]:







									'''simple docstring'''





									super().test_inference_batch_single_identical(expected_max_diff=3e-3		)




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






		'''simple docstring'''
		def 						__lowercase      (						self							:       str		)      ->			Any:







									'''simple docstring'''





									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()
		def 						__lowercase      (						self							:       List[Any]      ,		UpperCamelCase_							:       Union[str, Any]      ,		UpperCamelCase_							:       int="cpu"      ,		UpperCamelCase_							:       Union[str, Any]=torch.floataa      ,		UpperCamelCase_							:       Any=0		)      ->			List[Any]:







									'''simple docstring'''





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







									'''simple docstring'''





									_lowercase							:		List[str]				=   StableDiffusionImageVariationPipeline.from_pretrained(
									    '''lambdalabs/sd-image-variations-diffusers'''      ,		safety_checker=_UpperCAmelCase		)
									_lowercase							:		List[str]				=   sd_pipe.to(_UpperCAmelCase		)
									sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

									_lowercase							:		str				=   self.get_inputs(_UpperCAmelCase		)
									_lowercase							:		Tuple				=   sd_pipe(**_UpperCAmelCase		).images
									_lowercase							:		Union[str, Any]				=   image[0, -3:, -3:, -1].flatten()

									assert image.shape == (1, 512, 512, 3)
									_lowercase							:		List[str]				=   np.array([0.84_491, 0.90_789, 0.75_708, 0.78_734, 0.83_485, 0.70_099, 0.66_938, 0.68_727, 0.61_379]		)
									assert np.abs(image_slice - expected_slice		).max() < 6e-3
		def 						__lowercase      (						self							:       Optional[int]		)      ->			str:







									'''simple docstring'''





									_lowercase							:		List[Any]				=   0

									def callback_fn(UpperCamelCase_							:       Tuple      ,		UpperCamelCase_							:       Union[str, Any]      ,		UpperCamelCase_							:       str		) -> None:
																_lowercase							:		Dict				=   True
																nonlocal number_of_steps
																number_of_steps += 1
																if step == 1:
																							_lowercase							:		Optional[Any]				=   latents.detach().cpu().numpy()
																							assert latents.shape == (1, 4, 64, 64)
																							_lowercase							:		int				=   latents[0, -3:, -3:, -1]
																							_lowercase							:		Tuple				=   np.array(
																							    [-0.1_621, 0.2_837, -0.7_979, -0.1_221, -1.3_057, 0.7_681, -2.1_191, 0.0_464, 1.6_309]		)

																							assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5e-2
																elif step == 2:
																							_lowercase							:		Dict				=   latents.detach().cpu().numpy()
																							assert latents.shape == (1, 4, 64, 64)
																							_lowercase							:		Any				=   latents[0, -3:, -3:, -1]
																							_lowercase							:		Optional[int]				=   np.array([0.6_299, 1.7_500, 1.1_992, -2.1_582, -1.8_994, 0.7_334, -0.7_090, 1.0_137, 1.5_273]		)

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

									_lowercase							:		Union[str, Any]				=   False

									_lowercase							:		Optional[int]				=   StableDiffusionImageVariationPipeline.from_pretrained(
									    '''fusing/sd-image-variations-diffusers'''      ,		safety_checker=_UpperCAmelCase      ,		torch_dtype=torch.floataa      ,		)
									pipe.to(_UpperCAmelCase		)
									pipe.set_progress_bar_config(disable=_UpperCAmelCase		)
									pipe.enable_attention_slicing()

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





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







									'''simple docstring'''





									torch.cuda.empty_cache()
									torch.cuda.reset_max_memory_allocated()
									torch.cuda.reset_peak_memory_stats()

									_lowercase							:		int				=   'fusing/sd-image-variations-diffusers'
									_lowercase							:		Tuple				=   StableDiffusionImageVariationPipeline.from_pretrained(
									    _UpperCAmelCase      ,		safety_checker=_UpperCAmelCase      ,		torch_dtype=torch.floataa		)
									_lowercase							:		Tuple				=   pipe.to(_UpperCAmelCase		)
									pipe.set_progress_bar_config(disable=_UpperCAmelCase		)
									pipe.enable_attention_slicing(1		)
									pipe.enable_sequential_cpu_offload()

									_lowercase							:		Optional[int]				=   self.get_inputs(_UpperCAmelCase      ,		dtype=torch.floataa		)
									_lowercase							:		Dict				=   pipe(**_UpperCAmelCase		)

									_lowercase							:		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 	_lowerCAmelCase     (    unittest.TestCase       ):






		'''simple docstring'''
		def 						__lowercase      (						self							:       List[Any]		)      ->			List[str]:







									'''simple docstring'''





									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()
		def 						__lowercase      (						self							:       Any      ,		UpperCamelCase_							:       Dict      ,		UpperCamelCase_							:       str="cpu"      ,		UpperCamelCase_							:       Optional[Any]=torch.floataa      ,		UpperCamelCase_							:       Dict=0		)      ->			Optional[int]:







									'''simple docstring'''





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







									'''simple docstring'''





									_lowercase							:		List[str]				=   StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''		)
									sd_pipe.to(_UpperCAmelCase		)
									sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

									_lowercase							:		List[str]				=   self.get_inputs(_UpperCAmelCase		)
									_lowercase							:		int				=   sd_pipe(**_UpperCAmelCase		).images[0]

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





		def 						__lowercase      (						self							:       Optional[int]		)      ->			Any:







									'''simple docstring'''





									_lowercase							:		Any				=   StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''		)
									_lowercase							:		Dict				=   DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config		)
									sd_pipe.to(_UpperCAmelCase		)
									sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

									_lowercase							:		Tuple				=   self.get_inputs(_UpperCAmelCase		)
									_lowercase							:		Tuple				=   25
									_lowercase							:		List[Any]				=   sd_pipe(**_UpperCAmelCase		).images[0]

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






from __future__ import annotations





def  _snake_case		(snake_case_					:     int  )							->		int | None:
  for r in range(1    ,       snake_case_  ):
    __UpperCAmelCase	:  List[str]		=     []
    for x in range(n - 1  ):
      __UpperCAmelCase	:  Union[str, Any]		=     pow(snake_case_    ,       snake_case_    ,       snake_case_  )
      if val in li:
        break
      li.append(snake_case_  )
    else:
      return r
  return None


if __name__ == "__main__":
      _SCREAMING_SNAKE_CASE							 =			int(input('''Enter a prime number q: '''))
      _SCREAMING_SNAKE_CASE							 =			find_primitive(q)
      if a is None:
            print(F'''Cannot find the primitive for the value: {a!r}''')
      else:
            _SCREAMING_SNAKE_CASE							 =			int(input('''Enter private key of A: '''))
            _SCREAMING_SNAKE_CASE							 =			pow(a, a_private, q)
            _SCREAMING_SNAKE_CASE							 =			int(input('''Enter private key of B: '''))
            _SCREAMING_SNAKE_CASE							 =			pow(a, b_private, q)

            _SCREAMING_SNAKE_CASE							 =			pow(b_public, a_private, q)
            _SCREAMING_SNAKE_CASE							 =			pow(a_public, b_private, q)

            print('''The key value generated by A is: ''', a_secret)
            print('''The key value generated by B is: ''', b_secret)





