

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







    """simple docstring"""





    lowerCamelCase			=							StableDiffusionXLImgaImgPipeline
    lowerCamelCase			=							TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
    lowerCamelCase			=							PipelineTesterMixin.required_optional_params - {'latents'}
    lowerCamelCase			=							TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
    lowerCamelCase			=							IMAGE_TO_IMAGE_IMAGE_PARAMS
    lowerCamelCase			=							IMAGE_TO_IMAGE_IMAGE_PARAMS


    def 						snake_case__     (      self :       Union[str, Any]		)->    Optional[Any]:





        '''simple docstring'''







        torch.manual_seed(0		)
        A__          =   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=lowercase_,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,)
        A__          =   EulerDiscreteScheduler(
            beta_start=0.00_085,beta_end=0.012,steps_offset=1,beta_schedule='scaled_linear',timestep_spacing='leading',)
        torch.manual_seed(0		)
        A__          =   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		)
        A__          =   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,)
        A__          =   CLIPTextModel(lowercase_		)
        A__          =   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip',local_files_only=lowercase_		)

        A__          =   CLIPTextModelWithProjection(lowercase_		)
        A__          =   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip',local_files_only=lowercase_		)

        A__          =   {
            '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 :       List[str],lowercase_ :       Optional[Any],lowercase_ :       str=0		)->    Optional[int]:





        '''simple docstring'''







        A__          =   floats_tensor((1, 3, 3_2, 3_2),rng=random.Random(lowercase_		)		).to(lowercase_		)
        A__          =   image / 2 + 0.5
        if str(lowercase_		).startswith('mps'		):
            A__          =   torch.manual_seed(lowercase_		)
        else:
            A__          =   torch.Generator(device=lowercase_		).manual_seed(lowercase_		)
        A__          =   {
            '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 :       Union[str, Any]		)->    str:





        '''simple docstring'''







        A__          =   'cpu'  # ensure determinism for the device-dependent torch.Generator
        A__          =   self.get_dummy_components()
        A__          =   StableDiffusionXLImgaImgPipeline(**lowercase_		)
        A__          =   sd_pipe.to(lowercase_		)
        sd_pipe.set_progress_bar_config(disable=lowercase_		)

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

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

        A__          =   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 						snake_case__     (      self :       List[Any]		)->    str:





        '''simple docstring'''







        super().test_attention_slicing_forward_pass(expected_max_diff=3E-3		)


    def 						snake_case__     (      self :       Tuple		)->    Dict:





        '''simple docstring'''







        super().test_inference_batch_single_identical(expected_max_diff=3E-3		)


    def 						snake_case__     (      self :       int		)->    Tuple:





        '''simple docstring'''







        pass


    def 						snake_case__     (      self :       List[str]		)->    Tuple:





        '''simple docstring'''







        A__          =   self.get_dummy_components()
        A__          =   StableDiffusionXLImgaImgPipeline(**lowercase_		)
        A__          =   sd_pipe.to(lowercase_		)
        A__          =   sd_pipe.to(lowercase_		)
        sd_pipe.set_progress_bar_config(disable=lowercase_		)

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

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

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

        (
            (
            A__
        )      ,						(
            A__
        )      ,						(
            A__
        )      ,						(
            A__
        )      ,						
        )          =   sd_pipe.encode_prompt(lowercase_,negative_prompt=lowercase_		)

        A__          =   sd_pipe(
            **lowercase_,prompt_embeds=lowercase_,negative_prompt_embeds=lowercase_,pooled_prompt_embeds=lowercase_,negative_pooled_prompt_embeds=lowercase_,)
        A__          =   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 :       Union[str, Any]		)->    List[Any]:





        '''simple docstring'''







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


    def 						snake_case__     (      self :       Optional[Any],lowercase_ :       Union[str, Any],lowercase_ :       Any="cpu",lowercase_ :       str=torch.floataa,lowercase_ :       str=0		)->    List[str]:





        '''simple docstring'''







        A__          =   torch.Generator(device=lowercase_		).manual_seed(lowercase_		)
        A__          =   np.random.RandomState(lowercase_		).standard_normal((1, 4, 6_4, 6_4)		)
        A__          =   torch.from_numpy(lowercase_		).to(device=lowercase_,dtype=lowercase_		)
        A__          =   {
            '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 :       List[Any]		)->    Dict:





        '''simple docstring'''







        A__          =   DiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-base'		)
        pipe.to(lowercase_		)
        pipe.set_progress_bar_config(disable=lowercase_		)

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

        assert image.shape == (1, 5_1_2, 5_1_2, 3)
        A__          =   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"""





import argparse
from collections import defaultdict







def 	__a     (		__lowerCamelCase,		__lowerCamelCase,		__lowerCamelCase,		__lowerCamelCase,		__lowerCamelCase  ):
   UpperCAmelCase_					:					int			     =    f"""{file}_{class_name}_{test_name}"""
   done_test[_id] += 1

   with open(__lowerCamelCase,		"r"  ) as f:
      UpperCAmelCase_					:					List[Any]			     =    f.readlines()

   UpperCAmelCase_					:					int			     =    f"""class {class_name}("""
   UpperCAmelCase_					:					Optional[Any]			     =    f"""{4 * " "}def {test_name}("""
   UpperCAmelCase_					:					Optional[Any]			     =    f"""{8 * " "}{correct_line.split()[0]}"""
   UpperCAmelCase_					:					Tuple			     =    f"""{16 * " "}{correct_line.split()[0]}"""
   UpperCAmelCase_					:					int			     =    False
   UpperCAmelCase_					:					Union[str, Any]			     =    False
   UpperCAmelCase_					:					str			     =    False
   UpperCAmelCase_					:					Optional[Any]			     =    False
   UpperCAmelCase_					:					List[str]			     =    0
   UpperCAmelCase_					:					Optional[int]			     =    0

   UpperCAmelCase_					:					int			     =    []
   for line in lines:
      if line.startswith(__lowerCamelCase  ):
         UpperCAmelCase_					:					Tuple			     =    True
      elif in_class and line.startswith(__lowerCamelCase  ):
         UpperCAmelCase_					:					Optional[int]			     =    True
      elif in_class and in_func and (line.startswith(__lowerCamelCase  ) or line.startswith(__lowerCamelCase  )):
         UpperCAmelCase_					:					Any			     =    len(line.split(correct_line.split()[0]  )[0]  )
         count += 1

         if count == done_test[_id]:
            UpperCAmelCase_					:					Union[str, Any]			     =    True

      if in_class and in_func and in_line:
         if ")" not in line:
            continue
         else:
            UpperCAmelCase_					:					Any			     =    True

      if in_class and in_func and in_line and insert_line:
         new_lines.append(f"""{spaces * " "}{correct_line}"""  )
         UpperCAmelCase_					:					int			     =    False
      else:
         new_lines.append(__lowerCamelCase  )

   with open(__lowerCamelCase,		"w"  ) as f:
      for line in new_lines:
         f.write(__lowerCamelCase  )







def 	__a     (		__lowerCamelCase,		__lowerCamelCase=None  ):
   if fail is not None:
      with open(__lowerCamelCase,		"r"  ) as f:
         UpperCAmelCase_					:					Tuple			     =    {l.strip() for l in f.readlines()}
   else:
      UpperCAmelCase_					:					str			     =    None

   with open(__lowerCamelCase,		"r"  ) as f:
      UpperCAmelCase_					:					Optional[int]			     =    f.readlines()

   UpperCAmelCase_					:					Any			     =    defaultdict(__lowerCamelCase  )
   for line in correct_lines:
      UpperCAmelCase_     ,   UpperCAmelCase_     ,   UpperCAmelCase_     ,   UpperCAmelCase_					:					Any			     =    line.split(";"  )
      if test_failures is None or "::".join([file, class_name, test_name]  ) in test_failures:
         overwrite_file(__lowerCamelCase,		__lowerCamelCase,		__lowerCamelCase,		__lowerCamelCase,		__lowerCamelCase  )


if __name__ == "__main__":
    _a		       =							argparse.ArgumentParser()
    parser.add_argument('--correct_filename', help='filename of tests with expected result')
    parser.add_argument('--fail_filename', help='filename of test failures', type=str, default=None)
    _a		       =							parser.parse_args()

    main(args.correct_filename, args.fail_filename)


