






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




       def 					a_					(					self      : str					):


            """simple docstring"""







            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						:       List[Any]				   =				DDIMScheduler(
                beta_start=0.0_0085	,    beta_end=0.012	,    beta_schedule="""scaled_linear"""	,    clip_sample=snake_case_	,    set_alpha_to_one=snake_case_	,    )
            torch.manual_seed(0					)
            __lowerCamelCase						:       Union[str, 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					)
            __lowerCamelCase						:       Optional[Any]				   =				CLIPTextConfig(
                bos_token_id=0	,    eos_token_id=2	,    hidden_size=32	,    intermediate_size=37	,    layer_norm_eps=1e-0_5	,    num_attention_heads=4	,    num_hidden_layers=5	,    pad_token_id=1	,    vocab_size=1000	,    hidden_act="""gelu"""	,    projection_dim=512	,    )
            __lowerCamelCase						:       Tuple				   =				CLIPTextModel(snake_case_					)
            __lowerCamelCase						:       Optional[int]				   =				CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""					)

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




       def 					a_					(					self      : int	,    A__      : Optional[int]	,    A__      : Tuple=0					):


            """simple docstring"""







            if str(snake_case_					).startswith("""mps"""					):
                 __lowerCamelCase						:       str				   =				torch.manual_seed(snake_case_					)
            else:
                 __lowerCamelCase						:       Union[str, Any]				   =				torch.Generator(device=snake_case_					).manual_seed(snake_case_					)
            __lowerCamelCase						:       str				   =				{
                "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      : Dict					):


            """simple docstring"""







            __lowerCamelCase						:       int				   =				"cpu"  # ensure determinism for the device-dependent torch.Generator
            __lowerCamelCase						:       Optional[Any]				   =				self.get_dummy_components()
            __lowerCamelCase						:       Tuple				   =				TextToVideoSDPipeline(**snake_case_					)
            __lowerCamelCase						:       List[str]				   =				sd_pipe.to(snake_case_					)
            sd_pipe.set_progress_bar_config(disable=snake_case_					)

            __lowerCamelCase						:       int				   =				self.get_dummy_inputs(snake_case_					)
            __lowerCamelCase						:       Union[str, Any]				   =				"np"
            __lowerCamelCase						:       Dict				   =				sd_pipe(**snake_case_					).frames
            __lowerCamelCase						:       Any				   =				frames[0][-3:, -3:, -1]

            assert frames[0].shape == (64, 64, 3)
            __lowerCamelCase						:       Dict				   =				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      : Optional[int]					):


            """simple docstring"""







            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 					a_					(					self      : Dict					):


            """simple docstring"""







            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 					a_					(					self      : int					):


            """simple docstring"""







            pass




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


            """simple docstring"""







            pass




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


            """simple docstring"""







            pass




       def 					a_					(					self      : Tuple					):


            """simple docstring"""







            return super().test_progress_bar()







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




       def 					a_					(					self      : int					):


            """simple docstring"""







            __lowerCamelCase						:       List[Any]				   =				load_numpy(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy"""					)

            __lowerCamelCase						:       int				   =				TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""					)
            __lowerCamelCase						:       str				   =				DPMSolverMultistepScheduler.from_config(pipe.scheduler.config					)
            __lowerCamelCase						:       Tuple				   =				pipe.to("""cuda"""					)

            __lowerCamelCase						:       List[Any]				   =				"Spiderman is surfing"
            __lowerCamelCase						:       Optional[int]				   =				torch.Generator(device="""cpu"""					).manual_seed(0					)

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

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




       def 					a_					(					self      : Optional[Any]					):


            """simple docstring"""







            __lowerCamelCase						:       Any				   =				load_numpy(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy"""					)

            __lowerCamelCase						:       str				   =				TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""					)
            __lowerCamelCase						:       int				   =				pipe.to("""cuda"""					)

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

            __lowerCamelCase						:       Any				   =				pipe(snake_case_	,    generator=snake_case_	,    num_inference_steps=2	,    output_type="""pt"""					).frames
            __lowerCamelCase						:       Optional[int]				   =				video_frames.cpu().numpy()

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











'''simple docstring'''

from transformers import HfArgumentParser, TensorFlowBenchmark, TensorFlowBenchmarkArguments
def     __lowercase  ()   ->     str:


     """simple docstring"""
     __lowerCamelCase						:       Any				   =				HfArgumentParser(_lowercase     )
     __lowerCamelCase						:       List[str]				   =				parser.parse_args_into_dataclasses()[0]
     __lowerCamelCase						:       Dict				   =				TensorFlowBenchmark(args=_lowercase     )
     try:
          __lowerCamelCase						:       List[str]				   =				parser.parse_args_into_dataclasses()[0]
     except ValueError as e:
          __lowerCamelCase						:       Tuple				   =				"""Arg --no_{0} is no longer used, please use --no-{0} instead."""
          __lowerCamelCase						:       List[Any]				   =				""" """.join(str(_lowercase     ).split(""" """     )[:-1]     )
          __lowerCamelCase						:       Tuple				   =				""""""
          __lowerCamelCase						:       List[str]				   =				eval(str(_lowercase     ).split(""" """     )[-1]     )
          __lowerCamelCase						:       Tuple				   =				[]
          for arg in depreciated_args:
               # arg[2:] removes '--'
               if arg[2:] in TensorFlowBenchmark.deprecated_args:
                    # arg[5:] removes '--no_'
                    full_error_msg += arg_error_msg.format(arg[5:]     )
               else:
                    wrong_args.append(_lowercase     )
          if len(_lowercase     ) > 0:
               __lowerCamelCase						:       Tuple				   =				full_error_msg + begin_error_msg + str(_lowercase     )
          raise ValueError(_lowercase     )
     benchmark.run()


if __name__ == "__main__":
   main()



