





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




   """simple docstring"""



   __A       :    Any	  =  StableDiffusionPanoramaPipeline
   __A       :    str	  =  TEXT_TO_IMAGE_PARAMS
   __A       :    int	  =  TEXT_TO_IMAGE_BATCH_PARAMS
   __A       :    int	  =  TEXT_TO_IMAGE_IMAGE_PARAMS
   __A       :    int	  =  TEXT_TO_IMAGE_IMAGE_PARAMS

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







          '''simple docstring'''

          torch.manual_seed(0)
          a__						:     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 ,  )
          a__						:     int			=    DDIMScheduler()
          torch.manual_seed(0)
          a__						:     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)
          a__						:     Optional[Any]			=    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 ,  )
          a__						:     Dict			=    CLIPTextModel(lowerCamelCase_)
          a__						:     Optional[Any]			=    CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

          a__						:     Optional[int]			=    {
              'unet': unet,
              'scheduler': scheduler,
              'vae': vae,
              'text_encoder': text_encoder,
              'tokenizer': tokenizer,
              'safety_checker': None,
              'feature_extractor': None,
          }
          return components

   def 							__lowercase		( self ,  lowercase ,  lowercase=0)      ->						Optional[int]:







          '''simple docstring'''

          a__						:     Dict			=    torch.manual_seed(lowerCamelCase_)
          a__						:     int			=    {
              '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 							__lowercase		( self)      ->						Optional[int]:







          '''simple docstring'''

          a__						:     int			=    'cpu'  # ensure determinism for the device-dependent torch.Generator
          a__						:     Any			=    self.get_dummy_components()
          a__						:     Any			=    StableDiffusionPanoramaPipeline(**lowerCamelCase_)
          a__						:     Tuple			=    sd_pipe.to(lowerCamelCase_)
          sd_pipe.set_progress_bar_config(disable=lowerCamelCase_)

          a__						:     Any			=    self.get_dummy_inputs(lowerCamelCase_)
          a__						:     int			=    sd_pipe(**lowerCamelCase_).images
          a__						:     Tuple			=    image[0, -3:, -3:, -1]
          assert image.shape == (1, 64, 64, 3)

          a__						:     Any			=    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 							__lowercase		( self)      ->						Optional[Any]:







          '''simple docstring'''

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

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







          '''simple docstring'''

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

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







          '''simple docstring'''

          a__						:     List[str]			=    'cpu'  # ensure determinism for the device-dependent torch.Generator
          a__						:     Optional[Any]			=    self.get_dummy_components()
          a__						:     Dict			=    StableDiffusionPanoramaPipeline(**lowerCamelCase_)
          a__						:     Dict			=    sd_pipe.to(lowerCamelCase_)
          sd_pipe.set_progress_bar_config(disable=lowerCamelCase_)

          a__						:     Union[str, Any]			=    self.get_dummy_inputs(lowerCamelCase_)
          a__						:     Union[str, Any]			=    'french fries'
          a__						:     Tuple			=    sd_pipe(**lowerCamelCase_ ,  negative_prompt=lowerCamelCase_)
          a__						:     Union[str, Any]			=    output.images
          a__						:     Optional[int]			=    image[0, -3:, -3:, -1]

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

          a__						:     List[str]			=    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 							__lowercase		( self)      ->						int:







          '''simple docstring'''

          a__						:     Dict			=    'cpu'  # ensure determinism for the device-dependent torch.Generator
          a__						:     Dict			=    self.get_dummy_components()
          a__						:     Dict			=    StableDiffusionPanoramaPipeline(**lowerCamelCase_)
          a__						:     Dict			=    sd_pipe.to(lowerCamelCase_)
          sd_pipe.set_progress_bar_config(disable=lowerCamelCase_)

          a__						:     Optional[Any]			=    self.get_dummy_inputs(lowerCamelCase_)
          a__						:     Any			=    sd_pipe(**lowerCamelCase_ ,  view_batch_size=2)
          a__						:     List[Any]			=    output.images
          a__						:     Optional[Any]			=    image[0, -3:, -3:, -1]

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

          a__						:     List[str]			=    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 							__lowercase		( self)      ->						Union[str, Any]:







          '''simple docstring'''

          a__						:     Any			=    'cpu'  # ensure determinism for the device-dependent torch.Generator
          a__						:     List[Any]			=    self.get_dummy_components()
          a__						:     int			=    EulerAncestralDiscreteScheduler(
              beta_start=0.0_00_85 ,  beta_end=0.0_12 ,  beta_schedule='scaled_linear')
          a__						:     str			=    StableDiffusionPanoramaPipeline(**lowerCamelCase_)
          a__						:     List[str]			=    sd_pipe.to(lowerCamelCase_)
          sd_pipe.set_progress_bar_config(disable=lowerCamelCase_)

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

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

          a__						:     Optional[int]			=    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 							__lowercase		( self)      ->						str:







          '''simple docstring'''

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

          a__						:     str			=    self.get_dummy_inputs(lowerCamelCase_)
          a__						:     List[Any]			=    sd_pipe(**lowerCamelCase_).images
          a__						:     Any			=    image[0, -3:, -3:, -1]

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

          a__						:     int			=    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     ):




   """simple docstring"""



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







          '''simple docstring'''

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

   def 							__lowercase		( self ,  lowercase=0)      ->						int:







          '''simple docstring'''

          a__						:     List[Any]			=    torch.manual_seed(lowerCamelCase_)
          a__						:     str			=    {
              'prompt': 'a photo of the dolomites',
              'generator': generator,
              'num_inference_steps': 3,
              'guidance_scale': 7.5,
              'output_type': 'numpy',
          }
          return inputs

   def 							__lowercase		( self)      ->						Any:







          '''simple docstring'''

          a__						:     str			=    'stabilityai/stable-diffusion-2-base'
          a__						:     List[Any]			=    DDIMScheduler.from_pretrained(lowerCamelCase_ ,  subfolder='scheduler')
          a__						:     Tuple			=    StableDiffusionPanoramaPipeline.from_pretrained(lowerCamelCase_ ,  scheduler=lowerCamelCase_ ,  safety_checker=lowerCamelCase_)
          pipe.to(lowerCamelCase_)
          pipe.set_progress_bar_config(disable=lowerCamelCase_)
          pipe.enable_attention_slicing()

          a__						:     int			=    self.get_inputs()
          a__						:     Tuple			=    pipe(**lowerCamelCase_).images
          a__						:     Optional[int]			=    image[0, -3:, -3:, -1].flatten()

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

          a__						:     str			=    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 							__lowercase		( self)      ->						List[str]:







          '''simple docstring'''

          a__						:     Any			=    StableDiffusionPanoramaPipeline.from_pretrained(
              'stabilityai/stable-diffusion-2-base' ,  safety_checker=lowerCamelCase_)
          a__						:     str			=    LMSDiscreteScheduler.from_config(pipe.scheduler.config)
          pipe.to(lowerCamelCase_)
          pipe.set_progress_bar_config(disable=lowerCamelCase_)
          pipe.enable_attention_slicing()

          a__						:     Dict			=    self.get_inputs()
          a__						:     List[Any]			=    pipe(**lowerCamelCase_).images
          a__						:     Optional[int]			=    image[0, -3:, -3:, -1].flatten()

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

          a__						:     Dict			=    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 							__lowercase		( self)      ->						Optional[int]:







          '''simple docstring'''

          a__						:     str			=    0

          def callback_fn(lowercase ,  lowercase ,  lowercase) -> None:
                 a__						:     List[Any]			=    True
                 nonlocal number_of_steps
                 number_of_steps += 1
                 if step == 1:
                        a__						:     Dict			=    latents.detach().cpu().numpy()
                        assert latents.shape == (1, 4, 64, 256)
                        a__						:     Optional[int]			=    latents[0, -3:, -3:, -1]

                        a__						:     Union[str, 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:
                        a__						:     str			=    latents.detach().cpu().numpy()
                        assert latents.shape == (1, 4, 64, 256)
                        a__						:     List[Any]			=    latents[0, -3:, -3:, -1]

                        a__						:     str			=    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

          a__						:     Any			=    False

          a__						:     Optional[int]			=    'stabilityai/stable-diffusion-2-base'
          a__						:     Tuple			=    DDIMScheduler.from_pretrained(lowerCamelCase_ ,  subfolder='scheduler')
          a__						:     Optional[int]			=    StableDiffusionPanoramaPipeline.from_pretrained(lowerCamelCase_ ,  scheduler=lowerCamelCase_ ,  safety_checker=lowerCamelCase_)
          a__						:     Optional[Any]			=    pipe.to(lowerCamelCase_)
          pipe.set_progress_bar_config(disable=lowerCamelCase_)
          pipe.enable_attention_slicing()

          a__						:     Optional[int]			=    self.get_inputs()
          pipe(**lowerCamelCase_ ,  callback=lowerCamelCase_ ,  callback_steps=1)
          assert callback_fn.has_been_called
          assert number_of_steps == 3

   def 							__lowercase		( self)      ->						str:







          '''simple docstring'''

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

          a__						:     str			=    'stabilityai/stable-diffusion-2-base'
          a__						:     Tuple			=    DDIMScheduler.from_pretrained(lowerCamelCase_ ,  subfolder='scheduler')
          a__						:     List[Any]			=    StableDiffusionPanoramaPipeline.from_pretrained(lowerCamelCase_ ,  scheduler=lowerCamelCase_ ,  safety_checker=lowerCamelCase_)
          a__						:     Any			=    pipe.to(lowerCamelCase_)
          pipe.set_progress_bar_config(disable=lowerCamelCase_)
          pipe.enable_attention_slicing(1)
          pipe.enable_sequential_cpu_offload()

          a__						:     Optional[Any]			=    self.get_inputs()
          a__						:     str			=    pipe(**lowerCamelCase_)

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









from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


lowercase				:	str      						=  {
    """configuration_vision_encoder_decoder""": ["""VisionEncoderDecoderConfig""", """VisionEncoderDecoderOnnxConfig"""]
}

try:
    if not is_torch_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowercase				:	Optional[Any]      						=  ["""VisionEncoderDecoderModel"""]

try:
    if not is_tf_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowercase				:	Optional[int]      						=  ["""TFVisionEncoderDecoderModel"""]

try:
    if not is_flax_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowercase				:	List[str]      						=  ["""FlaxVisionEncoderDecoderModel"""]

if TYPE_CHECKING:
    from .configuration_vision_encoder_decoder import VisionEncoderDecoderConfig, VisionEncoderDecoderOnnxConfig

    try:
        if not is_torch_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_vision_encoder_decoder import VisionEncoderDecoderModel

    try:
        if not is_tf_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_tf_vision_encoder_decoder import TFVisionEncoderDecoderModel

    try:
        if not is_flax_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_flax_vision_encoder_decoder import FlaxVisionEncoderDecoderModel

else:
    import sys

    lowercase				:	Optional[int]      						=  _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)


