





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__						(       lowercase__		,					unittest.TestCase    ):
      snake_case__       :  Tuple       =		TextToVideoSDPipeline
      snake_case__       :  Any       =		TEXT_TO_IMAGE_PARAMS
      snake_case__       :  Tuple       =		TEXT_TO_IMAGE_BATCH_PARAMS
      # No `output_type`.
      snake_case__       :  List[Any]       =		frozenset(
          [
              '''num_inference_steps''',
              '''generator''',
              '''latents''',
              '''return_dict''',
              '''callback''',
              '''callback_steps''',
          ]    )
      def  SCREAMING_SNAKE_CASE							( self       : Dict  )	->    Any:
             torch.manual_seed(0  )
             a_      :	Optional[Any]							=						UNetaDConditionModel(
                 block_out_channels=(3_2, 6_4, 6_4, 6_4)  ,							layers_per_block=2  ,							sample_size=3_2  ,							in_channels=4  ,							out_channels=4  ,							down_block_types=('CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'DownBlock3D')  ,							up_block_types=('UpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D')  ,							cross_attention_dim=3_2  ,							attention_head_dim=4  ,							)
             a_      :	Union[str, Any]							=						DDIMScheduler(
                 beta_start=0.00085  ,							beta_end=0.012  ,							beta_schedule='scaled_linear'  ,							clip_sample=SCREAMING_SNAKE_CASE__  ,							set_alpha_to_one=SCREAMING_SNAKE_CASE__  ,							)
             torch.manual_seed(0  )
             a_      :	Tuple							=						AutoencoderKL(
                 block_out_channels=[3_2, 6_4]  ,							in_channels=3  ,							out_channels=3  ,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']  ,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']  ,							latent_channels=4  ,							sample_size=1_2_8  ,							)
             torch.manual_seed(0  )
             a_      :	Optional[Any]							=						CLIPTextConfig(
                 bos_token_id=0  ,							eos_token_id=2  ,							hidden_size=3_2  ,							intermediate_size=3_7  ,							layer_norm_eps=1E-05  ,							num_attention_heads=4  ,							num_hidden_layers=5  ,							pad_token_id=1  ,							vocab_size=1_0_0_0  ,							hidden_act='gelu'  ,							projection_dim=5_1_2  ,							)
             a_      :	Any							=						CLIPTextModel(SCREAMING_SNAKE_CASE__  )
             a_      :	List[str]							=						CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )

             a_      :	Optional[int]							=						{
                 'unet': unet,
                 'scheduler': scheduler,
                 'vae': vae,
                 'text_encoder': text_encoder,
                 'tokenizer': tokenizer,
             }
             return components
      def  SCREAMING_SNAKE_CASE							( self       : List[Any]  ,							SCREAMING_SNAKE_CASE__       : Optional[int]  ,							SCREAMING_SNAKE_CASE__       : Any=0  )	->    Tuple:
             if str(SCREAMING_SNAKE_CASE__  ).startswith('mps'  ):
                    a_      :	Dict							=						torch.manual_seed(SCREAMING_SNAKE_CASE__  )
             else:
                    a_      :	Optional[Any]							=						torch.Generator(device=SCREAMING_SNAKE_CASE__  ).manual_seed(SCREAMING_SNAKE_CASE__  )
             a_      :	Dict							=						{
                 '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  SCREAMING_SNAKE_CASE							( self       : Any  )	->    Dict:
             a_      :	str							=						'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_      :	List[Any]							=						self.get_dummy_components()
             a_      :	Any							=						TextToVideoSDPipeline(**SCREAMING_SNAKE_CASE__  )
             a_      :	List[str]							=						sd_pipe.to(SCREAMING_SNAKE_CASE__  )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )

             a_      :	Optional[int]							=						self.get_dummy_inputs(SCREAMING_SNAKE_CASE__  )
             a_      :	Dict							=						'np'
             a_      :	List[Any]							=						sd_pipe(**SCREAMING_SNAKE_CASE__  ).frames
             a_      :	Union[str, Any]							=						frames[0][-3:, -3:, -1]

             assert frames[0].shape == (6_4, 6_4, 3)
             a_      :	str							=						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  SCREAMING_SNAKE_CASE							( self       : Dict  )	->    Optional[int]:
             self._test_attention_slicing_forward_pass(test_mean_pixel_difference=SCREAMING_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  SCREAMING_SNAKE_CASE							( self       : Optional[int]  )	->    Optional[int]:
             self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__  ,							expected_max_diff=1E-2  )
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'  )
      def  SCREAMING_SNAKE_CASE							( self       : List[Any]  )	->    Tuple:
             pass
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'  )
      def  SCREAMING_SNAKE_CASE							( self       : List[str]  )	->    str:
             pass
      @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'  )
      def  SCREAMING_SNAKE_CASE							( self       : Optional[int]  )	->    List[str]:
             pass

      def  SCREAMING_SNAKE_CASE							( self       : str  )	->    int:
             return super().test_progress_bar()
@slow
@skip_mps
class 				SCREAMING_SNAKE_CASE__						(       unittest.TestCase    ):
      def  SCREAMING_SNAKE_CASE							( self       : Tuple  )	->    Dict:
             a_      :	Any							=						load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'  )

             a_      :	List[Any]							=						TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'  )
             a_      :	Tuple							=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config  )
             a_      :	Any							=						pipe.to('cuda'  )

             a_      :	Optional[int]							=						'Spiderman is surfing'
             a_      :	Optional[Any]							=						torch.Generator(device='cpu'  ).manual_seed(0  )

             a_      :	int							=						pipe(SCREAMING_SNAKE_CASE__  ,							generator=SCREAMING_SNAKE_CASE__  ,							num_inference_steps=2_5  ,							output_type='pt'  ).frames
             a_      :	List[str]							=						video_frames.cpu().numpy()

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

      def  SCREAMING_SNAKE_CASE							( self       : int  )	->    List[Any]:
             a_      :	List[str]							=						load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'  )

             a_      :	Union[str, Any]							=						TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'  )
             a_      :	int							=						pipe.to('cuda'  )

             a_      :	List[str]							=						'Spiderman is surfing'
             a_      :	List[Any]							=						torch.Generator(device='cpu'  ).manual_seed(0  )

             a_      :	List[str]							=						pipe(SCREAMING_SNAKE_CASE__  ,							generator=SCREAMING_SNAKE_CASE__  ,							num_inference_steps=2  ,							output_type='pt'  ).frames
             a_      :	Optional[int]							=						video_frames.cpu().numpy()

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






from .dependency_versions_table import deps
from .utils.versions import require_version, require_version_core


# define which module versions we always want to check at run time
# (usually the ones defined in `install_requires` in setup.py)
#
# order specific notes:
# - tqdm must be checked before tokenizers

UpperCAmelCase_   :				Dict						      = [
    'python',
    'tqdm',
    'regex',
    'requests',
    'packaging',
    'filelock',
    'numpy',
    'tokenizers',
    'huggingface-hub',
    'safetensors',
    'accelerate',
    'pyyaml',
]

for pkg in pkgs_to_check_at_runtime:
   if pkg in deps:
      if pkg == "tokenizers":
         # must be loaded here, or else tqdm check may fail
         from .utils import is_tokenizers_available

         if not is_tokenizers_available():
            continue  # not required, check version only if installed
      elif pkg == "accelerate":
         # must be loaded here, or else tqdm check may fail
         from .utils import is_accelerate_available

         # Maybe switch to is_torch_available in the future here so that Accelerate is hard dep of
         # Transformers with PyTorch
         if not is_accelerate_available():
            continue  # not required, check version only if installed

      require_version_core(deps[pkg])
   else:
      raise ValueError(F'can\'t find {pkg} in {deps.keys()}, check dependency_versions_table.py')







def  SCREAMING_SNAKE_CASE_		(    __A				:    Optional[Any]						,					__A				:    int=None       )			->	int:

       """simple docstring"""

       require_version(deps[pkg]						,					__A       )


