




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







 '''simple docstring'''





 lowercase__				      =      StableDiffusionXLImgaImgPipeline
 lowercase__				      =      TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
 lowercase__				      =      PipelineTesterMixin.required_optional_params - {"latents"}
 lowercase__				      =      TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 lowercase__				      =      IMAGE_TO_IMAGE_IMAGE_PARAMS
 lowercase__				      =      IMAGE_TO_IMAGE_IMAGE_PARAMS


 def 				UpperCamelCase_     (       self:	Optional[Any]	):




   '''simple docstring'''



   torch.manual_seed(0	)
   _snake_case							:	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,   )
   _snake_case							:	List[str]			=						EulerDiscreteScheduler(
       beta_start=0.00_085,   beta_end=0.012,   steps_offset=1,   beta_schedule="""scaled_linear""",   timestep_spacing="""leading""",   )
   torch.manual_seed(0	)
   _snake_case							:	Optional[Any]			=						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							:	Union[str, Any]			=						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=1_000,   hidden_act="""gelu""",   projection_dim=32,   )
   _snake_case							:	Dict			=						CLIPTextModel(a_	)
   _snake_case							:	Dict			=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""",   local_files_only=a_	)

   _snake_case							:	Tuple			=						CLIPTextModelWithProjection(a_	)
   _snake_case							:	int			=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""",   local_files_only=a_	)

   _snake_case							:	Union[str, Any]			=						{
       """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 				UpperCamelCase_     (       self:	Union[str, Any],   a_:	str,   a_:	Dict=0	):




   '''simple docstring'''



   _snake_case							:	int			=						floats_tensor((1, 3, 32, 32),   rng=random.Random(a_	)	).to(a_	)
   _snake_case							:	Optional[int]			=						image / 2 + 0.5
   if str(a_	).startswith("""mps"""	):
     _snake_case							:	Optional[int]			=						torch.manual_seed(a_	)
   else:
     _snake_case							:	str			=						torch.Generator(device=a_	).manual_seed(a_	)
   _snake_case							:	str			=						{
       """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 				UpperCamelCase_     (       self:	int	):




   '''simple docstring'''



   _snake_case							:	Tuple			=						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
   _snake_case							:	Tuple			=						self.get_dummy_components()
   _snake_case							:	List[Any]			=						StableDiffusionXLImgaImgPipeline(**a_	)
   _snake_case							:	Optional[Any]			=						sd_pipe.to(a_	)
   sd_pipe.set_progress_bar_config(disable=a_	)

   _snake_case							:	str			=						self.get_dummy_inputs(a_	)
   _snake_case							:	List[Any]			=						sd_pipe(**a_	).images
   _snake_case							:	int			=						image[0, -3:, -3:, -1]

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

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




   '''simple docstring'''



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


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




   '''simple docstring'''



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


 def 				UpperCamelCase_     (       self:	str	):




   '''simple docstring'''



   pass


 def 				UpperCamelCase_     (       self:	Dict	):




   '''simple docstring'''



   _snake_case							:	Optional[Any]			=						self.get_dummy_components()
   _snake_case							:	Tuple			=						StableDiffusionXLImgaImgPipeline(**a_	)
   _snake_case							:	Union[str, Any]			=						sd_pipe.to(a_	)
   _snake_case							:	int			=						sd_pipe.to(a_	)
   sd_pipe.set_progress_bar_config(disable=a_	)

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

   _snake_case							:	Any			=						sd_pipe(**a_	)
   _snake_case							:	List[str]			=						output.images[0, -3:, -3:, -1]

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

   (
       _snake_case      
   )							:	str			=						sd_pipe.encode_prompt(a_,   negative_prompt=a_	)

   _snake_case							:	Any			=						sd_pipe(
       **a_,   prompt_embeds=a_,   negative_prompt_embeds=a_,   pooled_prompt_embeds=a_,   negative_pooled_prompt_embeds=a_,   )
   _snake_case							:	Optional[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       lowercase(  unittest.TestCase				):







 '''simple docstring'''





 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



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


 def 				UpperCamelCase_     (       self:	List[str],   a_:	str,   a_:	Dict="cpu",   a_:	str=torch.floataa,   a_:	str=0	):




   '''simple docstring'''



   _snake_case							:	str			=						torch.Generator(device=a_	).manual_seed(a_	)
   _snake_case							:	Any			=						np.random.RandomState(a_	).standard_normal((1, 4, 64, 64)	)
   _snake_case							:	str			=						torch.from_numpy(a_	).to(device=a_,   dtype=a_	)
   _snake_case							:	int			=						{
       """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 				UpperCamelCase_     (       self:	Union[str, Any]	):




   '''simple docstring'''



   _snake_case							:	str			=						DiffusionPipeline.from_pretrained("""stabilityai/stable-diffusion-2-base"""	)
   pipe.to(a_	)
   pipe.set_progress_bar_config(disable=a_	)

   _snake_case							:	int			=						self.get_inputs(a_	)
   _snake_case							:	Dict			=						pipe(**a_	).images
   _snake_case							:	Optional[int]			=						image[0, -3:, -3:, -1].flatten()

   assert image.shape == (1, 512, 512, 3)
   _snake_case							:	Optional[int]			=						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 ...processing_utils import ProcessorMixin






class       lowercase(  __a				):







 '''simple docstring'''





 lowercase__				      =      ["image_processor", "feature_extractor"]
 lowercase__				      =      "TvltImageProcessor"
 lowercase__				      =      "TvltFeatureExtractor"


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




   '''simple docstring'''



   super().__init__(image_processor=a_,   feature_extractor=a_	)

   _snake_case							:	Any			=						image_processor
   _snake_case							:	Dict			=						feature_extractor


 def __call__(       self:	int,   a_:	str=None,   a_:	Tuple=None,   a_:	Dict=None,   a_:	str=None,   a_:	Optional[int]=False,   a_:	Tuple=False,   *a_:	List[str],   **a_:	int,   ):




   '''simple docstring'''



   if images is None and audio is None:
     raise ValueError("""You need to specify either an `images` or `audio` input to process."""	)

   _snake_case							:	Optional[int]			=						None
   if images is not None:
     _snake_case							:	Tuple			=						self.image_processor(a_,   mask_pixel=a_,   *a_,   **a_	)
   if images_mixed is not None:
     _snake_case							:	Optional[int]			=						self.image_processor(a_,   is_mixed=a_,   *a_,   **a_	)
   if audio is not None:
     _snake_case							:	Any			=						self.feature_extractor(
         a_,   *a_,   sampling_rate=a_,   mask_audio=a_,   **a_	)

   _snake_case							:	List[str]			=						{}
   if audio is not None:
     output_dict.update(a_	)
   if images is not None:
     output_dict.update(a_	)
   if images_mixed_dict is not None:
     output_dict.update(a_	)
   return output_dict


 @property
 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



   _snake_case							:	Dict			=						self.image_processor.model_input_names
   _snake_case							:	List[str]			=						self.feature_extractor.model_input_names
   return list(dict.fromkeys(image_processor_input_names + feature_extractor_input_names	)	)


