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





     """simple docstring"""







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

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






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







           torch.manual_seed(0	)
           UpperCamelCase__     : List[Any]											=					UNetaDConditionModel(
               block_out_channels=(3_2, 6_4, 6_4, 6_4)    ,    layers_per_block=2    ,    sample_size=3_2    ,    in_channels=4    ,    out_channels=4    ,    down_block_types=('''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''DownBlock3D''')    ,    up_block_types=('''UpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''')    ,    cross_attention_dim=3_2    ,    attention_head_dim=4    ,    )
           UpperCamelCase__     : Any											=					DDIMScheduler(
               beta_start=0.00085    ,    beta_end=0.012    ,    beta_schedule='''scaled_linear'''    ,    clip_sample=__SCREAMING_SNAKE_CASE    ,    set_alpha_to_one=__SCREAMING_SNAKE_CASE    ,    )
           torch.manual_seed(0	)
           UpperCamelCase__     : Optional[Any]											=					AutoencoderKL(
               block_out_channels=[3_2, 6_4]    ,    in_channels=3    ,    out_channels=3    ,    down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']    ,    up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']    ,    latent_channels=4    ,    sample_size=1_2_8    ,    )
           torch.manual_seed(0	)
           UpperCamelCase__     : List[str]											=					CLIPTextConfig(
               bos_token_id=0    ,    eos_token_id=2    ,    hidden_size=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    ,    hidden_act='''gelu'''    ,    projection_dim=5_1_2    ,    )
           UpperCamelCase__     : List[str]											=					CLIPTextModel(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : str											=					CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''	)

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






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=0	)	-> Dict:




           """simple docstring"""







           UpperCamelCase__     : Optional[int]											=					floats_tensor((1, 3, 3, 3_2, 3_2)    ,    rng=random.Random(__SCREAMING_SNAKE_CASE	)	).to(__SCREAMING_SNAKE_CASE	)

           if str(__SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
                 UpperCamelCase__     : Union[str, Any]											=					torch.manual_seed(__SCREAMING_SNAKE_CASE	)
           else:
                 UpperCamelCase__     : Optional[int]											=					torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_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  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           UpperCamelCase__     : int											=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
           UpperCamelCase__     : Any											=					self.get_dummy_components()
           UpperCamelCase__     : Optional[Any]											=					VideoToVideoSDPipeline(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Optional[int]											=					sd_pipe.to(__SCREAMING_SNAKE_CASE	)
           sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : Dict											=					self.get_dummy_inputs(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Union[str, Any]											=					'''np'''
           UpperCamelCase__     : str											=					sd_pipe(**__SCREAMING_SNAKE_CASE	).frames
           UpperCamelCase__     : Any											=					frames[0][-3:, -3:, -1]

           assert frames[0].shape == (3_2, 3_2, 3)
           UpperCamelCase__     : str											=					np.array([1_0_6, 1_1_7, 1_1_3, 1_7_4, 1_3_7, 1_1_2, 1_4_8, 1_5_1, 1_3_1]	)

           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  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE    ,    expected_max_diff=5e-3	)






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




           """simple docstring"""







           pass






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




           """simple docstring"""







           pass






     @unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''	)
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Any:




           """simple docstring"""







           pass






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







           return super().test_progress_bar()




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





     """simple docstring"""







     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[int]:




           """simple docstring"""







           UpperCamelCase__     : Optional[Any]											=					VideoToVideoSDPipeline.from_pretrained('''cerspense/zeroscope_v2_XL'''    ,    torch_dtype=torch.floataa	)
           pipe.enable_model_cpu_offload()

           # 10 frames
           UpperCamelCase__     : Any											=					torch.Generator(device='''cpu'''	).manual_seed(0	)
           UpperCamelCase__     : Union[str, Any]											=					torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6)    ,    generator=__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Optional[int]											=					video.to('''cuda'''	)

           UpperCamelCase__     : Optional[int]											=					'''Spiderman is surfing'''

           UpperCamelCase__     : Dict											=					pipe(__SCREAMING_SNAKE_CASE    ,    video=__SCREAMING_SNAKE_CASE    ,    generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=3    ,    output_type='''pt'''	).frames

           UpperCamelCase__     : Optional[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




def 				SCREAMING_SNAKE_CASE_   (      UpperCamelCase__			,		UpperCamelCase__ = " "		):
      UpperCamelCase__     : Optional[int]											=					[]

      UpperCamelCase__     : int											=					0
      for index, char in enumerate(UpperCamelCase__		):
            if char == separator:
                  split_words.append(string[last_index:index]		)
                  UpperCamelCase__     : int											=					index + 1
            elif index + 1 == len(UpperCamelCase__		):
                  split_words.append(string[last_index : index + 1]		)
      return split_words


if __name__ == "__main__":
    from doctest import testmod

    testmod()


