






import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()





@skip_mps
class 						SCREAMING_SNAKE_CASE				( snake_case  ,	unittest.TestCase					):





							"""simple docstring"""

							A_						  =					TextToVideoSDPipeline
							A_						  =					TEXT_TO_IMAGE_PARAMS
							A_						  =					TEXT_TO_IMAGE_BATCH_PARAMS
							# No `output_type`.
							A_						  =					frozenset(
							    [
							        "num_inference_steps",
							        "generator",
							        "latents",
							        "return_dict",
							        "callback",
							        "callback_steps",
							    ]					)

							def       __A (  self:   Optional[Any]   )    ->				List[Any]:
												torch.manual_seed(0   )
												_A								= 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		,     )
												_A								= DDIMScheduler(
												    beta_start=0.00_085		,     beta_end=0.012		,     beta_schedule='''scaled_linear'''		,     clip_sample=__A		,     set_alpha_to_one=__A		,     )
												torch.manual_seed(0   )
												_A								= 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   )
												_A								= 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		,     )
												_A								= CLIPTextModel(__A   )
												_A								= CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''   )

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

							def       __A (  self:   Union[str, Any]		,     __A:   List[Any]		,     __A:   Tuple=0   )    ->				int:
												if str(__A   ).startswith('''mps'''   ):
																	_A								= torch.manual_seed(__A   )
												else:
																	_A								= torch.Generator(device=__A   ).manual_seed(__A   )
												_A								= {
												    '''prompt''': '''A painting of a squirrel eating a burger''',
												    '''generator''': generator,
												    '''num_inference_steps''': 2,
												    '''guidance_scale''': 6.0,
												    '''output_type''': '''pt''',
												}
												return inputs

							def       __A (  self:   List[Any]   )    ->				Dict:
												_A								= '''cpu'''  # ensure determinism for the device-dependent torch.Generator
												_A								= self.get_dummy_components()
												_A								= TextToVideoSDPipeline(**__A   )
												_A								= sd_pipe.to(__A   )
												sd_pipe.set_progress_bar_config(disable=__A   )

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

												assert frames[0].shape == (64, 64, 3)
												_A								= np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0]   )

												assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2

							def       __A (  self:   Dict   )    ->				str:
												self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__A		,     expected_max_diff=3e-3   )

							@unittest.skipIf(
							    torch_device != '''cuda''' or not is_xformers_available()		,     reason='''XFormers attention is only available with CUDA and `xformers` installed'''		,     )
							def       __A (  self:   Union[str, Any]   )    ->				List[Any]:
												self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__A		,     expected_max_diff=1e-2   )

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

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

							@unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''   )
							def       __A (  self:   Optional[int]   )    ->				List[str]:
												pass



							def       __A (  self:   List[Any]   )    ->				Dict:
												return super().test_progress_bar()





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





							"""simple docstring"""

							def       __A (  self:   List[Any]   )    ->				str:
												_A								= load_numpy(
												    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'''   )

												_A								= TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''   )
												_A								= DPMSolverMultistepScheduler.from_config(pipe.scheduler.config   )
												_A								= pipe.to('''cuda'''   )

												_A								= '''Spiderman is surfing'''
												_A								= torch.Generator(device='''cpu'''   ).manual_seed(0   )

												_A								= pipe(__A		,     generator=__A		,     num_inference_steps=25		,     output_type='''pt'''   ).frames
												_A								= video_frames.cpu().numpy()

												assert np.abs(expected_video - video   ).mean() < 5e-2



							def       __A (  self:   List[Any]   )    ->				Union[str, Any]:
												_A								= load_numpy(
												    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'''   )

												_A								= TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''   )
												_A								= pipe.to('''cuda'''   )

												_A								= '''Spiderman is surfing'''
												_A								= torch.Generator(device='''cpu'''   ).manual_seed(0   )

												_A								= pipe(__A		,     generator=__A		,     num_inference_steps=2		,     output_type='''pt'''   ).frames
												_A								= video_frames.cpu().numpy()

												assert np.abs(expected_video - video   ).mean() < 5e-2











from __future__ import annotations




def      __A					(     _lowercase					,			_lowercase					,			_lowercase					,			_lowercase ):  # noqa: E741

					'''simple docstring'''







					while r - l > 1:
										_A								= (l + r) // 2
										if v[m] >= key:
															_A								= m
										else:
															_A								= m  # noqa: E741
					return r




def      __A					(     _lowercase ):

					'''simple docstring'''







					if len(_lowercase ) == 0:
										return 0

					_A								= [0] * len(_lowercase )
					_A								= 1

					_A								= v[0]

					for i in range(1					,			len(_lowercase ) ):
										if v[i] < tail[0]:
															_A								= v[i]
										elif v[i] > tail[length - 1]:
															_A								= v[i]
															length += 1
										else:
															_A								= v[i]

					return length


if __name__ == "__main__":
				import doctest

				doctest.testmod()



