

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   __lowercase    (			UpperCAmelCase_		,	unittest.TestCase     ):




			"""simple docstring"""







			_UpperCAmelCase       :							List[Any]      							=					TextToVideoSDPipeline
			_UpperCAmelCase       :							List[str]      							=					TEXT_TO_IMAGE_PARAMS
			_UpperCAmelCase       :							Tuple      							=					TEXT_TO_IMAGE_BATCH_PARAMS
			# No `output_type`.
			_UpperCAmelCase       :							Any      							=					frozenset(
			    [
			        '''num_inference_steps''',
			        '''generator''',
			        '''latents''',
			        '''return_dict''',
			        '''callback''',
			        '''callback_steps''',
			    ]     )



			def 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[int]):
									torch.manual_seed(0)
									SCREAMING_SNAKE_CASE_:			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       ,						)
									SCREAMING_SNAKE_CASE_:			str  		=	DDIMScheduler(
									    beta_start=0.0_0085       ,						beta_end=0.012       ,						beta_schedule="scaled_linear"       ,						clip_sample=lowerCAmelCase__       ,						set_alpha_to_one=lowerCAmelCase__       ,						)
									torch.manual_seed(0)
									SCREAMING_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=128       ,						)
									torch.manual_seed(0)
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	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       ,						)
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	CLIPTextModel(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			Any  		=	CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

									SCREAMING_SNAKE_CASE_:			Dict  		=	{
									    "unet": unet,
									    "scheduler": scheduler,
									    "vae": vae,
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									}
									return components



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

									SCREAMING_SNAKE_CASE_:			List[str]  		=	self.get_dummy_inputs(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_:			str  		=	"np"
									SCREAMING_SNAKE_CASE_:			Dict  		=	sd_pipe(**lowerCAmelCase__).frames
									SCREAMING_SNAKE_CASE_:			Any  		=	frames[0][-3:, -3:, -1]

									assert frames[0].shape == (64, 64, 3)
									SCREAMING_SNAKE_CASE_:			Optional[int]  		=	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 	_SCREAMING_SNAKE_CASE      (			self  :					List[str]):
									self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowerCAmelCase__       ,						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 	_SCREAMING_SNAKE_CASE      (			self  :					Optional[int]):
									self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCAmelCase__       ,						expected_max_diff=1E-2)



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



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



			@unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline.")
			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[Any]):
									pass




			def 	_SCREAMING_SNAKE_CASE      (			self  :					Any):
									return super().test_progress_bar()







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




			"""simple docstring"""







			def 	_SCREAMING_SNAKE_CASE      (			self  :					List[str]):
									SCREAMING_SNAKE_CASE_:			int  		=	load_numpy(
									    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy")

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

									SCREAMING_SNAKE_CASE_:			str  		=	"Spiderman is surfing"
									SCREAMING_SNAKE_CASE_:			str  		=	torch.Generator(device="cpu").manual_seed(0)

									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	pipe(lowerCAmelCase__       ,						generator=lowerCAmelCase__       ,						num_inference_steps=25       ,						output_type="pt").frames
									SCREAMING_SNAKE_CASE_:			Tuple  		=	video_frames.cpu().numpy()

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




			def 	_SCREAMING_SNAKE_CASE      (			self  :					Tuple):
									SCREAMING_SNAKE_CASE_:			str  		=	load_numpy(
									    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy")

									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b")
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	pipe.to("cuda")

									SCREAMING_SNAKE_CASE_:			Dict  		=	"Spiderman is surfing"
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	torch.Generator(device="cpu").manual_seed(0)

									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	pipe(lowerCAmelCase__       ,						generator=lowerCAmelCase__       ,						num_inference_steps=2       ,						output_type="pt").frames
									SCREAMING_SNAKE_CASE_:			Tuple  		=	video_frames.cpu().numpy()

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









from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowerCAmelCase							:  Optional[int]            =						logging.get_logger(__name__)

lowerCAmelCase							:  int            =						{
    """google/fnet-base""": """https://huggingface.co/google/fnet-base/resolve/main/config.json""",
    """google/fnet-large""": """https://huggingface.co/google/fnet-large/resolve/main/config.json"""
    # See all FNet models at https://huggingface.co/models?filter=fnet
}



class   __lowercase    (			UpperCAmelCase_     ):




			"""simple docstring"""







			_UpperCAmelCase       :							int      							=					'''fnet'''



			def __init__(			self  :					Optional[Any]       ,						lowerCAmelCase__  :					List[str]=3_2000       ,						lowerCAmelCase__  :					Union[str, Any]=768       ,						lowerCAmelCase__  :					Optional[Any]=12       ,						lowerCAmelCase__  :					Tuple=3072       ,						lowerCAmelCase__  :					Any="gelu_new"       ,						lowerCAmelCase__  :					Union[str, Any]=0.1       ,						lowerCAmelCase__  :					Optional[Any]=512       ,						lowerCAmelCase__  :					int=4       ,						lowerCAmelCase__  :					List[Any]=0.02       ,						lowerCAmelCase__  :					Union[str, Any]=1E-12       ,						lowerCAmelCase__  :					Optional[int]=False       ,						lowerCAmelCase__  :					Any=512       ,						lowerCAmelCase__  :					Optional[int]=3       ,						lowerCAmelCase__  :					int=1       ,						lowerCAmelCase__  :					List[Any]=2       ,						**lowerCAmelCase__  :					Optional[int]       ,						):
									super().__init__(pad_token_id=lowerCAmelCase__       ,						bos_token_id=lowerCAmelCase__       ,						eos_token_id=lowerCAmelCase__       ,						**lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_:			str  		=	vocab_size
									SCREAMING_SNAKE_CASE_:			Optional[Any]  		=	max_position_embeddings
									SCREAMING_SNAKE_CASE_:			Tuple  		=	hidden_size
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	num_hidden_layers
									SCREAMING_SNAKE_CASE_:			Optional[int]  		=	intermediate_size
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	hidden_act
									SCREAMING_SNAKE_CASE_:			Any  		=	hidden_dropout_prob
									SCREAMING_SNAKE_CASE_:			int  		=	initializer_range
									SCREAMING_SNAKE_CASE_:			Optional[int]  		=	type_vocab_size
									SCREAMING_SNAKE_CASE_:			int  		=	layer_norm_eps
									SCREAMING_SNAKE_CASE_:			Union[str, Any]  		=	use_tpu_fourier_optimizations
									SCREAMING_SNAKE_CASE_:			int  		=	tpu_short_seq_length






