






"""simple docstring"""



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 					_UpperCAmelCase       ( _A   ,						unittest.TestCase							):
					SCREAMING_SNAKE_CASE_		:				Tuple							    =   TextToVideoSDPipeline
					SCREAMING_SNAKE_CASE_		:				Optional[Any]							    =   TEXT_TO_IMAGE_PARAMS
					SCREAMING_SNAKE_CASE_		:				Optional[int]							    =   TEXT_TO_IMAGE_BATCH_PARAMS
					# No `output_type`.
					SCREAMING_SNAKE_CASE_		:				Optional[Any]							    =   frozenset(
					    [
					        "num_inference_steps",
					        "generator",
					        "latents",
					        "return_dict",
					        "callback",
					        "callback_steps",
					    ]							)




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

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




					def   A  (				self     :						int    ,						A     :						List[str]    ,						A     :						List[str]=0   )  ->  Any:
							if str(A   ).startswith('''mps'''   ):
									lowercase_ : Optional[Any]							       =   torch.manual_seed(A   )
							else:
									lowercase_ : List[Any]							       =   torch.Generator(device=A   ).manual_seed(A   )
							lowercase_ : Union[str, Any]							       =   {
							    '''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     :						str   )  ->  Optional[Any]:
							lowercase_ : Tuple							       =   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
							lowercase_ : int							       =   self.get_dummy_components()
							lowercase_ : List[Any]							       =   TextToVideoSDPipeline(**A   )
							lowercase_ : Optional[Any]							       =   sd_pipe.to(A   )
							sd_pipe.set_progress_bar_config(disable=A   )

							lowercase_ : int							       =   self.get_dummy_inputs(A   )
							lowercase_ : Optional[Any]							       =   '''np'''
							lowercase_ : Tuple							       =   sd_pipe(**A   ).frames
							lowercase_ : Any							       =   frames[0][-3:, -3:, -1]

							assert frames[0].shape == (64, 64, 3)
							lowercase_ : Tuple							       =   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     :						int   )  ->  int:
							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     :						List[str]   )  ->  List[str]:
							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     :						Dict   )  ->  List[Any]:
							pass




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




					@unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''   )
					def   A  (				self     :						Optional[Any]   )  ->  Tuple:
							pass






					def   A  (				self     :						List[str]   )  ->  Tuple:
							return super().test_progress_bar()





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




					def   A  (				self     :						Any   )  ->  Union[str, Any]:
							lowercase_ : Optional[Any]							       =   load_numpy(
							    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'''   )

							lowercase_ : List[str]							       =   TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''   )
							lowercase_ : List[str]							       =   DPMSolverMultistepScheduler.from_config(pipe.scheduler.config   )
							lowercase_ : List[str]							       =   pipe.to('''cuda'''   )

							lowercase_ : List[str]							       =   '''Spiderman is surfing'''
							lowercase_ : List[Any]							       =   torch.Generator(device='''cpu'''   ).manual_seed(0   )

							lowercase_ : int							       =   pipe(A    ,						generator=A    ,						num_inference_steps=25    ,						output_type='''pt'''   ).frames
							lowercase_ : Union[str, Any]							       =   video_frames.cpu().numpy()

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






					def   A  (				self     :						Any   )  ->  Dict:
							lowercase_ : int							       =   load_numpy(
							    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'''   )

							lowercase_ : int							       =   TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''   )
							lowercase_ : List[Any]							       =   pipe.to('''cuda'''   )

							lowercase_ : Optional[Any]							       =   '''Spiderman is surfing'''
							lowercase_ : List[str]							       =   torch.Generator(device='''cpu'''   ).manual_seed(0   )

							lowercase_ : Union[str, Any]							       =   pipe(A    ,						generator=A    ,						num_inference_steps=2    ,						output_type='''pt'''   ).frames
							lowercase_ : Optional[Any]							       =   video_frames.cpu().numpy()

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











"""simple docstring"""



from __future__ import annotations

__A		:       List[Any]  				=		[-10, -5, 0, 5, 5.1, 11, 13, 21, 3, 4, -21, -10, -5, -1, 0]
__A		:       str  				=		[-5, 0, 5, 5.1, 11, 13, 21, -1, 4, -1, -10, -5, -1, 0, -1]



def 						lowercase    (				__snake_case				: list[float]	):
		lowercase_ : List[str]							       =   []
		lowercase_ : List[Any]							       =   len(__snake_case	)

		for i in range(__snake_case	):
				lowercase_ : float							       =   -1
				for j in range(i + 1		,					__snake_case	):
						if arr[i] < arr[j]:
								lowercase_ : List[str]							       =   arr[j]
								break
				result.append(__snake_case	)
		return result



def 						lowercase    (				__snake_case				: list[float]	):
		lowercase_ : List[str]							       =   []
		for i, outer in enumerate(__snake_case	):
				lowercase_ : float							       =   -1
				for inner in arr[i + 1 :]:
						if outer < inner:
								lowercase_ : List[Any]							       =   inner
								break
				result.append(__snake_case	)
		return result



def 						lowercase    (				__snake_case				: list[float]	):
		lowercase_ : List[str]							       =   len(__snake_case	)
		lowercase_ : list[float]							       =   []
		lowercase_ : list[float]							       =   [-1] * arr_size

		for index in reversed(range(__snake_case	)	):
				if stack:
						while stack[-1] <= arr[index]:
								stack.pop()
								if not stack:
										break
				if stack:
						lowercase_ : Optional[Any]							       =   stack[-1]
				stack.append(arr[index]	)
		return result


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

						testmod()
						print(next_greatest_element_slow(arr))
						print(next_greatest_element_fast(arr))
						print(next_greatest_element(arr))

						__A		:       int  				=		(
						    '''from __main__ import arr, next_greatest_element_slow, '''
						    '''next_greatest_element_fast, next_greatest_element'''
						)
						print(
						    '''next_greatest_element_slow():''',
						    timeit('''next_greatest_element_slow(arr)''', setup=setup),
						)
						print(
						    '''next_greatest_element_fast():''',
						    timeit('''next_greatest_element_fast(arr)''', setup=setup),
						)
						print(
						    '''     next_greatest_element():''',
						    timeit('''next_greatest_element(arr)''', setup=setup),
						)



