

'''simple docstring'''
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
@skip_mps
class 					_snake_case		(__SCREAMING_SNAKE_CASE			,			unittest.TestCase):
			__A  :	Union[str, Any]        =VideoToVideoSDPipeline
			__A  :	Tuple        =TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}) - {"image", "width", "height"}
			__A  :	Union[str, Any]        =TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}) - {"image"}
			__A  :	str        =PipelineTesterMixin.required_optional_params - {"latents"}
			__A  :	Dict        =False

			# No `output_type`.
			__A  :	Optional[int]        =frozenset(
			    [
			        "num_inference_steps",
			        "generator",
			        "latents",
			        "return_dict",
			        "callback",
			        "callback_steps",
			    ])






			def 		UpperCamelCase__							(  self     ):
				torch.manual_seed(0     )
				UpperCAmelCase_     :      Optional[int]						=  UNetaDConditionModel(
				    block_out_channels=(32, 64, 64, 64)			,layers_per_block=2			,sample_size=32			,in_channels=4			,out_channels=4			,down_block_types=("CrossAttnDownBlock3D", "CrossAttnDownBlock3D", "CrossAttnDownBlock3D", "DownBlock3D")			,up_block_types=("UpBlock3D", "CrossAttnUpBlock3D", "CrossAttnUpBlock3D", "CrossAttnUpBlock3D")			,cross_attention_dim=32			,attention_head_dim=4			,)
				UpperCAmelCase_     :      int						=  DDIMScheduler(
				    beta_start=0.00085			,beta_end=0.012			,beta_schedule="scaled_linear"			,clip_sample=_snake_case			,set_alpha_to_one=_snake_case			,)
				torch.manual_seed(0     )
				UpperCAmelCase_     :      Dict						=  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=1_28			,)
				torch.manual_seed(0     )
				UpperCAmelCase_     :      Dict						=  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=10_00			,hidden_act="gelu"			,projection_dim=5_12			,)
				UpperCAmelCase_     :      Union[str, Any]						=  CLIPTextModel(_snake_case     )
				UpperCAmelCase_     :      List[Any]						=  CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

				UpperCAmelCase_     :      Optional[int]						=  {
				    "unet": unet,
				    "scheduler": scheduler,
				    "vae": vae,
				    "text_encoder": text_encoder,
				    "tokenizer": tokenizer,
				}
				return components






			def 		UpperCamelCase__							(  self			,_snake_case			,_snake_case=0     ):
				# 3 frames
				UpperCAmelCase_     :      Dict						=  floats_tensor((1, 3, 3, 32, 32)			,rng=random.Random(_snake_case     )     ).to(_snake_case     )

				if str(_snake_case     ).startswith("mps"     ):
					UpperCAmelCase_     :      Tuple						=  torch.manual_seed(_snake_case     )
				else:
					UpperCAmelCase_     :      Tuple						=  torch.Generator(device=_snake_case     ).manual_seed(_snake_case     )
				UpperCAmelCase_     :      Union[str, Any]						=  {
				    "prompt": "A painting of a squirrel eating a burger",
				    "video": video,
				    "generator": generator,
				    "num_inference_steps": 2,
				    "guidance_scale": 6.0,
				    "output_type": "pt",
				}
				return inputs






			def 		UpperCamelCase__							(  self     ):
				UpperCAmelCase_     :      str						=  "cpu"  # ensure determinism for the device-dependent torch.Generator
				UpperCAmelCase_     :      Dict						=  self.get_dummy_components()
				UpperCAmelCase_     :      str						=  VideoToVideoSDPipeline(**_snake_case     )
				UpperCAmelCase_     :      int						=  sd_pipe.to(_snake_case     )
				sd_pipe.set_progress_bar_config(disable=_snake_case     )

				UpperCAmelCase_     :      Tuple						=  self.get_dummy_inputs(_snake_case     )
				UpperCAmelCase_     :      str						=  "np"
				UpperCAmelCase_     :      Dict						=  sd_pipe(**_snake_case     ).frames
				UpperCAmelCase_     :      Tuple						=  frames[0][-3:, -3:, -1]

				assert frames[0].shape == (32, 32, 3)
				UpperCAmelCase_     :      Dict						=  np.array([1_06, 1_17, 1_13, 1_74, 1_37, 1_12, 1_48, 1_51, 1_31]     )

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






			@unittest.skipIf(
			    torch_device != "cuda" or not is_xformers_available()			,reason="XFormers attention is only available with CUDA and `xformers` installed"			,)
			def 		UpperCamelCase__							(  self     ):
				self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=_snake_case			,expected_max_diff=5E-3     )






			@unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."     )
			def 		UpperCamelCase__							(  self     ):
				pass






			@unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."     )
			def 		UpperCamelCase__							(  self     ):
				pass






			@unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."     )
			def 		UpperCamelCase__							(  self     ):
				pass






			def 		UpperCamelCase__							(  self     ):
				return super().test_progress_bar()




@slow
@skip_mps
class 					_snake_case		(unittest.TestCase):






			def 		UpperCamelCase__							(  self     ):
				UpperCAmelCase_     :      Dict						=  VideoToVideoSDPipeline.from_pretrained("cerspense/zeroscope_v2_XL"			,torch_dtype=torch.floataa     )
				pipe.enable_model_cpu_offload()

				# 10 frames
				UpperCAmelCase_     :      str						=  torch.Generator(device="cpu"     ).manual_seed(0     )
				UpperCAmelCase_     :      int						=  torch.randn((1, 10, 3, 10_24, 5_76)			,generator=_snake_case     )
				UpperCAmelCase_     :      List[Any]						=  video.to("cuda"     )

				UpperCAmelCase_     :      List[Any]						=  "Spiderman is surfing"

				UpperCAmelCase_     :      Optional[Any]						=  pipe(_snake_case			,video=_snake_case			,generator=_snake_case			,num_inference_steps=3			,output_type="pt"     ).frames

				UpperCAmelCase_     :      Any						=  np.array([-1.0458984, -1.1279297, -0.9663086, -0.91503906, -0.75097656]     )
				assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array     ).sum() < 1E-2






"""simple docstring"""
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       lowerCAmelCase_		(				A__				,				A__				,				A__				,				unittest.TestCase ):



							'''simple docstring'''





							_snake_case						      =		StableUnCLIPPipeline
							_snake_case						      =		TEXT_TO_IMAGE_PARAMS
							_snake_case						      =		TEXT_TO_IMAGE_BATCH_PARAMS
							_snake_case						      =		TEXT_TO_IMAGE_IMAGE_PARAMS
							_snake_case						      =		TEXT_TO_IMAGE_IMAGE_PARAMS

							# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
							_snake_case						      =		False




							def    A__						( self				)       ->  Optional[Any]:
													__lowerCAmelCase     =  32
													__lowerCAmelCase     =  embedder_hidden_size

													# prior components

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

													torch.manual_seed(0				)
													__lowerCAmelCase     =  CLIPTextModelWithProjection(
													    CLIPTextConfig(
													        bos_token_id=0    ,				eos_token_id=2    ,				hidden_size=snake_case_    ,				projection_dim=snake_case_    ,				intermediate_size=37    ,				layer_norm_eps=1e-0_5    ,				num_attention_heads=4    ,				num_hidden_layers=5    ,				pad_token_id=1    ,				vocab_size=1_000    ,				)				)

													torch.manual_seed(0				)
													__lowerCAmelCase     =  PriorTransformer(
													    num_attention_heads=2    ,				attention_head_dim=12    ,				embedding_dim=snake_case_    ,				num_layers=1    ,				)

													torch.manual_seed(0				)
													__lowerCAmelCase     =  DDPMScheduler(
													    variance_type="""fixed_small_log"""    ,				prediction_type="""sample"""    ,				num_train_timesteps=1_000    ,				clip_sample=snake_case_    ,				clip_sample_range=5.0    ,				beta_schedule="""squaredcos_cap_v2"""    ,				)

													# regular denoising components

													torch.manual_seed(0				)
													__lowerCAmelCase     =  StableUnCLIPImageNormalizer(embedding_dim=snake_case_				)
													__lowerCAmelCase     =  DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""				)

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

													torch.manual_seed(0				)
													__lowerCAmelCase     =  CLIPTextModel(
													    CLIPTextConfig(
													        bos_token_id=0    ,				eos_token_id=2    ,				hidden_size=snake_case_    ,				projection_dim=32    ,				intermediate_size=37    ,				layer_norm_eps=1e-0_5    ,				num_attention_heads=4    ,				num_hidden_layers=5    ,				pad_token_id=1    ,				vocab_size=1_000    ,				)				)

													torch.manual_seed(0				)
													__lowerCAmelCase     =  UNetaDConditionModel(
													    sample_size=32    ,				in_channels=4    ,				out_channels=4    ,				down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")    ,				up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")    ,				block_out_channels=(32, 64)    ,				attention_head_dim=(2, 4)    ,				class_embed_type="""projection"""    ,				projection_class_embeddings_input_dim=embedder_projection_dim * 2    ,				cross_attention_dim=snake_case_    ,				layers_per_block=1    ,				upcast_attention=snake_case_    ,				use_linear_projection=snake_case_    ,				)

													torch.manual_seed(0				)
													__lowerCAmelCase     =  DDIMScheduler(
													    beta_schedule="""scaled_linear"""    ,				beta_start=0.00_085    ,				beta_end=0.012    ,				prediction_type="""v_prediction"""    ,				set_alpha_to_one=snake_case_    ,				steps_offset=1    ,				)

													torch.manual_seed(0				)
													__lowerCAmelCase     =  AutoencoderKL()

													__lowerCAmelCase     =  {
													    # 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    A__						( self    ,				snake_case_    ,				snake_case_=0				)       ->  Optional[Any]:
													if str(snake_case_				).startswith("""mps"""				):
																			__lowerCAmelCase     =  torch.manual_seed(snake_case_				)
													else:
																			__lowerCAmelCase     =  torch.Generator(device=snake_case_				).manual_seed(snake_case_				)
													__lowerCAmelCase     =  {
													    """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    A__						( self				)       ->  Tuple:
													__lowerCAmelCase     =  torch_device == """cpu"""

													self._test_attention_slicing_forward_pass(test_max_difference=snake_case_				)




							def    A__						( self				)       ->  Tuple:
													__lowerCAmelCase     =  torch_device in ["""cpu""", """mps"""]

													self._test_inference_batch_single_identical(test_max_difference=snake_case_				)
@slow
@require_torch_gpu
class       lowerCAmelCase_		(				unittest.TestCase ):



							'''simple docstring'''





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




							def    A__						( self				)       ->  Tuple:
													__lowerCAmelCase     =  load_numpy(
													    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""				)

													__lowerCAmelCase     =  StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""    ,				torch_dtype=torch.floataa				)
													pipe.to(snake_case_				)
													pipe.set_progress_bar_config(disable=snake_case_				)
													# 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()

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

													__lowerCAmelCase     =  output.images[0]

													assert image.shape == (768, 768, 3)

													assert_mean_pixel_difference(snake_case_    ,				snake_case_				)




							def    A__						( self				)       ->  Tuple:
													torch.cuda.empty_cache()
													torch.cuda.reset_max_memory_allocated()
													torch.cuda.reset_peak_memory_stats()

													__lowerCAmelCase     =  StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""    ,				torch_dtype=torch.floataa				)
													__lowerCAmelCase     =  pipe.to(snake_case_				)
													pipe.set_progress_bar_config(disable=snake_case_				)
													pipe.enable_attention_slicing()
													pipe.enable_sequential_cpu_offload()

													__lowerCAmelCase     =  pipe(
													    """anime turtle"""    ,				prior_num_inference_steps=2    ,				num_inference_steps=2    ,				output_type="""np"""    ,				)

													__lowerCAmelCase     =  torch.cuda.max_memory_allocated()
													# make sure that less than 7 GB is allocated
													assert mem_bytes < 7 * 10**9
