





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      lowerCamelCase(						__snake_case			, unittest.TestCase	):
     '''simple docstring'''


     __magic_name__   =			TextToVideoSDPipeline
     __magic_name__   =			TEXT_TO_IMAGE_PARAMS
     __magic_name__   =			TEXT_TO_IMAGE_BATCH_PARAMS
     # No `output_type`.
     __magic_name__   =			frozenset(
         [
             'num_inference_steps',
             'generator',
             'latents',
             'return_dict',
             'callback',
             'callback_steps',
         ]	)




     def        lowerCAmelCase__		(    self							):
         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_0085				,     beta_end=0.012				,     beta_schedule='scaled_linear'				,     clip_sample=snake_case_				,     set_alpha_to_one=snake_case_				,     )
         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(snake_case_							)
         _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				,     snake_case_				,     snake_case_=0							):
         if str(snake_case_							).startswith('mps'							):
             _A     =							torch.manual_seed(snake_case_							)
         else:
             _A     =							torch.Generator(device=snake_case_							).manual_seed(snake_case_							)
         _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							):
         _A     =							'cpu'  # ensure determinism for the device-dependent torch.Generator
         _A     =							self.get_dummy_components()
         _A     =							TextToVideoSDPipeline(**snake_case_							)
         _A     =							sd_pipe.to(snake_case_							)
         sd_pipe.set_progress_bar_config(disable=snake_case_							)

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

         assert frames[0].shape == (64, 64, 3)
         _A     =							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        lowerCAmelCase__		(    self							):
         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        lowerCAmelCase__		(    self							):
         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        lowerCAmelCase__		(    self							):
         pass




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




     @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'							)
     def        lowerCAmelCase__		(    self							):
         pass




     def        lowerCAmelCase__		(    self							):
         return super().test_progress_bar()







@slow
@skip_mps
class      lowerCamelCase(						unittest.TestCase	):
     '''simple docstring'''


     def        lowerCAmelCase__		(    self							):
         _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(snake_case_				,     generator=snake_case_				,     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							):
         _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(snake_case_				,     generator=snake_case_				,     num_inference_steps=2				,     output_type='pt'							).frames
         _A     =							video_frames.cpu().numpy()

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








# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse

from ...utils.dataclasses import (
    ComputeEnvironment,
    DistributedType,
    DynamoBackend,
    PrecisionType,
    SageMakerDistributedType,
)
from ..menu import BulletMenu


a_     :    str     		=   [
    "EAGER",
    "AOT_EAGER",
    "INDUCTOR",
    "NVFUSER",
    "AOT_NVFUSER",
    "AOT_CUDAGRAPHS",
    "OFI",
    "FX2TRT",
    "ONNXRT",
    "IPEX",
]


def 							__lowerCAmelCase     (							_UpperCamelCase		:	int		,				_UpperCamelCase		:	List[Any]=None		,				_UpperCamelCase		:	Dict=None		,				_UpperCamelCase		:	str=None						)		->				Any:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE		    =      True
 while ask_again:
  SCREAMING_SNAKE_CASE		    =      input(_UpperCamelCase						)
  try:
   if default is not None and len(_UpperCamelCase						) == 0:
    return default
   return convert_value(_UpperCamelCase						) if convert_value is not None else result
  except Exception:
   if error_message is not None:
    print(_UpperCamelCase						)


def 							__lowerCAmelCase     (							_UpperCamelCase		:	Optional[Any]		,				_UpperCamelCase		:	Union[str, Any]=[]		,				_UpperCamelCase		:	Optional[int]=None		,				_UpperCamelCase		:	Optional[Any]=0						)		->				List[Any]:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE		    =      BulletMenu(_UpperCamelCase		,				_UpperCamelCase						)
 SCREAMING_SNAKE_CASE		    =      menu.run(default_choice=_UpperCamelCase						)
 return convert_value(_UpperCamelCase						) if convert_value is not None else result


def 							__lowerCAmelCase     (							_UpperCamelCase		:	List[str]						)		->				int:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE		    =      int(_UpperCamelCase						)
 return ComputeEnvironment(['LOCAL_MACHINE', 'AMAZON_SAGEMAKER'][value]						)


def 							__lowerCAmelCase     (							_UpperCamelCase		:	Tuple						)		->				Tuple:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE		    =      int(_UpperCamelCase						)
 return DistributedType(['NO', 'MULTI_CPU', 'MULTI_XPU', 'MULTI_GPU', 'MULTI_NPU', 'TPU'][value]						)


def 							__lowerCAmelCase     (							_UpperCamelCase		:	List[Any]						)		->				List[str]:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE		    =      int(_UpperCamelCase						)
 return DynamoBackend(DYNAMO_BACKENDS[value]						).value


def 							__lowerCAmelCase     (							_UpperCamelCase		:	Union[str, Any]						)		->				Dict:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE		    =      int(_UpperCamelCase						)
 return PrecisionType(['no', 'fp16', 'bf16', 'fp8'][value]						)


def 							__lowerCAmelCase     (							_UpperCamelCase		:	Tuple						)		->				str:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE		    =      int(_UpperCamelCase						)
 return SageMakerDistributedType(['NO', 'DATA_PARALLEL', 'MODEL_PARALLEL'][value]						)


def 							__lowerCAmelCase     (							_UpperCamelCase		:	Dict						)		->				str:
 '''simple docstring'''


 return {"yes": True, "no": False}[value.lower()]




class   UpperCamelCase			(  argparse.RawDescriptionHelpFormatter ):

    def UpperCamelCase      (      self			:   Union[str, Any]				,     snake_case__			:   Tuple				,     snake_case__			:   Optional[Any]				,     snake_case__			:   List[str]				,     snake_case__			:   Dict     ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE		    =      super()._format_usage(snake_case__				,     snake_case__				,     snake_case__				,     snake_case__     )
     SCREAMING_SNAKE_CASE		    =      usage.replace('<command> [<args>] '				,     ''     )
     return usage

