





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       (  _a      ,   unittest.TestCase      ):
				__A								=			VideoToVideoSDPipeline
				__A								=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"""video"""}      ) - {"""image""", """width""", """height"""}
				__A								=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""video"""}      ) - {"""image"""}
				__A								=			PipelineTesterMixin.required_optional_params - {"""latents"""}
				__A								=			False

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


				def _snake_case    (						self							):
								torch.manual_seed(0							)
								lowerCamelCase			=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			=DDIMScheduler(
								    beta_start=0.0_0_0_8_5				,	beta_end=0.0_1_2				,	beta_schedule="""scaled_linear"""				,	clip_sample=UpperCAmelCase_				,	set_alpha_to_one=UpperCAmelCase_				,	)
								torch.manual_seed(0							)
								lowerCamelCase			=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			=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			=CLIPTextModel(UpperCAmelCase_							)
								lowerCamelCase			=CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""							)

								lowerCamelCase			={
								    'unet': unet,
								    'scheduler': scheduler,
								    'vae': vae,
								    'text_encoder': text_encoder,
								    'tokenizer': tokenizer,
								}
								return components


				def _snake_case    (						self				,	UpperCAmelCase_				,	UpperCAmelCase_=0							):
								lowerCamelCase			=floats_tensor((1, 3, 3, 32, 32)				,	rng=random.Random(UpperCAmelCase_							)							).to(UpperCAmelCase_							)

								if str(UpperCAmelCase_							).startswith("""mps"""							):
												lowerCamelCase			=torch.manual_seed(UpperCAmelCase_							)
								else:
												lowerCamelCase			=torch.Generator(device=UpperCAmelCase_							).manual_seed(UpperCAmelCase_							)
								lowerCamelCase			={
								    '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							):
								lowerCamelCase			='cpu'  # ensure determinism for the device-dependent torch.Generator
								lowerCamelCase			=self.get_dummy_components()
								lowerCamelCase			=VideoToVideoSDPipeline(**UpperCAmelCase_							)
								lowerCamelCase			=sd_pipe.to(UpperCAmelCase_							)
								sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_							)

								lowerCamelCase			=self.get_dummy_inputs(UpperCAmelCase_							)
								lowerCamelCase			='np'
								lowerCamelCase			=sd_pipe(**UpperCAmelCase_							).frames
								lowerCamelCase			=frames[0][-3:, -3:, -1]

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


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


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


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


				def _snake_case    (						self							):
								return super().test_progress_bar()




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


				def _snake_case    (						self							):
								lowerCamelCase			=VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL"""				,	torch_dtype=torch.floataa							)
								pipe.enable_model_cpu_offload()

								# 10 frames
								lowerCamelCase			=torch.Generator(device="""cpu"""							).manual_seed(0							)
								lowerCamelCase			=torch.randn((1, 10, 3, 1024, 576)				,	generator=UpperCAmelCase_							)
								lowerCamelCase			=video.to("""cuda"""							)

								lowerCamelCase			='Spiderman is surfing'

								lowerCamelCase			=pipe(UpperCAmelCase_				,	video=UpperCAmelCase_				,	generator=UpperCAmelCase_				,	num_inference_steps=3				,	output_type="""pt"""							).frames

								lowerCamelCase			=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 typing import List, Optional, Tuple, Union

import torch

from ...utils import logging, randn_tensor
from ..pipeline_utils import AudioPipelineOutput, DiffusionPipeline


UpperCAmelCase__  :     Dict   			=logging.get_logger(__name__)  # pylint: disable=invalid-name

class  __A       (  a      ):


				def __init__(						self				,	UpperCAmelCase_				,	UpperCAmelCase_							):
								super().__init__()
								self.register_modules(unet=UpperCAmelCase_				,	scheduler=UpperCAmelCase_							)


				@torch.no_grad()
				def __call__(						self				,	UpperCAmelCase_ = 1				,	UpperCAmelCase_ = 100				,	UpperCAmelCase_ = None				,	UpperCAmelCase_ = None				,	UpperCAmelCase_ = True				,	):

								if audio_length_in_s is None:
												lowerCamelCase			=self.unet.config.sample_size / self.unet.config.sample_rate

								lowerCamelCase			=audio_length_in_s * self.unet.config.sample_rate

								lowerCamelCase			=2 ** len(self.unet.up_blocks							)
								if sample_size < 3 * down_scale_factor:
												raise ValueError(
												    f"""{audio_length_in_s} is too small. Make sure it's bigger or equal to"""
												    f""" {3 * down_scale_factor / self.unet.config.sample_rate}."""							)

								lowerCamelCase			=int(UpperCAmelCase_							)
								if sample_size % down_scale_factor != 0:
												lowerCamelCase			=(
												    (audio_length_in_s * self.unet.config.sample_rate) // down_scale_factor + 1
												) * down_scale_factor
												logger.info(
												    f"""{audio_length_in_s} is increased to {sample_size / self.unet.config.sample_rate} so that it can be handled"""
												    f""" by the model. It will be cut to {original_sample_size / self.unet.config.sample_rate} after the denoising"""
												    """ process."""							)
								lowerCamelCase			=int(UpperCAmelCase_							)

								lowerCamelCase			=next(iter(self.unet.parameters()							)							).dtype
								lowerCamelCase			=(batch_size, self.unet.config.in_channels, sample_size)
								if isinstance(UpperCAmelCase_				,	UpperCAmelCase_							) and len(UpperCAmelCase_							) != batch_size:
												raise ValueError(
												    f"""You have passed a list of generators of length {len(UpperCAmelCase_							)}, but requested an effective batch"""
												    f""" size of {batch_size}. Make sure the batch size matches the length of the generators."""							)

								lowerCamelCase			=randn_tensor(UpperCAmelCase_				,	generator=UpperCAmelCase_				,	device=self.device				,	dtype=UpperCAmelCase_							)

								# set step values
								self.scheduler.set_timesteps(UpperCAmelCase_				,	device=audio.device							)
								lowerCamelCase			=self.scheduler.timesteps.to(UpperCAmelCase_							)

								for t in self.progress_bar(self.scheduler.timesteps							):
												# 1. predict noise model_output
												lowerCamelCase			=self.unet(UpperCAmelCase_				,	UpperCAmelCase_							).sample

												# 2. compute previous image: x_t -> t_t-1
												lowerCamelCase			=self.scheduler.step(UpperCAmelCase_				,	UpperCAmelCase_				,	UpperCAmelCase_							).prev_sample

								lowerCamelCase			=audio.clamp(-1				,	1							).float().cpu().numpy()

								lowerCamelCase			=audio[:, :, :original_sample_size]

								if not return_dict:
												return (audio,)

								return AudioPipelineOutput(audios=UpperCAmelCase_							)





