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 						__a (     _snake_case	,    unittest.TestCase				):
	SCREAMING_SNAKE_CASE       =       TextToVideoSDPipeline
	SCREAMING_SNAKE_CASE       =       TEXT_TO_IMAGE_PARAMS
	SCREAMING_SNAKE_CASE       =       TEXT_TO_IMAGE_BATCH_PARAMS
	# No `output_type`.
	SCREAMING_SNAKE_CASE       =       frozenset(
	    [
	        "num_inference_steps",
	        "generator",
	        "latents",
	        "return_dict",
	        "callback",
	        "callback_steps",
	    ]				)



	def 		UpperCamelCase							(						self							:    Optional[Any])->							List[Any]:
						torch.manual_seed(0)
						__lowerCAmelCase	   =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				,	)
						__lowerCAmelCase	   =DDIMScheduler(
						    beta_start=0.0_0_0_8_5				,	beta_end=0.0_1_2				,	beta_schedule="""scaled_linear"""				,	clip_sample=snake_case_				,	set_alpha_to_one=snake_case_				,	)
						torch.manual_seed(0)
						__lowerCAmelCase	   =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)
						__lowerCAmelCase	   =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				,	)
						__lowerCAmelCase	   =CLIPTextModel(snake_case_)
						__lowerCAmelCase	   =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

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



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

						__lowerCAmelCase	   =self.get_dummy_inputs(snake_case_)
						__lowerCAmelCase	   ="np"
						__lowerCAmelCase	   =sd_pipe(**snake_case_).frames
						__lowerCAmelCase	   =frames[0][-3:, -3:, -1]

						assert frames[0].shape == (64, 64, 3)
						__lowerCAmelCase	   =np.array([1_5_8.0, 1_6_0.0, 1_5_3.0, 1_2_5.0, 1_0_0.0, 1_2_1.0, 1_1_1.0, 9_3.0, 1_1_3.0])

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



	def 		UpperCamelCase							(						self							:    Any)->							int:
						self._test_attention_slicing_forward_pass(test_mean_pixel_difference=snake_case_				,	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 		UpperCamelCase							(						self							:    List[Any])->							Tuple:
						self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=snake_case_				,	expected_max_diff=1e-2)



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



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



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



	def 		UpperCamelCase							(						self							:    Optional[int])->							Dict:
						return super().test_progress_bar()






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



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

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

						__lowerCAmelCase	   ="Spiderman is surfing"
						__lowerCAmelCase	   =torch.Generator(device="""cpu""").manual_seed(0)

						__lowerCAmelCase	   =pipe(snake_case_				,	generator=snake_case_				,	num_inference_steps=25				,	output_type="""pt""").frames
						__lowerCAmelCase	   =video_frames.cpu().numpy()

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



	def 		UpperCamelCase							(						self							:    Optional[Any])->							List[Any]:
						__lowerCAmelCase	   =load_numpy(
						    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy""")

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

						__lowerCAmelCase	   ="Spiderman is surfing"
						__lowerCAmelCase	   =torch.Generator(device="""cpu""").manual_seed(0)

						__lowerCAmelCase	   =pipe(snake_case_				,	generator=snake_case_				,	num_inference_steps=2				,	output_type="""pt""").frames
						__lowerCAmelCase	   =video_frames.cpu().numpy()

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









import PIL.Image
import PIL.ImageOps
from packaging import version
from PIL import Image


if version.parse(version.parse(PIL.__version__).base_version) >= version.parse('''9.1.0'''):
				lowercase_	       =			{
				    '''linear''': PIL.Image.Resampling.BILINEAR,
				    '''bilinear''': PIL.Image.Resampling.BILINEAR,
				    '''bicubic''': PIL.Image.Resampling.BICUBIC,
				    '''lanczos''': PIL.Image.Resampling.LANCZOS,
				    '''nearest''': PIL.Image.Resampling.NEAREST,
				}
else:
				lowercase_	       =			{
				    '''linear''': PIL.Image.LINEAR,
				    '''bilinear''': PIL.Image.BILINEAR,
				    '''bicubic''': PIL.Image.BICUBIC,
				    '''lanczos''': PIL.Image.LANCZOS,
				    '''nearest''': PIL.Image.NEAREST,
				}






def 					__lowerCAmelCase		(  __lowerCamelCase	:    List[str]	)  ->						int:
					__lowerCAmelCase	   =(images / 2 + 0.5).clamp(0				,   1	)
					__lowerCAmelCase	   =images.cpu().permute(0				,   2				,   3				,   1	).float().numpy()
					__lowerCAmelCase	   =numpy_to_pil(__lowerCamelCase	)
					return images






def 					__lowerCAmelCase		(  __lowerCamelCase	:    str	)  ->						str:
					if images.ndim == 3:
										__lowerCAmelCase	   =images[None, ...]
					__lowerCAmelCase	   =(images * 255).round().astype("""uint8"""	)
					if images.shape[-1] == 1:
										# special case for grayscale (single channel) images
										__lowerCAmelCase	   =[Image.fromarray(image.squeeze()				,   mode="""L"""	) for image in images]
					else:
										__lowerCAmelCase	   =[Image.fromarray(__lowerCamelCase	) for image in images]

					return pil_images




