


from typing import Dict, Iterable, 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_DEFAULT_MEAN,
    IMAGENET_DEFAULT_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


_snake_case  :	List[str]			      =       logging.get_logger(__name__)

class        __magic_name__	(snake_case_   ):




							"""simple docstring"""


							__UpperCAmelCase					:							Dict  						=						["""pixel_values"""]

							def __init__(							self :						List[str]					,			lowerCamelCase :						List[str] = True					,			lowerCamelCase :						List[str] = None					,			lowerCamelCase :						Tuple = PILImageResampling.BICUBIC					,			lowerCamelCase :						Dict = True					,			lowerCamelCase :						List[str] = None					,			lowerCamelCase :						Any = True					,			lowerCamelCase :						Optional[Any] = 1 / 255					,			lowerCamelCase :						List[str] = True					,			lowerCamelCase :						int = IMAGENET_DEFAULT_MEAN					,			lowerCamelCase :						List[Any] = IMAGENET_DEFAULT_STD					,			**lowerCamelCase :						List[str]					,			)			->					int:
														super().__init__(**lowerCamelCase       )
														__snake_case	:    Optional[int]											=     size if size is not None else {"shortest_edge": 224}
														__snake_case	:    List[Any]											=     get_size_dict(lowerCamelCase					,			default_to_square=lowerCamelCase       )
														__snake_case	:    Tuple											=     crop_size if crop_size is not None else {"height": 224, "width": 224}
														__snake_case	:    Optional[Any]											=     get_size_dict(lowerCamelCase					,			param_name="crop_size"       )

														__snake_case	:    Tuple											=     do_resize
														__snake_case	:    Optional[Any]											=     size
														__snake_case	:    Union[str, Any]											=     resample
														__snake_case	:    List[str]											=     do_center_crop
														__snake_case	:    Optional[int]											=     crop_size
														__snake_case	:    List[Any]											=     do_rescale
														__snake_case	:    Optional[Any]											=     rescale_factor
														__snake_case	:    str											=     do_normalize
														__snake_case	:    Tuple											=     image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
														__snake_case	:    Optional[Any]											=     image_std if image_std is not None else IMAGENET_DEFAULT_STD

							def 					__snake_case      (							self :						List[Any]					,			lowerCamelCase :						Any					,			lowerCamelCase :						Optional[Any]					,			lowerCamelCase :						Optional[int] = PILImageResampling.BICUBIC					,			lowerCamelCase :						List[Any] = None					,			**lowerCamelCase :						Any					,			)			->					List[str]:
														__snake_case	:    List[Any]											=     get_size_dict(lowerCamelCase					,			default_to_square=lowerCamelCase       )
														# size_dict is a dict with either keys "height" and "width" or "shortest_edge"
														if "shortest_edge" in size:
																					__snake_case	:    Any											=     int((256 / 224) * size["shortest_edge"]       )
																					__snake_case	:    str											=     get_resize_output_image_size(lowerCamelCase					,			size=lowerCamelCase					,			default_to_square=lowerCamelCase       )
																					__snake_case	:    str											=     {"height": output_size[0], "width": output_size[1]}
														if "height" not in size_dict or "width" not in size_dict:
																					raise ValueError(
																					    F'Size dict must have keys \'height\' and \'width\' or \'shortest_edge\'. Got {size_dict.keys()}'       )
														return resize(
														    lowerCamelCase					,			size=(size_dict["height"], size_dict["width"])					,			resample=lowerCamelCase					,			data_format=lowerCamelCase					,			**lowerCamelCase       )

							def 					__snake_case      (							self :						List[str]					,			lowerCamelCase :						Optional[Any]					,			lowerCamelCase :						Dict					,			lowerCamelCase :						Union[str, Any] = None					,			**lowerCamelCase :						Union[str, Any]					,			)			->					List[Any]:
														__snake_case	:    Optional[int]											=     get_size_dict(lowerCamelCase       )
														if "height" not in size or "width" not in size:
																					raise ValueError(F'Size dict must have keys \'height\' and \'width\'. Got {size.keys()}'       )
														return center_crop(lowerCamelCase					,			size=(size["height"], size["width"])					,			data_format=lowerCamelCase					,			**lowerCamelCase       )

							def 					__snake_case      (							self :						int					,			lowerCamelCase :						str					,			lowerCamelCase :						Tuple					,			lowerCamelCase :						int = None					,			**lowerCamelCase :						Tuple					,			)			->					Dict:
														return rescale(lowerCamelCase					,			scale=lowerCamelCase					,			data_format=lowerCamelCase					,			**lowerCamelCase       )

							def 					__snake_case      (							self :						List[Any]					,			lowerCamelCase :						str					,			lowerCamelCase :						Dict					,			lowerCamelCase :						Tuple					,			lowerCamelCase :						str = None					,			**lowerCamelCase :						Union[str, Any]					,			)			->					Tuple:
														return normalize(lowerCamelCase					,			mean=lowerCamelCase					,			std=lowerCamelCase					,			data_format=lowerCamelCase					,			**lowerCamelCase       )






							def 					__snake_case      (							self :						Dict					,			lowerCamelCase :						Union[str, Any]					,			lowerCamelCase :						Optional[int] = None					,			lowerCamelCase :						Any = None					,			lowerCamelCase :						Optional[Any] = None					,			lowerCamelCase :						List[Any] = None					,			lowerCamelCase :						List[str] = None					,			lowerCamelCase :						List[str] = None					,			lowerCamelCase :						Any = None					,			lowerCamelCase :						Union[str, Any] = None					,			lowerCamelCase :						Optional[Any] = None					,			lowerCamelCase :						Dict = None					,			lowerCamelCase :						str = None					,			lowerCamelCase :						Union[str, Any] = ChannelDimension.FIRST					,			**lowerCamelCase :						Tuple					,			)			->					Optional[Any]:
														__snake_case	:    str											=     do_resize if do_resize is not None else self.do_resize
														__snake_case	:    List[str]											=     resample if resample is not None else self.resample
														__snake_case	:    Tuple											=     do_center_crop if do_center_crop is not None else self.do_center_crop
														__snake_case	:    int											=     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	:    Tuple											=     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	:    Dict											=     image_std if image_std is not None else self.image_std

														__snake_case	:    Optional[Any]											=     size if size is not None else self.size
														__snake_case	:    List[str]											=     get_size_dict(lowerCamelCase					,			default_to_square=lowerCamelCase       )
														__snake_case	:    List[str]											=     crop_size if crop_size is not None else self.crop_size
														__snake_case	:    Dict											=     get_size_dict(lowerCamelCase					,			param_name="crop_size"       )

														__snake_case	:    Any											=     make_list_of_images(lowerCamelCase       )

														if not valid_images(lowerCamelCase       ):
																					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_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.
														__snake_case	:    Optional[int]											=     [to_numpy_array(lowerCamelCase       ) for image in images]

														if do_resize:
																					__snake_case	:    Any											=     [self.resize(lowerCamelCase					,			lowerCamelCase					,			lowerCamelCase       ) for image in images]

														if do_center_crop:
																					__snake_case	:    List[Any]											=     [self.center_crop(lowerCamelCase					,			lowerCamelCase       ) for image in images]

														if do_rescale:
																					__snake_case	:    Tuple											=     [self.rescale(lowerCamelCase					,			lowerCamelCase       ) for image in images]

														if do_normalize:
																					__snake_case	:    Optional[Any]											=     [self.normalize(lowerCamelCase					,			lowerCamelCase					,			lowerCamelCase       ) for image in images]

														__snake_case	:    int											=     [to_channel_dimension_format(lowerCamelCase					,			lowerCamelCase       ) for image in images]

														__snake_case	:    Optional[int]											=     {"pixel_values": images}
														return BatchFeature(data=lowerCamelCase					,			tensor_type=lowerCamelCase       )





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	(_lowerCAmelCase  ,       _lowerCAmelCase  ,       unittest.TestCase   ):




							"""simple docstring"""


							__UpperCAmelCase					:							Optional[Any]  						=						StableDiffusionXLImgaImgPipeline
							__UpperCAmelCase					:							Any  						=						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
							__UpperCAmelCase					:							Dict  						=						PipelineTesterMixin.required_optional_params - {"latents"}
							__UpperCAmelCase					:							Union[str, Any]  						=						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
							__UpperCAmelCase					:							Tuple  						=						IMAGE_TO_IMAGE_IMAGE_PARAMS
							__UpperCAmelCase					:							str  						=						IMAGE_TO_IMAGE_IMAGE_PARAMS

							def 					__snake_case      (							self :						Optional[Any]       )			->					Tuple:
														torch.manual_seed(0       )
														__snake_case	:    Optional[int]											=     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")					,			attention_head_dim=(2, 4)					,			use_linear_projection=lowerCamelCase					,			addition_embed_type="text_time"					,			addition_time_embed_dim=8					,			transformer_layers_per_block=(1, 2)					,			projection_class_embeddings_input_dim=80					,			cross_attention_dim=64					,			)
														__snake_case	:    Tuple											=     EulerDiscreteScheduler(
														    beta_start=0.0_00_85					,			beta_end=0.0_12					,			steps_offset=1					,			beta_schedule="scaled_linear"					,			timestep_spacing="leading"					,			)
														torch.manual_seed(0       )
														__snake_case	:    int											=     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       )
														__snake_case	:    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=1000					,			hidden_act="gelu"					,			projection_dim=32					,			)
														__snake_case	:    List[str]											=     CLIPTextModel(lowerCamelCase       )
														__snake_case	:    Any											=     CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"					,			local_files_only=lowerCamelCase       )

														__snake_case	:    List[str]											=     CLIPTextModelWithProjection(lowerCamelCase       )
														__snake_case	:    List[Any]											=     CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"					,			local_files_only=lowerCamelCase       )

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

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

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

														__snake_case	:    str											=     np.array([0.46_56, 0.48_40, 0.44_39, 0.66_98, 0.55_74, 0.45_24, 0.57_99, 0.59_43, 0.51_65]       )

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

							def 					__snake_case      (							self :						str       )			->					Optional[Any]:
														super().test_attention_slicing_forward_pass(expected_max_diff=3E-3       )

							def 					__snake_case      (							self :						Any       )			->					Dict:
														super().test_inference_batch_single_identical(expected_max_diff=3E-3       )

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






							def 					__snake_case      (							self :						Tuple       )			->					Union[str, Any]:
														__snake_case	:    str											=     self.get_dummy_components()
														__snake_case	:    List[Any]											=     StableDiffusionXLImgaImgPipeline(**lowerCamelCase       )
														__snake_case	:    Optional[Any]											=     sd_pipe.to(lowerCamelCase       )
														__snake_case	:    int											=     sd_pipe.to(lowerCamelCase       )
														sd_pipe.set_progress_bar_config(disable=lowerCamelCase       )

														# forward without prompt embeds
														__snake_case	:    List[str]											=     self.get_dummy_inputs(lowerCamelCase       )
														__snake_case	:    str											=     3 * ["this is a negative prompt"]
														__snake_case	:    Any											=     negative_prompt
														__snake_case	:    Optional[Any]											=     3 * [inputs["prompt"]]

														__snake_case	:    int											=     sd_pipe(**lowerCamelCase       )
														__snake_case	:    List[Any]											=     output.images[0, -3:, -3:, -1]

														# forward with prompt embeds
														__snake_case	:    List[Any]											=     self.get_dummy_inputs(lowerCamelCase       )
														__snake_case	:    Optional[Any]											=     3 * ["this is a negative prompt"]
														__snake_case	:    int											=     3 * [inputs.pop("prompt"       )]

														(
														    (
														    __snake_case
														)				,    (
														    __snake_case
														)				,    (
														    __snake_case
														)				,    (
														    __snake_case
														)				,    
														)	:    Dict											=     sd_pipe.encode_prompt(lowerCamelCase					,			negative_prompt=lowerCamelCase       )

														__snake_case	:    Tuple											=     sd_pipe(
														    **lowerCamelCase					,			prompt_embeds=lowerCamelCase					,			negative_prompt_embeds=lowerCamelCase					,			pooled_prompt_embeds=lowerCamelCase					,			negative_pooled_prompt_embeds=lowerCamelCase					,			)
														__snake_case	:    List[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   ):




							"""simple docstring"""


							def 					__snake_case      (							self :						Optional[int]       )			->					Dict:
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()

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

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

														assert image.shape == (1, 512, 512, 3)
														__snake_case	:    Optional[int]											=     np.array([0.4_94_93, 0.4_78_96, 0.4_07_98, 0.5_42_14, 0.5_32_12, 0.4_82_02, 0.4_76_56, 0.4_63_29, 0.4_85_06]       )
														assert np.abs(image_slice - expected_slice       ).max() < 7E-3

