






"""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__      (							snake_case__,       unittest.TestCase       ):
			_UpperCAmelCase		:Dict             =			VideoToVideoSDPipeline
			_UpperCAmelCase		:Optional[Any]             =			TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}       ) - {"image", "width", "height"}
			_UpperCAmelCase		:int             =			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}       ) - {"image"}
			_UpperCAmelCase		:Optional[int]             =			PipelineTesterMixin.required_optional_params - {"latents"}
			_UpperCAmelCase		:Optional[Any]             =			False

			# No `output_type`.
			_UpperCAmelCase		:Optional[Any]             =			frozenset(
			    [
			        "num_inference_steps",
			        "generator",
			        "latents",
			        "return_dict",
			        "callback",
			        "callback_steps",
			    ]       )


			def      UpperCAmelCase__						(				self					:  str ):
										torch.manual_seed(0 )
										lowerCamelCase_	:							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       ,	)
										lowerCamelCase_	:							Dict									=DDIMScheduler(
										    beta_start=0.00_085       ,	beta_end=0.012       ,	beta_schedule="scaled_linear"       ,	clip_sample=snake_case__       ,	set_alpha_to_one=snake_case__       ,	)
										torch.manual_seed(0 )
										lowerCamelCase_	:							Optional[int]									=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 )
										lowerCamelCase_	:							Dict									=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=1000       ,	hidden_act="gelu"       ,	projection_dim=512       ,	)
										lowerCamelCase_	:							Union[str, Any]									=CLIPTextModel(snake_case__ )
										lowerCamelCase_	:							Tuple									=CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip" )

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


			def      UpperCAmelCase__						(				self					:  Tuple       ,	snake_case__					:  Tuple       ,	snake_case__					:  Tuple=0 ):
										# 3 frames
										lowerCamelCase_	:							Any									=floats_tensor((1, 3, 3, 32, 32)       ,	rng=random.Random(snake_case__ ) ).to(snake_case__ )

										if str(snake_case__ ).startswith("mps" ):
																	lowerCamelCase_	:							Any									=torch.manual_seed(snake_case__ )
										else:
																	lowerCamelCase_	:							Tuple									=torch.Generator(device=snake_case__ ).manual_seed(snake_case__ )
										lowerCamelCase_	:							Tuple									={
										    "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					:  Optional[Any] ):
										lowerCamelCase_	:							Dict									="cpu"  # ensure determinism for the device-dependent torch.Generator
										lowerCamelCase_	:							Optional[int]									=self.get_dummy_components()
										lowerCamelCase_	:							str									=VideoToVideoSDPipeline(**snake_case__ )
										lowerCamelCase_	:							Optional[Any]									=sd_pipe.to(snake_case__ )
										sd_pipe.set_progress_bar_config(disable=snake_case__ )

										lowerCamelCase_	:							List[str]									=self.get_dummy_inputs(snake_case__ )
										lowerCamelCase_	:							Optional[int]									="np"
										lowerCamelCase_	:							Union[str, Any]									=sd_pipe(**snake_case__ ).frames
										lowerCamelCase_	:							Any									=frames[0][-3:, -3:, -1]

										assert frames[0].shape == (32, 32, 3)
										lowerCamelCase_	:							Optional[Any]									=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					:  Dict ):
										self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=snake_case__       ,	expected_max_diff=5E-3 )


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


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


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


			def      UpperCAmelCase__						(				self					:  Optional[int] ):
										return super().test_progress_bar()







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


			def      UpperCAmelCase__						(				self					:  Optional[int] ):
										lowerCamelCase_	:							Union[str, Any]									=VideoToVideoSDPipeline.from_pretrained("cerspense/zeroscope_v2_XL"       ,	torch_dtype=torch.floataa )
										pipe.enable_model_cpu_offload()

										# 10 frames
										lowerCamelCase_	:							Optional[int]									=torch.Generator(device="cpu" ).manual_seed(0 )
										lowerCamelCase_	:							List[str]									=torch.randn((1, 10, 3, 1024, 576)       ,	generator=snake_case__ )
										lowerCamelCase_	:							int									=video.to("cuda" )

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

										lowerCamelCase_	:							Dict									=pipe(snake_case__       ,	video=snake_case__       ,	generator=snake_case__       ,	num_inference_steps=3       ,	output_type="pt" ).frames

										lowerCamelCase_	:							Tuple									=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"""



def 			_snake_case    (					lowerCamelCase__     :   int  )    ->     int:

							if not isinstance(lowerCamelCase__    ,						lowerCamelCase__  ):
														raise TypeError("only integers accepted as input"  )
							else:
														lowerCamelCase_	:							str									=str(abs(lowerCamelCase__  )  )
														lowerCamelCase_	:							Tuple									=[list(lowerCamelCase__  ) for char in range(len(lowerCamelCase__  )  )]
														for index in range(len(lowerCamelCase__  )  ):
																					num_transpositions[index].pop(lowerCamelCase__  )
														return max(
														    int("".join(list(lowerCamelCase__  )  )  ) for transposition in num_transpositions  )


if __name__ == "__main__":
						__import__('doctest').testmod()



