






'''simple docstring'''






import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




class a_    (    UpperCamelCase_		,      UpperCamelCase_		,      unittest.TestCase							):
   _snake_case        =							StableDiffusionXLImgaImgPipeline
   _snake_case        =							TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
   _snake_case        =							PipelineTesterMixin.required_optional_params - {"""latents"""}
   _snake_case        =							TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
   _snake_case        =							IMAGE_TO_IMAGE_IMAGE_PARAMS
   _snake_case        =							IMAGE_TO_IMAGE_IMAGE_PARAMS







   def     SCREAMING_SNAKE_CASE__					(self)			->      Any:
      """simple docstring"""






      torch.manual_seed(0)
      __snake_case				:      int     							= UNetaDConditionModel(
          block_out_channels=(3_2, 6_4)					,   layers_per_block=2					,   sample_size=3_2					,   in_channels=4					,   out_channels=4					,   down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')					,   up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')					,   attention_head_dim=(2, 4)					,   use_linear_projection=__a					,   addition_embed_type='text_time'					,   addition_time_embed_dim=8					,   transformer_layers_per_block=(1, 2)					,   projection_class_embeddings_input_dim=8_0					,   cross_attention_dim=6_4					,   )
      __snake_case				:      Any     							= EulerDiscreteScheduler(
          beta_start=0.00_085					,   beta_end=0.012					,   steps_offset=1					,   beta_schedule='scaled_linear'					,   timestep_spacing='leading'					,   )
      torch.manual_seed(0)
      __snake_case				:      List[str]     							= AutoencoderKL(
          block_out_channels=[3_2, 6_4]					,   in_channels=3					,   out_channels=3					,   down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']					,   up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']					,   latent_channels=4					,   sample_size=1_2_8					,   )
      torch.manual_seed(0)
      __snake_case				:      str     							= CLIPTextConfig(
          bos_token_id=0					,   eos_token_id=2					,   hidden_size=3_2					,   intermediate_size=3_7					,   layer_norm_eps=1E-05					,   num_attention_heads=4					,   num_hidden_layers=5					,   pad_token_id=1					,   vocab_size=1_0_0_0					,   hidden_act='gelu'					,   projection_dim=3_2					,   )
      __snake_case				:      Any     							= CLIPTextModel(__a)
      __snake_case				:      List[Any]     							= CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					,   local_files_only=__a)

      __snake_case				:      int     							= CLIPTextModelWithProjection(__a)
      __snake_case				:      Tuple     							= CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					,   local_files_only=__a)

      __snake_case				:      Union[str, Any]     							= {
          'unet': unet,
          'scheduler': scheduler,
          'vae': vae,
          'text_encoder': text_encoder,
          'tokenizer': tokenizer,
          'text_encoder_2': text_encoder_a,
          'tokenizer_2': tokenizer_a,
          # "safety_checker": None,
          # "feature_extractor": None,
      }
      return components







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a=0)			->      Tuple:
      """simple docstring"""






      __snake_case				:      Optional[Any]     							= floats_tensor((1, 3, 3_2, 3_2)					,   rng=random.Random(__a)).to(__a)
      __snake_case				:      Optional[int]     							= image / 2 + 0.5
      if str(__a).startswith('mps'):
         __snake_case				:      Dict     							= torch.manual_seed(__a)
      else:
         __snake_case				:      Dict     							= torch.Generator(device=__a).manual_seed(__a)
      __snake_case				:      List[Any]     							= {
          'prompt': 'A painting of a squirrel eating a burger',
          'image': image,
          'generator': generator,
          'num_inference_steps': 2,
          'guidance_scale': 5.0,
          'output_type': 'numpy',
          'strength': 0.75,
      }
      return inputs







   def     SCREAMING_SNAKE_CASE__					(self)			->      Any:
      """simple docstring"""






      __snake_case				:      str     							= 'cpu'  # ensure determinism for the device-dependent torch.Generator
      __snake_case				:      Optional[Any]     							= self.get_dummy_components()
      __snake_case				:      int     							= StableDiffusionXLImgaImgPipeline(**__a)
      __snake_case				:      Optional[int]     							= sd_pipe.to(__a)
      sd_pipe.set_progress_bar_config(disable=__a)

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

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

      __snake_case				:      Optional[int]     							= np.array([0.4_656, 0.4_840, 0.4_439, 0.6_698, 0.5_574, 0.4_524, 0.5_799, 0.5_943, 0.5_165])

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







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[Any]:
      """simple docstring"""






      super().test_attention_slicing_forward_pass(expected_max_diff=3E-3)







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[Any]:
      """simple docstring"""






      super().test_inference_batch_single_identical(expected_max_diff=3E-3)







   def     SCREAMING_SNAKE_CASE__					(self)			->      Optional[Any]:
      """simple docstring"""






      pass







   def     SCREAMING_SNAKE_CASE__					(self)			->      str:
      """simple docstring"""






      __snake_case				:      Optional[int]     							= self.get_dummy_components()
      __snake_case				:      Tuple     							= StableDiffusionXLImgaImgPipeline(**__a)
      __snake_case				:      Optional[int]     							= sd_pipe.to(__a)
      __snake_case				:      Optional[Any]     							= sd_pipe.to(__a)
      sd_pipe.set_progress_bar_config(disable=__a)

      # forward without prompt embeds
      __snake_case				:      Union[str, Any]     							= self.get_dummy_inputs(__a)
      __snake_case				:      int     							= 3 * ['this is a negative prompt']
      __snake_case				:      Optional[Any]     							= negative_prompt
      __snake_case				:      str     							= 3 * [inputs['prompt']]

      __snake_case				:      Dict     							= sd_pipe(**__a)
      __snake_case				:      Union[str, Any]     							= output.images[0, -3:, -3:, -1]

      # forward with prompt embeds
      __snake_case				:      int     							= self.get_dummy_inputs(__a)
      __snake_case				:      Tuple     							= 3 * ['this is a negative prompt']
      __snake_case				:      Union[str, Any]     							= 3 * [inputs.pop('prompt')]

      (
          (
          __snake_case
      )							,(
          __snake_case
      )							,(
          __snake_case
      )							,(
          __snake_case
      )							,
      )				:      Tuple     							= sd_pipe.encode_prompt(__a					,   negative_prompt=__a)

      __snake_case				:      str     							= sd_pipe(
          **__a					,   prompt_embeds=__a					,   negative_prompt_embeds=__a					,   pooled_prompt_embeds=__a					,   negative_pooled_prompt_embeds=__a					,   )
      __snake_case				:      str     							= output.images[0, -3:, -3:, -1]

      # make sure that it's equal
      assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()).max() < 1E-4




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







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[str]:
      """simple docstring"""






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







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a="cpu"					,   __a=torch.floataa					,   __a=0)			->      Any:
      """simple docstring"""






      __snake_case				:      int     							= torch.Generator(device=__a).manual_seed(__a)
      __snake_case				:      Tuple     							= np.random.RandomState(__a).standard_normal((1, 4, 6_4, 6_4))
      __snake_case				:      Tuple     							= torch.from_numpy(__a).to(device=__a					,   dtype=__a)
      __snake_case				:      Tuple     							= {
          'prompt': 'a photograph of an astronaut riding a horse',
          'latents': latents,
          'generator': generator,
          'num_inference_steps': 3,
          'guidance_scale': 7.5,
          'output_type': 'numpy',
      }
      return inputs







   def     SCREAMING_SNAKE_CASE__					(self)			->      Tuple:
      """simple docstring"""






      __snake_case				:      Optional[int]     							= DiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-base')
      pipe.to(__a)
      pipe.set_progress_bar_config(disable=__a)

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

      assert image.shape == (1, 5_1_2, 5_1_2, 3)
      __snake_case				:      Optional[int]     							= np.array([0.49_493, 0.47_896, 0.40_798, 0.54_214, 0.53_212, 0.48_202, 0.47_656, 0.46_329, 0.48_506])
      assert np.abs(image_slice - expected_slice).max() < 7E-3







'''simple docstring'''






from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
      import PIL


__A    	=   logging.get_logger(__name__)




class a_    (    UpperCamelCase_							):
   _snake_case        =							["""pixel_values"""]







   def __init__(self					,   __a = True					,   __a = None					,   __a = None					,   __a = PILImageResampling.BILINEAR					,   __a = True					,   __a = 1 / 2_5_5					,   __a = True					,   __a = None					,   __a = None					,   **__a					,   )			->      None:
      """simple docstring"""






      super().__init__(**__a)
      __snake_case				:      Tuple     							= size if size is not None else {'shortest_edge': 3_8_4}
      __snake_case				:      List[Any]     							= get_size_dict(__a					,   default_to_square=__a)

      __snake_case				:      int     							= do_resize
      __snake_case				:      List[str]     							= size
      # Default value set here for backwards compatibility where the value in config is None
      __snake_case				:      Any     							= crop_pct if crop_pct is not None else 2_2_4 / 2_5_6
      __snake_case				:      Tuple     							= resample
      __snake_case				:      Dict     							= do_rescale
      __snake_case				:      Any     							= rescale_factor
      __snake_case				:      str     							= do_normalize
      __snake_case				:      Union[str, Any]     							= image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
      __snake_case				:      Dict     							= image_std if image_std is not None else IMAGENET_STANDARD_STD







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a					,   __a					,   __a = PILImageResampling.BICUBIC					,   __a = None					,   **__a					,   )			->      np.ndarray:
      """simple docstring"""






      __snake_case				:      Dict     							= get_size_dict(__a					,   default_to_square=__a)
      if "shortest_edge" not in size:
         raise ValueError(F"""Size dictionary must contain 'shortest_edge' key. Got {size.keys()}""")
      __snake_case				:      List[str]     							= size['shortest_edge']

      if shortest_edge < 3_8_4:
         # maintain same ratio, resizing shortest edge to shortest_edge/crop_pct
         __snake_case				:      Any     							= int(shortest_edge / crop_pct)
         __snake_case				:      Any     							= get_resize_output_image_size(__a					,   size=__a					,   default_to_square=__a)
         __snake_case				:      int     							= resize(image=__a					,   size=__a					,   resample=__a					,   data_format=__a					,   **__a)
         # then crop to (shortest_edge, shortest_edge)
         return center_crop(image=__a					,   size=(shortest_edge, shortest_edge)					,   data_format=__a					,   **__a)
      else:
         # warping (no cropping) when evaluated at 384 or larger
         return resize(
             __a					,   size=(shortest_edge, shortest_edge)					,   resample=__a					,   data_format=__a					,   **__a)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a					,   __a = None					,   **__a					,   )			->      Any:
      """simple docstring"""






      return rescale(__a					,   scale=__a					,   data_format=__a					,   **__a)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a					,   __a					,   __a = None					,   **__a					,   )			->      np.ndarray:
      """simple docstring"""






      return normalize(__a					,   mean=__a					,   std=__a					,   data_format=__a					,   **__a)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   __a = ChannelDimension.FIRST					,   **__a					,   )			->      PIL.Image.Image:
      """simple docstring"""






      __snake_case				:      Optional[int]     							= do_resize if do_resize is not None else self.do_resize
      __snake_case				:      Dict     							= crop_pct if crop_pct is not None else self.crop_pct
      __snake_case				:      Tuple     							= resample if resample is not None else self.resample
      __snake_case				:      Any     							= do_rescale if do_rescale is not None else self.do_rescale
      __snake_case				:      Any     							= rescale_factor if rescale_factor is not None else self.rescale_factor
      __snake_case				:      Union[str, Any]     							= do_normalize if do_normalize is not None else self.do_normalize
      __snake_case				:      Optional[int]     							= image_mean if image_mean is not None else self.image_mean
      __snake_case				:      Optional[Any]     							= image_std if image_std is not None else self.image_std

      __snake_case				:      List[str]     							= size if size is not None else self.size
      __snake_case				:      Any     							= get_size_dict(__a					,   default_to_square=__a)

      __snake_case				:      Dict     							= make_list_of_images(__a)

      if not valid_images(__a):
         raise ValueError(
             'Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '
             'torch.Tensor, tf.Tensor or jax.ndarray.')

      if do_resize and size is None or resample is None:
         raise ValueError('Size and resample must be specified if do_resize is True.')

      if do_resize and size["shortest_edge"] < 3_8_4 and crop_pct is None:
         raise ValueError('crop_pct must be specified if size < 384.')

      if do_rescale and rescale_factor is None:
         raise ValueError('Rescale factor must be specified if do_rescale is True.')

      if do_normalize and (image_mean is None or image_std is None):
         raise ValueError('Image mean and std must be specified if do_normalize is True.')

      # All transformations expect numpy arrays.
      __snake_case				:      Tuple     							= [to_numpy_array(__a) for image in images]

      if do_resize:
         __snake_case				:      Optional[int]     							= [self.resize(image=__a					,   size=__a					,   crop_pct=__a					,   resample=__a) for image in images]

      if do_rescale:
         __snake_case				:      Optional[int]     							= [self.rescale(image=__a					,   scale=__a) for image in images]

      if do_normalize:
         __snake_case				:      Any     							= [self.normalize(image=__a					,   mean=__a					,   std=__a) for image in images]

      __snake_case				:      Dict     							= [to_channel_dimension_format(__a					,   __a) for image in images]

      __snake_case				:      Union[str, Any]     							= {'pixel_values': images}
      return BatchFeature(data=__a					,   tensor_type=__a)