





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






							"""simple docstring"""

							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      A_  (      self				:     Union[str, Any]    )	->Tuple:
													torch.manual_seed(0    )
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					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=a							,	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							,	)
													SCREAMING_SNAKE_CASE__     :				str   =					EulerDiscreteScheduler(
													    beta_start=0.0_0085							,	beta_end=0.012							,	steps_offset=1							,	beta_schedule="scaled_linear"							,	timestep_spacing="leading"							,	)
													torch.manual_seed(0    )
													SCREAMING_SNAKE_CASE__     :				Tuple   =					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=1_28							,	)
													torch.manual_seed(0    )
													SCREAMING_SNAKE_CASE__     :				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=10_00							,	hidden_act="gelu"							,	projection_dim=32							,	)
													SCREAMING_SNAKE_CASE__     :				Tuple   =					CLIPTextModel(a    )
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"							,	local_files_only=a    )

													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					CLIPTextModelWithProjection(a    )
													SCREAMING_SNAKE_CASE__     :				str   =					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"							,	local_files_only=a    )

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

													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					self.get_dummy_inputs(a    )
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					sd_pipe(**a    ).images
													SCREAMING_SNAKE_CASE__     :				Tuple   =					image[0, -3:, -3:, -1]

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

													SCREAMING_SNAKE_CASE__     :				List[Any]   =					np.array([0.4656, 0.4840, 0.4439, 0.6698, 0.5574, 0.4524, 0.5799, 0.5943, 0.5165]    )

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

							def      A_  (      self				:     Tuple    )	->List[Any]:
													super().test_attention_slicing_forward_pass(expected_max_diff=3E-3    )

							def      A_  (      self				:     Any    )	->Union[str, Any]:
													super().test_inference_batch_single_identical(expected_max_diff=3E-3    )

							def      A_  (      self				:     Dict    )	->int:
													pass






							def      A_  (      self				:     Optional[int]    )	->Dict:
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					self.get_dummy_components()
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					StableDiffusionXLImgaImgPipeline(**a    )
													SCREAMING_SNAKE_CASE__     :				str   =					sd_pipe.to(a    )
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					sd_pipe.to(a    )
													sd_pipe.set_progress_bar_config(disable=a    )

													# forward without prompt embeds
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					self.get_dummy_inputs(a    )
													SCREAMING_SNAKE_CASE__     :				Any   =					3 * ["this is a negative prompt"]
													SCREAMING_SNAKE_CASE__     :				Dict   =					negative_prompt
													SCREAMING_SNAKE_CASE__     :				str   =					3 * [inputs["prompt"]]

													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					sd_pipe(**a    )
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					output.images[0, -3:, -3:, -1]

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

													(
													    (
													    SCREAMING_SNAKE_CASE__
													),							(
													    SCREAMING_SNAKE_CASE__
													),							(
													    SCREAMING_SNAKE_CASE__
													),							(
													    SCREAMING_SNAKE_CASE__
													),							
													)     :				Optional[Any]   =					sd_pipe.encode_prompt(a							,	negative_prompt=a    )

													SCREAMING_SNAKE_CASE__     :				Tuple   =					sd_pipe(
													    **a							,	prompt_embeds=a							,	negative_prompt_embeds=a							,	pooled_prompt_embeds=a							,	negative_pooled_prompt_embeds=a							,	)
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					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      A_  (      self				:     Optional[int]    )	->Optional[int]:
													super().tearDown()
													gc.collect()
													torch.cuda.empty_cache()

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

													SCREAMING_SNAKE_CASE__     :				List[str]   =					self.get_inputs(a    )
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					pipe(**a    ).images
													SCREAMING_SNAKE_CASE__     :				str   =					image[0, -3:, -3:, -1].flatten()

													assert image.shape == (1, 5_12, 5_12, 3)
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					np.array([0.4_9493, 0.4_7896, 0.4_0798, 0.5_4214, 0.5_3212, 0.4_8202, 0.4_7656, 0.4_6329, 0.4_8506]    )
													assert np.abs(image_slice - expected_slice    ).max() < 7E-3






# tests directory-specific settings - this file is run automatically
# by pytest before any tests are run

import doctest
import sys
import warnings
from os.path import abspath, dirname, join

import _pytest

from transformers.testing_utils import HfDoctestModule, HfDocTestParser


# allow having multiple repository checkouts and not needing to remember to rerun
# 'pip install -e .[dev]' when switching between checkouts and running tests.
__lowercase       :List[Any]       =  abspath(join(dirname(__file__), "src"))
sys.path.insert(1, git_repo_path)

# silence FutureWarning warnings in tests since often we can't act on them until
# they become normal warnings - i.e. the tests still need to test the current functionality
warnings.simplefilter(action="ignore", category=FutureWarning)







def       UpperCAmelCase							(     _lowerCamelCase      :  int     ):





						'''simple docstring'''


						config.addinivalue_line(
						    "markers"			,   "is_pt_tf_cross_test: mark test to run only when PT and TF interactions are tested"     )
						config.addinivalue_line(
						    "markers"			,   "is_pt_flax_cross_test: mark test to run only when PT and FLAX interactions are tested"     )
						config.addinivalue_line("markers"			,   "is_pipeline_test: mark test to run only when pipelines are tested"     )
						config.addinivalue_line("markers"			,   "is_staging_test: mark test to run only in the staging environment"     )
						config.addinivalue_line("markers"			,   "accelerate_tests: mark test that require accelerate"     )
						config.addinivalue_line("markers"			,   "tool_tests: mark the tool tests that are run on their specific schedule"     )







def       UpperCAmelCase							(     _lowerCamelCase      :  str     ):





						'''simple docstring'''


						from transformers.testing_utils import pytest_addoption_shared

						pytest_addoption_shared(_lowerCamelCase     )







def       UpperCAmelCase							(     _lowerCamelCase      :  Tuple     ):





						'''simple docstring'''


						from transformers.testing_utils import pytest_terminal_summary_main

						SCREAMING_SNAKE_CASE__     :				List[str]   =					terminalreporter.config.getoption("--make-reports"     )
						if make_reports:
												pytest_terminal_summary_main(_lowerCamelCase			,   id=_lowerCamelCase     )







def       UpperCAmelCase							(     _lowerCamelCase      :  List[Any]			,   _lowerCamelCase      :  Dict     ):





						'''simple docstring'''


						if exitstatus == 5:
												SCREAMING_SNAKE_CASE__     :				List[str]   =					0


# Doctest custom flag to ignore output.
__lowercase       :Optional[Any]       =  doctest.register_optionflag("IGNORE_RESULT")

__lowercase       :Dict       =  doctest.OutputChecker



class    _a     (				lowercase__							):






							"""simple docstring"""

							def      A_  (      self				:     Dict							,	a				:     List[str]							,	a				:     Dict							,	a				:     int    )	->Optional[Any]:
													if IGNORE_RESULT & optionflags:
																			return True
													return OutputChecker.check_output(self							,	a							,	a							,	a    )







__lowercase       :Any       =  CustomOutputChecker
__lowercase       :Any       =  HfDoctestModule



__lowercase       :int       =  HfDocTestParser