

'''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"""




  SCREAMING_SNAKE_CASE       :    List[str]							     =     TextToVideoSDPipeline
  SCREAMING_SNAKE_CASE       :    List[str]							     =     TEXT_TO_IMAGE_PARAMS
  SCREAMING_SNAKE_CASE       :    Dict							     =     TEXT_TO_IMAGE_BATCH_PARAMS
  # No `output_type`.
  SCREAMING_SNAKE_CASE       :    Optional[int]							     =     frozenset(
      [
          'num_inference_steps',
          'generator',
          'latents',
          'return_dict',
          'callback',
          'callback_steps',
      ] )






  def    SCREAMING_SNAKE_CASE ( self  :	Optional[int]     ):
   torch.manual_seed(0     )
   __lowercase				    =					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   ,)
   __lowercase				    =					DDIMScheduler(
       beta_start=0.0_0_0_8_5   ,beta_end=0.0_1_2   ,beta_schedule='''scaled_linear'''   ,clip_sample=lowercase__   ,set_alpha_to_one=lowercase__   ,)
   torch.manual_seed(0     )
   __lowercase				    =					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     )
   __lowercase				    =					CLIPTextConfig(
       bos_token_id=0   ,eos_token_id=2   ,hidden_size=3_2   ,intermediate_size=3_7   ,layer_norm_eps=1e-0_5   ,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   ,)
   __lowercase				    =					CLIPTextModel(lowercase__     )
   __lowercase				    =					CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''     )

   __lowercase				    =					{
       '''unet''': unet,
       '''scheduler''': scheduler,
       '''vae''': vae,
       '''text_encoder''': text_encoder,
       '''tokenizer''': tokenizer,
   }
   return components






  def    SCREAMING_SNAKE_CASE ( self  :	List[str]   ,lowercase__  :	int   ,lowercase__  :	List[str]=0     ):
   if str(lowercase__     ).startswith('''mps'''     ):
    __lowercase				    =					torch.manual_seed(lowercase__     )
   else:
    __lowercase				    =					torch.Generator(device=lowercase__     ).manual_seed(lowercase__     )
   __lowercase				    =					{
       '''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  :	Optional[int]     ):
   __lowercase				    =					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
   __lowercase				    =					self.get_dummy_components()
   __lowercase				    =					TextToVideoSDPipeline(**lowercase__     )
   __lowercase				    =					sd_pipe.to(lowercase__     )
   sd_pipe.set_progress_bar_config(disable=lowercase__     )

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

   assert frames[0].shape == (6_4, 6_4, 3)
   __lowercase				    =					np.array([1_5_8.0, 1_6_0.0, 1_5_3.0, 1_2_5.0, 1_0_0.0, 1_2_1.0, 1_1_1.0, 9_3.0, 1_1_3.0]     )

   assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-2






  def    SCREAMING_SNAKE_CASE ( self  :	Optional[Any]     ):
   self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowercase__   ,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  :	Any     ):
   self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowercase__   ,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[str]     ):
   pass






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






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






  def    SCREAMING_SNAKE_CASE ( self  :	List[str]     ):
   return super().test_progress_bar()






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




  """simple docstring"""




  def    SCREAMING_SNAKE_CASE ( self  :	int     ):
   __lowercase				    =					load_numpy(
       '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'''     )

   __lowercase				    =					TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''     )
   __lowercase				    =					DPMSolverMultistepScheduler.from_config(pipe.scheduler.config     )
   __lowercase				    =					pipe.to('''cuda'''     )

   __lowercase				    =					'''Spiderman is surfing'''
   __lowercase				    =					torch.Generator(device='''cpu'''     ).manual_seed(0     )

   __lowercase				    =					pipe(lowercase__   ,generator=lowercase__   ,num_inference_steps=2_5   ,output_type='''pt'''     ).frames
   __lowercase				    =					video_frames.cpu().numpy()

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






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

   __lowercase				    =					TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''     )
   __lowercase				    =					pipe.to('''cuda'''     )

   __lowercase				    =					'''Spiderman is surfing'''
   __lowercase				    =					torch.Generator(device='''cpu'''     ).manual_seed(0     )

   __lowercase				    =					pipe(lowercase__   ,generator=lowercase__   ,num_inference_steps=2   ,output_type='''pt'''     ).frames
   __lowercase				    =					video_frames.cpu().numpy()

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





'''simple docstring'''

from __future__ import annotations

from typing import Any
class     __lowerCAmelCase		:





      '''simple docstring'''






      def __init__(self       :       Optional[Any]  ,		UpperCamelCase       :       int = 6					):

             '''simple docstring'''







             lowercase__      =					None
             lowercase__      =					None
             self.create_linked_list(UpperCamelCase					)



      def  UpperCamelCase__						(self       :       List[str]  ,		UpperCamelCase       :       int					):

             '''simple docstring'''







             lowercase__      =					Node()
             lowercase__      =					current_node
             lowercase__      =					current_node
             lowercase__      =					current_node
             for _ in range(1  ,		UpperCamelCase					):
                    lowercase__      =					Node()
                    lowercase__      =					current_node
                    lowercase__      =					previous_node
                    lowercase__      =					current_node
             lowercase__      =					self.front
             lowercase__      =					previous_node



      def  UpperCamelCase__						(self       :       Union[str, Any]					):

             '''simple docstring'''







             return (
                 self.front == self.rear
                 and self.front is not None
                 and self.front.data is None
             )



      def  UpperCamelCase__						(self       :       Dict					):

             '''simple docstring'''







             self.check_can_perform_operation()
             return self.front.data if self.front else None



      def  UpperCamelCase__						(self       :       Any  ,		UpperCamelCase       :       Any					):

             '''simple docstring'''







             if self.rear is None:
                    return

             self.check_is_full()
             if not self.is_empty():
                    lowercase__      =					self.rear.next
             if self.rear:
                    lowercase__      =					data



      def  UpperCamelCase__						(self       :       List[Any]					):

             '''simple docstring'''







             self.check_can_perform_operation()
             if self.rear is None or self.front is None:
                    return None
             if self.front == self.rear:
                    lowercase__      =					self.front.data
                    lowercase__      =					None
                    return data

             lowercase__      =					self.front
             lowercase__      =					old_front.next
             lowercase__      =					old_front.data
             lowercase__      =					None
             return data



      def  UpperCamelCase__						(self       :       Union[str, Any]					):

             '''simple docstring'''







             if self.is_empty():
                    raise Exception('''Empty Queue'''					)






      def  UpperCamelCase__						(self       :       int					):

             '''simple docstring'''







             if self.rear and self.rear.next == self.front:
                    raise Exception('''Full Queue'''					)






class     __lowerCAmelCase		:





      '''simple docstring'''






      def __init__(self       :       Any					):

             '''simple docstring'''







             lowercase__      =					None
             lowercase__      =					None
             lowercase__      =					None


if __name__ == "__main__":
 import doctest

 doctest.testmod()




