






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



	'''simple docstring'''




	lowerCAmelCase						=					VideoToVideoSDPipeline
	lowerCAmelCase						=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'''video'''}							) - {'''image''', '''width''', '''height'''}
	lowerCAmelCase						=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''video'''}							) - {'''image'''}
	lowerCAmelCase						=					PipelineTesterMixin.required_optional_params - {'''latents'''}
	lowerCAmelCase						=					False

	# No `output_type`.
	lowerCAmelCase						=					frozenset(
	    [
	        '''num_inference_steps''',
	        '''generator''',
	        '''latents''',
	        '''return_dict''',
	        '''callback''',
	        '''callback_steps''',
	    ]							)




	def 				_UpperCamelCase     (			self   )			->       Tuple:
								torch.manual_seed(0   )
								snake_case_           =				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			,     )
								snake_case_           =				DDIMScheduler(
								    beta_start=0.00_085			,     beta_end=0.012			,     beta_schedule='scaled_linear'			,     clip_sample=a			,     set_alpha_to_one=a			,     )
								torch.manual_seed(0   )
								snake_case_           =				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   )
								snake_case_           =				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			,     )
								snake_case_           =				CLIPTextModel(a   )
								snake_case_           =				CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   )

								snake_case_           =				{
								    'unet': unet,
								    'scheduler': scheduler,
								    'vae': vae,
								    'text_encoder': text_encoder,
								    'tokenizer': tokenizer,
								}
								return components




	def 				_UpperCamelCase     (			self			,     a			,     a=0   )			->       Optional[Any]:
								# 3 frames
								snake_case_           =				floats_tensor((1, 3, 3, 32, 32)			,     rng=random.Random(a   )   ).to(a   )

								if str(a   ).startswith('mps'   ):
															snake_case_           =				torch.manual_seed(a   )
								else:
															snake_case_           =				torch.Generator(device=a   ).manual_seed(a   )
								snake_case_           =				{
								    '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   )			->       Optional[Any]:
								snake_case_           =				'cpu'  # ensure determinism for the device-dependent torch.Generator
								snake_case_           =				self.get_dummy_components()
								snake_case_           =				VideoToVideoSDPipeline(**a   )
								snake_case_           =				sd_pipe.to(a   )
								sd_pipe.set_progress_bar_config(disable=a   )

								snake_case_           =				self.get_dummy_inputs(a   )
								snake_case_           =				'np'
								snake_case_           =				sd_pipe(**a   ).frames
								snake_case_           =				frames[0][-3:, -3:, -1]

								assert frames[0].shape == (32, 32, 3)
								snake_case_           =				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   )			->       Any:
								self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=a			,     expected_max_diff=5E-3   )




	@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'   )
	def 				_UpperCamelCase     (			self   )			->       Any:
								pass




	@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'   )
	def 				_UpperCamelCase     (			self   )			->       Optional[Any]:
								pass




	@unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'   )
	def 				_UpperCamelCase     (			self   )			->       Union[str, Any]:
								pass






	def 				_UpperCamelCase     (			self   )			->       Union[str, Any]:
								return super().test_progress_bar()






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



	'''simple docstring'''




	def 				_UpperCamelCase     (			self   )			->       Dict:
								snake_case_           =				VideoToVideoSDPipeline.from_pretrained('cerspense/zeroscope_v2_XL'			,     torch_dtype=torch.floataa   )
								pipe.enable_model_cpu_offload()

								# 10 frames
								snake_case_           =				torch.Generator(device='cpu'   ).manual_seed(0   )
								snake_case_           =				torch.randn((1, 10, 3, 10_24, 5_76)			,     generator=a   )
								snake_case_           =				video.to('cuda'   )

								snake_case_           =				'Spiderman is surfing'

								snake_case_           =				pipe(a			,     video=a			,     generator=a			,     num_inference_steps=3			,     output_type='pt'   ).frames

								snake_case_           =				np.array([-1.0_458_984, -1.1_279_297, -0.9_663_086, -0.91_503_906, -0.75_097_656]   )
								assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array   ).sum() < 1E-2













import copy
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Optional, Union




@dataclass
class 					UpperCamelCase_ :



	'''simple docstring'''




	lowerCAmelCase						=					None
	lowerCAmelCase						=					False
	lowerCAmelCase						=					False
	lowerCAmelCase						=					False
	lowerCAmelCase						=					None
	lowerCAmelCase						=					None
	lowerCAmelCase						=					False
	lowerCAmelCase						=					False
	lowerCAmelCase						=					False
	lowerCAmelCase						=					True
	lowerCAmelCase						=					None
	lowerCAmelCase						=					1
	lowerCAmelCase						=					None
	lowerCAmelCase						=					False
	lowerCAmelCase						=					None
	lowerCAmelCase						=					None




	def 				_UpperCamelCase     (			self   )			->       "DownloadConfig":
								return self.__class__(**{k: copy.deepcopy(a   ) for k, v in self.__dict__.items()}   )





