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       snake_case					(      lowercase__   ,						unittest.TestCase						):
  lowercase_			=     TextToVideoSDPipeline
  lowercase_			=     TEXT_TO_IMAGE_PARAMS
  lowercase_			=     TEXT_TO_IMAGE_BATCH_PARAMS
  # No `output_type`.
  lowercase_			=     frozenset(
      [
          'num_inference_steps',
          'generator',
          'latents',
          'return_dict',
          'callback',
          'callback_steps',
      ]						)




  def 	__lowercase(      self     :	int				)->       Optional[int]:




       """simple docstring"""
       torch.manual_seed(0				)
       SCREAMING_SNAKE_CASE__				:				Tuple			=		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     , )
       SCREAMING_SNAKE_CASE__				:				List[Any]			=		DDIMScheduler(
           beta_start=0.0_0085     , beta_end=0.012     , beta_schedule='scaled_linear'     , clip_sample=a_     , set_alpha_to_one=a_     , )
       torch.manual_seed(0				)
       SCREAMING_SNAKE_CASE__				:				Optional[Any]			=		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				)
       SCREAMING_SNAKE_CASE__				:				Dict			=		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=1000     , hidden_act='gelu'     , projection_dim=512     , )
       SCREAMING_SNAKE_CASE__				:				Any			=		CLIPTextModel(a_				)
       SCREAMING_SNAKE_CASE__				:				Optional[Any]			=		CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'				)

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




  def 	__lowercase(      self     :	Dict     , a_     :	Optional[Any]     , a_     :	Dict=0				)->       Any:




       """simple docstring"""
       if str(a_				).startswith('mps'				):
            SCREAMING_SNAKE_CASE__				:				Union[str, Any]			=		torch.manual_seed(a_				)
       else:
            SCREAMING_SNAKE_CASE__				:				Any			=		torch.Generator(device=a_				).manual_seed(a_				)
       SCREAMING_SNAKE_CASE__				:				int			=		{
           '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 	__lowercase(      self     :	Union[str, Any]				)->       Tuple:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				List[str]			=		'cpu'  # ensure determinism for the device-dependent torch.Generator
       SCREAMING_SNAKE_CASE__				:				Dict			=		self.get_dummy_components()
       SCREAMING_SNAKE_CASE__				:				Optional[Any]			=		TextToVideoSDPipeline(**a_				)
       SCREAMING_SNAKE_CASE__				:				int			=		sd_pipe.to(a_				)
       sd_pipe.set_progress_bar_config(disable=a_				)

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

       assert frames[0].shape == (64, 64, 3)
       SCREAMING_SNAKE_CASE__				:				Union[str, 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 	__lowercase(      self     :	Dict				)->       Tuple:




       """simple docstring"""
       self._test_attention_slicing_forward_pass(test_mean_pixel_difference=a_     , 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 	__lowercase(      self     :	Optional[int]				)->       Tuple:




       """simple docstring"""
       self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=a_     , expected_max_diff=1e-2				)




  @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'				)
  def 	__lowercase(      self     :	Tuple				)->       List[Any]:




       """simple docstring"""
       pass




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




       """simple docstring"""
       pass




  @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'				)
  def 	__lowercase(      self     :	Tuple				)->       Union[str, Any]:




       """simple docstring"""
       pass




  def 	__lowercase(      self     :	List[str]				)->       Optional[Any]:




       """simple docstring"""
       return super().test_progress_bar()




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




  def 	__lowercase(      self     :	Any				)->       int:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				Tuple			=		load_numpy(
           'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'				)

       SCREAMING_SNAKE_CASE__				:				str			=		TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'				)
       SCREAMING_SNAKE_CASE__				:				Dict			=		DPMSolverMultistepScheduler.from_config(pipe.scheduler.config				)
       SCREAMING_SNAKE_CASE__				:				Any			=		pipe.to('cuda'				)

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

       SCREAMING_SNAKE_CASE__				:				Tuple			=		pipe(a_     , generator=a_     , num_inference_steps=25     , output_type='pt'				).frames
       SCREAMING_SNAKE_CASE__				:				int			=		video_frames.cpu().numpy()

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




  def 	__lowercase(      self     :	Any				)->       Any:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				int			=		load_numpy(
           'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'				)

       SCREAMING_SNAKE_CASE__				:				Optional[Any]			=		TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'				)
       SCREAMING_SNAKE_CASE__				:				Tuple			=		pipe.to('cuda'				)

       SCREAMING_SNAKE_CASE__				:				Union[str, Any]			=		'Spiderman is surfing'
       SCREAMING_SNAKE_CASE__				:				int			=		torch.Generator(device='cpu'				).manual_seed(0				)

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

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




import unittest

import numpy as np
import requests

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
      import torch

      from transformers.pytorch_utils import is_torch_greater_or_equal_than_1_11
else:
      SCREAMING_SNAKE_CASE__		:					Any						=					False

if is_vision_available():
      from PIL import Image

      from transformers import PixaStructImageProcessor

class       snake_case					(      unittest.TestCase						):




  def __init__(      self     :	List[Any]     , a_     :	Optional[int]     , a_     :	Dict=7     , a_     :	Any=3     , a_     :	Any=18     , a_     :	int=30     , a_     :	int=400     , a_     :	List[Any]=None     , a_     :	int=True     , a_     :	int=True     , a_     :	Dict=None     , )->       Optional[int]:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				Dict			=		size if size is not None else {'height': 20, 'width': 20}
       SCREAMING_SNAKE_CASE__				:				str			=		parent
       SCREAMING_SNAKE_CASE__				:				Union[str, Any]			=		batch_size
       SCREAMING_SNAKE_CASE__				:				Any			=		num_channels
       SCREAMING_SNAKE_CASE__				:				Optional[Any]			=		image_size
       SCREAMING_SNAKE_CASE__				:				List[str]			=		min_resolution
       SCREAMING_SNAKE_CASE__				:				Dict			=		max_resolution
       SCREAMING_SNAKE_CASE__				:				List[Any]			=		size
       SCREAMING_SNAKE_CASE__				:				Tuple			=		do_normalize
       SCREAMING_SNAKE_CASE__				:				Optional[Any]			=		do_convert_rgb
       SCREAMING_SNAKE_CASE__				:				List[str]			=		[512, 1024, 2048, 4096]
       SCREAMING_SNAKE_CASE__				:				Union[str, Any]			=		patch_size if patch_size is not None else {'height': 16, 'width': 16}




  def 	__lowercase(      self     :	Optional[Any]				)->       str:




       """simple docstring"""
       return {"do_normalize": self.do_normalize, "do_convert_rgb": self.do_convert_rgb}




  def 	__lowercase(      self     :	Dict				)->       Dict:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				Optional[Any]			=		'https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/australia.jpg'
       SCREAMING_SNAKE_CASE__				:				str			=		Image.open(requests.get(a_     , stream=a_				).raw				).convert('RGB'				)
       return raw_image




@unittest.skipIf(
    not is_torch_greater_or_equal_than_1_11   ,						reason='`Pix2StructImageProcessor` requires `torch>=1.11.0`.'   ,						)
@require_torch
@require_vision
class       snake_case					(      UpperCamelCase_   ,						unittest.TestCase						):
  lowercase_			=     PixaStructImageProcessor if is_vision_available() else None




  def 	__lowercase(      self     :	List[str]				)->       Optional[Any]:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				int			=		PixaStructImageProcessingTester(self				)




  @property
  def 	__lowercase(      self     :	Dict				)->       Tuple:




       """simple docstring"""
       return self.image_processor_tester.prepare_image_processor_dict()




  def 	__lowercase(      self     :	Any				)->       Tuple:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				Optional[int]			=		self.image_processing_class(**self.image_processor_dict				)
       self.assertTrue(hasattr(a_     , 'do_normalize'				)				)
       self.assertTrue(hasattr(a_     , 'do_convert_rgb'				)				)




  def 	__lowercase(      self     :	List[Any]				)->       Tuple:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				Dict			=		self.image_processor_tester.prepare_dummy_image()

       SCREAMING_SNAKE_CASE__				:				Union[str, Any]			=		self.image_processing_class(**self.image_processor_dict				)
       SCREAMING_SNAKE_CASE__				:				List[Any]			=		2048

       SCREAMING_SNAKE_CASE__				:				Optional[int]			=		image_processor(a_     , return_tensors='pt'     , max_patches=a_				)
       self.assertTrue(torch.allclose(inputs.flattened_patches.mean()     , torch.tensor(0.0606				)     , atol=1e-3     , rtol=1e-3				)				)




  def 	__lowercase(      self     :	Any				)->       Tuple:




       """simple docstring"""
       # Initialize image_processor
       SCREAMING_SNAKE_CASE__				:				Union[str, Any]			=		self.image_processing_class(**self.image_processor_dict				)
       # create random PIL images
       SCREAMING_SNAKE_CASE__				:				Dict			=		prepare_image_inputs(self.image_processor_tester     , equal_resolution=a_				)
       for image in image_inputs:
            self.assertIsInstance(a_     , Image.Image				)

       # Test not batched input
       SCREAMING_SNAKE_CASE__				:				str			=		(
           (self.image_processor_tester.patch_size['height'] * self.image_processor_tester.patch_size['width'])
           * self.image_processor_tester.num_channels
       ) + 2

       for max_patch in self.image_processor_tester.max_patches:
            # Test not batched input
            SCREAMING_SNAKE_CASE__				:				List[str]			=		image_processor(
                image_inputs[0]     , return_tensors='pt'     , max_patches=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (1, max_patch, expected_hidden_dim)     , )

            # Test batched
            SCREAMING_SNAKE_CASE__				:				Tuple			=		image_processor(
                a_     , return_tensors='pt'     , max_patches=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (self.image_processor_tester.batch_size, max_patch, expected_hidden_dim)     , )




  def 	__lowercase(      self     :	Any				)->       Any:




       """simple docstring"""
       # Initialize image_processor
       SCREAMING_SNAKE_CASE__				:				Tuple			=		self.image_processing_class(**self.image_processor_dict				)
       # create random PIL images
       SCREAMING_SNAKE_CASE__				:				Dict			=		prepare_image_inputs(self.image_processor_tester     , equal_resolution=a_				)
       for image in image_inputs:
            self.assertIsInstance(a_     , Image.Image				)

       # Test not batched input
       SCREAMING_SNAKE_CASE__				:				str			=		(
           (self.image_processor_tester.patch_size['height'] * self.image_processor_tester.patch_size['width'])
           * self.image_processor_tester.num_channels
       ) + 2

       SCREAMING_SNAKE_CASE__				:				int			=		True

       for max_patch in self.image_processor_tester.max_patches:
            # Test not batched input
            with self.assertRaises(a_				):
                 SCREAMING_SNAKE_CASE__				:				Dict			=		image_processor(
                     image_inputs[0]     , return_tensors='pt'     , max_patches=a_				).flattened_patches

            SCREAMING_SNAKE_CASE__				:				List[Any]			=		'Hello'

            SCREAMING_SNAKE_CASE__				:				List[Any]			=		image_processor(
                image_inputs[0]     , return_tensors='pt'     , max_patches=a_     , header_text=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (1, max_patch, expected_hidden_dim)     , )

            # Test batched
            SCREAMING_SNAKE_CASE__				:				Any			=		image_processor(
                a_     , return_tensors='pt'     , max_patches=a_     , header_text=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (self.image_processor_tester.batch_size, max_patch, expected_hidden_dim)     , )




  def 	__lowercase(      self     :	List[Any]				)->       Dict:




       """simple docstring"""
       # Initialize image_processor
       SCREAMING_SNAKE_CASE__				:				Optional[int]			=		self.image_processing_class(**self.image_processor_dict				)
       # create random numpy tensors
       SCREAMING_SNAKE_CASE__				:				List[Any]			=		prepare_image_inputs(self.image_processor_tester     , equal_resolution=a_     , numpify=a_				)
       for image in image_inputs:
            self.assertIsInstance(a_     , np.ndarray				)

       SCREAMING_SNAKE_CASE__				:				str			=		(
           (self.image_processor_tester.patch_size['height'] * self.image_processor_tester.patch_size['width'])
           * self.image_processor_tester.num_channels
       ) + 2

       for max_patch in self.image_processor_tester.max_patches:
            # Test not batched input
            SCREAMING_SNAKE_CASE__				:				str			=		image_processor(
                image_inputs[0]     , return_tensors='pt'     , max_patches=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (1, max_patch, expected_hidden_dim)     , )

            # Test batched
            SCREAMING_SNAKE_CASE__				:				int			=		image_processor(
                a_     , return_tensors='pt'     , max_patches=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (self.image_processor_tester.batch_size, max_patch, expected_hidden_dim)     , )




  def 	__lowercase(      self     :	str				)->       Optional[Any]:




       """simple docstring"""
       # Initialize image_processor
       SCREAMING_SNAKE_CASE__				:				Any			=		self.image_processing_class(**self.image_processor_dict				)
       # create random PyTorch tensors
       SCREAMING_SNAKE_CASE__				:				List[Any]			=		prepare_image_inputs(self.image_processor_tester     , equal_resolution=a_     , torchify=a_				)
       for image in image_inputs:
            self.assertIsInstance(a_     , torch.Tensor				)

       # Test not batched input
       SCREAMING_SNAKE_CASE__				:				Any			=		(
           (self.image_processor_tester.patch_size['height'] * self.image_processor_tester.patch_size['width'])
           * self.image_processor_tester.num_channels
       ) + 2

       for max_patch in self.image_processor_tester.max_patches:
            # Test not batched input
            SCREAMING_SNAKE_CASE__				:				Optional[Any]			=		image_processor(
                image_inputs[0]     , return_tensors='pt'     , max_patches=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (1, max_patch, expected_hidden_dim)     , )

            # Test batched
            SCREAMING_SNAKE_CASE__				:				List[Any]			=		image_processor(
                a_     , return_tensors='pt'     , max_patches=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (self.image_processor_tester.batch_size, max_patch, expected_hidden_dim)     , )




@unittest.skipIf(
    not is_torch_greater_or_equal_than_1_11   ,						reason='`Pix2StructImageProcessor` requires `torch>=1.11.0`.'   ,						)
@require_torch
@require_vision
class       snake_case					(      UpperCamelCase_   ,						unittest.TestCase						):
  lowercase_			=     PixaStructImageProcessor if is_vision_available() else None




  def 	__lowercase(      self     :	Union[str, Any]				)->       Optional[Any]:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				str			=		PixaStructImageProcessingTester(self     , num_channels=4				)
       SCREAMING_SNAKE_CASE__				:				Dict			=		3




  @property
  def 	__lowercase(      self     :	Any				)->       Any:




       """simple docstring"""
       return self.image_processor_tester.prepare_image_processor_dict()




  def 	__lowercase(      self     :	Dict				)->       Optional[int]:




       """simple docstring"""
       SCREAMING_SNAKE_CASE__				:				int			=		self.image_processing_class(**self.image_processor_dict				)
       self.assertTrue(hasattr(a_     , 'do_normalize'				)				)
       self.assertTrue(hasattr(a_     , 'do_convert_rgb'				)				)




  def 	__lowercase(      self     :	str				)->       Union[str, Any]:




       """simple docstring"""
       # Initialize image_processor
       SCREAMING_SNAKE_CASE__				:				str			=		self.image_processing_class(**self.image_processor_dict				)
       # create random PIL images
       SCREAMING_SNAKE_CASE__				:				Tuple			=		prepare_image_inputs(self.image_processor_tester     , equal_resolution=a_				)
       for image in image_inputs:
            self.assertIsInstance(a_     , Image.Image				)

       # Test not batched input
       SCREAMING_SNAKE_CASE__				:				Dict			=		(
           (self.image_processor_tester.patch_size['height'] * self.image_processor_tester.patch_size['width'])
           * (self.image_processor_tester.num_channels - 1)
       ) + 2

       for max_patch in self.image_processor_tester.max_patches:
            # Test not batched input
            SCREAMING_SNAKE_CASE__				:				Optional[int]			=		image_processor(
                image_inputs[0]     , return_tensors='pt'     , max_patches=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (1, max_patch, expected_hidden_dim)     , )

            # Test batched
            SCREAMING_SNAKE_CASE__				:				Tuple			=		image_processor(
                a_     , return_tensors='pt'     , max_patches=a_				).flattened_patches
            self.assertEqual(
                encoded_images.shape     , (self.image_processor_tester.batch_size, max_patch, expected_hidden_dim)     , )



