




from manim import *







class  UpperCAmelCase(	snake_case_	):



	"""simple docstring"""






	def        __a      (	self							)   ->     Dict:





								"""simple docstring"""


								lowercase__    :			Dict     =				Rectangle(height=0.5			, width=0.5							)
								lowercase__    :			Any     =				Rectangle(height=0.46			, width=0.46							).set_stroke(width=0							)

								lowercase__    :			Any     =				[mem.copy() for i in range(6							)]
								lowercase__    :			Tuple     =				[mem.copy() for i in range(6							)]
								lowercase__    :			Optional[int]     =				VGroup(*snake_case_							).arrange(snake_case_			, buff=0							)
								lowercase__    :			Tuple     =				VGroup(*snake_case_							).arrange(snake_case_			, buff=0							)
								lowercase__    :			Optional[int]     =				VGroup(snake_case_			, snake_case_							).arrange(snake_case_			, buff=0							)
								lowercase__    :			int     =				Text("CPU"			, font_size=24							)
								lowercase__    :			str     =				Group(snake_case_			, snake_case_							).arrange(snake_case_			, buff=0.5			, aligned_edge=snake_case_							)
								cpu.move_to([-2.5, -0.5, 0]							)
								self.add(snake_case_							)

								lowercase__    :			str     =				[mem.copy() for i in range(1							)]
								lowercase__    :			int     =				VGroup(*snake_case_							).arrange(snake_case_			, buff=0							)
								lowercase__    :			Optional[Any]     =				Text("GPU"			, font_size=24							)
								lowercase__    :			Optional[Any]     =				Group(snake_case_			, snake_case_							).arrange(snake_case_			, buff=0.5			, aligned_edge=snake_case_							)
								gpu.align_to(snake_case_			, snake_case_							)
								gpu.set_x(gpu.get_x() - 1							)

								self.add(snake_case_							)

								lowercase__    :			Dict     =				[mem.copy() for i in range(6							)]
								lowercase__    :			int     =				VGroup(*snake_case_							).arrange(snake_case_			, buff=0							)

								lowercase__    :			Dict     =				Text("Model"			, font_size=24							)
								lowercase__    :			List[Any]     =				Group(snake_case_			, snake_case_							).arrange(snake_case_			, buff=0.5			, aligned_edge=snake_case_							)
								model.move_to([3, -1.0, 0]							)

								self.play(
								    Create(snake_case_			, run_time=1							)			, Create(snake_case_			, run_time=1							)			, Create(snake_case_			, run_time=1							)			, )

								lowercase__    :			Tuple     =				MarkupText(
								    f"""First, an empty model skeleton is loaded\ninto <span fgcolor='{YELLOW}'>memory</span> without using much RAM."""			, font_size=24			, )

								lowercase__    :			Optional[Any]     =				Square(side_length=2.2							)
								key.move_to([-5, 2, 0]							)

								lowercase__    :			List[str]     =				MarkupText(
								    f"""<b>Key:</b>\n\n<span fgcolor='{YELLOW}'>●</span> Empty Model"""			, font_size=18			, )

								key_text.move_to([-5, 2.4, 0]							)

								step_a.move_to([2, 2, 0]							)
								self.play(Write(snake_case_			, run_time=2.5							)			, Write(snake_case_							)			, Write(snake_case_							)							)

								self.add(snake_case_							)

								lowercase__    :			List[Any]     =				[]
								lowercase__    :			Dict     =				[]
								lowercase__    :			Any     =				[]
								for i, rect in enumerate(snake_case_							):
															lowercase__    :			List[str]     =				Rectangle(height=0.46			, width=0.46							).set_stroke(width=0.0							).set_fill(snake_case_			, opacity=0.7							)
															cpu_target.move_to(snake_case_							)
															cpu_target.generate_target()
															lowercase__    :			int     =				0.46 / 4
															lowercase__    :			Any     =				0.46 / 3

															if i == 0:
																						cpu_target.target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT							)			, buff=0.02			, direction=snake_case_							)
																						cpu_target.target.set_x(cpu_target.target.get_x() + 0.1							)
															elif i == 3:
																						cpu_target.target.next_to(cpu_targs[0].target			, direction=snake_case_			, buff=0.0							)
															else:
																						cpu_target.target.next_to(cpu_targs[i - 1].target			, direction=snake_case_			, buff=0.0							)
															cpu_targs.append(snake_case_							)

															first_animations.append(rect.animate(run_time=0.5							).set_stroke(snake_case_							)							)
															second_animations.append(MoveToTarget(snake_case_			, run_time=1.5							)							)

								self.play(*snake_case_							)
								self.play(*snake_case_							)

								self.wait()




import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import (
    DiffusionPipeline,
    UnCLIPImageVariationPipeline,
    UnCLIPScheduler,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, load_image, require_torch_gpu, skip_mps

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()






class     __a			(  SCREAMING_SNAKE_CASE						,     unittest.TestCase					):
	SCREAMING_SNAKE_CASE  	=							UnCLIPImageVariationPipeline
	SCREAMING_SNAKE_CASE  	=							IMAGE_VARIATION_PARAMS - {"height", "width", "guidance_scale"}
	SCREAMING_SNAKE_CASE  	=							IMAGE_VARIATION_BATCH_PARAMS

	SCREAMING_SNAKE_CASE  	=							[
	    "generator",
	    "return_dict",
	    "decoder_num_inference_steps",
	    "super_res_num_inference_steps",
	]
	SCREAMING_SNAKE_CASE  	=							False



	@property
	def 		UpperCamelCase				(    self				:     int)->				Union[str, Any]:
						return 32



	@property
	def 		UpperCamelCase				(    self				:     List[Any])->				List[str]:
						return 32



	@property
	def 		UpperCamelCase				(    self				:     Dict)->				Dict:
						return self.time_input_dim



	@property
	def 		UpperCamelCase				(    self				:     Dict)->				Optional[int]:
						return self.time_input_dim * 4



	@property
	def 		UpperCamelCase				(    self				:     str)->				str:
						return 1_00



	@property
	def 		UpperCamelCase				(    self				:     Optional[Any])->				Union[str, Any]:
						__lowerCAmelCase						 =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")
						return tokenizer



	@property
	def 		UpperCamelCase				(    self				:     List[str])->				List[Any]:
						torch.manual_seed(0)
						__lowerCAmelCase						 =CLIPTextConfig(
						    bos_token_id=0					,						eos_token_id=2					,						hidden_size=self.text_embedder_hidden_size					,						projection_dim=self.text_embedder_hidden_size					,						intermediate_size=37					,						layer_norm_eps=1e-05					,						num_attention_heads=4					,						num_hidden_layers=5					,						pad_token_id=1					,						vocab_size=10_00					,						)
						return CLIPTextModelWithProjection(snake_case_)



	@property
	def 		UpperCamelCase				(    self				:     List[str])->				Dict:
						torch.manual_seed(0)
						__lowerCAmelCase						 =CLIPVisionConfig(
						    hidden_size=self.text_embedder_hidden_size					,						projection_dim=self.text_embedder_hidden_size					,						num_hidden_layers=5					,						num_attention_heads=4					,						image_size=32					,						intermediate_size=37					,						patch_size=1					,						)
						return CLIPVisionModelWithProjection(snake_case_)



	@property
	def 		UpperCamelCase				(    self				:     str)->				Union[str, Any]:
						torch.manual_seed(0)

						__lowerCAmelCase						 ={
						    """clip_embeddings_dim""": self.text_embedder_hidden_size,
						    """time_embed_dim""": self.time_embed_dim,
						    """cross_attention_dim""": self.cross_attention_dim,
						}

						__lowerCAmelCase						 =UnCLIPTextProjModel(**snake_case_)
						return model



	@property
	def 		UpperCamelCase				(    self				:     Optional[int])->				List[str]:
						torch.manual_seed(0)

						__lowerCAmelCase						 ={
						    """sample_size""": 32,
						    # RGB in channels
						    """in_channels""": 3,
						    # Out channels is double in channels because predicts mean and variance
						    """out_channels""": 6,
						    """down_block_types""": ("""ResnetDownsampleBlock2D""", """SimpleCrossAttnDownBlock2D"""),
						    """up_block_types""": ("""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""),
						    """mid_block_type""": """UNetMidBlock2DSimpleCrossAttn""",
						    """block_out_channels""": (self.block_out_channels_a, self.block_out_channels_a * 2),
						    """layers_per_block""": 1,
						    """cross_attention_dim""": self.cross_attention_dim,
						    """attention_head_dim""": 4,
						    """resnet_time_scale_shift""": """scale_shift""",
						    """class_embed_type""": """identity""",
						}

						__lowerCAmelCase						 =UNetaDConditionModel(**snake_case_)
						return model



	@property
	def 		UpperCamelCase				(    self				:     int)->				str:
						return {
						    "sample_size": 64,
						    "layers_per_block": 1,
						    "down_block_types": ("ResnetDownsampleBlock2D", "ResnetDownsampleBlock2D"),
						    "up_block_types": ("ResnetUpsampleBlock2D", "ResnetUpsampleBlock2D"),
						    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
						    "in_channels": 6,
						    "out_channels": 3,
						}



	@property
	def 		UpperCamelCase				(    self				:     Optional[int])->				List[Any]:
						torch.manual_seed(0)

						__lowerCAmelCase						 =UNetaDModel(**self.dummy_super_res_kwargs)
						return model



	@property
	def 		UpperCamelCase				(    self				:     Tuple)->				int:
						# seeded differently to get different unet than `self.dummy_super_res_first`
						torch.manual_seed(1)

						__lowerCAmelCase						 =UNetaDModel(**self.dummy_super_res_kwargs)
						return model



	def 		UpperCamelCase				(    self				:     List[Any])->				Tuple:
						__lowerCAmelCase						 =self.dummy_decoder
						__lowerCAmelCase						 =self.dummy_text_proj
						__lowerCAmelCase						 =self.dummy_text_encoder
						__lowerCAmelCase						 =self.dummy_tokenizer
						__lowerCAmelCase						 =self.dummy_super_res_first
						__lowerCAmelCase						 =self.dummy_super_res_last

						__lowerCAmelCase						 =UnCLIPScheduler(
						    variance_type="""learned_range"""					,						prediction_type="""epsilon"""					,						num_train_timesteps=10_00					,						)

						__lowerCAmelCase						 =UnCLIPScheduler(
						    variance_type="""fixed_small_log"""					,						prediction_type="""epsilon"""					,						num_train_timesteps=10_00					,						)

						__lowerCAmelCase						 =CLIPImageProcessor(crop_size=32					,						size=32)

						__lowerCAmelCase						 =self.dummy_image_encoder

						return {
						    "decoder": decoder,
						    "text_encoder": text_encoder,
						    "tokenizer": tokenizer,
						    "text_proj": text_proj,
						    "feature_extractor": feature_extractor,
						    "image_encoder": image_encoder,
						    "super_res_first": super_res_first,
						    "super_res_last": super_res_last,
						    "decoder_scheduler": decoder_scheduler,
						    "super_res_scheduler": super_res_scheduler,
						}



	def 		UpperCamelCase				(    self				:     Optional[int]					,						snake_case_				:     Optional[int]					,						snake_case_				:     Optional[int]=0					,						snake_case_				:     List[str]=True)->				List[str]:
						__lowerCAmelCase						 =floats_tensor((1, 3, 32, 32)					,						rng=random.Random(snake_case_)).to(snake_case_)
						if str(snake_case_).startswith("""mps"""):
											__lowerCAmelCase						 =torch.manual_seed(snake_case_)
						else:
											__lowerCAmelCase						 =torch.Generator(device=snake_case_).manual_seed(snake_case_)

						if pil_image:
											__lowerCAmelCase						 =input_image * 0.5 + 0.5
											__lowerCAmelCase						 =input_image.clamp(0					,						1)
											__lowerCAmelCase						 =input_image.cpu().permute(0					,						2					,						3					,						1).float().numpy()
											__lowerCAmelCase						 =DiffusionPipeline.numpy_to_pil(snake_case_)[0]

						return {
						    "image": input_image,
						    "generator": generator,
						    "decoder_num_inference_steps": 2,
						    "super_res_num_inference_steps": 2,
						    "output_type": "np",
						}



	def 		UpperCamelCase				(    self				:     Optional[Any])->				str:
						__lowerCAmelCase						 ="""cpu"""

						__lowerCAmelCase						 =self.get_dummy_components()

						__lowerCAmelCase						 =self.pipeline_class(**snake_case_)
						__lowerCAmelCase						 =pipe.to(snake_case_)

						pipe.set_progress_bar_config(disable=snake_case_)

						__lowerCAmelCase						 =self.get_dummy_inputs(snake_case_					,						pil_image=snake_case_)

						__lowerCAmelCase						 =pipe(**snake_case_)
						__lowerCAmelCase						 =output.images

						__lowerCAmelCase						 =self.get_dummy_inputs(snake_case_					,						pil_image=snake_case_)

						__lowerCAmelCase						 =pipe(
						    **snake_case_					,						return_dict=snake_case_					,						)[0]

						__lowerCAmelCase						 =image[0, -3:, -3:, -1]
						__lowerCAmelCase						 =image_from_tuple[0, -3:, -3:, -1]

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

						__lowerCAmelCase						 =np.array(
						    [
						        0.9_9_9_7,
						        0.0_0_0_2,
						        0.9_9_9_7,
						        0.9_9_9_7,
						        0.9_9_6_9,
						        0.0_0_2_3,
						        0.9_9_9_7,
						        0.9_9_6_9,
						        0.9_9_7_0,
						    ])

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



	def 		UpperCamelCase				(    self				:     Optional[Any])->				List[Any]:
						__lowerCAmelCase						 ="""cpu"""

						__lowerCAmelCase						 =self.get_dummy_components()

						__lowerCAmelCase						 =self.pipeline_class(**snake_case_)
						__lowerCAmelCase						 =pipe.to(snake_case_)

						pipe.set_progress_bar_config(disable=snake_case_)

						__lowerCAmelCase						 =self.get_dummy_inputs(snake_case_					,						pil_image=snake_case_)

						__lowerCAmelCase						 =pipe(**snake_case_)
						__lowerCAmelCase						 =output.images

						__lowerCAmelCase						 =self.get_dummy_inputs(snake_case_					,						pil_image=snake_case_)

						__lowerCAmelCase						 =pipe(
						    **snake_case_					,						return_dict=snake_case_					,						)[0]

						__lowerCAmelCase						 =image[0, -3:, -3:, -1]
						__lowerCAmelCase						 =image_from_tuple[0, -3:, -3:, -1]

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

						__lowerCAmelCase						 =np.array([0.9_9_9_7, 0.0_0_0_3, 0.9_9_9_7, 0.9_9_9_7, 0.9_9_7_0, 0.0_0_2_4, 0.9_9_9_7, 0.9_9_7_1, 0.9_9_7_1])

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



	def 		UpperCamelCase				(    self				:     int)->				Optional[Any]:
						__lowerCAmelCase						 ="""cpu"""

						__lowerCAmelCase						 =self.get_dummy_components()

						__lowerCAmelCase						 =self.pipeline_class(**snake_case_)
						__lowerCAmelCase						 =pipe.to(snake_case_)

						pipe.set_progress_bar_config(disable=snake_case_)

						__lowerCAmelCase						 =self.get_dummy_inputs(snake_case_					,						pil_image=snake_case_)
						__lowerCAmelCase						 =[
						    pipeline_inputs["""image"""],
						    pipeline_inputs["""image"""],
						]

						__lowerCAmelCase						 =pipe(**snake_case_)
						__lowerCAmelCase						 =output.images

						__lowerCAmelCase						 =self.get_dummy_inputs(snake_case_					,						pil_image=snake_case_)
						__lowerCAmelCase						 =[
						    tuple_pipeline_inputs["""image"""],
						    tuple_pipeline_inputs["""image"""],
						]

						__lowerCAmelCase						 =pipe(
						    **snake_case_					,						return_dict=snake_case_					,						)[0]

						__lowerCAmelCase						 =image[0, -3:, -3:, -1]
						__lowerCAmelCase						 =image_from_tuple[0, -3:, -3:, -1]

						assert image.shape == (2, 64, 64, 3)

						__lowerCAmelCase						 =np.array(
						    [
						        0.9_9_9_7,
						        0.9_9_8_9,
						        0.0_0_0_8,
						        0.0_0_2_1,
						        0.9_9_6_0,
						        0.0_0_1_8,
						        0.0_0_1_4,
						        0.0_0_0_2,
						        0.9_9_3_3,
						    ])

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



	def 		UpperCamelCase				(    self				:     Tuple)->				Optional[int]:
						__lowerCAmelCase						 =torch.device("""cpu""")






						class     __a			:
							SCREAMING_SNAKE_CASE  	=							1


						__lowerCAmelCase						 =self.get_dummy_components()

						__lowerCAmelCase						 =self.pipeline_class(**snake_case_)
						__lowerCAmelCase						 =pipe.to(snake_case_)

						pipe.set_progress_bar_config(disable=snake_case_)

						__lowerCAmelCase						 =torch.Generator(device=snake_case_).manual_seed(0)
						__lowerCAmelCase						 =pipe.decoder.dtype
						__lowerCAmelCase						 =1

						__lowerCAmelCase						 =(
						    batch_size,
						    pipe.decoder.config.in_channels,
						    pipe.decoder.config.sample_size,
						    pipe.decoder.config.sample_size,
						)
						__lowerCAmelCase						 =pipe.prepare_latents(
						    snake_case_					,						dtype=snake_case_					,						device=snake_case_					,						generator=snake_case_					,						latents=snake_case_					,						scheduler=DummyScheduler())

						__lowerCAmelCase						 =(
						    batch_size,
						    pipe.super_res_first.config.in_channels // 2,
						    pipe.super_res_first.config.sample_size,
						    pipe.super_res_first.config.sample_size,
						)
						__lowerCAmelCase						 =pipe.prepare_latents(
						    snake_case_					,						dtype=snake_case_					,						device=snake_case_					,						generator=snake_case_					,						latents=snake_case_					,						scheduler=DummyScheduler())

						__lowerCAmelCase						 =self.get_dummy_inputs(snake_case_					,						pil_image=snake_case_)

						__lowerCAmelCase						 =pipe(
						    **snake_case_					,						decoder_latents=snake_case_					,						super_res_latents=snake_case_).images

						__lowerCAmelCase						 =self.get_dummy_inputs(snake_case_					,						pil_image=snake_case_)
						# Don't pass image, instead pass embedding
						__lowerCAmelCase						 =pipeline_inputs.pop("""image""")
						__lowerCAmelCase						 =pipe.image_encoder(snake_case_).image_embeds

						__lowerCAmelCase						 =pipe(
						    **snake_case_					,						decoder_latents=snake_case_					,						super_res_latents=snake_case_					,						image_embeddings=snake_case_					,						).images

						# make sure passing text embeddings manually is identical
						assert np.abs(img_out_a - img_out_a).max() < 1e-4



	@skip_mps
	def 		UpperCamelCase				(    self				:     Optional[int])->				Optional[int]:
						__lowerCAmelCase						 =torch_device == """cpu"""

						# Check is relaxed because there is not a torch 2.0 sliced attention added kv processor
						__lowerCAmelCase						 =1e-2

						self._test_attention_slicing_forward_pass(
						    test_max_difference=snake_case_					,						expected_max_diff=snake_case_)



	@skip_mps
	def 		UpperCamelCase				(    self				:     int)->				int:
						__lowerCAmelCase						 =torch_device == """cpu"""
						__lowerCAmelCase						 =True
						__lowerCAmelCase						 =[
						    """decoder_num_inference_steps""",
						    """super_res_num_inference_steps""",
						]

						self._test_inference_batch_single_identical(
						    test_max_difference=snake_case_					,						relax_max_difference=snake_case_					,						additional_params_copy_to_batched_inputs=snake_case_					,						)



	def 		UpperCamelCase				(    self				:     List[Any])->				int:
						__lowerCAmelCase						 =[
						    """decoder_num_inference_steps""",
						    """super_res_num_inference_steps""",
						]

						if torch_device == "mps":
											# TODO: MPS errors with larger batch sizes
											__lowerCAmelCase						 =[2, 3]
											self._test_inference_batch_consistent(
											    batch_sizes=snake_case_					,						additional_params_copy_to_batched_inputs=snake_case_					,						)
						else:
											self._test_inference_batch_consistent(
											    additional_params_copy_to_batched_inputs=snake_case_)



	@skip_mps
	def 		UpperCamelCase				(    self				:     Optional[Any])->				Tuple:
						return super().test_dict_tuple_outputs_equivalent()



	@skip_mps
	def 		UpperCamelCase				(    self				:     Optional[int])->				Tuple:
						return super().test_save_load_local()



	@skip_mps
	def 		UpperCamelCase				(    self				:     Union[str, Any])->				Any:
						return super().test_save_load_optional_components()






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



	def 		UpperCamelCase				(    self				:     Any)->				Optional[Any]:
						# clean up the VRAM after each test
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()



	def 		UpperCamelCase				(    self				:     Optional[Any])->				List[Any]:
						__lowerCAmelCase						 =load_image(
						    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png""")
						__lowerCAmelCase						 =load_numpy(
						    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
						    """/unclip/karlo_v1_alpha_cat_variation_fp16.npy""")

						__lowerCAmelCase						 =UnCLIPImageVariationPipeline.from_pretrained(
						    """kakaobrain/karlo-v1-alpha-image-variations"""					,						torch_dtype=torch.floataa)
						__lowerCAmelCase						 =pipeline.to(snake_case_)
						pipeline.set_progress_bar_config(disable=snake_case_)

						__lowerCAmelCase						 =torch.Generator(device="""cpu""").manual_seed(0)
						__lowerCAmelCase						 =pipeline(
						    snake_case_					,						generator=snake_case_					,						output_type="""np"""					,						)

						__lowerCAmelCase						 =output.images[0]

						assert image.shape == (2_56, 2_56, 3)

						assert_mean_pixel_difference(snake_case_					,						snake_case_					,						15)




