



'''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 							UpperCAmelCase__	(     SCREAMING_SNAKE_CASE_				,      SCREAMING_SNAKE_CASE_				,      unittest.TestCase):
 __SCREAMING_SNAKE_CASE          =					StableDiffusionXLImgaImgPipeline
 __SCREAMING_SNAKE_CASE          =					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
 __SCREAMING_SNAKE_CASE          =					PipelineTesterMixin.required_optional_params - {'''latents'''}
 __SCREAMING_SNAKE_CASE          =					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 __SCREAMING_SNAKE_CASE          =					IMAGE_TO_IMAGE_IMAGE_PARAMS
 __SCREAMING_SNAKE_CASE          =					IMAGE_TO_IMAGE_IMAGE_PARAMS



 def  __lowerCamelCase					(	self      )	->				str:
  torch.manual_seed(0      )
  __UpperCamelCase					   =	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=__snake_case		,						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		,						)
  __UpperCamelCase					   =	EulerDiscreteScheduler(
      beta_start=0.00_085		,						beta_end=0.012		,						steps_offset=1		,						beta_schedule="""scaled_linear"""		,						timestep_spacing="""leading"""		,						)
  torch.manual_seed(0      )
  __UpperCamelCase					   =	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      )
  __UpperCamelCase					   =	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		,						)
  __UpperCamelCase					   =	CLIPTextModel(__snake_case      )
  __UpperCamelCase					   =	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		,						local_files_only=__snake_case      )

  __UpperCamelCase					   =	CLIPTextModelWithProjection(__snake_case      )
  __UpperCamelCase					   =	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		,						local_files_only=__snake_case      )

  __UpperCamelCase					   =	{
      """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  __lowerCamelCase					(	self		,						lowercase		,						lowercase=0      )	->				int:
  __UpperCamelCase					   =	floats_tensor((1, 3, 3_2, 3_2)		,						rng=random.Random(__snake_case      )      ).to(__snake_case      )
  __UpperCamelCase					   =	image / 2 + 0.5
  if str(__snake_case      ).startswith("""mps"""      ):
   __UpperCamelCase					   =	torch.manual_seed(__snake_case      )
  else:
   __UpperCamelCase					   =	torch.Generator(device=__snake_case      ).manual_seed(__snake_case      )
  __UpperCamelCase					   =	{
      """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  __lowerCamelCase					(	self      )	->				List[Any]:
  __UpperCamelCase					   =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
  __UpperCamelCase					   =	self.get_dummy_components()
  __UpperCamelCase					   =	StableDiffusionXLImgaImgPipeline(**__snake_case      )
  __UpperCamelCase					   =	sd_pipe.to(__snake_case      )
  sd_pipe.set_progress_bar_config(disable=__snake_case      )

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

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

  __UpperCamelCase					   =	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  __lowerCamelCase					(	self      )	->				Optional[Any]:
  super().test_attention_slicing_forward_pass(expected_max_diff=3E-3      )



 def  __lowerCamelCase					(	self      )	->				List[str]:
  super().test_inference_batch_single_identical(expected_max_diff=3E-3      )



 def  __lowerCamelCase					(	self      )	->				Optional[int]:
  pass




 def  __lowerCamelCase					(	self      )	->				int:
  __UpperCamelCase					   =	self.get_dummy_components()
  __UpperCamelCase					   =	StableDiffusionXLImgaImgPipeline(**__snake_case      )
  __UpperCamelCase					   =	sd_pipe.to(__snake_case      )
  __UpperCamelCase					   =	sd_pipe.to(__snake_case      )
  sd_pipe.set_progress_bar_config(disable=__snake_case      )

  # forward without prompt embeds
  __UpperCamelCase					   =	self.get_dummy_inputs(__snake_case      )
  __UpperCamelCase					   =	3 * ["""this is a negative prompt"""]
  __UpperCamelCase					   =	negative_prompt
  __UpperCamelCase					   =	3 * [inputs["""prompt"""]]

  __UpperCamelCase					   =	sd_pipe(**__snake_case      )
  __UpperCamelCase					   =	output.images[0, -3:, -3:, -1]

  # forward with prompt embeds
  __UpperCamelCase					   =	self.get_dummy_inputs(__snake_case      )
  __UpperCamelCase					   =	3 * ["""this is a negative prompt"""]
  __UpperCamelCase					   =	3 * [inputs.pop("""prompt"""      )]

  (
      __UpperCamelCase			
  )					   =	sd_pipe.encode_prompt(__snake_case		,						negative_prompt=__snake_case      )

  __UpperCamelCase					   =	sd_pipe(
      **__snake_case		,						prompt_embeds=__snake_case		,						negative_prompt_embeds=__snake_case		,						pooled_prompt_embeds=__snake_case		,						negative_pooled_prompt_embeds=__snake_case		,						)
  __UpperCamelCase					   =	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 							UpperCAmelCase__	(     unittest.TestCase):



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



 def  __lowerCamelCase					(	self		,						lowercase		,						lowercase="cpu"		,						lowercase=torch.floataa		,						lowercase=0      )	->				List[str]:
  __UpperCamelCase					   =	torch.Generator(device=__snake_case      ).manual_seed(__snake_case      )
  __UpperCamelCase					   =	np.random.RandomState(__snake_case      ).standard_normal((1, 4, 6_4, 6_4)      )
  __UpperCamelCase					   =	torch.from_numpy(__snake_case      ).to(device=__snake_case		,						dtype=__snake_case      )
  __UpperCamelCase					   =	{
      """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  __lowerCamelCase					(	self      )	->				Optional[int]:
  __UpperCamelCase					   =	DiffusionPipeline.from_pretrained("""stabilityai/stable-diffusion-2-base"""      )
  pipe.to(__snake_case      )
  pipe.set_progress_bar_config(disable=__snake_case      )

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

  assert image.shape == (1, 5_1_2, 5_1_2, 3)
  __UpperCamelCase					   =	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 .feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from .utils import PaddingStrategy, TensorType, is_tf_tensor, is_torch_tensor, logging, to_numpy


__lowerCamelCase							=  logging.get_logger(__name__)



class       __A		(   SCREAMING_SNAKE_CASE_			):

					def __init__(			self  :      Union[str, Any]     ,		__snake_case  :      int     ,		__snake_case  :      int     ,		__snake_case  :      float     ,		**__snake_case  :      Optional[Any]					)			->							str:
											__magic_name__:	Optional[Any]							=       feature_size
											__magic_name__:	List[Any]							=       sampling_rate
											__magic_name__:	Tuple							=       padding_value

											__magic_name__:	int							=       kwargs.pop("""padding_side"""     ,		"""right"""					)
											__magic_name__:	Optional[Any]							=       kwargs.pop("""return_attention_mask"""     ,		__snake_case					)

											super().__init__(**__snake_case					)

					def   lowerCamelCase__   (			self  :      Union[str, Any]     ,		__snake_case  :      Union[
					        BatchFeature,
					        List[BatchFeature],
					        Dict[str, BatchFeature],
					        Dict[str, List[BatchFeature]],
					        List[Dict[str, BatchFeature]],
					    ]     ,		__snake_case  :      Union[bool, str, PaddingStrategy] = True     ,		__snake_case  :      Optional[int] = None     ,		__snake_case  :      bool = False     ,		__snake_case  :      Optional[int] = None     ,		__snake_case  :      Optional[bool] = None     ,		__snake_case  :      Optional[Union[str, TensorType]] = None     ,		)			->							BatchFeature:
											# If we have a list of dicts, let's convert it in a dict of lists
											# We do this to allow using this method as a collate_fn function in PyTorch Dataloader
											if isinstance(__snake_case     ,		(list, tuple)					) and isinstance(processed_features[0]     ,		(dict, BatchFeature)					):
																	__magic_name__:	Union[str, Any]							=       {
																	    key: [example[key] for example in processed_features] for key in processed_features[0].keys()
																	}

											# The model's main input name, usually `input_values`, has be passed for padding
											if self.model_input_names[0] not in processed_features:
																	raise ValueError(
																	    """You should supply an instance of `transformers.BatchFeature` or list of `transformers.BatchFeature`"""
																	    F' to this method that includes {self.model_input_names[0]}, but you provided'
																	    F' {list(processed_features.keys()					)}'					)

											__magic_name__:	Any							=       processed_features[self.model_input_names[0]]
											__magic_name__:	Tuple							=       (
											    return_attention_mask if return_attention_mask is not None else self.return_attention_mask
											)

											if len(__snake_case					) == 0:
																	if return_attention_mask:
																							__magic_name__:	Optional[Any]							=       []
																	return processed_features

											# If we have PyTorch/TF tensors or lists as inputs, we cast them as Numpy arrays
											# and rebuild them afterwards if no return_tensors is specified
											# Note that we lose the specific device the tensor may be on for PyTorch

											__magic_name__:	Tuple							=       required_input[0]
											if isinstance(__snake_case     ,		(list, tuple)					):
																	# first_element might be an empty list/tuple in some edge cases so we grab the first non empty element.
																	__magic_name__:	str							=       0
																	while len(required_input[index]					) == 0:
																							index += 1
																	if index < len(__snake_case					):
																							__magic_name__:	Optional[int]							=       required_input[index][0]

											if return_tensors is None:
																	if is_tf_tensor(__snake_case					):
																							__magic_name__:	List[str]							=       """tf"""
																	elif is_torch_tensor(__snake_case					):
																							__magic_name__:	Any							=       """pt"""
																	elif isinstance(__snake_case     ,		(int, float, list, tuple, np.ndarray)					):
																							__magic_name__:	int							=       """np"""
																	else:
																							raise ValueError(
																							    F'type of {first_element} unknown: {type(__snake_case					)}. '
																							    """Should be one of a python, numpy, pytorch or tensorflow object."""					)

											for key, value in processed_features.items():
																	if isinstance(value[0]     ,		(int, float)					):
																							__magic_name__:	List[str]							=       to_numpy(__snake_case					)
																	else:
																							__magic_name__:	Any							=       [to_numpy(__snake_case					) for v in value]

        # Convert padding_strategy in PaddingStrategy
											__magic_name__:	str							=       self._get_padding_strategies(padding=__snake_case     ,		max_length=__snake_case					)

											__magic_name__:	str							=       processed_features[self.model_input_names[0]]

											__magic_name__:	str							=       len(__snake_case					)
											if not all(len(__snake_case					) == batch_size for v in processed_features.values()					):
																	raise ValueError("""Some items in the output dictionary have a different batch size than others."""					)

											__magic_name__:	List[Any]							=       []
											for i in range(__snake_case					):
																	__magic_name__:	List[Any]							=       {k: v[i] for k, v in processed_features.items()}
																	# truncation
																	__magic_name__:	List[Any]							=       self._truncate(
																	    __snake_case     ,		max_length=__snake_case     ,		pad_to_multiple_of=__snake_case     ,		truncation=__snake_case     ,		)
																	truncated_inputs.append(__snake_case					)

											if padding_strategy == PaddingStrategy.LONGEST:
																	# make sure that `max_length` cannot be longer than the longest truncated length
																	__magic_name__:	Union[str, Any]							=       max(len(input_slice[self.model_input_names[0]]					) for input_slice in truncated_inputs					)
																	__magic_name__:	Union[str, Any]							=       PaddingStrategy.MAX_LENGTH

											__magic_name__:	List[str]							=       {}
											for i in range(__snake_case					):
																	# padding
																	__magic_name__:	str							=       self._pad(
																	    truncated_inputs[i]     ,		max_length=__snake_case     ,		padding_strategy=__snake_case     ,		pad_to_multiple_of=__snake_case     ,		return_attention_mask=__snake_case     ,		)

																	for key, value in outputs.items():
																							if key not in batch_outputs:
																													__magic_name__:	Optional[Any]							=       []
																							if value.dtype is np.dtype(np.floataa					):
																													__magic_name__:	Any							=       value.astype(np.floataa					)
																							batch_outputs[key].append(__snake_case					)

											return BatchFeature(__snake_case     ,		tensor_type=__snake_case					)

					def   lowerCamelCase__   (			self  :      Tuple     ,		__snake_case  :      Union[Dict[str, np.ndarray], BatchFeature]     ,		__snake_case  :      Optional[int] = None     ,		__snake_case  :      PaddingStrategy = PaddingStrategy.DO_NOT_PAD     ,		__snake_case  :      Optional[int] = None     ,		__snake_case  :      Optional[bool] = None     ,		)			->							dict:
											__magic_name__:	List[str]							=       processed_features[self.model_input_names[0]]

											if padding_strategy == PaddingStrategy.LONGEST:
																	__magic_name__:	List[Any]							=       len(__snake_case					)

											if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
																	__magic_name__:	List[str]							=       ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

											__magic_name__:	str							=       padding_strategy != PaddingStrategy.DO_NOT_PAD and len(__snake_case					) < max_length

											if return_attention_mask and "attention_mask" not in processed_features:
																	__magic_name__:	int							=       np.ones(len(__snake_case					)     ,		dtype=np.intaa					)

											if needs_to_be_padded:
																	__magic_name__:	str							=       max_length - len(__snake_case					)
																	if self.padding_side == "right":
																							if return_attention_mask:
																													__magic_name__:	List[Any]							=       np.pad(
																													    processed_features["""attention_mask"""]     ,		(0, difference)					)
																							__magic_name__:	Union[str, Any]							=       ((0, difference), (0, 0)) if self.feature_size > 1 else (0, difference)
																							__magic_name__:	int							=       np.pad(
																							    __snake_case     ,		__snake_case     ,		"""constant"""     ,		constant_values=self.padding_value					)
																	elif self.padding_side == "left":
																							if return_attention_mask:
																													__magic_name__:	Dict							=       np.pad(
																													    processed_features["""attention_mask"""]     ,		(difference, 0)					)
																							__magic_name__:	Optional[Any]							=       ((difference, 0), (0, 0)) if self.feature_size > 1 else (difference, 0)
																							__magic_name__:	int							=       np.pad(
																							    __snake_case     ,		__snake_case     ,		"""constant"""     ,		constant_values=self.padding_value					)
																	else:
																							raise ValueError("""Invalid padding strategy:""" + str(self.padding_side					)					)

											return processed_features

					def   lowerCamelCase__   (			self  :      Optional[Any]     ,		__snake_case  :      Union[Dict[str, np.ndarray], BatchFeature]     ,		__snake_case  :      Optional[int] = None     ,		__snake_case  :      Optional[int] = None     ,		__snake_case  :      Optional[bool] = None     ,		)			->							int:
											if not truncation:
																	return processed_features
											elif truncation and max_length is None:
																	raise ValueError("""When setting ``truncation=True``, make sure that ``max_length`` is defined."""					)

											__magic_name__:	Dict							=       processed_features[self.model_input_names[0]]

											# find `max_length` that fits `pad_to_multiple_of`
											if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
																	__magic_name__:	Optional[Any]							=       ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

											__magic_name__:	Tuple							=       len(__snake_case					) > max_length

											if needs_to_be_truncated:
																	__magic_name__:	Any							=       processed_features[self.model_input_names[0]][:max_length]
																	if "attention_mask" in processed_features:
																							__magic_name__:	List[Any]							=       processed_features["""attention_mask"""][:max_length]

											return processed_features






					def   lowerCamelCase__   (			self  :      List[Any]     ,		__snake_case  :      int=False     ,		__snake_case  :      Tuple=None					)			->							Optional[Any]:

											# Get padding strategy
											if padding is not False:
																	if padding is True:
																							__magic_name__:	Optional[Any]							=       PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
																	elif not isinstance(__snake_case     ,		__snake_case					):
																							__magic_name__:	Tuple							=       PaddingStrategy(__snake_case					)
																	elif isinstance(__snake_case     ,		__snake_case					):
																							__magic_name__:	Dict							=       padding
											else:
																	__magic_name__:	int							=       PaddingStrategy.DO_NOT_PAD

											# Set max length if needed
											if max_length is None:
																	if padding_strategy == PaddingStrategy.MAX_LENGTH:
																							raise ValueError(
																							    F'When setting ``padding={PaddingStrategy.MAX_LENGTH}``, make sure that max_length is defined'					)

        # Test if we have a padding value
											if padding_strategy != PaddingStrategy.DO_NOT_PAD and (self.padding_value is None):
																	raise ValueError(
																	    """Asking to pad but the feature_extractor does not have a padding value. Please select a value to use"""
																	    """ as `padding_value`. For example: `feature_extractor.padding_value = 0.0`."""					)

											return padding_strategy



