



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







						"""simple docstring"""







						_A		:							str							       =		StableDiffusionPanoramaPipeline
						_A		:							Dict							       =		TEXT_TO_IMAGE_PARAMS
						_A		:							Optional[int]							       =		TEXT_TO_IMAGE_BATCH_PARAMS
						_A		:							List[str]							       =		TEXT_TO_IMAGE_IMAGE_PARAMS
						_A		:							List[str]							       =		TEXT_TO_IMAGE_IMAGE_PARAMS







						def 						__UpperCamelCase							(self       ):
											torch.manual_seed(0       )
											snake_case_					:		List[Any] 					=					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   ,		)
											snake_case_					:		str 					=					DDIMScheduler()
											torch.manual_seed(0       )
											snake_case_					:		List[str] 					=					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       )
											snake_case_					:		Optional[int] 					=					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=10_00   ,		)
											snake_case_					:		List[Any] 					=					CLIPTextModel(_UpperCAmelCase       )
											snake_case_					:		int 					=					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

											snake_case_					:		Union[str, Any] 					=					{
											    '''unet''': unet,
											    '''scheduler''': scheduler,
											    '''vae''': vae,
											    '''text_encoder''': text_encoder,
											    '''tokenizer''': tokenizer,
											    '''safety_checker''': None,
											    '''feature_extractor''': None,
											}
											return components







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__=0       ):
											snake_case_					:		Optional[int] 					=					torch.manual_seed(_UpperCAmelCase       )
											snake_case_					:		Tuple 					=					{
											    '''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       ):
											snake_case_					:		Dict 					=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
											snake_case_					:		Union[str, Any] 					=					self.get_dummy_components()
											snake_case_					:		Optional[Any] 					=					StableDiffusionPanoramaPipeline(**_UpperCAmelCase       )
											snake_case_					:		Tuple 					=					sd_pipe.to(_UpperCAmelCase       )
											sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

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

											snake_case_					:		List[str] 					=					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 						__UpperCamelCase							(self       ):
											super().test_inference_batch_consistent(batch_sizes=[1, 2]       )







						def 						__UpperCamelCase							(self       ):
											super().test_inference_batch_single_identical(batch_size=2   ,		expected_max_diff=3.25e-3       )







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

											snake_case_					:		Optional[int] 					=					self.get_dummy_inputs(_UpperCAmelCase       )
											snake_case_					:		Dict 					=					'''french fries'''
											snake_case_					:		List[Any] 					=					sd_pipe(**_UpperCAmelCase   ,		negative_prompt=_UpperCAmelCase       )
											snake_case_					:		List[str] 					=					output.images
											snake_case_					:		str 					=					image[0, -3:, -3:, -1]

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

											snake_case_					:		int 					=					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 						__UpperCamelCase							(self       ):
											snake_case_					:		Tuple 					=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
											snake_case_					:		List[Any] 					=					self.get_dummy_components()
											snake_case_					:		Union[str, Any] 					=					StableDiffusionPanoramaPipeline(**_UpperCAmelCase       )
											snake_case_					:		Any 					=					sd_pipe.to(_UpperCAmelCase       )
											sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

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

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

											snake_case_					:		Optional[int] 					=					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 						__UpperCamelCase							(self       ):
											snake_case_					:		int 					=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
											snake_case_					:		Any 					=					self.get_dummy_components()
											snake_case_					:		Optional[Any] 					=					EulerAncestralDiscreteScheduler(
											    beta_start=0.00085   ,		beta_end=0.012   ,		beta_schedule="""scaled_linear"""       )
											snake_case_					:		List[Any] 					=					StableDiffusionPanoramaPipeline(**_UpperCAmelCase       )
											snake_case_					:		Optional[Any] 					=					sd_pipe.to(_UpperCAmelCase       )
											sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

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

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

											snake_case_					:		int 					=					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 						__UpperCamelCase							(self       ):
											snake_case_					:		List[str] 					=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
											snake_case_					:		Dict 					=					self.get_dummy_components()
											snake_case_					:		str 					=					PNDMScheduler(
											    beta_start=0.00085   ,		beta_end=0.012   ,		beta_schedule="""scaled_linear"""   ,		skip_prk_steps=_UpperCAmelCase       )
											snake_case_					:		Dict 					=					StableDiffusionPanoramaPipeline(**_UpperCAmelCase       )
											snake_case_					:		Dict 					=					sd_pipe.to(_UpperCAmelCase       )
											sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

											snake_case_					:		str 					=					self.get_dummy_inputs(_UpperCAmelCase       )
											snake_case_					:		Union[str, Any] 					=					sd_pipe(**_UpperCAmelCase       ).images
											snake_case_					:		Dict 					=					image[0, -3:, -3:, -1]

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

											snake_case_					:		Union[str, Any] 					=					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    __lowercase     (					unittest.TestCase):







						"""simple docstring"""







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







						def 						__UpperCamelCase							(self   ,		lowercase__=0       ):
											snake_case_					:		List[Any] 					=					torch.manual_seed(_UpperCAmelCase       )
											snake_case_					:		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       ):
											snake_case_					:		Union[str, Any] 					=					'''stabilityai/stable-diffusion-2-base'''
											snake_case_					:		List[str] 					=					DDIMScheduler.from_pretrained(_UpperCAmelCase   ,		subfolder="""scheduler"""       )
											snake_case_					:		Optional[Any] 					=					StableDiffusionPanoramaPipeline.from_pretrained(_UpperCAmelCase   ,		scheduler=_UpperCAmelCase   ,		safety_checker=_UpperCAmelCase       )
											pipe.to(_UpperCAmelCase       )
											pipe.set_progress_bar_config(disable=_UpperCAmelCase       )
											pipe.enable_attention_slicing()

											snake_case_					:		Dict 					=					self.get_inputs()
											snake_case_					:		Union[str, Any] 					=					pipe(**_UpperCAmelCase       ).images
											snake_case_					:		Tuple 					=					image[0, -3:, -3:, -1].flatten()

											assert image.shape == (1, 5_12, 20_48, 3)

											snake_case_					:		Dict 					=					np.array(
											    [
											        0.36968392,
											        0.27025372,
											        0.32446766,
											        0.28379387,
											        0.36363274,
											        0.30733347,
											        0.27100027,
											        0.27054125,
											        0.25536096,
											    ]       )

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







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Dict 					=					StableDiffusionPanoramaPipeline.from_pretrained(
											    """stabilityai/stable-diffusion-2-base"""   ,		safety_checker=_UpperCAmelCase       )
											snake_case_					:		List[Any] 					=					LMSDiscreteScheduler.from_config(pipe.scheduler.config       )
											pipe.to(_UpperCAmelCase       )
											pipe.set_progress_bar_config(disable=_UpperCAmelCase       )
											pipe.enable_attention_slicing()

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

											assert image.shape == (1, 5_12, 20_48, 3)

											snake_case_					:		int 					=					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       ):
											snake_case_					:		List[str] 					=					0

											def callback_fn(lowercase__   ,		lowercase__   ,		lowercase__       ) -> None:
																snake_case_					:		List[Any] 					=					True
																nonlocal number_of_steps
																number_of_steps += 1
																if step == 1:
																					snake_case_					:		List[str] 					=					latents.detach().cpu().numpy()
																					assert latents.shape == (1, 4, 64, 2_56)
																					snake_case_					:		int 					=					latents[0, -3:, -3:, -1]

																					snake_case_					:		List[str] 					=					np.array(
																					    [
																					        0.18681869,
																					        0.33907816,
																					        0.5361276,
																					        0.14432865,
																					        -0.02856611,
																					        -0.73941123,
																					        0.23397987,
																					        0.47322682,
																					        -0.37823164,
																					    ]       )
																					assert np.abs(latents_slice.flatten() - expected_slice       ).max() < 5e-2
																elif step == 2:
																					snake_case_					:		str 					=					latents.detach().cpu().numpy()
																					assert latents.shape == (1, 4, 64, 2_56)
																					snake_case_					:		str 					=					latents[0, -3:, -3:, -1]

																					snake_case_					:		Union[str, Any] 					=					np.array(
																					    [
																					        0.18539645,
																					        0.33987248,
																					        0.5378559,
																					        0.14437142,
																					        -0.02455261,
																					        -0.7338317,
																					        0.23990755,
																					        0.47356272,
																					        -0.3786505,
																					    ]       )

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

											snake_case_					:		Union[str, Any] 					=					False

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

											snake_case_					:		Optional[Any] 					=					self.get_inputs()
											pipe(**_UpperCAmelCase   ,		callback=_UpperCAmelCase   ,		callback_steps=1       )
											assert callback_fn.has_been_called
											assert number_of_steps == 3







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

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

											snake_case_					:		List[str] 					=					self.get_inputs()
											snake_case_					:		List[Any] 					=					pipe(**_UpperCAmelCase       )

											snake_case_					:		str 					=					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 collections import OrderedDict
from typing import TYPE_CHECKING, Any, Mapping, Optional, Union

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


if TYPE_CHECKING:
			from ... import FeatureExtractionMixin, PreTrainedTokenizerBase, TensorType


a_			   =					logging.get_logger(__name__)

a_			   =					{
    '''microsoft/deberta-v2-xlarge''': '''https://huggingface.co/microsoft/deberta-v2-xlarge/resolve/main/config.json''',
    '''microsoft/deberta-v2-xxlarge''': '''https://huggingface.co/microsoft/deberta-v2-xxlarge/resolve/main/config.json''',
    '''microsoft/deberta-v2-xlarge-mnli''': (
        '''https://huggingface.co/microsoft/deberta-v2-xlarge-mnli/resolve/main/config.json'''
    ),
    '''microsoft/deberta-v2-xxlarge-mnli''': (
        '''https://huggingface.co/microsoft/deberta-v2-xxlarge-mnli/resolve/main/config.json'''
    ),
}







class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						_A		:							Dict							       =		"""deberta-v2"""







						def __init__(self   ,		lowercase__=12_81_00   ,		lowercase__=15_36   ,		lowercase__=24   ,		lowercase__=24   ,		lowercase__=61_44   ,		lowercase__="gelu"   ,		lowercase__=0.1   ,		lowercase__=0.1   ,		lowercase__=5_12   ,		lowercase__=0   ,		lowercase__=0.02   ,		lowercase__=1e-7   ,		lowercase__=False   ,		lowercase__=-1   ,		lowercase__=0   ,		lowercase__=True   ,		lowercase__=None   ,		lowercase__=0   ,		lowercase__="gelu"   ,		**lowercase__   ,		):
											super().__init__(**lowercase__       )

											snake_case_					:		Union[str, Any] 					=					hidden_size
											snake_case_					:		str 					=					num_hidden_layers
											snake_case_					:		Tuple 					=					num_attention_heads
											snake_case_					:		Dict 					=					intermediate_size
											snake_case_					:		Optional[int] 					=					hidden_act
											snake_case_					:		Union[str, Any] 					=					hidden_dropout_prob
											snake_case_					:		Any 					=					attention_probs_dropout_prob
											snake_case_					:		List[Any] 					=					max_position_embeddings
											snake_case_					:		Union[str, Any] 					=					type_vocab_size
											snake_case_					:		Union[str, Any] 					=					initializer_range
											snake_case_					:		List[Any] 					=					relative_attention
											snake_case_					:		Dict 					=					max_relative_positions
											snake_case_					:		Optional[int] 					=					pad_token_id
											snake_case_					:		List[str] 					=					position_biased_input

											# Backwards compatibility
											if type(lowercase__       ) == str:
																snake_case_					:		Union[str, Any] 					=					[x.strip() for x in pos_att_type.lower().split("""|"""       )]

											snake_case_					:		Optional[int] 					=					pos_att_type
											snake_case_					:		List[str] 					=					vocab_size
											snake_case_					:		Tuple 					=					layer_norm_eps

											snake_case_					:		List[Any] 					=					kwargs.get("""pooler_hidden_size"""   ,		lowercase__       )
											snake_case_					:		List[str] 					=					pooler_dropout
											snake_case_					:		int 					=					pooler_hidden_act







class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						@property
						def 						__UpperCamelCase							(self       ):
											if self.task == "multiple-choice":
																snake_case_					:		List[Any] 					=					{0: """batch""", 1: """choice""", 2: """sequence"""}
											else:
																snake_case_					:		int 					=					{0: """batch""", 1: """sequence"""}
											if self._config.type_vocab_size > 0:
																return OrderedDict(
																    [("""input_ids""", dynamic_axis), ("""attention_mask""", dynamic_axis), ("""token_type_ids""", dynamic_axis)]       )
											else:
																return OrderedDict([("""input_ids""", dynamic_axis), ("""attention_mask""", dynamic_axis)]       )







						@property
						def 						__UpperCamelCase							(self       ):
											return 12







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = -1   ,		lowercase__ = -1   ,		lowercase__ = -1   ,		lowercase__ = False   ,		lowercase__ = None   ,		lowercase__ = 3   ,		lowercase__ = 40   ,		lowercase__ = 40   ,		lowercase__ = None   ,		):
											snake_case_					:		str 					=					super().generate_dummy_inputs(preprocessor=lowercase__   ,		framework=lowercase__       )
											if self._config.type_vocab_size == 0 and "token_type_ids" in dummy_inputs:
																del dummy_inputs["token_type_ids"]
											return dummy_inputs




