




import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionModelEditingPipeline,
    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 PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





@skip_mps
class 		A__				(   SCREAMING_SNAKE_CASE__		,     SCREAMING_SNAKE_CASE__		,     SCREAMING_SNAKE_CASE__		,     unittest.TestCase ):
	lowerCamelCase__		: Any				    =StableDiffusionModelEditingPipeline
	lowerCamelCase__		: Union[str, Any]				    =TEXT_TO_IMAGE_PARAMS
	lowerCamelCase__		: Dict				    =TEXT_TO_IMAGE_BATCH_PARAMS
	lowerCamelCase__		: Tuple				    =TEXT_TO_IMAGE_IMAGE_PARAMS
	lowerCamelCase__		: Optional[int]				    =TEXT_TO_IMAGE_IMAGE_PARAMS






	def  lowercase      (	self						)      ->   List[Any]:




						"""simple docstring"""






						torch.manual_seed(0						)
						__magic_name__					:							Optional[int]		   =							UNetaDConditionModel(
						    block_out_channels=(32, 64)				,     layers_per_block=2				,     sample_size=32				,     in_channels=4				,     out_channels=4				,     down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')				,     up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')				,     cross_attention_dim=32				,     )
						__magic_name__					:							Any		   =							DDIMScheduler()
						torch.manual_seed(0						)
						__magic_name__					:							Any		   =							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						)
						__magic_name__					:							List[str]		   =							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				,     )
						__magic_name__					:							Optional[Any]		   =							CLIPTextModel(snake_case__						)
						__magic_name__					:							int		   =							CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''						)

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






	def  lowercase      (	self				,     lowerCamelCase				,     lowerCamelCase=0						)      ->   Optional[int]:




						"""simple docstring"""






						__magic_name__					:							Optional[int]		   =							torch.manual_seed(snake_case__						)
						__magic_name__					:							Tuple		   =							{
						    "prompt": "A field of roses",
						    "generator": generator,
						    # Setting height and width to None to prevent OOMs on CPU.
						    "height": None,
						    "width": None,
						    "num_inference_steps": 2,
						    "guidance_scale": 6.0,
						    "output_type": "numpy",
						}
						return inputs






	def  lowercase      (	self						)      ->   List[str]:




						"""simple docstring"""






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

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

						__magic_name__					:							List[str]		   =							np.array([0.4_7_5_5, 0.5_1_3_2, 0.4_9_7_6, 0.3_9_0_4, 0.3_5_5_4, 0.4_7_6_5, 0.5_1_3_9, 0.5_1_5_8, 0.4_8_8_9]						)

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






	def  lowercase      (	self						)      ->   int:




						"""simple docstring"""






						__magic_name__					:							Optional[int]		   =							"cpu"  # ensure determinism for the device-dependent torch.Generator
						__magic_name__					:							str		   =							self.get_dummy_components()
						__magic_name__					:							Tuple		   =							StableDiffusionModelEditingPipeline(**snake_case__						)
						__magic_name__					:							Union[str, Any]		   =							sd_pipe.to(snake_case__						)
						sd_pipe.set_progress_bar_config(disable=snake_case__						)

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

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

						__magic_name__					:							str		   =							np.array([0.4_9_9_2, 0.5_1_0_1, 0.5_0_0_4, 0.3_9_4_9, 0.3_6_0_4, 0.4_7_3_5, 0.5_2_1_6, 0.5_2_0_4, 0.4_9_1_3]						)

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






	def  lowercase      (	self						)      ->   List[str]:




						"""simple docstring"""






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

						__magic_name__					:							Dict		   =							self.get_dummy_inputs(snake_case__						)
						__magic_name__					:							Optional[Any]		   =							sd_pipe(**snake_case__						).images
						__magic_name__					:							Any		   =							image[0, -3:, -3:, -1]

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

						__magic_name__					:							Any		   =							np.array([0.4_7_4_7, 0.5_3_7_2, 0.4_7_7_9, 0.4_9_8_2, 0.5_5_4_3, 0.4_8_1_6, 0.5_2_3_8, 0.4_9_0_4, 0.5_0_2_7]						)

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






	def  lowercase      (	self						)      ->   int:




						"""simple docstring"""






						__magic_name__					:							Union[str, Any]		   =							"cpu"  # ensure determinism for the device-dependent torch.Generator
						__magic_name__					:							Dict		   =							self.get_dummy_components()
						__magic_name__					:							int		   =							PNDMScheduler()
						__magic_name__					:							Optional[Any]		   =							StableDiffusionModelEditingPipeline(**snake_case__						)
						__magic_name__					:							Tuple		   =							sd_pipe.to(snake_case__						)
						sd_pipe.set_progress_bar_config(disable=snake_case__						)

						__magic_name__					:							List[Any]		   =							self.get_dummy_inputs(snake_case__						)
						# the pipeline does not expect pndm so test if it raises error.
						with self.assertRaises(snake_case__						):
											__magic_name__					:							List[str]		   =							sd_pipe(**snake_case__						).images






	def  lowercase      (	self						)      ->   Union[str, Any]:




						"""simple docstring"""






						super().test_inference_batch_single_identical(expected_max_diff=5e-3						)






	def  lowercase      (	self						)      ->   List[str]:




						"""simple docstring"""






						super().test_attention_slicing_forward_pass(expected_max_diff=5e-3						)





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






	def  lowercase      (	self						)      ->   Optional[int]:




						"""simple docstring"""






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






	def  lowercase      (	self				,     lowerCamelCase=0						)      ->   Tuple:




						"""simple docstring"""






						__magic_name__					:							Dict		   =							torch.manual_seed(snake_case__						)
						__magic_name__					:							Union[str, Any]		   =							{
						    "prompt": "A field of roses",
						    "generator": generator,
						    "num_inference_steps": 3,
						    "guidance_scale": 7.5,
						    "output_type": "numpy",
						}
						return inputs






	def  lowercase      (	self						)      ->   str:




						"""simple docstring"""






						__magic_name__					:							Optional[int]		   =							"CompVis/stable-diffusion-v1-4"
						__magic_name__					:							int		   =							StableDiffusionModelEditingPipeline.from_pretrained(snake_case__				,     safety_checker=snake_case__						)
						pipe.to(snake_case__						)
						pipe.set_progress_bar_config(disable=snake_case__						)
						pipe.enable_attention_slicing()

						__magic_name__					:							List[Any]		   =							self.get_inputs()
						__magic_name__					:							int		   =							pipe(**snake_case__						).images
						__magic_name__					:							Optional[Any]		   =							image[0, -3:, -3:, -1].flatten()

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

						__magic_name__					:							Tuple		   =							np.array(
						    [0.6_7_4_9_4_9_6, 0.6_3_8_6_4_5_3, 0.5_1_4_4_3_2_6_7, 0.6_6_0_9_4_9_0_5, 0.6_1_9_2_1_2_1_5, 0.5_4_9_1_3_3_2, 0.5_7_4_4_4_1_7, 0.5_8_0_7_5_1_0_6, 0.5_1_7_4_6_5_8]						)

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

						# make sure image changes after editing
						pipe.edit_model('''A pack of roses'''				,     '''A pack of blue roses'''						)

						__magic_name__					:							int		   =							pipe(**snake_case__						).images
						__magic_name__					:							int		   =							image[0, -3:, -3:, -1].flatten()

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

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






	def  lowercase      (	self						)      ->   Union[str, Any]:




						"""simple docstring"""






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

						__magic_name__					:							Optional[Any]		   =							"CompVis/stable-diffusion-v1-4"
						__magic_name__					:							str		   =							DDIMScheduler.from_pretrained(snake_case__				,     subfolder='''scheduler'''						)
						__magic_name__					:							List[str]		   =							StableDiffusionModelEditingPipeline.from_pretrained(
						    snake_case__				,     scheduler=snake_case__				,     safety_checker=snake_case__						)
						__magic_name__					:							int		   =							pipe.to(snake_case__						)
						pipe.set_progress_bar_config(disable=snake_case__						)
						pipe.enable_attention_slicing(1						)
						pipe.enable_sequential_cpu_offload()

						__magic_name__					:							List[str]		   =							self.get_inputs()
						__magic_name__					:							Optional[int]		   =							pipe(**snake_case__						)

						__magic_name__					:							int		   =							torch.cuda.max_memory_allocated()
						# make sure that less than 4.4 GB is allocated
						assert mem_bytes < 4.4 * 10**9











"""simple docstring"""







import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device


if is_torch_available():
			import torch

			from transformers import XLMProphetNetForConditionalGeneration, XLMProphetNetTokenizer


@require_torch
class     __UpperCAmelCase(   unittest.TestCase							):




						"""simple docstring"""




						@slow
						def 		UpperCAmelCase_       (   self			):
												'''simple docstring'''
												lowercase__				:      Optional[int]=       XLMProphetNetForConditionalGeneration.from_pretrained("microsoft/xprophetnet-large-wiki100-cased"			)
												model.to(snake_case__			)

												# encoder-decoder outputs
												lowercase__				:      Dict=       torch.tensor([[17, 96208, 103471, 2]]			).to(snake_case__			)
												lowercase__				:      Optional[Any]=       torch.tensor(
												    [[2, 250, 9953, 34, 69489, 1620, 32, 118424, 624, 210, 105, 2913, 1032, 351]]			).to(snake_case__			)
												lowercase__				:      Dict=       model(
												    input_ids=snake_case__       ,  attention_mask=snake_case__       ,  encoder_outputs=snake_case__       ,  decoder_input_ids=snake_case__			)
												lowercase__				:      int=       output[0]
												lowercase__				:      str=       torch.Size((1, 14, 250012)			)
												self.assertEqual(output_predited_logis.shape       ,  snake_case__			)
												lowercase__				:      Union[str, Any]=       torch.tensor(
												    [[[-6.39_86, -8.23_91, 12.51_89], [-6.32_89, -8.08_64, 12.62_11], [-6.24_18, -8.04_45, 12.79_68]]]			).to(snake_case__			)
												self.assertTrue(torch.allclose(output_predited_logis[:, :3, :3]       ,  snake_case__       ,  atol=1e-4			)			)

												# encoder outputs
												lowercase__				:      Tuple=       model.prophetnet.encoder(snake_case__			)[0]
												lowercase__				:      Union[str, Any]=       torch.tensor(
												    [[[-1.42_60, -0.76_28, 0.84_53], [-1.47_19, -0.13_91, 0.78_07], [-1.76_78, 0.01_14, 0.46_46]]]			).to(snake_case__			)
												lowercase__				:      List[Any]=       torch.Size((1, 4, 1024)			)
												self.assertEqual(encoder_outputs.shape       ,  snake_case__			)
												self.assertTrue(torch.allclose(encoder_outputs[:, :3, :3]       ,  snake_case__       ,  atol=1e-4			)			)

												# decoder outputs
												lowercase__				:      int=       model.prophetnet.decoder(
												    snake_case__       ,  encoder_hidden_states=snake_case__       ,  )
												lowercase__				:      Union[str, Any]=       decoder_outputs[1].view(1       ,  model.config.ngram       ,  14       ,  -1			)
												lowercase__				:      List[Any]=       model.lm_head(snake_case__			)
												lowercase__				:      str=       predicting_streams_logits[:, 0]
												self.assertTrue(torch.allclose(next_first_stream_logits[:, :3, :3]       ,  snake_case__       ,  atol=1e-4			)			)




						@slow
						def 		UpperCAmelCase_       (   self			):
												'''simple docstring'''
												lowercase__				:      Union[str, Any]=       XLMProphetNetForConditionalGeneration.from_pretrained(
												    "microsoft/xprophetnet-large-wiki100-cased-xglue-ntg"			)
												model.to(snake_case__			)

												lowercase__				:      Dict=       torch.tensor([[17, 96208, 103471, 2]]			).to(snake_case__			)
												lowercase__				:      List[str]=       torch.tensor(
												    [[2, 250, 9953, 34, 69489, 1620, 32, 118424, 624, 210, 105, 2913, 1032, 351]]			).to(snake_case__			)
												lowercase__				:      List[Any]=       model(
												    input_ids=snake_case__       ,  attention_mask=snake_case__       ,  encoder_outputs=snake_case__       ,  decoder_input_ids=snake_case__			)
												lowercase__				:      str=       output[0]
												lowercase__				:      Optional[Any]=       torch.Size((1, 14, 250012)			)
												self.assertEqual(output_predited_logis.shape       ,  snake_case__			)
												# compare the actual values for a slice.
												lowercase__				:      str=       torch.tensor(
												    [[[-9.22_53, -9.71_73, -6.35_29], [-7.67_01, -9.01_45, -1.93_82], [-8.01_95, -7.00_04, -0.15_23]]]			).to(snake_case__			)

												self.assertTrue(torch.allclose(output_predited_logis[:, :3, :3]       ,  snake_case__       ,  atol=1e-4			)			)




						@slow
						def 		UpperCAmelCase_       (   self			):
												'''simple docstring'''
												lowercase__				:      Tuple=       XLMProphetNetForConditionalGeneration.from_pretrained(
												    "microsoft/xprophetnet-large-wiki100-cased-xglue-ntg"			)
												model.to(snake_case__			)
												lowercase__				:      Optional[int]=       512

												lowercase__				:      int=       XLMProphetNetTokenizer.from_pretrained("microsoft/xprophetnet-large-wiki100-cased-xglue-ntg"			)

												lowercase__				:      str=       (
												    "Microsoft Corporation intends to officially end free support for the Windows 7 operating system after"
												    " January 14, 2020, according to the official portal of the organization. From that day, users of this"
												    " system will not be able to receive security updates, which could make their computers vulnerable to"
												    " cyber attacks."
												)
												lowercase__				:      Tuple=       (
												    "орпорация Microsoft намерена официально прекратить бесплатную поддержку операционной системы Windows 7"
												    " после 14 января 2020 года, сообщается на официальном портале организации . С указанного дня пользователи"
												    " этой системы не смогут получать обновления безопасности, из-за чего их компьютеры могут стать уязвимыми"
												    " к кибератакам."
												)
												lowercase__				:      Optional[int]=       (
												    "根据该组织的官方门户网站，微软公司打算在2020年1月14日之后正式终止对Windows 7操作系统的免费支持。从那时起，该系统的用户将无法接收安全更新，这可能会使他们的计算机容易受到网络攻击。"
												)

												lowercase__				:      Dict=       tokenizer(
												    [EN_SENTENCE, RU_SENTENCE, ZH_SENTENCE]       ,  padding=snake_case__       ,  max_length=255       ,  return_tensors="pt"			).input_ids
												lowercase__				:      Any=       input_ids.to(snake_case__			)

												lowercase__				:      List[Any]=       model.generate(
												    snake_case__       ,  num_beams=10       ,  length_penalty=1.0       ,  no_repeat_ngram_size=3       ,  early_stopping=snake_case__			)
												lowercase__				:      str=       [tokenizer.decode(snake_case__       ,  skip_special_tokens=snake_case__			) for g in summary_ids]
												lowercase__				:      Dict=       "Microsoft to end Windows 7 free support after January 14, 2020"
												lowercase__				:      int=       "Microsoft намерена прекратить бесплатную поддержку Windows 7 после 14 января 2020 года"
												lowercase__				:      List[str]=       "微软打算终止对Windows 7操作系统的免费支持"
												self.assertListEqual(
												    [EXPECTED_TITLE_EN, EXPECTED_TITLE_RU, EXPECTED_TITLE_ZH]       ,  snake_case__       ,  )

												lowercase__				:      List[str]=       model.generate(
												    snake_case__       ,  num_beams=1       ,  length_penalty=1.0       ,  no_repeat_ngram_size=3       ,  early_stopping=snake_case__			)
												lowercase__				:      Optional[Any]=       [
												    tokenizer.convert_ids_to_tokens(snake_case__       ,  skip_special_tokens=snake_case__			) for g in summary_ids_beama
												]
												lowercase__				:      List[str]=       "▁Microsoft ▁to ▁end ▁free ▁support ▁for ▁Windows ▁7".split(" "			)
												lowercase__				:      List[Any]=       "▁Microsoft ▁намерен а ▁прекрати ть ▁бес плат ную ▁поддержку ▁Windows ▁7 ▁после ▁14 ▁января ▁2020 ▁года".split(
												    " "			)
												lowercase__				:      int=       "微软 公司 打算 终止 对 Windows ▁7 操作 系统的 免费 支持".split(" "			)
												self.assertListEqual(
												    [EXPECTED_TITLE_EN_BEAM1_TOK, EXPECTED_TITLE_RU_BEAM1_TOK, EXPECTED_TITLE_ZH_BEAM1_TOK]       ,  snake_case__       ,  )






