





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





   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      torch.manual_seed(0      )
      a__	:						List[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					,			)
      a__	:						Optional[int]      		=  DDIMScheduler(
          beta_start=0.00_085					,			beta_end=0.012					,			beta_schedule="scaled_linear"					,			clip_sample=snake_case					,			set_alpha_to_one=snake_case					,			)
      torch.manual_seed(0      )
      a__	:						str      		=  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__	:						Optional[int]      		=  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=1_000					,			hidden_act="gelu"					,			projection_dim=512					,			)
      a__	:						Optional[int]      		=  CLIPTextModel(snake_case      )
      a__	:						Optional[Any]      		=  CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"      )

      a__	:						Dict      		=  {
          "unet": unet,
          "scheduler": scheduler,
          "vae": vae,
          "text_encoder": text_encoder,
          "tokenizer": tokenizer,
      }
      return components





   def 	_snake_case ( self					,			snake_case					,			snake_case=0      )				->  Optional[Any]:
      """simple docstring"""





      if str(snake_case      ).startswith("mps"      ):
         a__	:						Union[str, Any]      		=  torch.manual_seed(snake_case      )
      else:
         a__	:						Dict      		=  torch.Generator(device=snake_case      ).manual_seed(snake_case      )
      a__	:						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 	_snake_case ( self      )				->  Optional[int]:
      """simple docstring"""





      a__	:						Tuple      		=  "cpu"  # ensure determinism for the device-dependent torch.Generator
      a__	:						Dict      		=  self.get_dummy_components()
      a__	:						List[Any]      		=  TextToVideoSDPipeline(**snake_case      )
      a__	:						Dict      		=  sd_pipe.to(snake_case      )
      sd_pipe.set_progress_bar_config(disable=snake_case      )

      a__	:						Union[str, Any]      		=  self.get_dummy_inputs(snake_case      )
      a__	:						List[Any]      		=  "np"
      a__	:						Union[str, Any]      		=  sd_pipe(**snake_case      ).frames
      a__	:						Dict      		=  frames[0][-3:, -3:, -1]

      assert frames[0].shape == (64, 64, 3)
      a__	:						Optional[Any]      		=  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 	_snake_case ( self      )				->  Dict:
      """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 	_snake_case ( self      )				->  Tuple:
      """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 	_snake_case ( self      )				->  str:
      """simple docstring"""





      pass





   @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."      )
   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      pass





   @unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."      )
   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





      pass





   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      return super().test_progress_bar()







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





   def 	_snake_case ( self      )				->  Union[str, Any]:
      """simple docstring"""





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

      a__	:						int      		=  TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b"      )
      a__	:						List[str]      		=  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config      )
      a__	:						List[Any]      		=  pipe.to("cuda"      )

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

      a__	:						str      		=  pipe(snake_case					,			generator=snake_case					,			num_inference_steps=25					,			output_type="pt"      ).frames
      a__	:						Dict      		=  video_frames.cpu().numpy()

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





   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  load_numpy(
          "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy"      )

      a__	:						List[str]      		=  TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b"      )
      a__	:						List[Any]      		=  pipe.to("cuda"      )

      a__	:						Union[str, Any]      		=  "Spiderman is surfing"
      a__	:						Optional[Any]      		=  torch.Generator(device="cpu"      ).manual_seed(0      )

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

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








import tempfile
import unittest

from make_student import create_student_by_copying_alternating_layers

from transformers import AutoConfig
from transformers.file_utils import cached_property
from transformers.testing_utils import require_torch


SCREAMING_SNAKE_CASE__       :			List[Any]			      =						"""sshleifer/bart-tiny-random"""
SCREAMING_SNAKE_CASE__       :			Tuple			      =						"""patrickvonplaten/t5-tiny-random"""







@require_torch
class   __lowerCAmelCase			(					unittest.TestCase  ):





   @cached_property
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      return AutoConfig.from_pretrained(snake_case      )





   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





      a__		,     *a__	:						int      		=  create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=1					,			d=1      )
      self.assertEqual(student.config.num_hidden_layers					,			1      )





   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      a__		,     *a__	:						int      		=  create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=1					,			d=snake_case      )





   def 	_snake_case ( self      )				->  List[str]:
      """simple docstring"""





      a__		,     *a__	:						Any      		=  create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=1					,			d=snake_case      )
      self.assertEqual(student.config.encoder_layers					,			1      )
      self.assertEqual(student.config.decoder_layers					,			self.teacher_config.encoder_layers      )





   def 	_snake_case ( self      )				->  Optional[int]:
      """simple docstring"""





      a__		,     *a__	:						List[str]      		=  create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=1					,			d=1      )
      self.assertEqual(student.config.encoder_layers					,			1      )
      self.assertEqual(student.config.decoder_layers					,			1      )





   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      with self.assertRaises(snake_case      ):
         create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=snake_case					,			d=snake_case      )

