'''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 		SCREAMING_SNAKE_CASE     (					SCREAMING_SNAKE_CASE   , unittest.TestCase	):







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

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





				def  snake_case__				(						self					:						Optional[int]					)							->List[str]:


										'''simple docstring'''




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

										_UpperCamelCase			:			List[Any]   			=						{
										    "unet": unet,
										    "scheduler": scheduler,
										    "vae": vae,
										    "text_encoder": text_encoder,
										    "tokenizer": tokenizer,
										}
										return components





				def  snake_case__				(						self					:						List[Any]       ,		lowercase__					:						Tuple       ,		lowercase__					:						Tuple=0					)							->Any:


										'''simple docstring'''




										_UpperCamelCase			:			Union[str, Any]   			=						floats_tensor((1, 3, 3, 32, 32)       ,		rng=random.Random(lowercase__					)					).to(lowercase__					)

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


										'''simple docstring'''




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

										_UpperCamelCase			:			Union[str, Any]   			=						self.get_dummy_inputs(lowercase__					)
										_UpperCamelCase			:			Tuple   			=						"np"
										_UpperCamelCase			:			List[Any]   			=						sd_pipe(**lowercase__					).frames
										_UpperCamelCase			:			Tuple   			=						frames[0][-3:, -3:, -1]

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


										'''simple docstring'''




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





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


										'''simple docstring'''




										pass





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


										'''simple docstring'''




										pass





				@unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."					)
				def  snake_case__				(						self					:						Dict					)							->str:


										'''simple docstring'''




										pass







				def  snake_case__				(						self					:						int					)							->Optional[Any]:


										'''simple docstring'''




										return super().test_progress_bar()





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







				'''simple docstring'''





				def  snake_case__				(						self					:						Any					)							->str:


										'''simple docstring'''




										_UpperCamelCase			:			int   			=						VideoToVideoSDPipeline.from_pretrained("cerspense/zeroscope_v2_XL"       ,		torch_dtype=torch.floataa					)
										pipe.enable_model_cpu_offload()

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

										_UpperCamelCase			:			Optional[int]   			=						"Spiderman is surfing"

										_UpperCamelCase			:			int   			=						pipe(lowercase__       ,		video=lowercase__       ,		generator=lowercase__       ,		num_inference_steps=3       ,		output_type="pt"					).frames

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



'''simple docstring'''





from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


lowerCAmelCase_	:		Union[str, Any]			=     {"""configuration_opt""": ["""OPT_PRETRAINED_CONFIG_ARCHIVE_MAP""", """OPTConfig"""]}

try:
			if not is_torch_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowerCAmelCase_	:		Tuple			=     [
			    """OPT_PRETRAINED_MODEL_ARCHIVE_LIST""",
			    """OPTForCausalLM""",
			    """OPTModel""",
			    """OPTPreTrainedModel""",
			    """OPTForSequenceClassification""",
			    """OPTForQuestionAnswering""",
			]

try:
			if not is_tf_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowerCAmelCase_	:		str			=     ["""TFOPTForCausalLM""", """TFOPTModel""", """TFOPTPreTrainedModel"""]

try:
			if not is_flax_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowerCAmelCase_	:		List[Any]			=     [
			    """FlaxOPTForCausalLM""",
			    """FlaxOPTModel""",
			    """FlaxOPTPreTrainedModel""",
			]


if TYPE_CHECKING:
			from .configuration_opt import OPT_PRETRAINED_CONFIG_ARCHIVE_MAP, OPTConfig

			try:
						if not is_torch_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_opt import (
						    OPT_PRETRAINED_MODEL_ARCHIVE_LIST,
						    OPTForCausalLM,
						    OPTForQuestionAnswering,
						    OPTForSequenceClassification,
						    OPTModel,
						    OPTPreTrainedModel,
						)

			try:
						if not is_tf_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_tf_opt import TFOPTForCausalLM, TFOPTModel, TFOPTPreTrainedModel

			try:
						if not is_flax_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_flax_opt import FlaxOPTForCausalLM, FlaxOPTModel, FlaxOPTPreTrainedModel

else:
			import sys

			lowerCAmelCase_	:		Any			=     _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)


