


"""simple docstring"""







import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPanoramaPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






@skip_mps
class 				__A    (	_SCREAMING_SNAKE_CASE,   _SCREAMING_SNAKE_CASE,   unittest.TestCase      ):






 """simple docstring"""




 __lowerCAmelCase			 =  StableDiffusionPanoramaPipeline
 __lowerCAmelCase			 =  TEXT_TO_IMAGE_PARAMS
 __lowerCAmelCase			 =  TEXT_TO_IMAGE_BATCH_PARAMS
 __lowerCAmelCase			 =  TEXT_TO_IMAGE_IMAGE_PARAMS
 __lowerCAmelCase			 =  TEXT_TO_IMAGE_IMAGE_PARAMS


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					str:
     torch.manual_seed(0						)
     a					=UNetaDConditionModel(
         block_out_channels=(32, 64)			,   layers_per_block=1			,   sample_size=32			,   in_channels=4			,   out_channels=4			,   down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')			,   up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')			,   cross_attention_dim=32			,   )
     a					=DDIMScheduler()
     torch.manual_seed(0						)
     a					=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						)
     a					=CLIPTextConfig(
         bos_token_id=0			,   eos_token_id=2			,   hidden_size=32			,   intermediate_size=37			,   layer_norm_eps=1E-0_5			,   num_attention_heads=4			,   num_hidden_layers=5			,   pad_token_id=1			,   vocab_size=1000			,   )
     a					=CLIPTextModel(__A						)
     a					=CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''						)

     a					={
         '''unet''': unet,
         '''scheduler''': scheduler,
         '''vae''': vae,
         '''text_encoder''': text_encoder,
         '''tokenizer''': tokenizer,
         '''safety_checker''': None,
         '''feature_extractor''': None,
     }
     return components


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A=0						) ->					List[Any]:
     a					=torch.manual_seed(__A						)
     a					={
         '''prompt''': '''a photo of the dolomites''',
         '''generator''': generator,
         # Setting height and width to None to prevent OOMs on CPU.
         '''height''': None,
         '''width''': None,
         '''num_inference_steps''': 1,
         '''guidance_scale''': 6.0,
         '''output_type''': '''numpy''',
     }
     return inputs


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[Any]:
     a					='''cpu'''  # ensure determinism for the device-dependent torch.Generator
     a					=self.get_dummy_components()
     a					=StableDiffusionPanoramaPipeline(**__A						)
     a					=sd_pipe.to(__A						)
     sd_pipe.set_progress_bar_config(disable=__A						)

     a					=self.get_dummy_inputs(__A						)
     a					=sd_pipe(**__A						).images
     a					=image[0, -3:, -3:, -1]
     assert image.shape == (1, 64, 64, 3)

     a					=np.array([0.6_186, 0.5_374, 0.4_915, 0.4_135, 0.4_114, 0.4_563, 0.5_128, 0.4_977, 0.4_757]						)

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


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					int:
     super().test_inference_batch_consistent(batch_sizes=[1, 2]						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[Any]:
     super().test_inference_batch_single_identical(batch_size=2			,   expected_max_diff=3.2_5E-3						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[str]:
     a					='''cpu'''  # ensure determinism for the device-dependent torch.Generator
     a					=self.get_dummy_components()
     a					=StableDiffusionPanoramaPipeline(**__A						)
     a					=sd_pipe.to(__A						)
     sd_pipe.set_progress_bar_config(disable=__A						)

     a					=self.get_dummy_inputs(__A						)
     a					='''french fries'''
     a					=sd_pipe(**__A			,   negative_prompt=__A						)
     a					=output.images
     a					=image[0, -3:, -3:, -1]

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

     a					=np.array([0.6_187, 0.5_375, 0.4_915, 0.4_136, 0.4_114, 0.4_563, 0.5_128, 0.4_976, 0.4_757]						)

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


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					str:
     a					='''cpu'''  # ensure determinism for the device-dependent torch.Generator
     a					=self.get_dummy_components()
     a					=StableDiffusionPanoramaPipeline(**__A						)
     a					=sd_pipe.to(__A						)
     sd_pipe.set_progress_bar_config(disable=__A						)

     a					=self.get_dummy_inputs(__A						)
     a					=sd_pipe(**__A			,   view_batch_size=2						)
     a					=output.images
     a					=image[0, -3:, -3:, -1]

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

     a					=np.array([0.6_187, 0.5_375, 0.4_915, 0.4_136, 0.4_114, 0.4_563, 0.5_128, 0.4_976, 0.4_757]						)

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


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Optional[int]:
     a					='''cpu'''  # ensure determinism for the device-dependent torch.Generator
     a					=self.get_dummy_components()
     a					=EulerAncestralDiscreteScheduler(
         beta_start=0.00_085			,   beta_end=0.012			,   beta_schedule='''scaled_linear'''						)
     a					=StableDiffusionPanoramaPipeline(**__A						)
     a					=sd_pipe.to(__A						)
     sd_pipe.set_progress_bar_config(disable=__A						)

     a					=self.get_dummy_inputs(__A						)
     a					=sd_pipe(**__A						).images
     a					=image[0, -3:, -3:, -1]

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

     a					=np.array([0.4_024, 0.6_510, 0.4_901, 0.5_378, 0.5_813, 0.5_622, 0.4_795, 0.4_467, 0.4_952]						)

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





 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Tuple:
     a					='''cpu'''  # ensure determinism for the device-dependent torch.Generator
     a					=self.get_dummy_components()
     a					=PNDMScheduler(
         beta_start=0.00_085			,   beta_end=0.012			,   beta_schedule='''scaled_linear'''			,   skip_prk_steps=__A						)
     a					=StableDiffusionPanoramaPipeline(**__A						)
     a					=sd_pipe.to(__A						)
     sd_pipe.set_progress_bar_config(disable=__A						)

     a					=self.get_dummy_inputs(__A						)
     a					=sd_pipe(**__A						).images
     a					=image[0, -3:, -3:, -1]

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

     a					=np.array([0.6_391, 0.6_291, 0.4_861, 0.5_134, 0.5_552, 0.4_578, 0.5_032, 0.5_023, 0.4_539]						)

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






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






 """simple docstring"""




 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Any:
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A=0						) ->					Tuple:
     a					=torch.manual_seed(__A						)
     a					={
         '''prompt''': '''a photo of the dolomites''',
         '''generator''': generator,
         '''num_inference_steps''': 3,
         '''guidance_scale''': 7.5,
         '''output_type''': '''numpy''',
     }
     return inputs


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					str:
     a					='''stabilityai/stable-diffusion-2-base'''
     a					=DDIMScheduler.from_pretrained(__A			,   subfolder='''scheduler'''						)
     a					=StableDiffusionPanoramaPipeline.from_pretrained(__A			,   scheduler=__A			,   safety_checker=__A						)
     pipe.to(__A						)
     pipe.set_progress_bar_config(disable=__A						)
     pipe.enable_attention_slicing()

     a					=self.get_inputs()
     a					=pipe(**__A						).images
     a					=image[0, -3:, -3:, -1].flatten()

     assert image.shape == (1, 512, 2048, 3)

     a					=np.array(
         [
             0.36_968_392,
             0.27_025_372,
             0.32_446_766,
             0.28_379_387,
             0.36_363_274,
             0.30_733_347,
             0.27_100_027,
             0.27_054_125,
             0.25_536_096,
         ]						)

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


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					int:
     a					=StableDiffusionPanoramaPipeline.from_pretrained(
         '''stabilityai/stable-diffusion-2-base'''			,   safety_checker=__A						)
     a					=LMSDiscreteScheduler.from_config(pipe.scheduler.config						)
     pipe.to(__A						)
     pipe.set_progress_bar_config(disable=__A						)
     pipe.enable_attention_slicing()

     a					=self.get_inputs()
     a					=pipe(**__A						).images
     a					=image[0, -3:, -3:, -1].flatten()

     assert image.shape == (1, 512, 2048, 3)

     a					=np.array(
         [
             [
                 0.0,
                 0.0,
                 0.0,
                 0.0,
                 0.0,
                 0.0,
                 0.0,
                 0.0,
                 0.0,
             ]
         ]						)

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


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					int:
     a					=0

     def callback_fn(__A			,   __A			,   __A						) -> None:
         a					=True
         nonlocal number_of_steps
         number_of_steps += 1
         if step == 1:
             a					=latents.detach().cpu().numpy()
             assert latents.shape == (1, 4, 64, 256)
             a					=latents[0, -3:, -3:, -1]

             a					=np.array(
                 [
                     0.18_681_869,
                     0.33_907_816,
                     0.5_361_276,
                     0.14_432_865,
                     -0.02_856_611,
                     -0.73_941_123,
                     0.23_397_987,
                     0.47_322_682,
                     -0.37_823_164,
                 ]						)
             assert np.abs(latents_slice.flatten() - expected_slice						).max() < 5E-2
         elif step == 2:
             a					=latents.detach().cpu().numpy()
             assert latents.shape == (1, 4, 64, 256)
             a					=latents[0, -3:, -3:, -1]

             a					=np.array(
                 [
                     0.18_539_645,
                     0.33_987_248,
                     0.5_378_559,
                     0.14_437_142,
                     -0.02_455_261,
                     -0.7_338_317,
                     0.23_990_755,
                     0.47_356_272,
                     -0.3_786_505,
                 ]						)

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

     a					=False

     a					='''stabilityai/stable-diffusion-2-base'''
     a					=DDIMScheduler.from_pretrained(__A			,   subfolder='''scheduler'''						)
     a					=StableDiffusionPanoramaPipeline.from_pretrained(__A			,   scheduler=__A			,   safety_checker=__A						)
     a					=pipe.to(__A						)
     pipe.set_progress_bar_config(disable=__A						)
     pipe.enable_attention_slicing()

     a					=self.get_inputs()
     pipe(**__A			,   callback=__A			,   callback_steps=1						)
     assert callback_fn.has_been_called
     assert number_of_steps == 3





 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Union[str, Any]:
     torch.cuda.empty_cache()
     torch.cuda.reset_max_memory_allocated()
     torch.cuda.reset_peak_memory_stats()

     a					='''stabilityai/stable-diffusion-2-base'''
     a					=DDIMScheduler.from_pretrained(__A			,   subfolder='''scheduler'''						)
     a					=StableDiffusionPanoramaPipeline.from_pretrained(__A			,   scheduler=__A			,   safety_checker=__A						)
     a					=pipe.to(__A						)
     pipe.set_progress_bar_config(disable=__A						)
     pipe.enable_attention_slicing(1						)
     pipe.enable_sequential_cpu_offload()

     a					=self.get_inputs()
     a					=pipe(**__A						)

     a					=torch.cuda.max_memory_allocated()
     # make sure that less than 5.2 GB is allocated
     assert mem_bytes < 5.5 * 10**9



"""simple docstring"""







from __future__ import annotations






class 				__A    :






 """simple docstring"""




 def __init__(		self			,   __A = 0						) ->					Dict:
     a					=key


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A						) ->					list[str]:
     assert isinstance(__A			,   __A						) and isinstance(__A			,   __A						)

     a					=key or self.__key or 1

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

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


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A						) ->					list[str]:
     assert isinstance(__A			,   __A						) and isinstance(__A			,   __A						)

     a					=key or self.__key or 1

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

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


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A = 0						) ->					str:
     assert isinstance(__A			,   __A						) and isinstance(__A			,   __A						)

     a					=key or self.__key or 1

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

     # This will be returned
     a					=''''''

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

     return ans


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A = 0						) ->					str:
     assert isinstance(__A			,   __A						) and isinstance(__A			,   __A						)

     a					=key or self.__key or 1

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

     # This will be returned
     a					=''''''

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

     return ans


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A = 0						) ->					bool:
     assert isinstance(__A			,   __A						) and isinstance(__A			,   __A						)

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

     except OSError:
         return False

     return True





 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A						) ->					bool:
     assert isinstance(__A			,   __A						) and isinstance(__A			,   __A						)

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

     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")