


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


       """simple docstring"""




       A__											=			StableDiffusionPanoramaPipeline
       A__											=			TEXT_TO_IMAGE_PARAMS
       A__											=			TEXT_TO_IMAGE_BATCH_PARAMS
       A__											=			TEXT_TO_IMAGE_IMAGE_PARAMS
       A__											=			TEXT_TO_IMAGE_IMAGE_PARAMS





       def 						lowerCAmelCase     (   self							:			List[Any]       ):


             '''simple docstring'''



             torch.manual_seed(0       )
             lowerCamelCase__		:				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					,      )
             lowerCamelCase__		:				int    =  DDIMScheduler()
             torch.manual_seed(0       )
             lowerCamelCase__		:				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       )
             lowerCamelCase__		:				int    =  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					,      )
             lowerCamelCase__		:				Union[str, Any]    =  CLIPTextModel(SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				Union[str, Any]    =  CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"       )

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





       def 						lowerCAmelCase     (   self							:			Optional[int]					,      __lowerCamelCase							:			List[str]					,      __lowerCamelCase							:			int=0       ):


             '''simple docstring'''



             lowerCamelCase__		:				Any    =  torch.manual_seed(SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				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 						lowerCAmelCase     (   self							:			List[Any]       ):


             '''simple docstring'''



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

             lowerCamelCase__		:				Union[str, Any]    =  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				Union[str, Any]    =  sd_pipe(**SCREAMING_SNAKE_CASE__       ).images
             lowerCamelCase__		:				Any    =  image[0, -3:, -3:, -1]
             assert image.shape == (1, 64, 64, 3)

             lowerCamelCase__		:				Tuple    =  np.array([0.6_1_8_6, 0.5_3_7_4, 0.4_9_1_5, 0.4_1_3_5, 0.4_1_1_4, 0.4_5_6_3, 0.5_1_2_8, 0.4_9_7_7, 0.4_7_5_7]       )

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





       def 						lowerCAmelCase     (   self							:			Optional[int]       ):


             '''simple docstring'''



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





       def 						lowerCAmelCase     (   self							:			Tuple       ):


             '''simple docstring'''



             super().test_inference_batch_single_identical(batch_size=2					,      expected_max_diff=3.2_5E-3       )





       def 						lowerCAmelCase     (   self							:			Optional[Any]       ):


             '''simple docstring'''



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

             lowerCamelCase__		:				List[Any]    =  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				List[str]    =  "french fries"
             lowerCamelCase__		:				Optional[int]    =  sd_pipe(**SCREAMING_SNAKE_CASE__					,      negative_prompt=SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				List[str]    =  output.images
             lowerCamelCase__		:				int    =  image[0, -3:, -3:, -1]

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

             lowerCamelCase__		:				Optional[Any]    =  np.array([0.6_1_8_7, 0.5_3_7_5, 0.4_9_1_5, 0.4_1_3_6, 0.4_1_1_4, 0.4_5_6_3, 0.5_1_2_8, 0.4_9_7_6, 0.4_7_5_7]       )

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





       def 						lowerCAmelCase     (   self							:			Any       ):


             '''simple docstring'''



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

             lowerCamelCase__		:				str    =  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				Tuple    =  sd_pipe(**SCREAMING_SNAKE_CASE__					,      view_batch_size=2       )
             lowerCamelCase__		:				int    =  output.images
             lowerCamelCase__		:				str    =  image[0, -3:, -3:, -1]

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

             lowerCamelCase__		:				int    =  np.array([0.6_1_8_7, 0.5_3_7_5, 0.4_9_1_5, 0.4_1_3_6, 0.4_1_1_4, 0.4_5_6_3, 0.5_1_2_8, 0.4_9_7_6, 0.4_7_5_7]       )

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





       def 						lowerCAmelCase     (   self							:			Tuple       ):


             '''simple docstring'''



             lowerCamelCase__		:				Union[str, Any]    =  "cpu"  # ensure determinism for the device-dependent torch.Generator
             lowerCamelCase__		:				Dict    =  self.get_dummy_components()
             lowerCamelCase__		:				Tuple    =  EulerAncestralDiscreteScheduler(
                 beta_start=0.0_0_0_8_5					,      beta_end=0.0_1_2					,      beta_schedule="scaled_linear"       )
             lowerCamelCase__		:				List[Any]    =  StableDiffusionPanoramaPipeline(**SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				int    =  sd_pipe.to(SCREAMING_SNAKE_CASE__       )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )

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

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

             lowerCamelCase__		:				Union[str, Any]    =  np.array([0.4_0_2_4, 0.6_5_1_0, 0.4_9_0_1, 0.5_3_7_8, 0.5_8_1_3, 0.5_6_2_2, 0.4_7_9_5, 0.4_4_6_7, 0.4_9_5_2]       )

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





       def 						lowerCAmelCase     (   self							:			List[Any]       ):


             '''simple docstring'''



             lowerCamelCase__		:				int    =  "cpu"  # ensure determinism for the device-dependent torch.Generator
             lowerCamelCase__		:				Any    =  self.get_dummy_components()
             lowerCamelCase__		:				Optional[int]    =  PNDMScheduler(
                 beta_start=0.0_0_0_8_5					,      beta_end=0.0_1_2					,      beta_schedule="scaled_linear"					,      skip_prk_steps=SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				Tuple    =  StableDiffusionPanoramaPipeline(**SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				Optional[Any]    =  sd_pipe.to(SCREAMING_SNAKE_CASE__       )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )

             lowerCamelCase__		:				Union[str, Any]    =  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				Tuple    =  sd_pipe(**SCREAMING_SNAKE_CASE__       ).images
             lowerCamelCase__		:				Any    =  image[0, -3:, -3:, -1]

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

             lowerCamelCase__		:				Dict    =  np.array([0.6_3_9_1, 0.6_2_9_1, 0.4_8_6_1, 0.5_1_3_4, 0.5_5_5_2, 0.4_5_7_8, 0.5_0_3_2, 0.5_0_2_3, 0.4_5_3_9]       )

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







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


       """simple docstring"""




       def 						lowerCAmelCase     (   self							:			Any       ):


             '''simple docstring'''



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





       def 						lowerCAmelCase     (   self							:			List[Any]					,      __lowerCamelCase							:			str=0       ):


             '''simple docstring'''



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





       def 						lowerCAmelCase     (   self							:			Union[str, Any]       ):


             '''simple docstring'''



             lowerCamelCase__		:				Tuple    =  "stabilityai/stable-diffusion-2-base"
             lowerCamelCase__		:				List[Any]    =  DDIMScheduler.from_pretrained(SCREAMING_SNAKE_CASE__					,      subfolder="scheduler"       )
             lowerCamelCase__		:				Dict    =  StableDiffusionPanoramaPipeline.from_pretrained(SCREAMING_SNAKE_CASE__					,      scheduler=SCREAMING_SNAKE_CASE__					,      safety_checker=SCREAMING_SNAKE_CASE__       )
             pipe.to(SCREAMING_SNAKE_CASE__       )
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )
             pipe.enable_attention_slicing()

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

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

             lowerCamelCase__		:				Optional[Any]    =  np.array(
                 [
                     0.3_6_9_6_8_3_9_2,
                     0.2_7_0_2_5_3_7_2,
                     0.3_2_4_4_6_7_6_6,
                     0.2_8_3_7_9_3_8_7,
                     0.3_6_3_6_3_2_7_4,
                     0.3_0_7_3_3_3_4_7,
                     0.2_7_1_0_0_0_2_7,
                     0.2_7_0_5_4_1_2_5,
                     0.2_5_5_3_6_0_9_6,
                 ]       )

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





       def 						lowerCAmelCase     (   self							:			List[str]       ):


             '''simple docstring'''



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

             lowerCamelCase__		:				Dict    =  self.get_inputs()
             lowerCamelCase__		:				str    =  pipe(**SCREAMING_SNAKE_CASE__       ).images
             lowerCamelCase__		:				str    =  image[0, -3:, -3:, -1].flatten()

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

             lowerCamelCase__		:				List[str]    =  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 						lowerCAmelCase     (   self							:			Tuple       ):


             '''simple docstring'''



             lowerCamelCase__		:				Tuple    =  0

             def callback_fn(__lowerCamelCase							:			int					,      __lowerCamelCase							:			int					,      __lowerCamelCase							:			torch.FloatTensor       ) -> None:
                   lowerCamelCase__		:				Tuple    =  True
                   nonlocal number_of_steps
                   number_of_steps += 1
                   if step == 1:
                         lowerCamelCase__		:				List[Any]    =  latents.detach().cpu().numpy()
                         assert latents.shape == (1, 4, 64, 256)
                         lowerCamelCase__		:				str    =  latents[0, -3:, -3:, -1]

                         lowerCamelCase__		:				Optional[int]    =  np.array(
                             [
                                 0.1_8_6_8_1_8_6_9,
                                 0.3_3_9_0_7_8_1_6,
                                 0.5_3_6_1_2_7_6,
                                 0.1_4_4_3_2_8_6_5,
                                 -0.0_2_8_5_6_6_1_1,
                                 -0.7_3_9_4_1_1_2_3,
                                 0.2_3_3_9_7_9_8_7,
                                 0.4_7_3_2_2_6_8_2,
                                 -0.3_7_8_2_3_1_6_4,
                             ]       )
                         assert np.abs(latents_slice.flatten() - expected_slice       ).max() < 5E-2
                   elif step == 2:
                         lowerCamelCase__		:				List[str]    =  latents.detach().cpu().numpy()
                         assert latents.shape == (1, 4, 64, 256)
                         lowerCamelCase__		:				str    =  latents[0, -3:, -3:, -1]

                         lowerCamelCase__		:				Optional[int]    =  np.array(
                             [
                                 0.1_8_5_3_9_6_4_5,
                                 0.3_3_9_8_7_2_4_8,
                                 0.5_3_7_8_5_5_9,
                                 0.1_4_4_3_7_1_4_2,
                                 -0.0_2_4_5_5_2_6_1,
                                 -0.7_3_3_8_3_1_7,
                                 0.2_3_9_9_0_7_5_5,
                                 0.4_7_3_5_6_2_7_2,
                                 -0.3_7_8_6_5_0_5,
                             ]       )

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

             lowerCamelCase__		:				Any    =  False

             lowerCamelCase__		:				Dict    =  "stabilityai/stable-diffusion-2-base"
             lowerCamelCase__		:				Optional[int]    =  DDIMScheduler.from_pretrained(SCREAMING_SNAKE_CASE__					,      subfolder="scheduler"       )
             lowerCamelCase__		:				List[Any]    =  StableDiffusionPanoramaPipeline.from_pretrained(SCREAMING_SNAKE_CASE__					,      scheduler=SCREAMING_SNAKE_CASE__					,      safety_checker=SCREAMING_SNAKE_CASE__       )
             lowerCamelCase__		:				Tuple    =  pipe.to(SCREAMING_SNAKE_CASE__       )
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )
             pipe.enable_attention_slicing()

             lowerCamelCase__		:				List[str]    =  self.get_inputs()
             pipe(**SCREAMING_SNAKE_CASE__					,      callback=SCREAMING_SNAKE_CASE__					,      callback_steps=1       )
             assert callback_fn.has_been_called
             assert number_of_steps == 3





       def 						lowerCAmelCase     (   self							:			Union[str, Any]       ):


             '''simple docstring'''



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

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

             lowerCamelCase__		:				Tuple    =  self.get_inputs()
             lowerCamelCase__		:				Any    =  pipe(**SCREAMING_SNAKE_CASE__       )

             lowerCamelCase__		:				Dict    =  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_torch_available


SCREAMING_SNAKE_CASE__   :			Optional[Any]												=    {"configuration_wavlm": ["WAVLM_PRETRAINED_CONFIG_ARCHIVE_MAP", "WavLMConfig"]}

try:
					if not is_torch_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					SCREAMING_SNAKE_CASE__   :			int												=    [
					    "WAVLM_PRETRAINED_MODEL_ARCHIVE_LIST",
					    "WavLMForAudioFrameClassification",
					    "WavLMForCTC",
					    "WavLMForSequenceClassification",
					    "WavLMForXVector",
					    "WavLMModel",
					    "WavLMPreTrainedModel",
					]

if TYPE_CHECKING:
					from .configuration_wavlm import WAVLM_PRETRAINED_CONFIG_ARCHIVE_MAP, WavLMConfig

					try:
										if not is_torch_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_wavlm import (
										    WAVLM_PRETRAINED_MODEL_ARCHIVE_LIST,
										    WavLMForAudioFrameClassification,
										    WavLMForCTC,
										    WavLMForSequenceClassification,
										    WavLMForXVector,
										    WavLMModel,
										    WavLMPreTrainedModel,
										)

else:
					import sys

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



