





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__   (       __snake_case       ,   __snake_case       ,   unittest.TestCase			):
      _UpperCAmelCase	:Optional[Any]						    =   StableDiffusionPanoramaPipeline
      _UpperCAmelCase	:List[Any]						    =   TEXT_TO_IMAGE_PARAMS
      _UpperCAmelCase	:Optional[Any]						    =   TEXT_TO_IMAGE_BATCH_PARAMS
      _UpperCAmelCase	:Any						    =   TEXT_TO_IMAGE_IMAGE_PARAMS
      _UpperCAmelCase	:List[Any]						    =   TEXT_TO_IMAGE_IMAGE_PARAMS





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       torch.manual_seed(0   )
       UpperCamelCase	:					Dict       	=   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       ,							)
       UpperCamelCase	:					Optional[int]       	=   DDIMScheduler()
       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	:					List[str]       	=   CLIPTextConfig(
           bos_token_id=0       ,							eos_token_id=2       ,							hidden_size=32       ,							intermediate_size=37       ,							layer_norm_eps=1e-05       ,							num_attention_heads=4       ,							num_hidden_layers=5       ,							pad_token_id=1       ,							vocab_size=1000       ,							)
       UpperCamelCase	:					Any       	=   CLIPTextModel(A_   )
       UpperCamelCase	:					int       	=   CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"   )

       UpperCamelCase	:					Dict       	=   {
           "unet": unet,
           "scheduler": scheduler,
           "vae": vae,
           "text_encoder": text_encoder,
           "tokenizer": tokenizer,
           "safety_checker": None,
           "feature_extractor": None,
       }
       return components





      def        __UpperCamelCase( self       ,							A_       ,							A_=0   ):
       '''simple docstring'''




       UpperCamelCase	:					int       	=   torch.manual_seed(A_   )
       UpperCamelCase	:					List[str]       	=   {
           "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        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Union[str, Any]       	=   "cpu"  # ensure determinism for the device-dependent torch.Generator
       UpperCamelCase	:					Optional[Any]       	=   self.get_dummy_components()
       UpperCamelCase	:					Dict       	=   StableDiffusionPanoramaPipeline(**A_   )
       UpperCamelCase	:					Tuple       	=   sd_pipe.to(A_   )
       sd_pipe.set_progress_bar_config(disable=A_   )

       UpperCamelCase	:					Optional[int]       	=   self.get_dummy_inputs(A_   )
       UpperCamelCase	:					str       	=   sd_pipe(**A_   ).images
       UpperCamelCase	:					Optional[Any]       	=   image[0, -3:, -3:, -1]
       assert image.shape == (1, 64, 64, 3)

       UpperCamelCase	:					Optional[int]       	=   np.array([0.61_86, 0.53_74, 0.49_15, 0.41_35, 0.41_14, 0.45_63, 0.51_28, 0.49_77, 0.47_57]   )

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





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       super().test_inference_batch_consistent(batch_sizes=[1, 2]   )





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       super().test_inference_batch_single_identical(batch_size=2       ,							expected_max_diff=3.25e-3   )





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




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

       UpperCamelCase	:					Dict       	=   self.get_dummy_inputs(A_   )
       UpperCamelCase	:					List[str]       	=   "french fries"
       UpperCamelCase	:					Optional[Any]       	=   sd_pipe(**A_       ,							negative_prompt=A_   )
       UpperCamelCase	:					Any       	=   output.images
       UpperCamelCase	:					Union[str, Any]       	=   image[0, -3:, -3:, -1]

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

       UpperCamelCase	:					Optional[int]       	=   np.array([0.61_87, 0.53_75, 0.49_15, 0.41_36, 0.41_14, 0.45_63, 0.51_28, 0.49_76, 0.47_57]   )

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





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Optional[int]       	=   "cpu"  # ensure determinism for the device-dependent torch.Generator
       UpperCamelCase	:					int       	=   self.get_dummy_components()
       UpperCamelCase	:					List[str]       	=   StableDiffusionPanoramaPipeline(**A_   )
       UpperCamelCase	:					Optional[Any]       	=   sd_pipe.to(A_   )
       sd_pipe.set_progress_bar_config(disable=A_   )

       UpperCamelCase	:					Optional[int]       	=   self.get_dummy_inputs(A_   )
       UpperCamelCase	:					Dict       	=   sd_pipe(**A_       ,							view_batch_size=2   )
       UpperCamelCase	:					Dict       	=   output.images
       UpperCamelCase	:					Optional[Any]       	=   image[0, -3:, -3:, -1]

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

       UpperCamelCase	:					Tuple       	=   np.array([0.61_87, 0.53_75, 0.49_15, 0.41_36, 0.41_14, 0.45_63, 0.51_28, 0.49_76, 0.47_57]   )

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





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					int       	=   "cpu"  # ensure determinism for the device-dependent torch.Generator
       UpperCamelCase	:					int       	=   self.get_dummy_components()
       UpperCamelCase	:					Optional[Any]       	=   EulerAncestralDiscreteScheduler(
           beta_start=0.0_00_85       ,							beta_end=0.0_12       ,							beta_schedule="scaled_linear"   )
       UpperCamelCase	:					Any       	=   StableDiffusionPanoramaPipeline(**A_   )
       UpperCamelCase	:					Dict       	=   sd_pipe.to(A_   )
       sd_pipe.set_progress_bar_config(disable=A_   )

       UpperCamelCase	:					Optional[Any]       	=   self.get_dummy_inputs(A_   )
       UpperCamelCase	:					Any       	=   sd_pipe(**A_   ).images
       UpperCamelCase	:					Tuple       	=   image[0, -3:, -3:, -1]

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

       UpperCamelCase	:					str       	=   np.array([0.40_24, 0.65_10, 0.49_01, 0.53_78, 0.58_13, 0.56_22, 0.47_95, 0.44_67, 0.49_52]   )

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





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					List[Any]       	=   "cpu"  # ensure determinism for the device-dependent torch.Generator
       UpperCamelCase	:					Any       	=   self.get_dummy_components()
       UpperCamelCase	:					int       	=   PNDMScheduler(
           beta_start=0.0_00_85       ,							beta_end=0.0_12       ,							beta_schedule="scaled_linear"       ,							skip_prk_steps=A_   )
       UpperCamelCase	:					List[Any]       	=   StableDiffusionPanoramaPipeline(**A_   )
       UpperCamelCase	:					Optional[Any]       	=   sd_pipe.to(A_   )
       sd_pipe.set_progress_bar_config(disable=A_   )

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

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

       UpperCamelCase	:					Optional[Any]       	=   np.array([0.63_91, 0.62_91, 0.48_61, 0.51_34, 0.55_52, 0.45_78, 0.50_32, 0.50_23, 0.45_39]   )

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



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





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




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





      def        __UpperCamelCase( self       ,							A_=0   ):
       '''simple docstring'''




       UpperCamelCase	:					Optional[Any]       	=   torch.manual_seed(A_   )
       UpperCamelCase	:					Optional[Any]       	=   {
           "prompt": "a photo of the dolomites",
           "generator": generator,
           "num_inference_steps": 3,
           "guidance_scale": 7.5,
           "output_type": "numpy",
       }
       return inputs





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Any       	=   "stabilityai/stable-diffusion-2-base"
       UpperCamelCase	:					str       	=   DDIMScheduler.from_pretrained(A_       ,							subfolder="scheduler"   )
       UpperCamelCase	:					str       	=   StableDiffusionPanoramaPipeline.from_pretrained(A_       ,							scheduler=A_       ,							safety_checker=A_   )
       pipe.to(A_   )
       pipe.set_progress_bar_config(disable=A_   )
       pipe.enable_attention_slicing()

       UpperCamelCase	:					List[str]       	=   self.get_inputs()
       UpperCamelCase	:					Union[str, Any]       	=   pipe(**A_   ).images
       UpperCamelCase	:					str       	=   image[0, -3:, -3:, -1].flatten()

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

       UpperCamelCase	:					Any       	=   np.array(
           [
               0.36_96_83_92,
               0.27_02_53_72,
               0.32_44_67_66,
               0.28_37_93_87,
               0.36_36_32_74,
               0.30_73_33_47,
               0.27_10_00_27,
               0.27_05_41_25,
               0.25_53_60_96,
           ]   )

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





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Optional[Any]       	=   StableDiffusionPanoramaPipeline.from_pretrained(
           "stabilityai/stable-diffusion-2-base"       ,							safety_checker=A_   )
       UpperCamelCase	:					Optional[Any]       	=   LMSDiscreteScheduler.from_config(pipe.scheduler.config   )
       pipe.to(A_   )
       pipe.set_progress_bar_config(disable=A_   )
       pipe.enable_attention_slicing()

       UpperCamelCase	:					str       	=   self.get_inputs()
       UpperCamelCase	:					Optional[int]       	=   pipe(**A_   ).images
       UpperCamelCase	:					Optional[int]       	=   image[0, -3:, -3:, -1].flatten()

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

       UpperCamelCase	:					List[Any]       	=   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        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					int       	=   0

       def callback_fn(A_       ,							A_       ,							A_   ) -> None:
        UpperCamelCase	:					Union[str, Any]       	=   True
        nonlocal number_of_steps
        number_of_steps += 1
        if step == 1:
         UpperCamelCase	:					str       	=   latents.detach().cpu().numpy()
         assert latents.shape == (1, 4, 64, 256)
         UpperCamelCase	:					Tuple       	=   latents[0, -3:, -3:, -1]

         UpperCamelCase	:					Any       	=   np.array(
             [
                 0.18_68_18_69,
                 0.33_90_78_16,
                 0.5_36_12_76,
                 0.14_43_28_65,
                 -0.02_85_66_11,
                 -0.73_94_11_23,
                 0.23_39_79_87,
                 0.47_32_26_82,
                 -0.37_82_31_64,
             ]   )
         assert np.abs(latents_slice.flatten() - expected_slice   ).max() < 5e-2
        elif step == 2:
         UpperCamelCase	:					Optional[int]       	=   latents.detach().cpu().numpy()
         assert latents.shape == (1, 4, 64, 256)
         UpperCamelCase	:					str       	=   latents[0, -3:, -3:, -1]

         UpperCamelCase	:					Union[str, Any]       	=   np.array(
             [
                 0.18_53_96_45,
                 0.33_98_72_48,
                 0.5_37_85_59,
                 0.14_43_71_42,
                 -0.02_45_52_61,
                 -0.7_33_83_17,
                 0.23_99_07_55,
                 0.47_35_62_72,
                 -0.3_78_65_05,
             ]   )

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

       UpperCamelCase	:					List[str]       	=   False

       UpperCamelCase	:					Optional[int]       	=   "stabilityai/stable-diffusion-2-base"
       UpperCamelCase	:					List[str]       	=   DDIMScheduler.from_pretrained(A_       ,							subfolder="scheduler"   )
       UpperCamelCase	:					int       	=   StableDiffusionPanoramaPipeline.from_pretrained(A_       ,							scheduler=A_       ,							safety_checker=A_   )
       UpperCamelCase	:					str       	=   pipe.to(A_   )
       pipe.set_progress_bar_config(disable=A_   )
       pipe.enable_attention_slicing()

       UpperCamelCase	:					int       	=   self.get_inputs()
       pipe(**A_       ,							callback=A_       ,							callback_steps=1   )
       assert callback_fn.has_been_called
       assert number_of_steps == 3





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




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

       UpperCamelCase	:					str       	=   "stabilityai/stable-diffusion-2-base"
       UpperCamelCase	:					List[str]       	=   DDIMScheduler.from_pretrained(A_       ,							subfolder="scheduler"   )
       UpperCamelCase	:					Any       	=   StableDiffusionPanoramaPipeline.from_pretrained(A_       ,							scheduler=A_       ,							safety_checker=A_   )
       UpperCamelCase	:					Union[str, Any]       	=   pipe.to(A_   )
       pipe.set_progress_bar_config(disable=A_   )
       pipe.enable_attention_slicing(1   )
       pipe.enable_sequential_cpu_offload()

       UpperCamelCase	:					Any       	=   self.get_inputs()
       UpperCamelCase	:					Optional[Any]       	=   pipe(**A_   )

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







def 		A_				(     _lowerCAmelCase     )						-> str:
 UpperCamelCase	:					List[Any]       	=   ""
 for ch in key:
  if ch == " " or ch not in key_no_dups and ch.isalpha():
   key_no_dups += ch
 return key_no_dups



def 		A_				(     _lowerCAmelCase     )						-> dict[str, str]:
 UpperCamelCase	:					Optional[Any]       	=   [chr(i + 65     ) for i in range(26     )]
 # Remove duplicate characters from key
 UpperCamelCase	:					Tuple       	=   remove_duplicates(key.upper()     )
 UpperCamelCase	:					int       	=   len(_lowerCAmelCase     )
 # First fill cipher with key characters
 UpperCamelCase	:					int       	=   {alphabet[i]: char for i, char in enumerate(_lowerCAmelCase     )}
 # Then map remaining characters in alphabet to
 # the alphabet from the beginning
 for i in range(len(_lowerCAmelCase     )							,   26     ):
  UpperCamelCase	:					Optional[Any]       	=   alphabet[i - offset]
  # Ensure we are not mapping letters to letters previously mapped
  while char in key:
   offset -= 1
   UpperCamelCase	:					List[str]       	=   alphabet[i - offset]
  UpperCamelCase	:					List[Any]       	=   char
 return cipher_alphabet



def 		A_				(     _lowerCAmelCase							,   _lowerCAmelCase     )						-> str:
 return "".join(cipher_map.get(_lowerCAmelCase							,   _lowerCAmelCase     ) for ch in message.upper()     )



def 		A_				(     _lowerCAmelCase							,   _lowerCAmelCase     )						-> str:
 UpperCamelCase	:					Union[str, Any]       	=   {v: k for k, v in cipher_map.items()}
 return "".join(rev_cipher_map.get(_lowerCAmelCase							,   _lowerCAmelCase     ) for ch in message.upper()     )



def 		A_				(     )						-> None:
 UpperCamelCase	:					int       	=   input("Enter message to encode or decode: "     ).strip()
 UpperCamelCase	:					str       	=   input("Enter keyword: "     ).strip()
 UpperCamelCase	:					Union[str, Any]       	=   input("Encipher or decipher? E/D:"     ).strip()[0].lower()
 try:
  UpperCamelCase	:					List[str]       	=   {"e": encipher, "d": decipher}[option]
 except KeyError:
  raise KeyError("invalid input option"     )
 UpperCamelCase	:					str       	=   create_cipher_map(_lowerCAmelCase     )
 print(func(_lowerCAmelCase							,   _lowerCAmelCase     )     )


if __name__ == "__main__":
    import doctest

    doctest.testmod()
    main()
