

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     snake_case_			(				_a			, _a			, unittest.TestCase							):


							"""simple docstring"""





							__UpperCAmelCase       =StableDiffusionPanoramaPipeline
							__UpperCAmelCase       =TEXT_TO_IMAGE_PARAMS
							__UpperCAmelCase       =TEXT_TO_IMAGE_BATCH_PARAMS
							__UpperCAmelCase       =TEXT_TO_IMAGE_IMAGE_PARAMS
							__UpperCAmelCase       =TEXT_TO_IMAGE_IMAGE_PARAMS




							def        A__  ( self							):
												torch.manual_seed(0							)
												__lowerCAmelCase					=							UNetaDConditionModel(
												    block_out_channels=(3_2, 6_4)					,						layers_per_block=1					,						sample_size=3_2					,						in_channels=4					,						out_channels=4					,						down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')					,						up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')					,						cross_attention_dim=3_2					,						)
												__lowerCAmelCase					=							DDIMScheduler()
												torch.manual_seed(0							)
												__lowerCAmelCase					=							AutoencoderKL(
												    block_out_channels=[3_2, 6_4]					,						in_channels=3					,						out_channels=3					,						down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']					,						up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']					,						latent_channels=4					,						)
												torch.manual_seed(0							)
												__lowerCAmelCase					=							CLIPTextConfig(
												    bos_token_id=0					,						eos_token_id=2					,						hidden_size=3_2					,						intermediate_size=3_7					,						layer_norm_eps=1e-0_5					,						num_attention_heads=4					,						num_hidden_layers=5					,						pad_token_id=1					,						vocab_size=1_0_0_0					,						)
												__lowerCAmelCase					=							CLIPTextModel(_A							)
												__lowerCAmelCase					=							CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'							)

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




							def        A__  ( self					,						_A					,						_A=0							):
												__lowerCAmelCase					=							torch.manual_seed(_A							)
												__lowerCAmelCase					=							{
												    '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        A__  ( self							):
												__lowerCAmelCase					=							'cpu'  # ensure determinism for the device-dependent torch.Generator
												__lowerCAmelCase					=							self.get_dummy_components()
												__lowerCAmelCase					=							StableDiffusionPanoramaPipeline(**_A							)
												__lowerCAmelCase					=							sd_pipe.to(_A							)
												sd_pipe.set_progress_bar_config(disable=_A							)

												__lowerCAmelCase					=							self.get_dummy_inputs(_A							)
												__lowerCAmelCase					=							sd_pipe(**_A							).images
												__lowerCAmelCase					=							image[0, -3:, -3:, -1]
												assert image.shape == (1, 6_4, 6_4, 3)

												__lowerCAmelCase					=							np.array([0.6186, 0.5374, 0.4915, 0.4135, 0.4114, 0.4563, 0.5128, 0.4977, 0.4757]							)

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




							def        A__  ( self							):
												super().test_inference_batch_consistent(batch_sizes=[1, 2]							)




							def        A__  ( self							):
												super().test_inference_batch_single_identical(batch_size=2					,						expected_max_diff=3.2_5e-3							)




							def        A__  ( self							):
												__lowerCAmelCase					=							'cpu'  # ensure determinism for the device-dependent torch.Generator
												__lowerCAmelCase					=							self.get_dummy_components()
												__lowerCAmelCase					=							StableDiffusionPanoramaPipeline(**_A							)
												__lowerCAmelCase					=							sd_pipe.to(_A							)
												sd_pipe.set_progress_bar_config(disable=_A							)

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

												assert image.shape == (1, 6_4, 6_4, 3)

												__lowerCAmelCase					=							np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]							)

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




							def        A__  ( self							):
												__lowerCAmelCase					=							'cpu'  # ensure determinism for the device-dependent torch.Generator
												__lowerCAmelCase					=							self.get_dummy_components()
												__lowerCAmelCase					=							StableDiffusionPanoramaPipeline(**_A							)
												__lowerCAmelCase					=							sd_pipe.to(_A							)
												sd_pipe.set_progress_bar_config(disable=_A							)

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

												assert image.shape == (1, 6_4, 6_4, 3)

												__lowerCAmelCase					=							np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]							)

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




							def        A__  ( self							):
												__lowerCAmelCase					=							'cpu'  # ensure determinism for the device-dependent torch.Generator
												__lowerCAmelCase					=							self.get_dummy_components()
												__lowerCAmelCase					=							EulerAncestralDiscreteScheduler(
												    beta_start=0.0_0085					,						beta_end=0.012					,						beta_schedule='scaled_linear'							)
												__lowerCAmelCase					=							StableDiffusionPanoramaPipeline(**_A							)
												__lowerCAmelCase					=							sd_pipe.to(_A							)
												sd_pipe.set_progress_bar_config(disable=_A							)

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

												assert image.shape == (1, 6_4, 6_4, 3)

												__lowerCAmelCase					=							np.array([0.4024, 0.6510, 0.4901, 0.5378, 0.5813, 0.5622, 0.4795, 0.4467, 0.4952]							)

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




							def        A__  ( self							):
												__lowerCAmelCase					=							'cpu'  # ensure determinism for the device-dependent torch.Generator
												__lowerCAmelCase					=							self.get_dummy_components()
												__lowerCAmelCase					=							PNDMScheduler(
												    beta_start=0.0_0085					,						beta_end=0.012					,						beta_schedule='scaled_linear'					,						skip_prk_steps=_A							)
												__lowerCAmelCase					=							StableDiffusionPanoramaPipeline(**_A							)
												__lowerCAmelCase					=							sd_pipe.to(_A							)
												sd_pipe.set_progress_bar_config(disable=_A							)

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

												assert image.shape == (1, 6_4, 6_4, 3)

												__lowerCAmelCase					=							np.array([0.6391, 0.6291, 0.4861, 0.5134, 0.5552, 0.4578, 0.5032, 0.5023, 0.4539]							)

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





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


							"""simple docstring"""





							def        A__  ( self							):
												super().tearDown()
												gc.collect()
												torch.cuda.empty_cache()




							def        A__  ( self					,						_A=0							):
												__lowerCAmelCase					=							torch.manual_seed(_A							)
												__lowerCAmelCase					=							{
												    'prompt': 'a photo of the dolomites',
												    'generator': generator,
												    'num_inference_steps': 3,
												    'guidance_scale': 7.5,
												    'output_type': 'numpy',
												}
												return inputs




							def        A__  ( self							):
												__lowerCAmelCase					=							'stabilityai/stable-diffusion-2-base'
												__lowerCAmelCase					=							DDIMScheduler.from_pretrained(_A					,						subfolder='scheduler'							)
												__lowerCAmelCase					=							StableDiffusionPanoramaPipeline.from_pretrained(_A					,						scheduler=_A					,						safety_checker=_A							)
												pipe.to(_A							)
												pipe.set_progress_bar_config(disable=_A							)
												pipe.enable_attention_slicing()

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

												assert image.shape == (1, 5_1_2, 2_0_4_8, 3)

												__lowerCAmelCase					=							np.array(
												    [
												        0.3696_8392,
												        0.2702_5372,
												        0.3244_6766,
												        0.2837_9387,
												        0.3636_3274,
												        0.3073_3347,
												        0.2710_0027,
												        0.2705_4125,
												        0.2553_6096,
												    ]							)

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




							def        A__  ( self							):
												__lowerCAmelCase					=							StableDiffusionPanoramaPipeline.from_pretrained(
												    'stabilityai/stable-diffusion-2-base'					,						safety_checker=_A							)
												__lowerCAmelCase					=							LMSDiscreteScheduler.from_config(pipe.scheduler.config							)
												pipe.to(_A							)
												pipe.set_progress_bar_config(disable=_A							)
												pipe.enable_attention_slicing()

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

												assert image.shape == (1, 5_1_2, 2_0_4_8, 3)

												__lowerCAmelCase					=							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        A__  ( self							):
												__lowerCAmelCase					=							0

												def callback_fn(_A					,						_A					,						_A							) -> None:
																	__lowerCAmelCase					=							True
																	nonlocal number_of_steps
																	number_of_steps += 1
																	if step == 1:
																						__lowerCAmelCase					=							latents.detach().cpu().numpy()
																						assert latents.shape == (1, 4, 6_4, 2_5_6)
																						__lowerCAmelCase					=							latents[0, -3:, -3:, -1]

																						__lowerCAmelCase					=							np.array(
																						    [
																						        0.1868_1869,
																						        0.3390_7816,
																						        0.536_1276,
																						        0.1443_2865,
																						        -0.0285_6611,
																						        -0.7394_1123,
																						        0.2339_7987,
																						        0.4732_2682,
																						        -0.3782_3164,
																						    ]							)
																						assert np.abs(latents_slice.flatten() - expected_slice							).max() < 5e-2
																	elif step == 2:
																						__lowerCAmelCase					=							latents.detach().cpu().numpy()
																						assert latents.shape == (1, 4, 6_4, 2_5_6)
																						__lowerCAmelCase					=							latents[0, -3:, -3:, -1]

																						__lowerCAmelCase					=							np.array(
																						    [
																						        0.1853_9645,
																						        0.3398_7248,
																						        0.537_8559,
																						        0.1443_7142,
																						        -0.0245_5261,
																						        -0.733_8317,
																						        0.2399_0755,
																						        0.4735_6272,
																						        -0.378_6505,
																						    ]							)

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

												__lowerCAmelCase					=							False

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

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




							def        A__  ( self							):
												torch.cuda.empty_cache()
												torch.cuda.reset_max_memory_allocated()
												torch.cuda.reset_peak_memory_stats()

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

												__lowerCAmelCase					=							self.get_inputs()
												__lowerCAmelCase					=							pipe(**_A							)

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









from typing import TYPE_CHECKING

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


lowerCamelCase      				=     {'''configuration_unispeech''': ['''UNISPEECH_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''UniSpeechConfig''']}

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		lowerCamelCase      				=     [
		    '''UNISPEECH_PRETRAINED_MODEL_ARCHIVE_LIST''',
		    '''UniSpeechForCTC''',
		    '''UniSpeechForPreTraining''',
		    '''UniSpeechForSequenceClassification''',
		    '''UniSpeechModel''',
		    '''UniSpeechPreTrainedModel''',
		]

if TYPE_CHECKING:
		from .configuration_unispeech import UNISPEECH_PRETRAINED_CONFIG_ARCHIVE_MAP, UniSpeechConfig

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_unispeech import (
				    UNISPEECH_PRETRAINED_MODEL_ARCHIVE_LIST,
				    UniSpeechForCTC,
				    UniSpeechForPreTraining,
				    UniSpeechForSequenceClassification,
				    UniSpeechModel,
				    UniSpeechPreTrainedModel,
				)

else:
		import sys

		lowerCamelCase      				=     _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)






