





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    _a     (				lowercase__					,							unittest.TestCase							):






							"""simple docstring"""

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

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

							def      A_  (      self				:     int    )	->Union[str, Any]:
													torch.manual_seed(0    )
													SCREAMING_SNAKE_CASE__     :				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							,	)
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					DDIMScheduler(
													    beta_start=0.0_0085							,	beta_end=0.012							,	beta_schedule="scaled_linear"							,	clip_sample=__lowerCamelCase							,	set_alpha_to_one=__lowerCamelCase							,	)
													torch.manual_seed(0    )
													SCREAMING_SNAKE_CASE__     :				Union[str, 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=1_28							,	)
													torch.manual_seed(0    )
													SCREAMING_SNAKE_CASE__     :				Union[str, 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							,	)
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					CLIPTextModel(__lowerCamelCase    )
													SCREAMING_SNAKE_CASE__     :				int   =					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"    )

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

							def      A_  (      self				:     str							,	a				:     Optional[int]							,	a				:     Tuple=0    )	->str:
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					floats_tensor((1, 3, 3, 32, 32)							,	rng=random.Random(__lowerCamelCase    )    ).to(__lowerCamelCase    )

													if str(__lowerCamelCase    ).startswith("mps"    ):
																			SCREAMING_SNAKE_CASE__     :				str   =					torch.manual_seed(__lowerCamelCase    )
													else:
																			SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					torch.Generator(device=__lowerCamelCase    ).manual_seed(__lowerCamelCase    )
													SCREAMING_SNAKE_CASE__     :				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      A_  (      self				:     Any    )	->Optional[Any]:
													SCREAMING_SNAKE_CASE__     :				Any   =					"cpu"  # ensure determinism for the device-dependent torch.Generator
													SCREAMING_SNAKE_CASE__     :				Any   =					self.get_dummy_components()
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					VideoToVideoSDPipeline(**__lowerCamelCase    )
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					sd_pipe.to(__lowerCamelCase    )
													sd_pipe.set_progress_bar_config(disable=__lowerCamelCase    )

													SCREAMING_SNAKE_CASE__     :				int   =					self.get_dummy_inputs(__lowerCamelCase    )
													SCREAMING_SNAKE_CASE__     :				int   =					"np"
													SCREAMING_SNAKE_CASE__     :				Tuple   =					sd_pipe(**__lowerCamelCase    ).frames
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					frames[0][-3:, -3:, -1]

													assert frames[0].shape == (32, 32, 3)
													SCREAMING_SNAKE_CASE__     :				int   =					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      A_  (      self				:     Dict    )	->int:
													self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__lowerCamelCase							,	expected_max_diff=5E-3    )

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

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

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






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







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






							"""simple docstring"""

							def      A_  (      self				:     Optional[int]    )	->Dict:
													SCREAMING_SNAKE_CASE__     :				Any   =					VideoToVideoSDPipeline.from_pretrained("cerspense/zeroscope_v2_XL"							,	torch_dtype=torch.floataa    )
													pipe.enable_model_cpu_offload()

													# 10 frames
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					torch.Generator(device="cpu"    ).manual_seed(0    )
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					torch.randn((1, 10, 3, 10_24, 5_76)							,	generator=__lowerCamelCase    )
													SCREAMING_SNAKE_CASE__     :				Dict   =					video.to("cuda"    )

													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					"Spiderman is surfing"

													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					pipe(__lowerCamelCase							,	video=__lowerCamelCase							,	generator=__lowerCamelCase							,	num_inference_steps=3							,	output_type="pt"    ).frames

													SCREAMING_SNAKE_CASE__     :				Dict   =					np.array([-1.045_8984, -1.127_9297, -0.966_3086, -0.9150_3906, -0.7509_7656]    )
													assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array    ).sum() < 1E-2






import unittest
from typing import Dict, List, Optional, Union

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
				import torch

if is_vision_available():
				from PIL import Image

				from transformers import BridgeTowerImageProcessor



class    _a     (				unittest.TestCase							):






							"""simple docstring"""

							def __init__(      self				:     Optional[int]							,	a				:     Any							,	a				:     bool = True							,	a				:     Dict[str, int] = None							,	a				:     int = 32							,	a				:     bool = True							,	a				:     Union[int, float] = 1 / 2_55							,	a				:     bool = True							,	a				:     bool = True							,	a				:     Optional[Union[float, List[float]]] = [0.4814_5466, 0.457_8275, 0.4082_1073]							,	a				:     Optional[Union[float, List[float]]] = [0.2686_2954, 0.2613_0258, 0.2757_7711]							,	a				:     bool = True							,	a				:     Any=7							,	a				:     str=30							,	a				:     Dict=4_00							,	a				:     Optional[int]=3							,	)	->int:
													SCREAMING_SNAKE_CASE__     :				int   =					parent
													SCREAMING_SNAKE_CASE__     :				Dict   =					do_resize
													SCREAMING_SNAKE_CASE__     :				List[str]   =					size if size is not None else {"shortest_edge": 2_88}
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					size_divisor
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					do_rescale
													SCREAMING_SNAKE_CASE__     :				Tuple   =					rescale_factor
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					do_normalize
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					do_center_crop
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					image_mean
													SCREAMING_SNAKE_CASE__     :				Dict   =					image_std
													SCREAMING_SNAKE_CASE__     :				List[str]   =					do_pad
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					batch_size
													SCREAMING_SNAKE_CASE__     :				int   =					num_channels
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					min_resolution
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					max_resolution

							def      A_  (      self				:     List[str]    )	->Tuple:
													return {
													    "image_mean": self.image_mean,
													    "image_std": self.image_std,
													    "do_normalize": self.do_normalize,
													    "do_resize": self.do_resize,
													    "size": self.size,
													    "size_divisor": self.size_divisor,
													}






							def      A_  (      self				:     int							,	a				:     Optional[int]							,	a				:     Union[str, Any]=False    )	->Optional[Any]:
													if not batched:
																			SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					self.size["shortest_edge"]
																			SCREAMING_SNAKE_CASE__     :				Dict   =					image_inputs[0]
																			if isinstance(a							,	Image.Image    ):
																									SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Dict   =					image.size
																			else:
																									SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				List[str]   =					image.shape[1], image.shape[2]
																			SCREAMING_SNAKE_CASE__     :				Any   =					size / min(a							,	a    )
																			if h < w:
																									SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Optional[int]   =					size, scale * w
																			else:
																									SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				List[str]   =					scale * h, size

																			SCREAMING_SNAKE_CASE__     :				List[Any]   =					int((13_33 / 8_00) * size    )
																			if max(a							,	a    ) > max_size:
																									SCREAMING_SNAKE_CASE__     :				List[Any]   =					max_size / max(a							,	a    )
																									SCREAMING_SNAKE_CASE__     :				int   =					newh * scale
																									SCREAMING_SNAKE_CASE__     :				Optional[int]   =					neww * scale

																			SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Any   =					int(newh + 0.5    ), int(neww + 0.5    )
																			SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Any   =					(
																			    newh // self.size_divisor * self.size_divisor,
																			    neww // self.size_divisor * self.size_divisor,
																			)

													else:
																			SCREAMING_SNAKE_CASE__     :				List[Any]   =					[]
																			for image in image_inputs:
																									SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				List[Any]   =					self.get_expected_values([image]    )
																									expected_values.append((expected_height, expected_width)    )
																			SCREAMING_SNAKE_CASE__     :				Tuple   =					max(a							,	key=lambda a    : item[0]    )[0]
																			SCREAMING_SNAKE_CASE__     :				Tuple   =					max(a							,	key=lambda a    : item[1]    )[1]

													return expected_height, expected_width







@require_torch
@require_vision
class    _a     (				lowercase__					,							unittest.TestCase							):






							"""simple docstring"""

							snake_case_							 =						BridgeTowerImageProcessor if is_vision_available() else None

							def      A_  (      self				:     List[Any]    )	->Union[str, Any]:
													SCREAMING_SNAKE_CASE__     :				Any   =					BridgeTowerImageProcessingTester(self    )

							@property
							def      A_  (      self				:     Optional[int]    )	->Optional[Any]:
													return self.image_processor_tester.prepare_image_processor_dict()

							def      A_  (      self				:     Tuple    )	->str:
													SCREAMING_SNAKE_CASE__     :				int   =					self.image_processing_class(**self.image_processor_dict    )
													self.assertTrue(hasattr(a							,	"image_mean"    )    )
													self.assertTrue(hasattr(a							,	"image_std"    )    )
													self.assertTrue(hasattr(a							,	"do_normalize"    )    )
													self.assertTrue(hasattr(a							,	"do_resize"    )    )
													self.assertTrue(hasattr(a							,	"size"    )    )
													self.assertTrue(hasattr(a							,	"size_divisor"    )    )

							def      A_  (      self				:     List[Any]    )	->List[Any]:
													pass

							def      A_  (      self				:     Tuple    )	->Optional[Any]:
													# Initialize image processor
													SCREAMING_SNAKE_CASE__     :				Dict   =					self.image_processing_class(**self.image_processor_dict    )
													# create random PIL images
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					prepare_image_inputs(self.image_processor_tester							,	equal_resolution=a    )
													for image in image_inputs:
																			self.assertIsInstance(a							,	Image.Image    )

													# Test not batched input
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					image_processing(image_inputs[0]							,	return_tensors="pt"    ).pixel_values

													SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					self.image_processor_tester.get_expected_values(a    )
													self.assertEqual(
													    encoded_images.shape							,	(1, self.image_processor_tester.num_channels, expected_height, expected_width)							,	)

													# Test batched
													SCREAMING_SNAKE_CASE__     :				int   =					image_processing(a							,	return_tensors="pt"    ).pixel_values

													SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Any   =					self.image_processor_tester.get_expected_values(a							,	batched=a    )
													self.assertEqual(
													    encoded_images.shape							,	(
													        self.image_processor_tester.batch_size,
													        self.image_processor_tester.num_channels,
													        expected_height,
													        expected_width,
													    )							,	)

							def      A_  (      self				:     Optional[int]    )	->Any:
													# Initialize image processor
													SCREAMING_SNAKE_CASE__     :				str   =					self.image_processing_class(**self.image_processor_dict    )
													# create random numpy tensors
													SCREAMING_SNAKE_CASE__     :				str   =					prepare_image_inputs(self.image_processor_tester							,	equal_resolution=a							,	numpify=a    )
													for image in image_inputs:
																			self.assertIsInstance(a							,	np.ndarray    )

													# Test not batched input
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					image_processing(image_inputs[0]							,	return_tensors="pt"    ).pixel_values

													SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				List[Any]   =					self.image_processor_tester.get_expected_values(a    )
													self.assertEqual(
													    encoded_images.shape							,	(1, self.image_processor_tester.num_channels, expected_height, expected_width)							,	)

													# Test batched
													SCREAMING_SNAKE_CASE__     :				Tuple   =					image_processing(a							,	return_tensors="pt"    ).pixel_values

													SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					self.image_processor_tester.get_expected_values(a							,	batched=a    )
													self.assertEqual(
													    encoded_images.shape							,	(
													        self.image_processor_tester.batch_size,
													        self.image_processor_tester.num_channels,
													        expected_height,
													        expected_width,
													    )							,	)






							def      A_  (      self				:     str    )	->Optional[int]:
													# Initialize image processor
													SCREAMING_SNAKE_CASE__     :				Dict   =					self.image_processing_class(**self.image_processor_dict    )
													# create random PyTorch tensors
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					prepare_image_inputs(self.image_processor_tester							,	equal_resolution=a							,	torchify=a    )
													for image in image_inputs:
																			self.assertIsInstance(a							,	torch.Tensor    )

													# Test not batched input
													SCREAMING_SNAKE_CASE__     :				Tuple   =					image_processing(image_inputs[0]							,	return_tensors="pt"    ).pixel_values

													SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Tuple   =					self.image_processor_tester.get_expected_values(a    )
													self.assertEqual(
													    encoded_images.shape							,	(1, self.image_processor_tester.num_channels, expected_height, expected_width)							,	)

													# Test batched
													SCREAMING_SNAKE_CASE__     :				Any   =					image_processing(a							,	return_tensors="pt"    ).pixel_values

													SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					self.image_processor_tester.get_expected_values(a							,	batched=a    )
													self.assertEqual(
													    encoded_images.shape							,	(
													        self.image_processor_tester.batch_size,
													        self.image_processor_tester.num_channels,
													        expected_height,
													        expected_width,
													    )							,	)