






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        __snake_case(					__UpperCamelCase     ,       __UpperCamelCase     ,       __UpperCamelCase     ,       unittest.TestCase      ):
 lowercase :       Union[str, Any]		      =		StableDiffusionImageVariationPipeline
 lowercase :       List[Any]		      =		IMAGE_VARIATION_PARAMS
 lowercase :       Any		      =		IMAGE_VARIATION_BATCH_PARAMS
 lowercase :       Optional[int]		      =		frozenset([]      )
 # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 lowercase :       Any		      =		frozenset([]      )



 def 						UpperCAmelCase    (	self   :   Dict	):

    """simple docstring"""






    torch.manual_seed(0	)
    lowercase__						:	Dict         = 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[str]         = PNDMScheduler(skip_prk_steps=UpperCAmelCase__	)
    torch.manual_seed(0	)
    lowercase__						:	Optional[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	)
    lowercase__						:	Optional[int]         = 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__						:	str         = CLIPVisionModelWithProjection(UpperCAmelCase__	)
    lowercase__						:	int         = CLIPImageProcessor(crop_size=32 ,		size=32	)

    lowercase__						:	List[Any]         = {
        """unet""": unet,
        """scheduler""": scheduler,
        """vae""": vae,
        """image_encoder""": image_encoder,
        """feature_extractor""": feature_extractor,
        """safety_checker""": None,
    }
    return components



 def 						UpperCAmelCase    (	self   :   List[Any] ,		UpperCAmelCase__   :   List[Any] ,		UpperCAmelCase__   :   Union[str, Any]=0	):

    """simple docstring"""






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



 def 						UpperCAmelCase    (	self   :   str	):

    """simple docstring"""






    lowercase__						:	Tuple         = """cpu"""  # ensure determinism for the device-dependent torch.Generator
    lowercase__						:	Dict         = self.get_dummy_components()
    lowercase__						:	int         = StableDiffusionImageVariationPipeline(**UpperCAmelCase__	)
    lowercase__						:	List[str]         = sd_pipe.to(UpperCAmelCase__	)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__	)

    lowercase__						:	Optional[int]         = self.get_dummy_inputs(UpperCAmelCase__	)
    lowercase__						:	Dict         = sd_pipe(**UpperCAmelCase__	).images
    lowercase__						:	Optional[int]         = image[0, -3:, -3:, -1]

    assert image.shape == (1, 64, 64, 3)
    lowercase__						:	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 						UpperCAmelCase    (	self   :   Optional[int]	):

    """simple docstring"""






    lowercase__						:	Union[str, Any]         = """cpu"""  # ensure determinism for the device-dependent torch.Generator
    lowercase__						:	Optional[int]         = self.get_dummy_components()
    lowercase__						:	Optional[Any]         = StableDiffusionImageVariationPipeline(**UpperCAmelCase__	)
    lowercase__						:	Any         = sd_pipe.to(UpperCAmelCase__	)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__	)

    lowercase__						:	Optional[Any]         = self.get_dummy_inputs(UpperCAmelCase__	)
    lowercase__						:	Any         = 2 * [inputs["""image"""]]
    lowercase__						:	str         = sd_pipe(**UpperCAmelCase__	)

    lowercase__						:	int         = output.images

    lowercase__						:	List[str]         = image[-1, -3:, -3:, -1]

    assert image.shape == (2, 64, 64, 3)
    lowercase__						:	Any         = 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 						UpperCAmelCase    (	self   :   Optional[int]	):

    """simple docstring"""






    super().test_inference_batch_single_identical(expected_max_diff=3e-3	)
@slow
@require_torch_gpu
class        __snake_case(					unittest.TestCase      ):



 def 						UpperCAmelCase    (	self   :   Any	):

    """simple docstring"""






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



 def 						UpperCAmelCase    (	self   :   List[Any] ,		UpperCAmelCase__   :   Optional[int] ,		UpperCAmelCase__   :   int="cpu" ,		UpperCAmelCase__   :   List[str]=torch.floataa ,		UpperCAmelCase__   :   List[str]=0	):

    """simple docstring"""






    lowercase__						:	str         = torch.Generator(device=UpperCAmelCase__	).manual_seed(UpperCAmelCase__	)
    lowercase__						:	List[str]         = load_image(
        """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
        """/stable_diffusion_imgvar/input_image_vermeer.png"""	)
    lowercase__						:	Dict         = np.random.RandomState(UpperCAmelCase__	).standard_normal((1, 4, 64, 64)	)
    lowercase__						:	int         = torch.from_numpy(UpperCAmelCase__	).to(device=UpperCAmelCase__ ,		dtype=UpperCAmelCase__	)
    lowercase__						:	str         = {
        """image""": init_image,
        """latents""": latents,
        """generator""": generator,
        """num_inference_steps""": 3,
        """guidance_scale""": 7.5,
        """output_type""": """numpy""",
    }
    return inputs



 def 						UpperCAmelCase    (	self   :   List[Any]	):

    """simple docstring"""






    lowercase__						:	Dict         = StableDiffusionImageVariationPipeline.from_pretrained(
        """lambdalabs/sd-image-variations-diffusers""" ,		safety_checker=UpperCAmelCase__	)
    lowercase__						:	str         = sd_pipe.to(UpperCAmelCase__	)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__	)

    lowercase__						:	Optional[Any]         = self.get_inputs(UpperCAmelCase__	)
    lowercase__						:	Tuple         = sd_pipe(**UpperCAmelCase__	).images
    lowercase__						:	List[str]         = image[0, -3:, -3:, -1].flatten()

    assert image.shape == (1, 512, 512, 3)
    lowercase__						:	Tuple         = 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 						UpperCAmelCase    (	self   :   List[str]	):

    """simple docstring"""






    lowercase__						:	Optional[int]         = 0

    def callback_fn(UpperCAmelCase__   :   int ,		UpperCAmelCase__   :   int ,		UpperCAmelCase__   :   torch.FloatTensor	) -> None:
       lowercase__						:	Union[str, Any]         = True
       nonlocal number_of_steps
       number_of_steps += 1
       if step == 1:
          lowercase__						:	List[Any]         = latents.detach().cpu().numpy()
          assert latents.shape == (1, 4, 64, 64)
          lowercase__						:	Optional[Any]         = latents[0, -3:, -3:, -1]
          lowercase__						:	List[Any]         = 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__						:	Tuple         = latents.detach().cpu().numpy()
          assert latents.shape == (1, 4, 64, 64)
          lowercase__						:	int         = latents[0, -3:, -3:, -1]
          lowercase__						:	Optional[Any]         = 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__						:	List[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__						:	Tuple         = 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 						UpperCAmelCase    (	self   :   str	):

    """simple docstring"""






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

    lowercase__						:	Dict         = """fusing/sd-image-variations-diffusers"""
    lowercase__						:	str         = StableDiffusionImageVariationPipeline.from_pretrained(
        UpperCAmelCase__ ,		safety_checker=UpperCAmelCase__ ,		torch_dtype=torch.floataa	)
    lowercase__						:	str         = pipe.to(UpperCAmelCase__	)
    pipe.set_progress_bar_config(disable=UpperCAmelCase__	)
    pipe.enable_attention_slicing(1	)
    pipe.enable_sequential_cpu_offload()

    lowercase__						:	List[str]         = self.get_inputs(UpperCAmelCase__ ,		dtype=torch.floataa	)
    lowercase__						:	str         = pipe(**UpperCAmelCase__	)

    lowercase__						:	Union[str, Any]         = 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        __snake_case(					unittest.TestCase      ):



 def 						UpperCAmelCase    (	self   :   Any	):

    """simple docstring"""






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



 def 						UpperCAmelCase    (	self   :   Optional[int] ,		UpperCAmelCase__   :   Dict ,		UpperCAmelCase__   :   Any="cpu" ,		UpperCAmelCase__   :   Tuple=torch.floataa ,		UpperCAmelCase__   :   Tuple=0	):

    """simple docstring"""






    lowercase__						:	Union[str, Any]         = torch.Generator(device=UpperCAmelCase__	).manual_seed(UpperCAmelCase__	)
    lowercase__						:	Optional[int]         = load_image(
        """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
        """/stable_diffusion_imgvar/input_image_vermeer.png"""	)
    lowercase__						:	List[str]         = 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""": 50,
        """guidance_scale""": 7.5,
        """output_type""": """numpy""",
    }
    return inputs



 def 						UpperCAmelCase    (	self   :   List[Any]	):

    """simple docstring"""






    lowercase__						:	Optional[int]         = StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers"""	)
    sd_pipe.to(UpperCAmelCase__	)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__	)

    lowercase__						:	Dict         = self.get_inputs(UpperCAmelCase__	)
    lowercase__						:	Optional[Any]         = sd_pipe(**UpperCAmelCase__	).images[0]

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







 def 						UpperCAmelCase    (	self   :   Optional[Any]	):

    """simple docstring"""






    lowercase__						:	List[str]         = StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers"""	)
    lowercase__						:	Optional[int]         = DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config	)
    sd_pipe.to(UpperCAmelCase__	)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__	)

    lowercase__						:	Optional[Any]         = self.get_inputs(UpperCAmelCase__	)
    lowercase__						:	int         = 25
    lowercase__						:	Any         = sd_pipe(**UpperCAmelCase__	).images[0]

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










import os
import sys
import time

from . import transposition_cipher as trans_cipher
def       a__						(  ):
   lowercase__						:	Union[str, Any]         = """Prehistoric Men.txt"""
   lowercase__						:	Optional[Any]         = """Output.txt"""
   lowercase__						:	List[str]         = int(input("""Enter key: """					)					)
   lowercase__						:	Tuple         = input("""Encrypt/Decrypt [e/d]: """					)

   if not os.path.exists(A					):
      print(F'''File {input_file} does not exist. Quitting...'''					)
      sys.exit()
   if os.path.exists(A					):
      print(F'''Overwrite {output_file}? [y/n]'''					)
      lowercase__						:	List[Any]         = input("""> """					)
      if not response.lower().startswith("""y"""					):
         sys.exit()

   lowercase__						:	str         = time.time()
   if mode.lower().startswith("""e"""					):
      with open(A					) as f:
         lowercase__						:	Optional[int]         = f.read()
      lowercase__						:	List[str]         = trans_cipher.encrypt_message(A		,				A					)
   elif mode.lower().startswith("""d"""					):
      with open(A					) as f:
         lowercase__						:	int         = f.read()
      lowercase__						:	int         = trans_cipher.decrypt_message(A		,				A					)

   with open(A		,				"""w"""					) as output_obj:
      output_obj.write(A					)

   lowercase__						:	Optional[Any]         = round(time.time() - start_time		,				2					)
   print(("""Done (""", total_time, """seconds )""")					)


if __name__ == "__main__":
 main()


