




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__	(lowerCAmelCase__    ,				unittest.TestCase     ):

    '''simple docstring'''



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







    def     _lowercase     (    self     )     ->      Optional[int]:
           torch.manual_seed(0     )
           lowerCamelCase   :      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     ,  )
           lowerCamelCase   :      Optional[Any]			 =					DDIMScheduler(
               beta_start=0.00085     ,  beta_end=0.012     ,  beta_schedule="scaled_linear"     ,  clip_sample=UpperCamelCase__     ,  set_alpha_to_one=UpperCamelCase__     ,  )
           torch.manual_seed(0     )
           lowerCamelCase   :      List[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=128     ,  )
           torch.manual_seed(0     )
           lowerCamelCase   :      List[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=1000     ,  hidden_act="gelu"     ,  projection_dim=512     ,  )
           lowerCamelCase   :      Any			 =					CLIPTextModel(UpperCamelCase__     )
           lowerCamelCase   :      Union[str, Any]			 =					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

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







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

           lowerCamelCase   :      Dict			 =					self.get_dummy_inputs(UpperCamelCase__     )
           lowerCamelCase   :      Dict			 =					"np"
           lowerCamelCase   :      Optional[int]			 =					sd_pipe(**UpperCamelCase__     ).frames
           lowerCamelCase   :      Dict			 =					frames[0][-3:, -3:, -1]

           assert frames[0].shape == (64, 64, 3)
           lowerCamelCase   :      Optional[Any]			 =					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     _lowercase     (    self     )     ->      Any:
           self._test_attention_slicing_forward_pass(test_mean_pixel_difference=UpperCamelCase__     ,  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     _lowercase     (    self     )     ->      Optional[Any]:
           self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=UpperCamelCase__     ,  expected_max_diff=1e-2     )







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







    @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."     )
    def     _lowercase     (    self     )     ->      Union[str, Any]:
           pass







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







    def     _lowercase     (    self     )     ->      int:
           return super().test_progress_bar()




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

    '''simple docstring'''



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

           lowerCamelCase   :      Optional[Any]			 =					TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b"     )
           lowerCamelCase   :      Optional[int]			 =					DPMSolverMultistepScheduler.from_config(pipe.scheduler.config     )
           lowerCamelCase   :      Tuple			 =					pipe.to("cuda"     )

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

           lowerCamelCase   :      Tuple			 =					pipe(UpperCamelCase__     ,  generator=UpperCamelCase__     ,  num_inference_steps=25     ,  output_type="pt"     ).frames
           lowerCamelCase   :      List[str]			 =					video_frames.cpu().numpy()

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







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

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

           lowerCamelCase   :      Optional[int]			 =					"Spiderman is surfing"
           lowerCamelCase   :      Union[str, Any]			 =					torch.Generator(device="cpu"     ).manual_seed(0     )

           lowerCamelCase   :      int			 =					pipe(UpperCamelCase__     ,  generator=UpperCamelCase__     ,  num_inference_steps=2     ,  output_type="pt"     ).frames
           lowerCamelCase   :      List[str]			 =					video_frames.cpu().numpy()

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












from typing import TYPE_CHECKING

from ....utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


SCREAMING_SNAKE_CASE__       :      Optional[Any]  =   {'configuration_mmbt': ['MMBTConfig']}

try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       SCREAMING_SNAKE_CASE__       :      List[Any]  =   ['MMBTForClassification', 'MMBTModel', 'ModalEmbeddings']


if TYPE_CHECKING:
       from .configuration_mmbt import MMBTConfig

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_mmbt import MMBTForClassification, MMBTModel, ModalEmbeddings

else:
       import sys

       SCREAMING_SNAKE_CASE__       :      int  =   _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)






