




"""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       lowercase(  __a ,    unittest.TestCase				):







 '''simple docstring'''





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

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


 def 				UpperCamelCase_     (       self:	str	):




   '''simple docstring'''



   torch.manual_seed(0	)
   _snake_case							:	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,   )
   _snake_case							:	Optional[Any]			=						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							:	Any			=						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=128,   )
   torch.manual_seed(0	)
   _snake_case							:	Optional[Any]			=						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=1_000,   hidden_act="""gelu""",   projection_dim=512,   )
   _snake_case							:	Optional[Any]			=						CLIPTextModel(a_	)
   _snake_case							:	Union[str, Any]			=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""	)

   _snake_case							:	Dict			=						{
       """unet""": unet,
       """scheduler""": scheduler,
       """vae""": vae,
       """text_encoder""": text_encoder,
       """tokenizer""": tokenizer,
   }
   return components


 def 				UpperCamelCase_     (       self:	Dict,   a_:	Tuple,   a_:	List[Any]=0	):




   '''simple docstring'''



   _snake_case							:	List[str]			=						floats_tensor((1, 3, 3, 32, 32),   rng=random.Random(a_	)	).to(a_	)

   if str(a_	).startswith("""mps"""	):
     _snake_case							:	Dict			=						torch.manual_seed(a_	)
   else:
     _snake_case							:	List[Any]			=						torch.Generator(device=a_	).manual_seed(a_	)
   _snake_case							:	int			=						{
       """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:	str	):




   '''simple docstring'''



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

   _snake_case							:	Optional[Any]			=						self.get_dummy_inputs(a_	)
   _snake_case							:	Tuple			=						"""np"""
   _snake_case							:	Tuple			=						sd_pipe(**a_	).frames
   _snake_case							:	Any			=						frames[0][-3:, -3:, -1]

   assert frames[0].shape == (32, 32, 3)
   _snake_case							:	int			=						np.array([106, 117, 113, 174, 137, 112, 148, 151, 131]	)

   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:	Tuple	):




   '''simple docstring'''



   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	):




   '''simple docstring'''



   pass


 @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""	)
 def 				UpperCamelCase_     (       self:	List[Any]	):




   '''simple docstring'''



   pass


 @unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""	)
 def 				UpperCamelCase_     (       self:	List[Any]	):




   '''simple docstring'''



   pass


 def 				UpperCamelCase_     (       self:	int	):




   '''simple docstring'''



   return super().test_progress_bar()







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







 '''simple docstring'''





 def 				UpperCamelCase_     (       self:	List[Any]	):




   '''simple docstring'''



   _snake_case							:	Union[str, Any]			=						VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL""",   torch_dtype=torch.floataa	)
   pipe.enable_model_cpu_offload()

   # 10 frames
   _snake_case							:	List[Any]			=						torch.Generator(device="""cpu"""	).manual_seed(0	)
   _snake_case							:	Any			=						torch.randn((1, 10, 3, 1_024, 576),   generator=a_	)
   _snake_case							:	Union[str, Any]			=						video.to("""cuda"""	)

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

   _snake_case							:	Any			=						pipe(a_,   video=a_,   generator=a_,   num_inference_steps=3,   output_type="""pt"""	).frames

   _snake_case							:	int			=						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








"""simple docstring"""


from abc import ABC, abstractmethod
from argparse import ArgumentParser






class       lowercase(  __a				):







 '''simple docstring'''





 @staticmethod
 @abstractmethod
 def 				UpperCamelCase_     (       a_:	ArgumentParser	):




   '''simple docstring'''



   raise NotImplementedError()


 @abstractmethod
 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



   raise NotImplementedError()


