
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   (	UpperCAmelCase_  ,						unittest.TestCase							):
	__lowerCAmelCase  :     List[str]        =			VideoToVideoSDPipeline
	__lowerCAmelCase  :     str        =			TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'video'}							) - {'image', 'width', 'height'}
	__lowerCAmelCase  :     Dict        =			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'video'}							) - {'image'}
	__lowerCAmelCase  :     Union[str, Any]        =			PipelineTesterMixin.required_optional_params - {'latents'}
	__lowerCAmelCase  :     Tuple        =			False

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


	def      lowercase__     (					self):


			'''simple docstring'''




			torch.manual_seed(0)
			lowercase__  :   Any    		=      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					,  )
			lowercase__  :   Optional[Any]    		=      DDIMScheduler(
			    beta_start=0.0_0_0_8_5					,  beta_end=0.0_1_2					,  beta_schedule="""scaled_linear"""					,  clip_sample=__a					,  set_alpha_to_one=__a					,  )
			torch.manual_seed(0)
			lowercase__  :   List[str]    		=      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)
			lowercase__  :   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=10_00					,  hidden_act="""gelu"""					,  projection_dim=5_12					,  )
			lowercase__  :   int    		=      CLIPTextModel(__a)
			lowercase__  :   Optional[Any]    		=      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

			lowercase__  :   str    		=      {
			    """unet""": unet,
			    """scheduler""": scheduler,
			    """vae""": vae,
			    """text_encoder""": text_encoder,
			    """tokenizer""": tokenizer,
			}
			return components


	def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_=0):


			'''simple docstring'''




			lowercase__  :   Tuple    		=      floats_tensor((1, 3, 3, 32, 32)					,  rng=random.Random(__a)).to(__a)

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


			'''simple docstring'''




			lowercase__  :   List[str]    		=      """cpu"""  # ensure determinism for the device-dependent torch.Generator
			lowercase__  :   Dict    		=      self.get_dummy_components()
			lowercase__  :   Optional[Any]    		=      VideoToVideoSDPipeline(**__a)
			lowercase__  :   Tuple    		=      sd_pipe.to(__a)
			sd_pipe.set_progress_bar_config(disable=__a)

			lowercase__  :   Tuple    		=      self.get_dummy_inputs(__a)
			lowercase__  :   int    		=      """np"""
			lowercase__  :   Tuple    		=      sd_pipe(**__a).frames
			lowercase__  :   Tuple    		=      frames[0][-3:, -3:, -1]

			assert frames[0].shape == (32, 32, 3)
			lowercase__  :   Any    		=      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      lowercase__     (					self):


			'''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      lowercase__     (					self):


			'''simple docstring'''




			pass


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


			'''simple docstring'''




			pass


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


			'''simple docstring'''




			pass



	def      lowercase__     (					self):


			'''simple docstring'''




			return super().test_progress_bar()



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


	def      lowercase__     (					self):


			'''simple docstring'''




			lowercase__  :   Optional[int]    		=      VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL"""					,  torch_dtype=torch.floataa)
			pipe.enable_model_cpu_offload()

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

			lowercase__  :   Tuple    		=      """Spiderman is surfing"""

			lowercase__  :   Optional[int]    		=      pipe(__a					,  video=__a					,  generator=__a					,  num_inference_steps=3					,  output_type="""pt""").frames

			lowercase__  :   int    		=      np.array([-1.0_4_5_8_9_8_4, -1.1_2_7_9_2_9_7, -0.9_6_6_3_0_8_6, -0.9_1_5_0_3_9_0_6, -0.7_5_0_9_7_6_5_6])
			assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array).sum() < 1E-2








from ..utils import DummyObject, requires_backends



class       _snake_case   (	metaclass=UpperCAmelCase_							):
 __lowerCAmelCase  :     int        =			['torch', 'transformers', 'onnx']


 def __init__(					self					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(self					,  ["""torch""", """transformers""", """onnx"""])


 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



class       _snake_case   (	metaclass=UpperCAmelCase_							):
 __lowerCAmelCase  :     Any        =			['torch', 'transformers', 'onnx']


 def __init__(					self					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(self					,  ["""torch""", """transformers""", """onnx"""])


 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



class       _snake_case   (	metaclass=UpperCAmelCase_							):
 __lowerCAmelCase  :     Union[str, Any]        =			['torch', 'transformers', 'onnx']


 def __init__(					self					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(self					,  ["""torch""", """transformers""", """onnx"""])


 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



class       _snake_case   (	metaclass=UpperCAmelCase_							):
 __lowerCAmelCase  :     List[Any]        =			['torch', 'transformers', 'onnx']


 def __init__(					self					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(self					,  ["""torch""", """transformers""", """onnx"""])


 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



class       _snake_case   (	metaclass=UpperCAmelCase_							):
 __lowerCAmelCase  :     Dict        =			['torch', 'transformers', 'onnx']


 def __init__(					self					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(self					,  ["""torch""", """transformers""", """onnx"""])


 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



class       _snake_case   (	metaclass=UpperCAmelCase_							):
 __lowerCAmelCase  :     List[str]        =			['torch', 'transformers', 'onnx']


 def __init__(					self					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(self					,  ["""torch""", """transformers""", """onnx"""])


 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])



 @classmethod
 def      lowercase__     (					cls					,  *SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   requires_backends(cls					,  ["""torch""", """transformers""", """onnx"""])






