





import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()

class        snake_case__    (__lowerCamelCase ,     __lowerCamelCase ,     __lowerCamelCase ,     unittest.TestCase		):







						"""simple docstring"""




						__lowerCAmelCase					:int      =					StableUnCLIPPipeline
						__lowerCAmelCase					:Optional[int]      =					TEXT_TO_IMAGE_PARAMS
						__lowerCAmelCase					:List[str]      =					TEXT_TO_IMAGE_BATCH_PARAMS
						__lowerCAmelCase					:Optional[int]      =					TEXT_TO_IMAGE_IMAGE_PARAMS
						__lowerCAmelCase					:Optional[Any]      =					TEXT_TO_IMAGE_IMAGE_PARAMS

						# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
						__lowerCAmelCase					:Union[str, Any]      =					False

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				List[Any]:
									"""simple docstring"""




									a__       :						Tuple							=  3_2
									a__       :						int							=  embedder_hidden_size

									# prior components

									torch.manual_seed(0  )
									a__       :						Dict							=  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

									torch.manual_seed(0  )
									a__       :						Union[str, Any]							=  CLIPTextModelWithProjection(
									    CLIPTextConfig(
									        bos_token_id=0 ,       eos_token_id=2 ,       hidden_size=__lowercase ,       projection_dim=__lowercase ,       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 ,       )  )

									torch.manual_seed(0  )
									a__       :						Any							=  PriorTransformer(
									    num_attention_heads=2 ,       attention_head_dim=1_2 ,       embedding_dim=__lowercase ,       num_layers=1 ,       )

									torch.manual_seed(0  )
									a__       :						Any							=  DDPMScheduler(
									    variance_type="""fixed_small_log""" ,       prediction_type="""sample""" ,       num_train_timesteps=1_0_0_0 ,       clip_sample=__lowercase ,       clip_sample_range=5.0 ,       beta_schedule="""squaredcos_cap_v2""" ,       )

									# regular denoising components

									torch.manual_seed(0  )
									a__       :						List[Any]							=  StableUnCLIPImageNormalizer(embedding_dim=__lowercase  )
									a__       :						int							=  DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""  )

									torch.manual_seed(0  )
									a__       :						List[Any]							=  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

									torch.manual_seed(0  )
									a__       :						int							=  CLIPTextModel(
									    CLIPTextConfig(
									        bos_token_id=0 ,       eos_token_id=2 ,       hidden_size=__lowercase ,       projection_dim=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 ,       )  )

									torch.manual_seed(0  )
									a__       :						Optional[int]							=  UNetaDConditionModel(
									    sample_size=3_2 ,       in_channels=4 ,       out_channels=4 ,       down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""") ,       up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""") ,       block_out_channels=(3_2, 6_4) ,       attention_head_dim=(2, 4) ,       class_embed_type="""projection""" ,       projection_class_embeddings_input_dim=embedder_projection_dim * 2 ,       cross_attention_dim=__lowercase ,       layers_per_block=1 ,       upcast_attention=__lowercase ,       use_linear_projection=__lowercase ,       )

									torch.manual_seed(0  )
									a__       :						Dict							=  DDIMScheduler(
									    beta_schedule="""scaled_linear""" ,       beta_start=0.0_0_0_8_5 ,       beta_end=0.0_1_2 ,       prediction_type="""v_prediction""" ,       set_alpha_to_one=__lowercase ,       steps_offset=1 ,       )

									torch.manual_seed(0  )
									a__       :						Optional[Any]							=  AutoencoderKL()

									a__       :						Union[str, Any]							=  {
									    # prior components
									    '''prior_tokenizer''': prior_tokenizer,
									    '''prior_text_encoder''': prior_text_encoder,
									    '''prior''': prior,
									    '''prior_scheduler''': prior_scheduler,
									    # image noising components
									    '''image_normalizer''': image_normalizer,
									    '''image_noising_scheduler''': image_noising_scheduler,
									    # regular denoising components
									    '''tokenizer''': tokenizer,
									    '''text_encoder''': text_encoder,
									    '''unet''': unet,
									    '''scheduler''': scheduler,
									    '''vae''': vae,
									}

									return components

						def 							SCREAMING_SNAKE_CASE__(       self ,       __lowercase ,       __lowercase=0  ) ->				Tuple:
									"""simple docstring"""




									if str(__lowercase  ).startswith("""mps"""  ):
												a__       :						List[Any]							=  torch.manual_seed(__lowercase  )
									else:
												a__       :						List[Any]							=  torch.Generator(device=__lowercase  ).manual_seed(__lowercase  )
									a__       :						List[Any]							=  {
									    '''prompt''': '''A painting of a squirrel eating a burger''',
									    '''generator''': generator,
									    '''num_inference_steps''': 2,
									    '''prior_num_inference_steps''': 2,
									    '''output_type''': '''numpy''',
									}
									return inputs

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				int:
									"""simple docstring"""




									a__       :						List[str]							=  torch_device == '''cpu'''

									self._test_attention_slicing_forward_pass(test_max_difference=__lowercase  )

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Union[str, Any]:
									"""simple docstring"""




									a__       :						str							=  torch_device in ['''cpu''', '''mps''']

									self._test_inference_batch_single_identical(test_max_difference=__lowercase  )




@slow
@require_torch_gpu
class        snake_case__    (unittest.TestCase		):







						"""simple docstring"""




						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Tuple:
									"""simple docstring"""




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

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Optional[int]:
									"""simple docstring"""




									a__       :						Optional[Any]							=  load_numpy(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""  )

									a__       :						Tuple							=  StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l""" ,       torch_dtype=torch.floataa  )
									pipe.to(__lowercase  )
									pipe.set_progress_bar_config(disable=__lowercase  )
									# stable unclip will oom when integration tests are run on a V100,
									# so turn on memory savings
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									a__       :						Dict							=  torch.Generator(device="""cpu"""  ).manual_seed(0  )
									a__       :						Optional[Any]							=  pipe("""anime turle""" ,       generator=__lowercase ,       output_type="""np"""  )

									a__       :						Union[str, Any]							=  output.images[0]

									assert image.shape == (7_6_8, 7_6_8, 3)

									assert_mean_pixel_difference(__lowercase ,       __lowercase  )

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Union[str, Any]:
									"""simple docstring"""




									torch.cuda.empty_cache()
									torch.cuda.reset_max_memory_allocated()
									torch.cuda.reset_peak_memory_stats()

									a__       :						List[str]							=  StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l""" ,       torch_dtype=torch.floataa  )
									a__       :						Dict							=  pipe.to(__lowercase  )
									pipe.set_progress_bar_config(disable=__lowercase  )
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									a__       :						Dict							=  pipe(
									    """anime turtle""" ,       prior_num_inference_steps=2 ,       num_inference_steps=2 ,       output_type="""np""" ,       )

									a__       :						Dict							=  torch.cuda.max_memory_allocated()
									# make sure that less than 7 GB is allocated
									assert mem_bytes < 7 * 1_0**9









import os
import sys
from contextlib import contextmanager


# Windows only
if os.name == "nt":
		import ctypes
		import msvcrt  # noqa

		class        snake_case__    (ctypes.Structure		):







								"""simple docstring"""




								__lowerCAmelCase					:Dict      =					[("size", ctypes.c_int), ("visible", ctypes.c_byte)]

def      lowerCAmelCase_					(  )							->	List[Any]:







			"""simple docstring"""
			if os.name == "nt":
						a__       :						int							=  CursorInfo()
						a__       :						Union[str, Any]							=  ctypes.windll.kernelaa.GetStdHandle(-11)
						ctypes.windll.kernelaa.GetConsoleCursorInfo(_lowercase						,       ctypes.byref(_lowercase))
						a__       :						List[str]							=  False
						ctypes.windll.kernelaa.SetConsoleCursorInfo(_lowercase						,       ctypes.byref(_lowercase))
			elif os.name == "posix":
						sys.stdout.write("""\033[?25l""")
						sys.stdout.flush()

def      lowerCAmelCase_					(  )							->	Optional[Any]:







			"""simple docstring"""
			if os.name == "nt":
						a__       :						List[Any]							=  CursorInfo()
						a__       :						Optional[int]							=  ctypes.windll.kernelaa.GetStdHandle(-11)
						ctypes.windll.kernelaa.GetConsoleCursorInfo(_lowercase						,       ctypes.byref(_lowercase))
						a__       :						Dict							=  True
						ctypes.windll.kernelaa.SetConsoleCursorInfo(_lowercase						,       ctypes.byref(_lowercase))
			elif os.name == "posix":
						sys.stdout.write("""\033[?25h""")
						sys.stdout.flush()





@contextmanager
def      lowerCAmelCase_					(  )							->	Any:







			"""simple docstring"""
			try:
						hide_cursor()
						yield
			finally:
						show_cursor()




