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

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



		def __UpperCamelCase				(				self )						->      Optional[int]:
				"""simple docstring"""
				torch.manual_seed(0 )
				UpperCamelCase						=  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						=  DDIMScheduler(
				    beta_start=0.0_0085       ,   beta_end=0.012       ,   beta_schedule='scaled_linear'       ,   clip_sample=A_       ,   set_alpha_to_one=A_       ,   )
				torch.manual_seed(0 )
				UpperCamelCase						=  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						=  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						=  CLIPTextModel(A_ )
				UpperCamelCase						=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

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



		def __UpperCamelCase				(				self       ,   A_       ,   A_=0 )						->      Any:
				"""simple docstring"""
				UpperCamelCase						=  floats_tensor((1, 3, 3, 32, 32)       ,   rng=random.Random(A_ ) ).to(A_ )

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

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

				assert frames[0].shape == (32, 32, 3)
				UpperCamelCase						=  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 )						->      int:
				"""simple docstring"""
				self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=A_       ,   expected_max_diff=5e-3 )



		@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.' )
		def __UpperCamelCase				(				self )						->      Any:
				"""simple docstring"""
				pass



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



		@unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.' )
		def __UpperCamelCase				(				self )						->      str:
				"""simple docstring"""
				pass



		def __UpperCamelCase				(				self )						->      int:
				"""simple docstring"""
				return super().test_progress_bar()






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



		def __UpperCamelCase				(				self )						->      Optional[int]:
				"""simple docstring"""
				UpperCamelCase						=  VideoToVideoSDPipeline.from_pretrained('cerspense/zeroscope_v2_XL'       ,   torch_dtype=torch.floataa )
				pipe.enable_model_cpu_offload()

				# 10 frames
				UpperCamelCase						=  torch.Generator(device='cpu' ).manual_seed(0 )
				UpperCamelCase						=  torch.randn((1, 10, 3, 1_024, 576)       ,   generator=A_ )
				UpperCamelCase						=  video.to('cuda' )

				UpperCamelCase						=  'Spiderman is surfing'

				UpperCamelCase						=  pipe(A_       ,   video=A_       ,   generator=A_       ,   num_inference_steps=3       ,   output_type='pt' ).frames

				UpperCamelCase						=  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


def    A (      lowercase				)     ->						str:
		'''simple docstring'''

		UpperCamelCase						=  int(lowercase				)
		if decimal in (0, 1):  # Exit cases for the recursion
				return str(lowercase				)
		UpperCamelCase			,						UpperCamelCase						=  divmod(lowercase					, 2				)
		return binary_recursive(lowercase				) + str(lowercase				)


def    A (      lowercase				)     ->						str:
		'''simple docstring'''

		UpperCamelCase						=  str(lowercase				).strip()
		if not number:
				raise ValueError('No input value was provided'				)
		UpperCamelCase						=  '-' if number.startswith('-'				) else ''
		UpperCamelCase						=  number.lstrip('-'				)
		if not number.isnumeric():
				raise ValueError('Input value is not an integer'				)
		return f'''{negative}0b{binary_recursive(int(lowercase				)				)}'''


if __name__ == "__main__":
					from doctest import testmod

					testmod()
