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

				# No `output_type`.
				__lowercase							: List[Any]  =					frozenset(
				    [
				        """num_inference_steps""",
				        """generator""",
				        """latents""",
				        """return_dict""",
				        """callback""",
				        """callback_steps""",
				    ])
				def 					lowerCamelCase__	(    self     ):
						torch.manual_seed(0     )
						_snake_case     :						int						=   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						,    )
						_snake_case     :						Optional[int]						=   DDIMScheduler(
						    beta_start=0.00085						,    beta_end=0.012						,    beta_schedule="scaled_linear"						,    clip_sample=lowerCamelCase__						,    set_alpha_to_one=lowerCamelCase__						,    )
						torch.manual_seed(0     )
						_snake_case     :						Optional[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     )
						_snake_case     :						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=10_00						,    hidden_act="gelu"						,    projection_dim=5_12						,    )
						_snake_case     :						int						=   CLIPTextModel(lowerCamelCase__     )
						_snake_case     :						Union[str, Any]						=   CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

						_snake_case     :						Optional[int]						=   {
						    '''unet''': unet,
						    '''scheduler''': scheduler,
						    '''vae''': vae,
						    '''text_encoder''': text_encoder,
						    '''tokenizer''': tokenizer,
						}
						return components
				def 					lowerCamelCase__	(    self						,    snake_case_						,    snake_case_=0     ):
						_snake_case     :						str						=   floats_tensor((1, 3, 3, 32, 32)						,    rng=random.Random(lowerCamelCase__     )     ).to(lowerCamelCase__     )

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

						_snake_case     :						Any						=   self.get_dummy_inputs(lowerCamelCase__     )
						_snake_case     :						str						=   '''np'''
						_snake_case     :						Any						=   sd_pipe(**lowerCamelCase__     ).frames
						_snake_case     :						Dict						=   frames[0][-3:, -3:, -1]

						assert frames[0].shape == (32, 32, 3)
						_snake_case     :						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 					lowerCamelCase__	(    self     ):
						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 					lowerCamelCase__	(    self     ):
						pass
				@unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."     )
				def 					lowerCamelCase__	(    self     ):
						pass
				@unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."     )
				def 					lowerCamelCase__	(    self     ):
						pass


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






@slow
@skip_mps
class      _UpperCAmelCase				(					unittest.TestCase):
				def 					lowerCamelCase__	(    self     ):
						_snake_case     :						Any						=   VideoToVideoSDPipeline.from_pretrained("cerspense/zeroscope_v2_XL"						,    torch_dtype=torch.floataa     )
						pipe.enable_model_cpu_offload()

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

						_snake_case     :						Optional[int]						=   '''Spiderman is surfing'''

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

						_snake_case     :						Any						=   np.array([-1.0458984, -1.1279297, -0.9663086, -0.91503906, -0.75097656]     )
						assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array     ).sum() < 1E-2



"""simple docstring"""
def 				a__   (   a   :				list       ,     a   :				int       ,     a   :				int = 0       ,     a   :				int = 0		):

  """simple docstring"""







  _snake_case     :						Optional[int]						=   right or len(a		) - 1
  if left > right:
    return -1
  elif list_data[left] == key:
    return left
  elif list_data[right] == key:
    return right
  else:
    return search(a       ,     a       ,     left + 1       ,     right - 1		)


if __name__ == "__main__":
    import doctest

    doctest.testmod()

