

'''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 	SCREAMING_SNAKE_CASE__			(				__A,__A,unittest.TestCase      ):
    _A			    =							StableDiffusionParadigmsPipeline
    _A			    =							TEXT_TO_IMAGE_PARAMS
    _A			    =							TEXT_TO_IMAGE_BATCH_PARAMS
    _A			    =							TEXT_TO_IMAGE_IMAGE_PARAMS
    _A			    =							TEXT_TO_IMAGE_IMAGE_PARAMS




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          torch.manual_seed(0			)
          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       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=lowerCamelCase__				,     )
          SCREAMING_SNAKE_CASE_      :			Any     =       DDIMParallelScheduler(
              beta_start=0.00085				,     beta_end=0.012				,     beta_schedule="scaled_linear"				,     clip_sample=lowerCamelCase__				,     set_alpha_to_one=lowerCamelCase__				,     )
          torch.manual_seed(0			)
          SCREAMING_SNAKE_CASE_      :			Optional[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				,     sample_size=128				,     )
          torch.manual_seed(0			)
          SCREAMING_SNAKE_CASE_      :			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				,     hidden_act="gelu"				,     projection_dim=512				,     )
          SCREAMING_SNAKE_CASE_      :			int     =       CLIPTextModel(lowerCamelCase__			)
          SCREAMING_SNAKE_CASE_      :			Dict     =       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"			)

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




    def   __lowerCamelCase      (					self				,     lowercase__				,     lowercase__=0			):






          """simple docstring"""






          if str(lowerCamelCase__			).startswith("mps"			):
                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       torch.manual_seed(lowerCamelCase__			)
          else:
                SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       torch.Generator(device=lowerCamelCase__			).manual_seed(lowerCamelCase__			)
          SCREAMING_SNAKE_CASE_      :			List[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			):






          """simple docstring"""






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

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

          SCREAMING_SNAKE_CASE_      :			List[str]     =       np.array([0.4773, 0.5417, 0.4723, 0.4925, 0.5631, 0.4752, 0.5240, 0.4935, 0.5023]			)

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




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Tuple     =       "cpu"  # ensure determinism for the device-dependent torch.Generator
          SCREAMING_SNAKE_CASE_      :			Any     =       self.get_dummy_components()
          torch.manual_seed(0			)
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       DDPMParallelScheduler()
          torch.manual_seed(0			)
          SCREAMING_SNAKE_CASE_      :			Tuple     =       StableDiffusionParadigmsPipeline(**lowerCamelCase__			)
          SCREAMING_SNAKE_CASE_      :			int     =       sd_pipe.to(lowerCamelCase__			)
          sd_pipe.set_progress_bar_config(disable=lowerCamelCase__			)

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

          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       np.array([0.3573, 0.4420, 0.4960, 0.4799, 0.3796, 0.3879, 0.4819, 0.4365, 0.4468]			)

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




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          super().test_inference_batch_consistent(batch_sizes=[1, 2]			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          super().test_inference_batch_single_identical(batch_size=2				,     expected_max_diff=3e-3			)







    def   __lowerCamelCase      (					self			):






          """simple docstring"""






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

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

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

          SCREAMING_SNAKE_CASE_      :			str     =       np.array([0.4771, 0.5420, 0.4683, 0.4918, 0.5636, 0.4725, 0.5230, 0.4923, 0.5015]			)

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






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




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






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




    def   __lowerCamelCase      (					self				,     lowercase__=0			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			List[str]     =       torch.Generator(device=lowerCamelCase__			).manual_seed(lowerCamelCase__			)
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       {
              "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			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Tuple     =       "stabilityai/stable-diffusion-2-base"
          SCREAMING_SNAKE_CASE_      :			Tuple     =       DDIMParallelScheduler.from_pretrained(lowerCamelCase__				,     subfolder="scheduler"			)
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       StableDiffusionParadigmsPipeline.from_pretrained(lowerCamelCase__				,     scheduler=lowerCamelCase__				,     safety_checker=lowerCamelCase__			)
          pipe.to(lowerCamelCase__			)
          pipe.set_progress_bar_config(disable=lowerCamelCase__			)
          pipe.enable_attention_slicing()

          SCREAMING_SNAKE_CASE_      :			Dict     =       self.get_inputs()
          SCREAMING_SNAKE_CASE_      :			Tuple     =       pipe(**lowerCamelCase__			).images
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       image[0, -3:, -3:, -1].flatten()

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

          SCREAMING_SNAKE_CASE_      :			List[Any]     =       np.array([0.9622, 0.9602, 0.9748, 0.9591, 0.9630, 0.9691, 0.9661, 0.9631, 0.9741]			)

          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


if is_torch_available():
  import torch

if is_vision_available():
  from PIL import Image

  from transformers import BlipImageProcessor






class        SCREAMING_SNAKE_CASE(				unittest.TestCase     ):

       def __init__(    self					,		lowerCamelCase__					,		lowerCamelCase__=7					,		lowerCamelCase__=3					,		lowerCamelCase__=18					,		lowerCamelCase__=30					,		lowerCamelCase__=400					,		lowerCamelCase__=True					,		lowerCamelCase__=None					,		lowerCamelCase__=True					,		lowerCamelCase__=False					,		lowerCamelCase__=[0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73]					,		lowerCamelCase__=[0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11]					,		lowerCamelCase__=True					,		)    ->					Union[str, Any]:

           """simple docstring"""



           __lowercase							=						size if size is not None else {"""height""": 20, """width""": 20}
           __lowercase							=						parent
           __lowercase							=						batch_size
           __lowercase							=						num_channels
           __lowercase							=						image_size
           __lowercase							=						min_resolution
           __lowercase							=						max_resolution
           __lowercase							=						do_resize
           __lowercase							=						size
           __lowercase							=						do_normalize
           __lowercase							=						image_mean
           __lowercase							=						image_std
           __lowercase							=						do_pad
           __lowercase							=						do_convert_rgb

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

           """simple docstring"""



           return {
               "do_resize": self.do_resize,
               "size": self.size,
               "do_normalize": self.do_normalize,
               "image_mean": self.image_mean,
               "image_std": self.image_std,
               "do_convert_rgb": self.do_convert_rgb,
               "do_pad": self.do_pad,
           }




       def     snake_case__			(    self					,		lowerCamelCase__=False					,		lowerCamelCase__=False					,		lowerCamelCase__=False     )    ->					Tuple:

           """simple docstring"""



           assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

           if equal_resolution:
               __lowercase							=						[]
               for i in range(self.batch_size     ):
                   image_inputs.append(
                       np.random.randint(
                           255					,		size=(self.num_channels, self.max_resolution, self.max_resolution)					,		dtype=np.uinta     )     )
           else:
               __lowercase							=						[]
               for i in range(self.batch_size     ):
                   __lowercase				,__lowercase							=						np.random.choice(np.arange(self.min_resolution					,		self.max_resolution     )					,		2     )
                   image_inputs.append(np.random.randint(255					,		size=(self.num_channels, width, height)					,		dtype=np.uinta     )     )

           if not numpify and not torchify:
               # PIL expects the channel dimension as last dimension
               __lowercase							=						[Image.fromarray(np.moveaxis(lowerCamelCase__					,		0					,		-1     )     ) for x in image_inputs]

           if torchify:
               __lowercase							=						[torch.from_numpy(lowerCamelCase__     ) for x in image_inputs]

           return image_inputs






@require_torch
@require_vision
class        SCREAMING_SNAKE_CASE(				__A   ,      unittest.TestCase     ):
       snake_case_ :     Optional[int]    =					BlipImageProcessor if is_vision_available() else None

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

           """simple docstring"""



           __lowercase							=						BlipImageProcessingTester(self     )

       @property
       def     snake_case__			(    self     )    ->					Optional[int]:

           """simple docstring"""



           return self.image_processor_tester.prepare_image_processor_dict()

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

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_resize"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""size"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_normalize"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""image_mean"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""image_std"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_convert_rgb"""     )     )

       def     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           pass

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

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           # create random PIL images
           __lowercase							=						self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__     )
           for image in image_inputs:
               self.assertIsInstance(lowerCamelCase__					,		Image.Image     )

           # Test not batched input
           __lowercase							=						image_processor(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
           __lowercase							=						image_processor(lowerCamelCase__					,		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     snake_case__			(    self     )    ->					int:

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           # create random numpy tensors
           __lowercase							=						self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__					,		numpify=lowerCamelCase__     )
           for image in image_inputs:
               self.assertIsInstance(lowerCamelCase__					,		np.ndarray     )

           # Test not batched input
           __lowercase							=						image_processor(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
           __lowercase							=						image_processor(lowerCamelCase__					,		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     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           # create random PyTorch tensors
           __lowercase							=						self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__					,		torchify=lowerCamelCase__     )
           for image in image_inputs:
               self.assertIsInstance(lowerCamelCase__					,		torch.Tensor     )

           # Test not batched input
           __lowercase							=						image_processor(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
           __lowercase							=						image_processor(lowerCamelCase__					,		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"""],
               )					,		)






@require_torch
@require_vision
class        SCREAMING_SNAKE_CASE(				__A   ,      unittest.TestCase     ):
       snake_case_ :     Union[str, Any]    =					BlipImageProcessor if is_vision_available() else None

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

           """simple docstring"""



           __lowercase							=						BlipImageProcessingTester(self					,		num_channels=4     )
           __lowercase							=						3

       @property
       def     snake_case__			(    self     )    ->					str:

           """simple docstring"""



           return self.image_processor_tester.prepare_image_processor_dict()

       def     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_resize"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""size"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_normalize"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""image_mean"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""image_std"""     )     )
           self.assertTrue(hasattr(lowerCamelCase__					,		"""do_convert_rgb"""     )     )

       def     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           pass




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

           """simple docstring"""



           __lowercase							=						self.image_processing_class(**self.image_processor_dict     )
           # create random PIL images
           __lowercase							=						self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__     )
           for image in image_inputs:
               self.assertIsInstance(lowerCamelCase__					,		Image.Image     )

           # Test not batched input
           __lowercase							=						image_processor(image_inputs[0]					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   1,
                   self.expected_encoded_image_num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)

           # Test batched
           __lowercase							=						image_processor(lowerCamelCase__					,		return_tensors="""pt"""     ).pixel_values
           self.assertEqual(
               encoded_images.shape					,		(
                   self.image_processor_tester.batch_size,
                   self.expected_encoded_image_num_channels,
                   self.image_processor_tester.size["""height"""],
                   self.image_processor_tester.size["""width"""],
               )					,		)




