

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





from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import rescale, resize, to_channel_dimension_format
from ...image_utils import (
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
		import PIL


SCREAMING_SNAKE_CASE__     					=			logging.get_logger(__name__)


def     lowercase	(    a     , a ):



		'''simple docstring'''




		SCREAMING_SNAKE_CASE_				:Dict				     =						b.T
		SCREAMING_SNAKE_CASE_				:str				     =						np.sum(np.square(a )     , axis=1 )
		SCREAMING_SNAKE_CASE_				:List[Any]				     =						np.sum(np.square(a )     , axis=0 )
		SCREAMING_SNAKE_CASE_				:Optional[int]				     =						np.matmul(a     , a )
		SCREAMING_SNAKE_CASE_				:Any				     =						aa[:, None] - 2 * ab + ba[None, :]
		return d


def     lowercase	(    a     , a ):



		'''simple docstring'''




		SCREAMING_SNAKE_CASE_				:str				     =						x.reshape(-1     , 3 )
		SCREAMING_SNAKE_CASE_				:str				     =						squared_euclidean_distance(a     , a )
		return np.argmin(a     , axis=1 )



class 						_UpperCAmelCase      (     lowercase							):
		lowerCamelCase_	:			List[str]       			=  ["""pixel_values"""]






		def __init__(						self				:     Dict   ,  UpperCAmelCase				:     Optional[Union[List[List[int]], np.ndarray]] = None   ,  UpperCAmelCase				:     bool = True   ,  UpperCAmelCase				:     Dict[str, int] = None   ,  UpperCAmelCase				:     PILImageResampling = PILImageResampling.BILINEAR   ,  UpperCAmelCase				:     bool = True   ,  UpperCAmelCase				:     bool = True   ,  **UpperCAmelCase				:     Union[str, Any]   ,  ):
				super().__init__(**UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						size if size is not None else {"height": 2_56, "width": 2_56}
				SCREAMING_SNAKE_CASE_				:Optional[int]				     =						get_size_dict(UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:int				     =						np.array(UpperCAmelCase) if clusters is not None else None
				SCREAMING_SNAKE_CASE_				:List[Any]				     =						do_resize
				SCREAMING_SNAKE_CASE_				:Dict				     =						size
				SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						resample
				SCREAMING_SNAKE_CASE_				:Dict				     =						do_normalize
				SCREAMING_SNAKE_CASE_				:Tuple				     =						do_color_quantize






		def _snake_case		(						self				:     Union[str, Any]   ,  UpperCAmelCase				:     np.ndarray   ,  UpperCAmelCase				:     Dict[str, int]   ,  UpperCAmelCase				:     PILImageResampling = PILImageResampling.BILINEAR   ,  UpperCAmelCase				:     Optional[Union[str, ChannelDimension]] = None   ,  **UpperCAmelCase				:     List[Any]   ,  ):
				SCREAMING_SNAKE_CASE_				:str				     =						get_size_dict(UpperCAmelCase)
				if "height" not in size or "width" not in size:
						raise ValueError(F"Size dictionary must contain both height and width keys. Got {size.keys()}")
				return resize(
				    UpperCAmelCase   ,  size=(size["height"], size["width"])   ,  resample=UpperCAmelCase   ,  data_format=UpperCAmelCase   ,  **UpperCAmelCase)






		def _snake_case		(						self				:     Union[str, Any]   ,  UpperCAmelCase				:     np.ndarray   ,  UpperCAmelCase				:     Optional[Union[str, ChannelDimension]] = None   ,  ):
				SCREAMING_SNAKE_CASE_				:int				     =						rescale(image=UpperCAmelCase   ,  scale=1 / 127.5   ,  data_format=UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:List[str]				     =						image - 1
				return image







		def _snake_case		(						self				:     List[str]   ,  UpperCAmelCase				:     ImageInput   ,  UpperCAmelCase				:     bool = None   ,  UpperCAmelCase				:     Dict[str, int] = None   ,  UpperCAmelCase				:     PILImageResampling = None   ,  UpperCAmelCase				:     bool = None   ,  UpperCAmelCase				:     Optional[bool] = None   ,  UpperCAmelCase				:     Optional[Union[List[List[int]], np.ndarray]] = None   ,  UpperCAmelCase				:     Optional[Union[str, TensorType]] = None   ,  UpperCAmelCase				:     Optional[Union[str, ChannelDimension]] = ChannelDimension.FIRST   ,  **UpperCAmelCase				:     Optional[Any]   ,  ):
				SCREAMING_SNAKE_CASE_				:Any				     =						do_resize if do_resize is not None else self.do_resize
				SCREAMING_SNAKE_CASE_				:Dict				     =						size if size is not None else self.size
				SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						get_size_dict(UpperCAmelCase)
				SCREAMING_SNAKE_CASE_				:Optional[int]				     =						resample if resample is not None else self.resample
				SCREAMING_SNAKE_CASE_				:Dict				     =						do_normalize if do_normalize is not None else self.do_normalize
				SCREAMING_SNAKE_CASE_				:List[str]				     =						do_color_quantize if do_color_quantize is not None else self.do_color_quantize
				SCREAMING_SNAKE_CASE_				:Tuple				     =						clusters if clusters is not None else self.clusters
				SCREAMING_SNAKE_CASE_				:Tuple				     =						np.array(UpperCAmelCase)

				SCREAMING_SNAKE_CASE_				:Dict				     =						make_list_of_images(UpperCAmelCase)

				if not valid_images(UpperCAmelCase):
						raise ValueError(
						    "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
						    "torch.Tensor, tf.Tensor or jax.ndarray.")

				if do_resize and size is None or resample is None:
						raise ValueError("Size and resample must be specified if do_resize is True.")

				if do_color_quantize and clusters is None:
						raise ValueError("Clusters must be specified if do_color_quantize is True.")

				# All transformations expect numpy arrays.
				SCREAMING_SNAKE_CASE_				:Union[str, Any]				     =						[to_numpy_array(UpperCAmelCase) for image in images]

				if do_resize:
						SCREAMING_SNAKE_CASE_				:Tuple				     =						[self.resize(image=UpperCAmelCase   ,  size=UpperCAmelCase   ,  resample=UpperCAmelCase) for image in images]

				if do_normalize:
						SCREAMING_SNAKE_CASE_				:int				     =						[self.normalize(image=UpperCAmelCase) for image in images]

				if do_color_quantize:
						SCREAMING_SNAKE_CASE_				:Optional[Any]				     =						[to_channel_dimension_format(UpperCAmelCase   ,  ChannelDimension.LAST) for image in images]
						# color quantize from (batch_size, height, width, 3) to (batch_size, height, width)
						SCREAMING_SNAKE_CASE_				:Optional[int]				     =						np.array(UpperCAmelCase)
						SCREAMING_SNAKE_CASE_				:Tuple				     =						color_quantize(UpperCAmelCase   ,  UpperCAmelCase).reshape(images.shape[:-1])

						# flatten to (batch_size, height*width)
						SCREAMING_SNAKE_CASE_				:Any				     =						images.shape[0]
						SCREAMING_SNAKE_CASE_				:Dict				     =						images.reshape(UpperCAmelCase   ,  -1)

						# We need to convert back to a list of images to keep consistent behaviour across processors.
						SCREAMING_SNAKE_CASE_				:Any				     =						list(UpperCAmelCase)
				else:
						SCREAMING_SNAKE_CASE_				:List[Any]				     =						[to_channel_dimension_format(UpperCAmelCase   ,  UpperCAmelCase) for image in images]

				SCREAMING_SNAKE_CASE_				:str				     =						{"input_ids": images}
				return BatchFeature(data=UpperCAmelCase   ,  tensor_type=UpperCAmelCase)






