
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"""





       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     SCREAMING_SNAKE_CASE  (					self	):
        torch.manual_seed(0	)
        _UpperCAmelCase 				=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=_SCREAMING_SNAKE_CASE       ,					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       ,					)
        _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=[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 				=CLIPTextConfig(
            bos_token_id=0       ,					eos_token_id=2       ,					hidden_size=32       ,					intermediate_size=37       ,					layer_norm_eps=1E-0_5       ,					num_attention_heads=4       ,					num_hidden_layers=5       ,					pad_token_id=1       ,					vocab_size=1000       ,					hidden_act="gelu"       ,					projection_dim=32       ,					)
        _UpperCAmelCase 				=CLIPTextModel(_SCREAMING_SNAKE_CASE	)
        _UpperCAmelCase 				=CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"       ,					local_files_only=_SCREAMING_SNAKE_CASE	)

        _UpperCAmelCase 				=CLIPTextModelWithProjection(_SCREAMING_SNAKE_CASE	)
        _UpperCAmelCase 				=CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"       ,					local_files_only=_SCREAMING_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     SCREAMING_SNAKE_CASE  (					self       ,					_snake_case       ,					_snake_case=0	):
        _UpperCAmelCase 				=floats_tensor((1, 3, 32, 32)       ,					rng=random.Random(_SCREAMING_SNAKE_CASE	)	).to(_SCREAMING_SNAKE_CASE	)
        _UpperCAmelCase 				=image / 2 + 0.5
        if str(_SCREAMING_SNAKE_CASE	).startswith("mps"	):
         _UpperCAmelCase 				=torch.manual_seed(_SCREAMING_SNAKE_CASE	)
        else:
         _UpperCAmelCase 				=torch.Generator(device=_SCREAMING_SNAKE_CASE	).manual_seed(_SCREAMING_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     SCREAMING_SNAKE_CASE  (					self	):
        _UpperCAmelCase 				='cpu'  # ensure determinism for the device-dependent torch.Generator
        _UpperCAmelCase 				=self.get_dummy_components()
        _UpperCAmelCase 				=StableDiffusionXLImgaImgPipeline(**_SCREAMING_SNAKE_CASE	)
        _UpperCAmelCase 				=sd_pipe.to(_SCREAMING_SNAKE_CASE	)
        sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE	)

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

        assert image.shape == (1, 32, 32, 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     SCREAMING_SNAKE_CASE  (					self	):
        super().test_attention_slicing_forward_pass(expected_max_diff=3E-3	)
       def     SCREAMING_SNAKE_CASE  (					self	):
        super().test_inference_batch_single_identical(expected_max_diff=3E-3	)
       def     SCREAMING_SNAKE_CASE  (					self	):
        pass

       def     SCREAMING_SNAKE_CASE  (					self	):
        _UpperCAmelCase 				=self.get_dummy_components()
        _UpperCAmelCase 				=StableDiffusionXLImgaImgPipeline(**_SCREAMING_SNAKE_CASE	)
        _UpperCAmelCase 				=sd_pipe.to(_SCREAMING_SNAKE_CASE	)
        _UpperCAmelCase 				=sd_pipe.to(_SCREAMING_SNAKE_CASE	)
        sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE	)

        # forward without prompt embeds
        _UpperCAmelCase 				=self.get_dummy_inputs(_SCREAMING_SNAKE_CASE	)
        _UpperCAmelCase 				=3 * ['this is a negative prompt']
        _UpperCAmelCase 				=negative_prompt
        _UpperCAmelCase 				=3 * [inputs['prompt']]

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

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

        (
            _UpperCAmelCase						
        ) 				=sd_pipe.encode_prompt(_SCREAMING_SNAKE_CASE       ,					negative_prompt=_SCREAMING_SNAKE_CASE	)

        _UpperCAmelCase 				=sd_pipe(
            **_SCREAMING_SNAKE_CASE       ,					prompt_embeds=_SCREAMING_SNAKE_CASE       ,					negative_prompt_embeds=_SCREAMING_SNAKE_CASE       ,					pooled_prompt_embeds=_SCREAMING_SNAKE_CASE       ,					negative_pooled_prompt_embeds=_SCREAMING_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 						_a   (						unittest.TestCase       ):

       """simple docstring"""





       def     SCREAMING_SNAKE_CASE  (					self	):
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()
       def     SCREAMING_SNAKE_CASE  (					self       ,					_snake_case       ,					_snake_case="cpu"       ,					_snake_case=torch.floataa       ,					_snake_case=0	):
        _UpperCAmelCase 				=torch.Generator(device=_SCREAMING_SNAKE_CASE	).manual_seed(_SCREAMING_SNAKE_CASE	)
        _UpperCAmelCase 				=np.random.RandomState(_SCREAMING_SNAKE_CASE	).standard_normal((1, 4, 64, 64)	)
        _UpperCAmelCase 				=torch.from_numpy(_SCREAMING_SNAKE_CASE	).to(device=_SCREAMING_SNAKE_CASE       ,					dtype=_SCREAMING_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     SCREAMING_SNAKE_CASE  (					self	):
        _UpperCAmelCase 				=DiffusionPipeline.from_pretrained("stabilityai/stable-diffusion-2-base"	)
        pipe.to(_SCREAMING_SNAKE_CASE	)
        pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE	)

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

        assert image.shape == (1, 512, 512, 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 TYPE_CHECKING

from ..utils import _LazyModule


A_						:    str				=      {
    'config': [
        'EXTERNAL_DATA_FORMAT_SIZE_LIMIT',
        'OnnxConfig',
        'OnnxConfigWithPast',
        'OnnxSeq2SeqConfigWithPast',
        'PatchingSpec',
    ],
    'convert': ['export', 'validate_model_outputs'],
    'features': ['FeaturesManager'],
    'utils': ['ParameterFormat', 'compute_serialized_parameters_size'],
}


if TYPE_CHECKING:
							from .config import (
							    EXTERNAL_DATA_FORMAT_SIZE_LIMIT,
							    OnnxConfig,
							    OnnxConfigWithPast,
							    OnnxSeqaSeqConfigWithPast,
							    PatchingSpec,
							)
							from .convert import export, validate_model_outputs
							from .features import FeaturesManager
							from .utils import ParameterFormat, compute_serialized_parameters_size

else:
							import sys

							A_						:    Optional[Any]				=      _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)





