


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




							"""simple docstring"""


							__UpperCAmelCase					:							List[str]  						=						StableDiffusionPanoramaPipeline
							__UpperCAmelCase					:							Union[str, Any]  						=						TEXT_TO_IMAGE_PARAMS
							__UpperCAmelCase					:							Union[str, Any]  						=						TEXT_TO_IMAGE_BATCH_PARAMS
							__UpperCAmelCase					:							int  						=						TEXT_TO_IMAGE_IMAGE_PARAMS
							__UpperCAmelCase					:							Tuple  						=						TEXT_TO_IMAGE_IMAGE_PARAMS

							def 					__snake_case      (							self :						Optional[int]       )			->					List[str]:
														torch.manual_seed(0       )
														__snake_case	:    int											=     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	:    Dict											=     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	:    List[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					,			)
														__snake_case	:    Dict											=     CLIPTextModel(lowerCamelCase       )
														__snake_case	:    Union[str, Any]											=     CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"       )

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

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

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

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

							def 					__snake_case      (							self :						str       )			->					int:
														super().test_inference_batch_single_identical(batch_size=2					,			expected_max_diff=3.25E-3       )

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

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

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

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

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

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

														__snake_case	:    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 					__snake_case      (							self :						Dict       )			->					int:
														__snake_case	:    str											=     "cpu"  # ensure determinism for the device-dependent torch.Generator
														__snake_case	:    List[str]											=     self.get_dummy_components()
														__snake_case	:    Any											=     EulerAncestralDiscreteScheduler(
														    beta_start=0.0_00_85					,			beta_end=0.0_12					,			beta_schedule="scaled_linear"       )
														__snake_case	:    List[Any]											=     StableDiffusionPanoramaPipeline(**lowerCamelCase       )
														__snake_case	:    List[Any]											=     sd_pipe.to(lowerCamelCase       )
														sd_pipe.set_progress_bar_config(disable=lowerCamelCase       )

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

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

														__snake_case	:    List[Any]											=     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 					__snake_case      (							self :						Optional[int]       )			->					Tuple:
														__snake_case	:    Dict											=     "cpu"  # ensure determinism for the device-dependent torch.Generator
														__snake_case	:    List[str]											=     self.get_dummy_components()
														__snake_case	:    Tuple											=     PNDMScheduler(
														    beta_start=0.0_00_85					,			beta_end=0.0_12					,			beta_schedule="scaled_linear"					,			skip_prk_steps=lowerCamelCase       )
														__snake_case	:    Dict											=     StableDiffusionPanoramaPipeline(**lowerCamelCase       )
														__snake_case	:    List[Any]											=     sd_pipe.to(lowerCamelCase       )
														sd_pipe.set_progress_bar_config(disable=lowerCamelCase       )

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

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

														__snake_case	:    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   ):




							"""simple docstring"""


							def 					__snake_case      (							self :						Union[str, Any]       )			->					str:
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()

							def 					__snake_case      (							self :						int					,			lowerCamelCase :						Optional[Any]=0       )			->					List[Any]:
														__snake_case	:    Any											=     torch.manual_seed(lowerCamelCase       )
														__snake_case	:    Optional[Any]											=     {
														    "prompt": "a photo of the dolomites",
														    "generator": generator,
														    "num_inference_steps": 3,
														    "guidance_scale": 7.5,
														    "output_type": "numpy",
														}
														return inputs

							def 					__snake_case      (							self :						int       )			->					List[Any]:
														__snake_case	:    str											=     "stabilityai/stable-diffusion-2-base"
														__snake_case	:    Any											=     DDIMScheduler.from_pretrained(lowerCamelCase					,			subfolder="scheduler"       )
														__snake_case	:    Dict											=     StableDiffusionPanoramaPipeline.from_pretrained(lowerCamelCase					,			scheduler=lowerCamelCase					,			safety_checker=lowerCamelCase       )
														pipe.to(lowerCamelCase       )
														pipe.set_progress_bar_config(disable=lowerCamelCase       )
														pipe.enable_attention_slicing()

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

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

														__snake_case	:    int											=     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 					__snake_case      (							self :						Tuple       )			->					Union[str, Any]:
														__snake_case	:    Optional[int]											=     StableDiffusionPanoramaPipeline.from_pretrained(
														    "stabilityai/stable-diffusion-2-base"					,			safety_checker=lowerCamelCase       )
														__snake_case	:    Dict											=     LMSDiscreteScheduler.from_config(pipe.scheduler.config       )
														pipe.to(lowerCamelCase       )
														pipe.set_progress_bar_config(disable=lowerCamelCase       )
														pipe.enable_attention_slicing()

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

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

														__snake_case	:    Tuple											=     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 					__snake_case      (							self :						Any       )			->					Optional[int]:
														__snake_case	:    Union[str, Any]											=     0

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

																												__snake_case	:    Tuple											=     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:
																												__snake_case	:    int											=     latents.detach().cpu().numpy()
																												assert latents.shape == (1, 4, 64, 256)
																												__snake_case	:    Dict											=     latents[0, -3:, -3:, -1]

																												__snake_case	:    List[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

														__snake_case	:    Dict											=     False

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

														__snake_case	:    Dict											=     self.get_inputs()
														pipe(**lowerCamelCase					,			callback=lowerCamelCase					,			callback_steps=1       )
														assert callback_fn.has_been_called
														assert number_of_steps == 3






							def 					__snake_case      (							self :						Any       )			->					Any:
														torch.cuda.empty_cache()
														torch.cuda.reset_max_memory_allocated()
														torch.cuda.reset_peak_memory_stats()

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

														__snake_case	:    int											=     self.get_inputs()
														__snake_case	:    Any											=     pipe(**lowerCamelCase       )

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





import warnings

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_snake_case  :	Optional[Any]			      =       logging.get_logger(__name__)

_snake_case  :	str			      =       {
    "xlnet-base-cased": "https://huggingface.co/xlnet-base-cased/resolve/main/config.json",
    "xlnet-large-cased": "https://huggingface.co/xlnet-large-cased/resolve/main/config.json",
}

class        a	(_lowerCAmelCase   ):




							"""simple docstring"""


							__UpperCAmelCase					:							List[Any]  						=						"xlnet"
							__UpperCAmelCase					:							Dict  						=						["mems"]
							__UpperCAmelCase					:							Tuple  						=						{
							    "n_token": "vocab_size",  # Backward compatibility
							    "hidden_size": "d_model",
							    "num_attention_heads": "n_head",
							    "num_hidden_layers": "n_layer",
							}

							def __init__(							self :						List[Any]					,			lowerCamelCase :						Optional[int]=32000					,			lowerCamelCase :						str=1024					,			lowerCamelCase :						int=24					,			lowerCamelCase :						List[Any]=16					,			lowerCamelCase :						Any=4096					,			lowerCamelCase :						Optional[Any]="gelu"					,			lowerCamelCase :						Optional[Any]=True					,			lowerCamelCase :						Optional[int]="bi"					,			lowerCamelCase :						Union[str, Any]=0.02					,			lowerCamelCase :						Union[str, Any]=1E-12					,			lowerCamelCase :						Dict=0.1					,			lowerCamelCase :						int=512					,			lowerCamelCase :						str=None					,			lowerCamelCase :						Dict=True					,			lowerCamelCase :						str=False					,			lowerCamelCase :						Any=False					,			lowerCamelCase :						Optional[Any]=-1					,			lowerCamelCase :						List[Any]=False					,			lowerCamelCase :						Optional[int]="last"					,			lowerCamelCase :						Dict=True					,			lowerCamelCase :						Union[str, Any]="tanh"					,			lowerCamelCase :						int=0.1					,			lowerCamelCase :						Any=5					,			lowerCamelCase :						List[str]=5					,			lowerCamelCase :						List[Any]=5					,			lowerCamelCase :						Tuple=1					,			lowerCamelCase :						Dict=2					,			**lowerCamelCase :						List[str]					,			)			->					List[Any]:
														__snake_case	:    Optional[Any]											=     vocab_size
														__snake_case	:    Any											=     d_model
														__snake_case	:    int											=     n_layer
														__snake_case	:    Optional[int]											=     n_head
														if d_model % n_head != 0:
																					raise ValueError(F'\'d_model % n_head\' ({d_model % n_head}) should be equal to 0'       )
														if "d_head" in kwargs:
																					if kwargs["d_head"] != d_model // n_head:
																												raise ValueError(
																												    F'`d_head` ({kwargs["d_head"]}) should be equal to `d_model // n_head` ({d_model // n_head})'       )
														__snake_case	:    Any											=     d_model // n_head
														__snake_case	:    Tuple											=     ff_activation
														__snake_case	:    int											=     d_inner
														__snake_case	:    Optional[Any]											=     untie_r
														__snake_case	:    Optional[Any]											=     attn_type

														__snake_case	:    List[str]											=     initializer_range
														__snake_case	:    Tuple											=     layer_norm_eps

														__snake_case	:    str											=     dropout
														__snake_case	:    Any											=     mem_len
														__snake_case	:    Tuple											=     reuse_len
														__snake_case	:    Union[str, Any]											=     bi_data
														__snake_case	:    List[Any]											=     clamp_len
														__snake_case	:    Optional[int]											=     same_length

														__snake_case	:    List[str]											=     summary_type
														__snake_case	:    Any											=     summary_use_proj
														__snake_case	:    Any											=     summary_activation
														__snake_case	:    Optional[int]											=     summary_last_dropout
														__snake_case	:    Dict											=     start_n_top
														__snake_case	:    Optional[Any]											=     end_n_top

														__snake_case	:    List[str]											=     bos_token_id
														__snake_case	:    Any											=     pad_token_id
														__snake_case	:    Optional[Any]											=     eos_token_id

														if "use_cache" in kwargs:
																					warnings.warn(
																					    "The `use_cache` argument is deprecated and will be removed in a future version, use `use_mems_eval`"
																					    " instead."					,			lowerCamelCase					,			)
																					__snake_case	:    Any											=     kwargs["use_cache"]

														__snake_case	:    int											=     use_mems_eval
														__snake_case	:    Optional[int]											=     use_mems_train
														super().__init__(pad_token_id=lowerCamelCase					,			bos_token_id=lowerCamelCase					,			eos_token_id=lowerCamelCase					,			**lowerCamelCase       )

							@property
							def 					__snake_case      (							self :						Any       )			->					Optional[int]:
														logger.info(F'The model {self.model_type} is one of the few models that has no sequence length limit.'       )
														return -1






							@max_position_embeddings.setter
							def 					__snake_case      (							self :						Union[str, Any]					,			lowerCamelCase :						Any       )			->					Optional[Any]:
														# Message copied from Transformer-XL documentation
														raise NotImplementedError(
														    F'The model {self.model_type} is one of the few models that has no sequence length limit.'       )

