

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     ):
				__SCREAMING_SNAKE_CASE   :					Dict			     =     StableDiffusionPanoramaPipeline
				__SCREAMING_SNAKE_CASE   :					Optional[Any]			     =     TEXT_TO_IMAGE_PARAMS
				__SCREAMING_SNAKE_CASE   :					str			     =     TEXT_TO_IMAGE_BATCH_PARAMS
				__SCREAMING_SNAKE_CASE   :					int			     =     TEXT_TO_IMAGE_IMAGE_PARAMS
				__SCREAMING_SNAKE_CASE   :					Dict			     =     TEXT_TO_IMAGE_IMAGE_PARAMS




				def        a				(   self			):
											torch.manual_seed(0			)
											snake_case_     							=       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_     							=       DDIMScheduler()
											torch.manual_seed(0			)
											snake_case_     							=       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_     							=       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 ,  )
											snake_case_     							=       CLIPTextModel(snake_case			)
											snake_case_     							=       CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'			)

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




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

											snake_case_     							=       self.get_dummy_inputs(snake_case			)
											snake_case_     							=       sd_pipe(**snake_case			).images
											snake_case_     							=       image[0, -3:, -3:, -1]
											assert image.shape == (1, 64, 64, 3)

											snake_case_     							=       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        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			):
											snake_case_     							=       'cpu'  # ensure determinism for the device-dependent torch.Generator
											snake_case_     							=       self.get_dummy_components()
											snake_case_     							=       StableDiffusionPanoramaPipeline(**snake_case			)
											snake_case_     							=       sd_pipe.to(snake_case			)
											sd_pipe.set_progress_bar_config(disable=snake_case			)

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

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

											snake_case_     							=       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        a				(   self			):
											snake_case_     							=       'cpu'  # ensure determinism for the device-dependent torch.Generator
											snake_case_     							=       self.get_dummy_components()
											snake_case_     							=       StableDiffusionPanoramaPipeline(**snake_case			)
											snake_case_     							=       sd_pipe.to(snake_case			)
											sd_pipe.set_progress_bar_config(disable=snake_case			)

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

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

											snake_case_     							=       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        a				(   self			):
											snake_case_     							=       'cpu'  # ensure determinism for the device-dependent torch.Generator
											snake_case_     							=       self.get_dummy_components()
											snake_case_     							=       EulerAncestralDiscreteScheduler(
											    beta_start=0.0_00_85 ,  beta_end=0.0_12 ,  beta_schedule='scaled_linear'			)
											snake_case_     							=       StableDiffusionPanoramaPipeline(**snake_case			)
											snake_case_     							=       sd_pipe.to(snake_case			)
											sd_pipe.set_progress_bar_config(disable=snake_case			)

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

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

											snake_case_     							=       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        a				(   self			):
											snake_case_     							=       'cpu'  # ensure determinism for the device-dependent torch.Generator
											snake_case_     							=       self.get_dummy_components()
											snake_case_     							=       PNDMScheduler(
											    beta_start=0.0_00_85 ,  beta_end=0.0_12 ,  beta_schedule='scaled_linear' ,  skip_prk_steps=snake_case			)
											snake_case_     							=       StableDiffusionPanoramaPipeline(**snake_case			)
											snake_case_     							=       sd_pipe.to(snake_case			)
											sd_pipe.set_progress_bar_config(disable=snake_case			)

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

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

											snake_case_     							=       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 					lowercase							(			unittest.TestCase     ):




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




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




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

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

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

											snake_case_     							=       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        a				(   self			):
											snake_case_     							=       StableDiffusionPanoramaPipeline.from_pretrained(
											    'stabilityai/stable-diffusion-2-base' ,  safety_checker=snake_case			)
											snake_case_     							=       LMSDiscreteScheduler.from_config(pipe.scheduler.config			)
											pipe.to(snake_case			)
											pipe.set_progress_bar_config(disable=snake_case			)
											pipe.enable_attention_slicing()

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

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

											snake_case_     							=       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			):
											snake_case_     							=       0

											def callback_fn(snake_case ,  snake_case ,  snake_case			) -> None:
																		snake_case_     							=       True
																		nonlocal number_of_steps
																		number_of_steps += 1
																		if step == 1:
																									snake_case_     							=       latents.detach().cpu().numpy()
																									assert latents.shape == (1, 4, 64, 256)
																									snake_case_     							=       latents[0, -3:, -3:, -1]

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

																									snake_case_     							=       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_     							=       False

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

											snake_case_     							=       self.get_inputs()
											pipe(**snake_case ,  callback=snake_case ,  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()

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

											snake_case_     							=       self.get_inputs()
											snake_case_     							=       pipe(**snake_case			)

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







import tempfile
import unittest

from make_student import create_student_by_copying_alternating_layers

from transformers import AutoConfig
from transformers.file_utils import cached_property
from transformers.testing_utils import require_torch


_UpperCAmelCase						:  Dict									=			"""sshleifer/bart-tiny-random"""
_UpperCAmelCase						:  Optional[Any]									=			"""patrickvonplaten/t5-tiny-random"""







@require_torch
class 					lowercase							(			unittest.TestCase     ):




				@cached_property
				def        a				(   self			):
											return AutoConfig.from_pretrained(snake_case			)




				def        a				(   self			):
											snake_case_							,       *snake_case_     							=       create_student_by_copying_alternating_layers(snake_case ,  tempfile.mkdtemp() ,  e=1 ,  d=1			)
											self.assertEqual(student.config.num_hidden_layers ,  1			)




				def        a				(   self			):
											snake_case_							,       *snake_case_     							=       create_student_by_copying_alternating_layers(snake_case ,  tempfile.mkdtemp() ,  e=1 ,  d=snake_case			)




				def        a				(   self			):
											snake_case_							,       *snake_case_     							=       create_student_by_copying_alternating_layers(snake_case ,  tempfile.mkdtemp() ,  e=1 ,  d=snake_case			)
											self.assertEqual(student.config.encoder_layers ,  1			)
											self.assertEqual(student.config.decoder_layers ,  self.teacher_config.encoder_layers			)




				def        a				(   self			):
											snake_case_							,       *snake_case_     							=       create_student_by_copying_alternating_layers(snake_case ,  tempfile.mkdtemp() ,  e=1 ,  d=1			)
											self.assertEqual(student.config.encoder_layers ,  1			)
											self.assertEqual(student.config.decoder_layers ,  1			)




				def        a				(   self			):
											with self.assertRaises(snake_case			):
																		create_student_by_copying_alternating_layers(snake_case ,  tempfile.mkdtemp() ,  e=snake_case ,  d=snake_case			)




