

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):
 _lowerCAmelCase									=	StableDiffusionParadigmsPipeline
 _lowerCAmelCase									=	TEXT_TO_IMAGE_PARAMS
 _lowerCAmelCase									=	TEXT_TO_IMAGE_BATCH_PARAMS
 _lowerCAmelCase									=	TEXT_TO_IMAGE_IMAGE_PARAMS
 _lowerCAmelCase									=	TEXT_TO_IMAGE_IMAGE_PARAMS




 def     a       (			self							):
   torch.manual_seed(0							)
   __UpperCAmelCase	:  Dict		=     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=_UpperCAmelCase  ,  )
   __UpperCAmelCase	:  List[str]		=     DDIMParallelScheduler(
       beta_start=0.0_0_0_8_5  ,  beta_end=0.0_1_2  ,  beta_schedule='scaled_linear'  ,  clip_sample=_UpperCAmelCase  ,  set_alpha_to_one=_UpperCAmelCase  ,  )
   torch.manual_seed(0							)
   __UpperCAmelCase	:  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							)
   __UpperCAmelCase	:  Optional[int]		=     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  ,  )
   __UpperCAmelCase	:  Optional[int]		=     CLIPTextModel(_UpperCAmelCase							)
   __UpperCAmelCase	:  int		=     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'							)

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




 def     a       (			self  ,  _UpperCAmelCase  ,  _UpperCAmelCase=0							):
   if str(_UpperCAmelCase							).startswith('mps'							):
     __UpperCAmelCase	:  Optional[int]		=     torch.manual_seed(_UpperCAmelCase							)
   else:
     __UpperCAmelCase	:  str		=     torch.Generator(device=_UpperCAmelCase							).manual_seed(_UpperCAmelCase							)
   __UpperCAmelCase	:  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     a       (			self							):
   __UpperCAmelCase	:  Optional[int]		=     'cpu'  # ensure determinism for the device-dependent torch.Generator
   __UpperCAmelCase	:  List[str]		=     self.get_dummy_components()
   __UpperCAmelCase	:  List[Any]		=     StableDiffusionParadigmsPipeline(**_UpperCAmelCase							)
   __UpperCAmelCase	:  List[str]		=     sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  Optional[int]		=     self.get_dummy_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  List[str]		=     sd_pipe(**_UpperCAmelCase							).images
   __UpperCAmelCase	:  List[str]		=     image[0, -3:, -3:, -1]
   assert image.shape == (1, 64, 64, 3)

   __UpperCAmelCase	:  Optional[Any]		=     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     a       (			self							):
   __UpperCAmelCase	:  List[str]		=     'cpu'  # ensure determinism for the device-dependent torch.Generator
   __UpperCAmelCase	:  Any		=     self.get_dummy_components()
   torch.manual_seed(0							)
   __UpperCAmelCase	:  List[Any]		=     DDPMParallelScheduler()
   torch.manual_seed(0							)
   __UpperCAmelCase	:  Dict		=     StableDiffusionParadigmsPipeline(**_UpperCAmelCase							)
   __UpperCAmelCase	:  Tuple		=     sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

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

   __UpperCAmelCase	:  List[Any]		=     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     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=3E-3							)




 def     a       (			self							):
   __UpperCAmelCase	:  Optional[Any]		=     'cpu'  # ensure determinism for the device-dependent torch.Generator
   __UpperCAmelCase	:  List[Any]		=     self.get_dummy_components()
   __UpperCAmelCase	:  List[Any]		=     StableDiffusionParadigmsPipeline(**_UpperCAmelCase							)
   __UpperCAmelCase	:  Optional[int]		=     sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  List[str]		=     self.get_dummy_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  int		=     'french fries'
   __UpperCAmelCase	:  str		=     sd_pipe(**_UpperCAmelCase  ,  negative_prompt=_UpperCAmelCase							)
   __UpperCAmelCase	:  Union[str, Any]		=     output.images
   __UpperCAmelCase	:  Optional[int]		=     image[0, -3:, -3:, -1]

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

   __UpperCAmelCase	:  Optional[Any]		=     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):




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




 def     a       (			self  ,  _UpperCAmelCase=0							):
   __UpperCAmelCase	:  List[Any]		=     torch.Generator(device=_UpperCAmelCase							).manual_seed(_UpperCAmelCase							)
   __UpperCAmelCase	:  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     a       (			self							):
   __UpperCAmelCase	:  str		=     'stabilityai/stable-diffusion-2-base'
   __UpperCAmelCase	:  Optional[int]		=     DDIMParallelScheduler.from_pretrained(_UpperCAmelCase  ,  subfolder='scheduler'							)
   __UpperCAmelCase	:  str		=     StableDiffusionParadigmsPipeline.from_pretrained(_UpperCAmelCase  ,  scheduler=_UpperCAmelCase  ,  safety_checker=_UpperCAmelCase							)
   pipe.to(_UpperCAmelCase							)
   pipe.set_progress_bar_config(disable=_UpperCAmelCase							)
   pipe.enable_attention_slicing()

   __UpperCAmelCase	:  List[str]		=     self.get_inputs()
   __UpperCAmelCase	:  Dict		=     pipe(**_UpperCAmelCase							).images
   __UpperCAmelCase	:  Union[str, Any]		=     image[0, -3:, -3:, -1].flatten()

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

   __UpperCAmelCase	:  Any		=     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







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 (
    get_resize_output_image_size,
    normalize,
    pad,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    get_image_size,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging
from ...utils.import_utils import is_vision_available


_UpperCAmelCase    			=					logging.get_logger(__name__)


if is_vision_available():
  import PIL

class 		__magic_name__				( lowercase_					):


  """simple docstring"""





  _UpperCamelCase						 = ["pixel_values"]


  def __init__(					self	,		a__ = True	,		a__ = None	,		a__ = PILImageResampling.BILINEAR	,		a__ = True	,		a__ = False	,		a__ = True	,		a__ = True	,		a__ = 1 / 2_55	,		a__ = True	,		a__ = None	,		a__ = None	,		**a__	,		):
    super().__init__(**a__		)

    _lowerCamelCase      =						size if size is not None else {'''height''': 25_60, '''width''': 19_20}
    if isinstance(a__	,		(tuple, list)		):
      # The previous feature extractor size parameter was in (width, height) format
      _lowerCamelCase      =						size[::-1]
    _lowerCamelCase      =						get_size_dict(a__		)

    _lowerCamelCase      =						do_resize
    _lowerCamelCase      =						size
    _lowerCamelCase      =						resample
    _lowerCamelCase      =						do_thumbnail
    _lowerCamelCase      =						do_align_long_axis
    _lowerCamelCase      =						do_pad
    _lowerCamelCase      =						do_rescale
    _lowerCamelCase      =						rescale_factor
    _lowerCamelCase      =						do_normalize
    _lowerCamelCase      =						image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
    _lowerCamelCase      =						image_std if image_std is not None else IMAGENET_STANDARD_STD


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__ = None		):
    _lowerCamelCase    ,    _lowerCamelCase      =						get_image_size(a__		)
    _lowerCamelCase    ,    _lowerCamelCase      =						size['''height'''], size['''width''']

    if (output_width < output_height and input_width > input_height) or (
        output_width > output_height and input_width < input_height
    ):
      _lowerCamelCase      =						np.rotaa(a__	,		3		)

    if data_format is not None:
      _lowerCamelCase      =						to_channel_dimension_format(a__	,		a__		)

    return image


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__ = False	,		a__ = None	,		):
    _lowerCamelCase    ,    _lowerCamelCase      =						size['''height'''], size['''width''']
    _lowerCamelCase    ,    _lowerCamelCase      =						get_image_size(a__		)

    _lowerCamelCase      =						output_width - input_width
    _lowerCamelCase      =						output_height - input_height

    if random_padding:
      _lowerCamelCase      =						np.random.randint(low=0	,		high=delta_height + 1		)
      _lowerCamelCase      =						np.random.randint(low=0	,		high=delta_width + 1		)
    else:
      _lowerCamelCase      =						delta_height // 2
      _lowerCamelCase      =						delta_width // 2

    _lowerCamelCase      =						delta_height - pad_top
    _lowerCamelCase      =						delta_width - pad_left

    _lowerCamelCase      =						((pad_top, pad_bottom), (pad_left, pad_right))
    return pad(a__	,		a__	,		data_format=a__		)


  def     _UpperCAmelCase  (					self	,		*a__	,		**a__		):
    logger.info('''pad is deprecated and will be removed in version 4.27. Please use pad_image instead.'''		)
    return self.pad_image(*a__	,		**a__		)


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__ = PILImageResampling.BICUBIC	,		a__ = None	,		**a__	,		):
    _lowerCamelCase    ,    _lowerCamelCase      =						get_image_size(a__		)
    _lowerCamelCase    ,    _lowerCamelCase      =						size['''height'''], size['''width''']

    # We always resize to the smallest of either the input or output size.
    _lowerCamelCase      =						min(a__	,		a__		)
    _lowerCamelCase      =						min(a__	,		a__		)

    if height == input_height and width == input_width:
      return image

    if input_height > input_width:
      _lowerCamelCase      =						int(input_width * height / input_height		)
    elif input_width > input_height:
      _lowerCamelCase      =						int(input_height * width / input_width		)

    return resize(
        a__	,		size=(height, width)	,		resample=a__	,		reducing_gap=2.0	,		data_format=a__	,		**a__		)


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__ = PILImageResampling.BICUBIC	,		a__ = None	,		**a__	,		):
    _lowerCamelCase      =						get_size_dict(a__		)
    _lowerCamelCase      =						min(size['''height''']	,		size['''width''']		)
    _lowerCamelCase      =						get_resize_output_image_size(a__	,		size=a__	,		default_to_square=a__		)
    _lowerCamelCase      =						resize(a__	,		size=a__	,		resample=a__	,		data_format=a__	,		**a__		)
    return resized_image


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__ = None	,		**a__	,		):
    return rescale(a__	,		scale=a__	,		data_format=a__	,		**a__		)


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__	,		a__ = None	,		**a__	,		):
    return normalize(a__	,		mean=a__	,		std=a__	,		data_format=a__	,		**a__		)


  def     _UpperCAmelCase  (					self	,		a__	,		a__ = None	,		a__ = None	,		a__ = None	,		a__ = None	,		a__ = None	,		a__ = None	,		a__ = False	,		a__ = None	,		a__ = None	,		a__ = None	,		a__ = None	,		a__ = None	,		a__ = None	,		a__ = ChannelDimension.FIRST	,		**a__	,		):
    _lowerCamelCase      =						do_resize if do_resize is not None else self.do_resize
    _lowerCamelCase      =						size if size is not None else self.size
    if isinstance(a__	,		(tuple, list)		):
      # Previous feature extractor had size in (width, height) format
      _lowerCamelCase      =						size[::-1]
    _lowerCamelCase      =						get_size_dict(a__		)
    _lowerCamelCase      =						resample if resample is not None else self.resample
    _lowerCamelCase      =						do_thumbnail if do_thumbnail is not None else self.do_thumbnail
    _lowerCamelCase      =						do_align_long_axis if do_align_long_axis is not None else self.do_align_long_axis
    _lowerCamelCase      =						do_pad if do_pad is not None else self.do_pad
    _lowerCamelCase      =						do_rescale if do_rescale is not None else self.do_rescale
    _lowerCamelCase      =						rescale_factor if rescale_factor is not None else self.rescale_factor
    _lowerCamelCase      =						do_normalize if do_normalize is not None else self.do_normalize
    _lowerCamelCase      =						image_mean if image_mean is not None else self.image_mean
    _lowerCamelCase      =						image_std if image_std is not None else self.image_std

    _lowerCamelCase      =						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:
      raise ValueError('''Size must be specified if do_resize is True.'''		)

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

    if do_pad and size is None:
      raise ValueError('''Size must be specified if do_pad 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.
    _lowerCamelCase      =						[to_numpy_array(a__		) for image in images]

    if do_align_long_axis:
      _lowerCamelCase      =						[self.align_long_axis(a__	,		size=a__		) for image in images]

    if do_resize:
      _lowerCamelCase      =						[self.resize(image=a__	,		size=a__	,		resample=a__		) for image in images]

    if do_thumbnail:
      _lowerCamelCase      =						[self.thumbnail(image=a__	,		size=a__		) for image in images]

    if do_pad:
      _lowerCamelCase      =						[self.pad_image(image=a__	,		size=a__	,		random_padding=a__		) for image in images]

    if do_rescale:
      _lowerCamelCase      =						[self.rescale(image=a__	,		scale=a__		) for image in images]

    if do_normalize:
      _lowerCamelCase      =						[self.normalize(image=a__	,		mean=a__	,		std=a__		) for image in images]

    _lowerCamelCase      =						[to_channel_dimension_format(a__	,		a__		) for image in images]

    _lowerCamelCase      =						{'''pixel_values''': images}
    return BatchFeature(data=a__	,		tensor_type=a__		)




