




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







     def       lowerCAmelCase_    (     self			)  ->   Optional[Any]:



          '''simple docstring'''



          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.0_0_0_8_5  ,		beta_end=0.0_1_2  ,		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=128  ,		)
          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=1000  ,		hidden_act="gelu"  ,		projection_dim=512  ,		)
          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       lowerCAmelCase_    (     self  ,		A  ,		A=0			)  ->   Optional[Any]:



          '''simple docstring'''



          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       lowerCAmelCase_    (     self			)  ->   Tuple:



          '''simple docstring'''



          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([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       lowerCAmelCase_    (     self			)  ->   Optional[int]:



          '''simple docstring'''



          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       lowerCAmelCase_    (     self			)  ->   Any:



          '''simple docstring'''



          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       lowerCAmelCase_    (     self			)  ->   Union[str, Any]:



          '''simple docstring'''



          pass







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



          '''simple docstring'''



          pass







     @unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."			)
     def       lowerCAmelCase_    (     self			)  ->   Tuple:



          '''simple docstring'''



          pass







     def       lowerCAmelCase_    (     self			)  ->   int:



          '''simple docstring'''



          return super().test_progress_bar()







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







     def       lowerCAmelCase_    (     self			)  ->   List[Any]:



          '''simple docstring'''



          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       lowerCAmelCase_    (     self			)  ->   List[Any]:



          '''simple docstring'''



          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







lowercase__				:      str	   = "\n# Installazione di Transformers\n! pip install transformers datasets\n# Per installare dalla fonte invece dell'ultima versione rilasciata, commenta il comando sopra e\n# rimuovi la modalità commento al comando seguente.\n# ! pip install git+https://github.com/huggingface/transformers.git\n"

lowercase__				:      Any	   = [{"type": "code", "content": INSTALL_CONTENT}]
lowercase__				:      Any	   = {
    "{processor_class}": "FakeProcessorClass",
    "{model_class}": "FakeModelClass",
    "{object_class}": "FakeObjectClass",
}

