




"""simple docstring"""





import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()



class      a__				(	__magic_name__							,    __magic_name__							,    __magic_name__							,    unittest.TestCase   ):
			lowercase_    =	StableUnCLIPPipeline
			lowercase_    =	TEXT_TO_IMAGE_PARAMS
			lowercase_    =	TEXT_TO_IMAGE_BATCH_PARAMS
			lowercase_    =	TEXT_TO_IMAGE_IMAGE_PARAMS
			lowercase_    =	TEXT_TO_IMAGE_IMAGE_PARAMS

			# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
			lowercase_    =	False


			def       a_     (      self      :    Optional[int]):


						"""simple docstring"""




						__UpperCAmelCase	: str		=					32
						__UpperCAmelCase	: Tuple		=					embedder_hidden_size

						# prior components

						torch.manual_seed(0)
						__UpperCAmelCase	: str		=					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

						torch.manual_seed(0)
						__UpperCAmelCase	: Dict		=					CLIPTextModelWithProjection(
						    CLIPTextConfig(
						        bos_token_id=0      ,  eos_token_id=2      ,  hidden_size=UpperCamelCase_      ,  projection_dim=UpperCamelCase_      ,  intermediate_size=37      ,  layer_norm_eps=1e-05      ,  num_attention_heads=4      ,  num_hidden_layers=5      ,  pad_token_id=1      ,  vocab_size=1000      ,  ))

						torch.manual_seed(0)
						__UpperCAmelCase	: Optional[int]		=					PriorTransformer(
						    num_attention_heads=2      ,  attention_head_dim=12      ,  embedding_dim=UpperCamelCase_      ,  num_layers=1      ,  )

						torch.manual_seed(0)
						__UpperCAmelCase	: Union[str, Any]		=					DDPMScheduler(
						    variance_type="fixed_small_log"      ,  prediction_type="sample"      ,  num_train_timesteps=1000      ,  clip_sample=UpperCamelCase_      ,  clip_sample_range=5.0      ,  beta_schedule="squaredcos_cap_v2"      ,  )

						# regular denoising components

						torch.manual_seed(0)
						__UpperCAmelCase	: Union[str, Any]		=					StableUnCLIPImageNormalizer(embedding_dim=UpperCamelCase_)
						__UpperCAmelCase	: Tuple		=					DDPMScheduler(beta_schedule="squaredcos_cap_v2")

						torch.manual_seed(0)
						__UpperCAmelCase	: Optional[int]		=					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

						torch.manual_seed(0)
						__UpperCAmelCase	: Tuple		=					CLIPTextModel(
						    CLIPTextConfig(
						        bos_token_id=0      ,  eos_token_id=2      ,  hidden_size=UpperCamelCase_      ,  projection_dim=32      ,  intermediate_size=37      ,  layer_norm_eps=1e-05      ,  num_attention_heads=4      ,  num_hidden_layers=5      ,  pad_token_id=1      ,  vocab_size=1000      ,  ))

						torch.manual_seed(0)
						__UpperCAmelCase	: Dict		=					UNetaDConditionModel(
						    sample_size=32      ,  in_channels=4      ,  out_channels=4      ,  down_block_types=("CrossAttnDownBlock2D", "DownBlock2D")      ,  up_block_types=("UpBlock2D", "CrossAttnUpBlock2D")      ,  block_out_channels=(32, 64)      ,  attention_head_dim=(2, 4)      ,  class_embed_type="projection"      ,  projection_class_embeddings_input_dim=embedder_projection_dim * 2      ,  cross_attention_dim=UpperCamelCase_      ,  layers_per_block=1      ,  upcast_attention=UpperCamelCase_      ,  use_linear_projection=UpperCamelCase_      ,  )

						torch.manual_seed(0)
						__UpperCAmelCase	: List[str]		=					DDIMScheduler(
						    beta_schedule="scaled_linear"      ,  beta_start=0.00085      ,  beta_end=0.012      ,  prediction_type="v_prediction"      ,  set_alpha_to_one=UpperCamelCase_      ,  steps_offset=1      ,  )

						torch.manual_seed(0)
						__UpperCAmelCase	: Optional[Any]		=					AutoencoderKL()

						__UpperCAmelCase	: List[Any]		=					{
						    # prior components
						    "prior_tokenizer": prior_tokenizer,
						    "prior_text_encoder": prior_text_encoder,
						    "prior": prior,
						    "prior_scheduler": prior_scheduler,
						    # image noising components
						    "image_normalizer": image_normalizer,
						    "image_noising_scheduler": image_noising_scheduler,
						    # regular denoising components
						    "tokenizer": tokenizer,
						    "text_encoder": text_encoder,
						    "unet": unet,
						    "scheduler": scheduler,
						    "vae": vae,
						}

						return components


			def       a_     (      self      :    str      ,  UpperCamelCase_      :    int      ,  UpperCamelCase_      :    Optional[int]=0):


						"""simple docstring"""




						if str(UpperCamelCase_).startswith("mps"):
									__UpperCAmelCase	: Optional[int]		=					torch.manual_seed(UpperCamelCase_)
						else:
									__UpperCAmelCase	: Dict		=					torch.Generator(device=UpperCamelCase_).manual_seed(UpperCamelCase_)
						__UpperCAmelCase	: Optional[int]		=					{
						    "prompt": "A painting of a squirrel eating a burger",
						    "generator": generator,
						    "num_inference_steps": 2,
						    "prior_num_inference_steps": 2,
						    "output_type": "numpy",
						}
						return inputs


			def       a_     (      self      :    Optional[Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Any		=					torch_device == "cpu"

						self._test_attention_slicing_forward_pass(test_max_difference=UpperCamelCase_)


			def       a_     (      self      :    Union[str, Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[Any]		=					torch_device in ["cpu", "mps"]

						self._test_inference_batch_single_identical(test_max_difference=UpperCamelCase_)






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


			def       a_     (      self      :    Optional[Any]):


						"""simple docstring"""




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


			def       a_     (      self      :    int):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy")

						__UpperCAmelCase	: List[str]		=					StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"      ,  torch_dtype=torch.floataa)
						pipe.to(UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)
						# stable unclip will oom when integration tests are run on a V100,
						# so turn on memory savings
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__UpperCAmelCase	: Optional[Any]		=					torch.Generator(device="cpu").manual_seed(0)
						__UpperCAmelCase	: str		=					pipe("anime turle"      ,  generator=UpperCamelCase_      ,  output_type="np")

						__UpperCAmelCase	: Optional[Any]		=					output.images[0]

						assert image.shape == (768, 768, 3)

						assert_mean_pixel_difference(UpperCamelCase_      ,  UpperCamelCase_)


			def       a_     (      self      :    Tuple):


						"""simple docstring"""




						torch.cuda.empty_cache()
						torch.cuda.reset_max_memory_allocated()
						torch.cuda.reset_peak_memory_stats()

						__UpperCAmelCase	: Any		=					StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"      ,  torch_dtype=torch.floataa)
						__UpperCAmelCase	: Any		=					pipe.to(UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__UpperCAmelCase	: int		=					pipe(
						    "anime turtle"      ,  prior_num_inference_steps=2      ,  num_inference_steps=2      ,  output_type="np"      ,  )

						__UpperCAmelCase	: Tuple		=					torch.cuda.max_memory_allocated()
						# make sure that less than 7 GB is allocated
						assert mem_bytes < 7 * 10**9








"""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,
    is_valid_image,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
					import PIL


A   =       logging.get_logger(__name__)

def 					_UpperCamelCase    (			UpperCamelCase							)			->			List[List[ImageInput]]:


			"""simple docstring"""
			if isinstance(UpperCamelCase				,       (list, tuple)							) and isinstance(videos[0]				,       (list, tuple)							) and is_valid_image(videos[0][0]							):
						return videos

			elif isinstance(UpperCamelCase				,       (list, tuple)							) and is_valid_image(videos[0]							):
						return [videos]

			elif is_valid_image(UpperCamelCase							):
						return [[videos]]

			raise ValueError(f"Could not make batched video from {videos}"							)





class      a__				(	__magic_name__   ):
			lowercase_    =	["pixel_values"]


			def __init__(      self      :    Tuple      ,  UpperCamelCase_      :    bool = True      ,  UpperCamelCase_      :    Dict[str, int] = None      ,  UpperCamelCase_      :    PILImageResampling = PILImageResampling.BILINEAR      ,  UpperCamelCase_      :    bool = True      ,  UpperCamelCase_      :    Dict[str, int] = None      ,  UpperCamelCase_      :    bool = True      ,  UpperCamelCase_      :    Union[int, float] = 1 / 255      ,  UpperCamelCase_      :    bool = True      ,  UpperCamelCase_      :    Optional[Union[float, List[float]]] = None      ,  UpperCamelCase_      :    Optional[Union[float, List[float]]] = None      ,  **UpperCamelCase_      :    List[str]      ,  ):


						"""simple docstring"""




						super().__init__(**UpperCamelCase_)
						__UpperCAmelCase	: Dict		=					size if size is not None else {"shortest_edge": 224}
						__UpperCAmelCase	: Union[str, Any]		=					get_size_dict(UpperCamelCase_      ,  default_to_square=UpperCamelCase_)
						__UpperCAmelCase	: Optional[int]		=					crop_size if crop_size is not None else {"height": 224, "width": 224}
						__UpperCAmelCase	: List[str]		=					get_size_dict(UpperCamelCase_      ,  param_name="crop_size")

						__UpperCAmelCase	: Optional[Any]		=					do_resize
						__UpperCAmelCase	: Union[str, Any]		=					size
						__UpperCAmelCase	: str		=					do_center_crop
						__UpperCAmelCase	: str		=					crop_size
						__UpperCAmelCase	: Any		=					resample
						__UpperCAmelCase	: List[Any]		=					do_rescale
						__UpperCAmelCase	: Union[str, Any]		=					rescale_factor
						__UpperCAmelCase	: List[Any]		=					do_normalize
						__UpperCAmelCase	: Dict		=					image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
						__UpperCAmelCase	: Tuple		=					image_std if image_std is not None else IMAGENET_STANDARD_STD


			def       a_     (      self      :    int      ,  UpperCamelCase_      :    np.ndarray      ,  UpperCamelCase_      :    Dict[str, int]      ,  UpperCamelCase_      :    PILImageResampling = PILImageResampling.BILINEAR      ,  UpperCamelCase_      :    Optional[Union[str, ChannelDimension]] = None      ,  **UpperCamelCase_      :    List[Any]      ,  ):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[Any]		=					get_size_dict(UpperCamelCase_      ,  default_to_square=UpperCamelCase_)
						if "shortest_edge" in size:
									__UpperCAmelCase	: Dict		=					get_resize_output_image_size(UpperCamelCase_      ,  size["shortest_edge"]      ,  default_to_square=UpperCamelCase_)
						elif "height" in size and "width" in size:
									__UpperCAmelCase	: Dict		=					(size["height"], size["width"])
						else:
									raise ValueError(F"Size must have 'height' and 'width' or 'shortest_edge' as keys. Got {size.keys()}")
						return resize(UpperCamelCase_      ,  size=UpperCamelCase_      ,  resample=UpperCamelCase_      ,  data_format=UpperCamelCase_      ,  **UpperCamelCase_)


			def       a_     (      self      :    List[str]      ,  UpperCamelCase_      :    np.ndarray      ,  UpperCamelCase_      :    Dict[str, int]      ,  UpperCamelCase_      :    Optional[Union[str, ChannelDimension]] = None      ,  **UpperCamelCase_      :    Tuple      ,  ):


						"""simple docstring"""




						__UpperCAmelCase	: int		=					get_size_dict(UpperCamelCase_)
						if "height" not in size or "width" not in size:
									raise ValueError(F"Size must have 'height' and 'width' as keys. Got {size.keys()}")
						return center_crop(UpperCamelCase_      ,  size=(size["height"], size["width"])      ,  data_format=UpperCamelCase_      ,  **UpperCamelCase_)


			def       a_     (      self      :    Optional[int]      ,  UpperCamelCase_      :    np.ndarray      ,  UpperCamelCase_      :    Union[int, float]      ,  UpperCamelCase_      :    Optional[Union[str, ChannelDimension]] = None      ,  **UpperCamelCase_      :    Optional[int]      ,  ):


						"""simple docstring"""




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


			def       a_     (      self      :    List[Any]      ,  UpperCamelCase_      :    np.ndarray      ,  UpperCamelCase_      :    Union[float, List[float]]      ,  UpperCamelCase_      :    Union[float, List[float]]      ,  UpperCamelCase_      :    Optional[Union[str, ChannelDimension]] = None      ,  **UpperCamelCase_      :    Union[str, Any]      ,  ):


						"""simple docstring"""




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


			def       a_     (      self      :    List[str]      ,  UpperCamelCase_      :    ImageInput      ,  UpperCamelCase_      :    bool = None      ,  UpperCamelCase_      :    Dict[str, int] = None      ,  UpperCamelCase_      :    PILImageResampling = None      ,  UpperCamelCase_      :    bool = None      ,  UpperCamelCase_      :    Dict[str, int] = None      ,  UpperCamelCase_      :    bool = None      ,  UpperCamelCase_      :    float = None      ,  UpperCamelCase_      :    bool = None      ,  UpperCamelCase_      :    Optional[Union[float, List[float]]] = None      ,  UpperCamelCase_      :    Optional[Union[float, List[float]]] = None      ,  UpperCamelCase_      :    Optional[ChannelDimension] = ChannelDimension.FIRST      ,  ):


						"""simple docstring"""




						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_center_crop and crop_size is None:
									raise ValueError("Crop size must be specified if do_center_crop is True.")

						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.
						__UpperCAmelCase	: List[str]		=					to_numpy_array(UpperCamelCase_)

						if do_resize:
									__UpperCAmelCase	: Tuple		=					self.resize(image=UpperCamelCase_      ,  size=UpperCamelCase_      ,  resample=UpperCamelCase_)

						if do_center_crop:
									__UpperCAmelCase	: str		=					self.center_crop(UpperCamelCase_      ,  size=UpperCamelCase_)

						if do_rescale:
									__UpperCAmelCase	: Union[str, Any]		=					self.rescale(image=UpperCamelCase_      ,  scale=UpperCamelCase_)

						if do_normalize:
									__UpperCAmelCase	: Optional[int]		=					self.normalize(image=UpperCamelCase_      ,  mean=UpperCamelCase_      ,  std=UpperCamelCase_)

						__UpperCAmelCase	: Any		=					to_channel_dimension_format(UpperCamelCase_      ,  UpperCamelCase_)
						return image


			def       a_     (      self      :    List[Any]      ,  UpperCamelCase_      :    ImageInput      ,  UpperCamelCase_      :    bool = None      ,  UpperCamelCase_      :    Dict[str, int] = None      ,  UpperCamelCase_      :    PILImageResampling = None      ,  UpperCamelCase_      :    bool = None      ,  UpperCamelCase_      :    Dict[str, int] = None      ,  UpperCamelCase_      :    bool = None      ,  UpperCamelCase_      :    float = None      ,  UpperCamelCase_      :    bool = None      ,  UpperCamelCase_      :    Optional[Union[float, List[float]]] = None      ,  UpperCamelCase_      :    Optional[Union[float, List[float]]] = None      ,  UpperCamelCase_      :    Optional[Union[str, TensorType]] = None      ,  UpperCamelCase_      :    ChannelDimension = ChannelDimension.FIRST      ,  **UpperCamelCase_      :    Optional[int]      ,  ):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[Any]		=					do_resize if do_resize is not None else self.do_resize
						__UpperCAmelCase	: Optional[Any]		=					resample if resample is not None else self.resample
						__UpperCAmelCase	: Any		=					do_center_crop if do_center_crop is not None else self.do_center_crop
						__UpperCAmelCase	: str		=					do_rescale if do_rescale is not None else self.do_rescale
						__UpperCAmelCase	: str		=					rescale_factor if rescale_factor is not None else self.rescale_factor
						__UpperCAmelCase	: Tuple		=					do_normalize if do_normalize is not None else self.do_normalize
						__UpperCAmelCase	: str		=					image_mean if image_mean is not None else self.image_mean
						__UpperCAmelCase	: Optional[Any]		=					image_std if image_std is not None else self.image_std

						__UpperCAmelCase	: Tuple		=					size if size is not None else self.size
						__UpperCAmelCase	: str		=					get_size_dict(UpperCamelCase_      ,  default_to_square=UpperCamelCase_)
						__UpperCAmelCase	: Any		=					crop_size if crop_size is not None else self.crop_size
						__UpperCAmelCase	: int		=					get_size_dict(UpperCamelCase_      ,  param_name="crop_size")

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

						__UpperCAmelCase	: Dict		=					make_batched(UpperCamelCase_)

						__UpperCAmelCase	: List[str]		=					[
						    [
						        self._preprocess_image(
						            image=UpperCamelCase_      ,  do_resize=UpperCamelCase_      ,  size=UpperCamelCase_      ,  resample=UpperCamelCase_      ,  do_center_crop=UpperCamelCase_      ,  crop_size=UpperCamelCase_      ,  do_rescale=UpperCamelCase_      ,  rescale_factor=UpperCamelCase_      ,  do_normalize=UpperCamelCase_      ,  image_mean=UpperCamelCase_      ,  image_std=UpperCamelCase_      ,  data_format=UpperCamelCase_      ,  )
						        for img in video
						    ]
						    for video in videos
						]

						__UpperCAmelCase	: int		=					{"pixel_values": videos}
						return BatchFeature(data=UpperCamelCase_      ,  tensor_type=UpperCamelCase_)


