



"""simple docstring"""
import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

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






					'''simple docstring'''





					_A		:      Dict						=					StableDiffusionParadigmsPipeline
					_A		:      Dict						=					TEXT_TO_IMAGE_PARAMS
					_A		:      List[Any]						=					TEXT_TO_IMAGE_BATCH_PARAMS
					_A		:      List[Any]						=					TEXT_TO_IMAGE_IMAGE_PARAMS
					_A		:      Dict						=					TEXT_TO_IMAGE_IMAGE_PARAMS






					def        lowerCAmelCase_    (	self:							Optional[Any]						)       ->     Optional[Any]:
											torch.manual_seed(0						)
											snake_case_			: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						,					attention_head_dim=(2, 4)						,					use_linear_projection=snake_case						,					)
											snake_case_			:Optional[Any]						=      DDIMParallelScheduler(
											    beta_start=0.0_0_0_8_5						,					beta_end=0.0_1_2						,					beta_schedule="""scaled_linear"""						,					clip_sample=snake_case						,					set_alpha_to_one=snake_case						,					)
											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						,					sample_size=128						,					)
											torch.manual_seed(0						)
											snake_case_			:Union[str, 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=1_000						,					hidden_act="""gelu"""						,					projection_dim=512						,					)
											snake_case_			:Optional[Any]						=      CLIPTextModel(snake_case						)
											snake_case_			:Optional[Any]						=      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""						)

											snake_case_			:Optional[int]						=      {
											    """unet""": unet,
											    """scheduler""": scheduler,
											    """vae""": vae,
											    """text_encoder""": text_encoder,
											    """tokenizer""": tokenizer,
											    """safety_checker""": None,
											    """feature_extractor""": None,
											}
											return components






					def        lowerCAmelCase_    (	self:							int						,					snake_case:							Union[str, Any]						,					snake_case:							Union[str, Any]=0						)       ->     List[Any]:
											if str(snake_case						).startswith("""mps"""						):
																	snake_case_			:Tuple						=      torch.manual_seed(snake_case						)
											else:
																	snake_case_			:Any						=      torch.Generator(device=snake_case						).manual_seed(snake_case						)
											snake_case_			:Union[str, Any]						=      {
											    """prompt""": """a photograph of an astronaut riding a horse""",
											    """generator""": generator,
											    """num_inference_steps""": 10,
											    """guidance_scale""": 6.0,
											    """output_type""": """numpy""",
											    """parallel""": 3,
											    """debug""": True,
											}
											return inputs






					def        lowerCAmelCase_    (	self:							Optional[Any]						)       ->     Optional[int]:
											snake_case_			:Tuple						=      """cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_			:Any						=      self.get_dummy_components()
											snake_case_			:Union[str, Any]						=      StableDiffusionParadigmsPipeline(**snake_case						)
											snake_case_			:Any						=      sd_pipe.to(snake_case						)
											sd_pipe.set_progress_bar_config(disable=snake_case						)

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

											snake_case_			:Optional[int]						=      np.array([0.4_7_7_3, 0.5_4_1_7, 0.4_7_2_3, 0.4_9_2_5, 0.5_6_3_1, 0.4_7_5_2, 0.5_2_4_0, 0.4_9_3_5, 0.5_0_2_3]						)

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






					def        lowerCAmelCase_    (	self:							int						)       ->     Optional[int]:
											snake_case_			:Tuple						=      """cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_			:Tuple						=      self.get_dummy_components()
											torch.manual_seed(0						)
											snake_case_			:Tuple						=      DDPMParallelScheduler()
											torch.manual_seed(0						)
											snake_case_			:Optional[int]						=      StableDiffusionParadigmsPipeline(**snake_case						)
											snake_case_			:List[Any]						=      sd_pipe.to(snake_case						)
											sd_pipe.set_progress_bar_config(disable=snake_case						)

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

											snake_case_			:str						=      np.array([0.3_5_7_3, 0.4_4_2_0, 0.4_9_6_0, 0.4_7_9_9, 0.3_7_9_6, 0.3_8_7_9, 0.4_8_1_9, 0.4_3_6_5, 0.4_4_6_8]						)

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






					def        lowerCAmelCase_    (	self:							Any						)       ->     Optional[int]:
											super().test_inference_batch_consistent(batch_sizes=[1, 2]						)






					def        lowerCAmelCase_    (	self:							Optional[int]						)       ->     Optional[int]:
											super().test_inference_batch_single_identical(batch_size=2						,					expected_max_diff=3E-3						)






					def        lowerCAmelCase_    (	self:							int						)       ->     Optional[Any]:
											snake_case_			:str						=      """cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_			:List[str]						=      self.get_dummy_components()
											snake_case_			:Dict						=      StableDiffusionParadigmsPipeline(**snake_case						)
											snake_case_			:Optional[int]						=      sd_pipe.to(snake_case						)
											sd_pipe.set_progress_bar_config(disable=snake_case						)

											snake_case_			:Optional[int]						=      self.get_dummy_inputs(snake_case						)
											snake_case_			:List[str]						=      """french fries"""
											snake_case_			:Any						=      sd_pipe(**snake_case						,					negative_prompt=snake_case						)
											snake_case_			:Union[str, 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.4_7_7_1, 0.5_4_2_0, 0.4_6_8_3, 0.4_9_1_8, 0.5_6_3_6, 0.4_7_2_5, 0.5_2_3_0, 0.4_9_2_3, 0.5_0_1_5]						)

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

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






					'''simple docstring'''





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






					def        lowerCAmelCase_    (	self:							Any						,					snake_case:							Optional[int]=0						)       ->     Tuple:
											snake_case_			:Dict						=      torch.Generator(device=snake_case						).manual_seed(snake_case						)
											snake_case_			:Tuple						=      {
											    """prompt""": """a photograph of an astronaut riding a horse""",
											    """generator""": generator,
											    """num_inference_steps""": 10,
											    """guidance_scale""": 7.5,
											    """output_type""": """numpy""",
											    """parallel""": 3,
											    """debug""": True,
											}
											return inputs






					def        lowerCAmelCase_    (	self:							str						)       ->     Any:
											snake_case_			:Any						=      """stabilityai/stable-diffusion-2-base"""
											snake_case_			:Optional[Any]						=      DDIMParallelScheduler.from_pretrained(snake_case						,					subfolder="""scheduler"""						)
											snake_case_			:Tuple						=      StableDiffusionParadigmsPipeline.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_			:int						=      self.get_inputs()
											snake_case_			:List[str]						=      pipe(**snake_case						).images
											snake_case_			:Optional[int]						=      image[0, -3:, -3:, -1].flatten()

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

											snake_case_			:Tuple						=      np.array([0.9_6_2_2, 0.9_6_0_2, 0.9_7_4_8, 0.9_5_9_1, 0.9_6_3_0, 0.9_6_9_1, 0.9_6_6_1, 0.9_6_3_1, 0.9_7_4_1]						)

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





"""simple docstring"""
import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
						import torch

if is_vision_available():
						from PIL import Image

						from transformers import ViTImageProcessor
class     lowerCamelCase  (      unittest.TestCase    ):






					'''simple docstring'''





					def __init__(	self:							int						,					snake_case:							List[str]						,					snake_case:							List[str]=7						,					snake_case:							Union[str, Any]=3						,					snake_case:							List[str]=18						,					snake_case:							Any=30						,					snake_case:							Any=400						,					snake_case:							Any=True						,					snake_case:							List[str]=None						,					snake_case:							Union[str, Any]=True						,					snake_case:							int=[0.5, 0.5, 0.5]						,					snake_case:							Tuple=[0.5, 0.5, 0.5]						,					)       ->     int:
											snake_case_			:Dict						=      size if size is not None else {"""height""": 18, """width""": 18}
											snake_case_			:Optional[Any]						=      parent
											snake_case_			:Tuple						=      batch_size
											snake_case_			:Tuple						=      num_channels
											snake_case_			:Optional[int]						=      image_size
											snake_case_			:Tuple						=      min_resolution
											snake_case_			:str						=      max_resolution
											snake_case_			:Any						=      do_resize
											snake_case_			:Union[str, Any]						=      size
											snake_case_			:Tuple						=      do_normalize
											snake_case_			:int						=      image_mean
											snake_case_			:List[Any]						=      image_std






					def        lowerCAmelCase_    (	self:							List[Any]						)       ->     List[Any]:
											return {
											    "image_mean": self.image_mean,
											    "image_std": self.image_std,
											    "do_normalize": self.do_normalize,
											    "do_resize": self.do_resize,
											    "size": self.size,
											}

@require_torch
@require_vision
class     lowerCamelCase  (      _lowerCAmelCase   ,						unittest.TestCase    ):






					'''simple docstring'''





					_A		:      Tuple						=					ViTImageProcessor if is_vision_available() else None






					def        lowerCAmelCase_    (	self:							str						)       ->     Optional[Any]:
											snake_case_			:Dict						=      ViTImageProcessingTester(self						)






					@property
					def        lowerCAmelCase_    (	self:							Tuple						)       ->     Optional[Any]:
											return self.image_processor_tester.prepare_image_processor_dict()






					def        lowerCAmelCase_    (	self:							str						)       ->     List[str]:
											snake_case_			:List[str]						=      self.image_processing_class(**self.image_processor_dict						)
											self.assertTrue(hasattr(snake_case						,					"""image_mean"""						)						)
											self.assertTrue(hasattr(snake_case						,					"""image_std"""						)						)
											self.assertTrue(hasattr(snake_case						,					"""do_normalize"""						)						)
											self.assertTrue(hasattr(snake_case						,					"""do_resize"""						)						)
											self.assertTrue(hasattr(snake_case						,					"""size"""						)						)






					def        lowerCAmelCase_    (	self:							Dict						)       ->     Union[str, Any]:
											snake_case_			:int						=      self.image_processing_class.from_dict(self.image_processor_dict						)
											self.assertEqual(image_processor.size						,					{"""height""": 18, """width""": 18}						)

											snake_case_			:int						=      self.image_processing_class.from_dict(self.image_processor_dict						,					size=42						)
											self.assertEqual(image_processor.size						,					{"""height""": 42, """width""": 42}						)






					def        lowerCAmelCase_    (	self:							Tuple						)       ->     int:
											pass






					def        lowerCAmelCase_    (	self:							Dict						)       ->     Union[str, Any]:
											# Initialize image_processing
											snake_case_			:Any						=      self.image_processing_class(**self.image_processor_dict						)
											# create random PIL images
											snake_case_			:str						=      prepare_image_inputs(self.image_processor_tester						,					equal_resolution=snake_case						)
											for image in image_inputs:
																	self.assertIsInstance(snake_case						,					Image.Image						)

											# Test not batched input
											snake_case_			:List[str]						=      image_processing(image_inputs[0]						,					return_tensors="""pt"""						).pixel_values
											self.assertEqual(
											    encoded_images.shape						,					(
											        1,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)

											# Test batched
											snake_case_			:Dict						=      image_processing(snake_case						,					return_tensors="""pt"""						).pixel_values
											self.assertEqual(
											    encoded_images.shape						,					(
											        self.image_processor_tester.batch_size,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)






					def        lowerCAmelCase_    (	self:							Optional[Any]						)       ->     Tuple:
											# Initialize image_processing
											snake_case_			:int						=      self.image_processing_class(**self.image_processor_dict						)
											# create random numpy tensors
											snake_case_			:str						=      prepare_image_inputs(self.image_processor_tester						,					equal_resolution=snake_case						,					numpify=snake_case						)
											for image in image_inputs:
																	self.assertIsInstance(snake_case						,					np.ndarray						)

											# Test not batched input
											snake_case_			:Any						=      image_processing(image_inputs[0]						,					return_tensors="""pt"""						).pixel_values
											self.assertEqual(
											    encoded_images.shape						,					(
											        1,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)

											# Test batched
											snake_case_			:Any						=      image_processing(snake_case						,					return_tensors="""pt"""						).pixel_values
											self.assertEqual(
											    encoded_images.shape						,					(
											        self.image_processor_tester.batch_size,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)






					def        lowerCAmelCase_    (	self:							List[Any]						)       ->     Optional[int]:
											# Initialize image_processing
											snake_case_			:Union[str, Any]						=      self.image_processing_class(**self.image_processor_dict						)
											# create random PyTorch tensors
											snake_case_			:List[Any]						=      prepare_image_inputs(self.image_processor_tester						,					equal_resolution=snake_case						,					torchify=snake_case						)
											for image in image_inputs:
																	self.assertIsInstance(snake_case						,					torch.Tensor						)

											# Test not batched input
											snake_case_			:Optional[Any]						=      image_processing(image_inputs[0]						,					return_tensors="""pt"""						).pixel_values
											self.assertEqual(
											    encoded_images.shape						,					(
											        1,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)

											# Test batched
											snake_case_			:List[Any]						=      image_processing(snake_case						,					return_tensors="""pt"""						).pixel_values
											self.assertEqual(
											    encoded_images.shape						,					(
											        self.image_processor_tester.batch_size,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)
