




"""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 							lowercase_			(	__lowerCAmelCase     ,      unittest.TestCase				):



 '''simple docstring'''
 UpperCAmelCase       :	List[str]     =      TextToVideoSDPipeline
 UpperCAmelCase       :	List[str]     =      TEXT_TO_IMAGE_PARAMS
 UpperCAmelCase       :	int     =      TEXT_TO_IMAGE_BATCH_PARAMS
 # No `output_type`.
 UpperCAmelCase       :	Any     =      frozenset(
     [
         '''num_inference_steps''',
         '''generator''',
         '''latents''',
         '''return_dict''',
         '''callback''',
         '''callback_steps''',
     ]				)

 def      lowerCAmelCase_				(					self			:			Optional[Any]						):
       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=_UpperCAmelCase       ,				set_alpha_to_one=_UpperCAmelCase       ,				)
       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-0_5       ,				num_attention_heads=4       ,				num_hidden_layers=5       ,				pad_token_id=1       ,				vocab_size=1_000       ,				hidden_act='gelu'       ,				projection_dim=512       ,				)
       _A     =  CLIPTextModel(_UpperCAmelCase						)
       _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			:			List[str]       ,				_UpperCAmelCase			:			Tuple       ,				_UpperCAmelCase			:			int=0						):
       if str(_UpperCAmelCase						).startswith('mps'						):
             _A     =  torch.manual_seed(_UpperCAmelCase						)
       else:
             _A     =  torch.Generator(device=_UpperCAmelCase						).manual_seed(_UpperCAmelCase						)
       _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			:			Optional[Any]						):
       _A     =  'cpu'  # ensure determinism for the device-dependent torch.Generator
       _A     =  self.get_dummy_components()
       _A     =  TextToVideoSDPipeline(**_UpperCAmelCase						)
       _A     =  sd_pipe.to(_UpperCAmelCase						)
       sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase						)

       _A     =  self.get_dummy_inputs(_UpperCAmelCase						)
       _A     =  'np'
       _A     =  sd_pipe(**_UpperCAmelCase						).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			:			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      lowerCAmelCase_				(					self			:			int						):
       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      lowerCAmelCase_				(					self			:			Optional[Any]						):
       pass

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

 @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'						)
 def      lowerCAmelCase_				(					self			:			Optional[int]						):
       pass



 def      lowerCAmelCase_				(					self			:			Union[str, Any]						):
       return super().test_progress_bar()





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



 '''simple docstring'''

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

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







"""simple docstring"""

import logging
from dataclasses import dataclass, field
from typing import Optional

from seqaseq_trainer import arg_to_scheduler

from transformers import TrainingArguments


a							      =     logging.getLogger(__name__)





@dataclass
class 							lowercase_			(	__lowerCAmelCase				):



 '''simple docstring'''
 UpperCAmelCase       :	Optional[float]     =      field(
     default=0.0     ,      metadata={'''help''': '''The label smoothing epsilon to apply (if not zero).'''}				)
 UpperCAmelCase       :	bool     =      field(default=__lowerCAmelCase     ,      metadata={'''help''': '''Whether to SortishSamler or not.'''}				)
 UpperCAmelCase       :	bool     =      field(
     default=__lowerCAmelCase     ,      metadata={'''help''': '''Whether to use generate to calculate generative metrics (ROUGE, BLEU).'''}				)
 UpperCAmelCase       :	bool     =      field(default=__lowerCAmelCase     ,      metadata={'''help''': '''whether to use adafactor'''}				)
 UpperCAmelCase       :	Optional[float]     =      field(
     default=__lowerCAmelCase     ,      metadata={'''help''': '''Encoder layer dropout probability. Goes into model.config.'''}				)
 UpperCAmelCase       :	Optional[float]     =      field(
     default=__lowerCAmelCase     ,      metadata={'''help''': '''Decoder layer dropout probability. Goes into model.config.'''}				)
 UpperCAmelCase       :	Optional[float]     =      field(default=__lowerCAmelCase     ,      metadata={'''help''': '''Dropout probability. Goes into model.config.'''}				)
 UpperCAmelCase       :	Optional[float]     =      field(
     default=__lowerCAmelCase     ,      metadata={'''help''': '''Attention dropout probability. Goes into model.config.'''}				)
 UpperCAmelCase       :	Optional[str]     =      field(
     default='''linear'''     ,      metadata={'''help''': f'''Which lr scheduler to use. Selected in {sorted(arg_to_scheduler.keys()				)}'''}     ,      )

