


'''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 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE  , unittest.TestCase    ):


  '''simple docstring'''







  a_  					=							TextToVideoSDPipeline
  a_  					=							TEXT_TO_IMAGE_PARAMS
  a_  					=							TEXT_TO_IMAGE_BATCH_PARAMS
  # No `output_type`.
  a_  					=							frozenset(
      [
          """num_inference_steps""",
          """generator""",
          """latents""",
          """return_dict""",
          """callback""",
          """callback_steps""",
      ]    )






  def        _a							(							self :       Any ):

        '''simple docstring'''




        torch.manual_seed(0 )
        A_						:  List[Any]          =				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_						:  Tuple          =				DDIMScheduler(
            beta_start=0.00085	,beta_end=0.012	,beta_schedule="""scaled_linear"""	,clip_sample=_a	,set_alpha_to_one=_a	,)
        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_						:  Union[str, Any]          =				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=1000	,hidden_act="""gelu"""	,projection_dim=512	,)
        A_						:  List[Any]          =				CLIPTextModel(_a )
        A_						:  str          =				CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

        A_						:  str          =				{
            """unet""": unet,
            """scheduler""": scheduler,
            """vae""": vae,
            """text_encoder""": text_encoder,
            """tokenizer""": tokenizer,
        }
        return components






  def        _a							(							self :       str	,_a :       List[Any]	,_a :       Union[str, Any]=0 ):

        '''simple docstring'''




        if str(_a ).startswith("""mps""" ):
              A_						:  Dict          =				torch.manual_seed(_a )
        else:
              A_						:  str          =				torch.Generator(device=_a ).manual_seed(_a )
        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        _a							(							self :       Optional[Any] ):

        '''simple docstring'''




        A_						:  Optional[int]          =				"""cpu"""  # ensure determinism for the device-dependent torch.Generator
        A_						:  str          =				self.get_dummy_components()
        A_						:  Optional[Any]          =				TextToVideoSDPipeline(**_a )
        A_						:  str          =				sd_pipe.to(_a )
        sd_pipe.set_progress_bar_config(disable=_a )

        A_						:  Optional[Any]          =				self.get_dummy_inputs(_a )
        A_						:  int          =				"""np"""
        A_						:  List[str]          =				sd_pipe(**_a ).frames
        A_						:  int          =				frames[0][-3:, -3:, -1]

        assert frames[0].shape == (64, 64, 3)
        A_						:  List[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        _a							(							self :       Any ):

        '''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        _a							(							self :       Union[str, Any] ):

        '''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        _a							(							self :       int ):

        '''simple docstring'''




        pass






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

        '''simple docstring'''




        pass






  @unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline.""" )
  def        _a							(							self :       Optional[Any] ):

        '''simple docstring'''




        pass






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

        '''simple docstring'''




        return super().test_progress_bar()







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


  '''simple docstring'''







  def        _a							(							self :       List[str] ):

        '''simple docstring'''




        A_						:  str          =				load_numpy(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy""" )

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

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

        A_						:  List[str]          =				pipe(_a	,generator=_a	,num_inference_steps=25	,output_type="""pt""" ).frames
        A_						:  str          =				video_frames.cpu().numpy()

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






  def        _a							(							self :       int ):

        '''simple docstring'''




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

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

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

        A_						:  List[str]          =				pipe(_a	,generator=_a	,num_inference_steps=2	,output_type="""pt""" ).frames
        A_						:  List[Any]          =				video_frames.cpu().numpy()

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




'''simple docstring'''


import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType



class 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							["""image_processor""", """tokenizer"""]
  a_  					=							"""ViltImageProcessor"""
  a_  					=							("""BertTokenizer""", """BertTokenizerFast""")






  def __init__(							self :       List[Any]	,_a :       Optional[Any]=None	,_a :       List[str]=None	,**_a :       Any ):

        '''simple docstring'''




        A_						:  Any          =				None
        if "feature_extractor" in kwargs:
              warnings.warn(
                  """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
                  """ instead."""	,_a	,)
              A_						:  List[str]          =				kwargs.pop("""feature_extractor""" )

        A_						:  List[Any]          =				image_processor if image_processor is not None else feature_extractor
        if image_processor is None:
              raise ValueError("""You need to specify an `image_processor`.""" )
        if tokenizer is None:
              raise ValueError("""You need to specify a `tokenizer`.""" )

        super().__init__(_a	,_a )
        A_						:  Optional[Any]          =				self.image_processor






  def __call__(							self :       Any	,_a :       Tuple	,_a :       Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None	,_a :       bool = True	,_a :       Union[bool, str, PaddingStrategy] = False	,_a :       Union[bool, str, TruncationStrategy] = None	,_a :       Optional[int] = None	,_a :       int = 0	,_a :       Optional[int] = None	,_a :       Optional[bool] = None	,_a :       Optional[bool] = None	,_a :       bool = False	,_a :       bool = False	,_a :       bool = False	,_a :       bool = False	,_a :       bool = True	,_a :       Optional[Union[str, TensorType]] = None	,**_a :       Tuple	,):

        '''simple docstring'''




        A_						:  int          =				self.tokenizer(
            text=_a	,add_special_tokens=_a	,padding=_a	,truncation=_a	,max_length=_a	,stride=_a	,pad_to_multiple_of=_a	,return_token_type_ids=_a	,return_attention_mask=_a	,return_overflowing_tokens=_a	,return_special_tokens_mask=_a	,return_offsets_mapping=_a	,return_length=_a	,verbose=_a	,return_tensors=_a	,**_a	,)
        # add pixel_values + pixel_mask
        A_						:  Optional[int]          =				self.image_processor(_a	,return_tensors=_a )
        encoding.update(_a )

        return encoding






  def        _a							(							self :       List[Any]	,*_a :       Any	,**_a :       Any ):

        '''simple docstring'''




        return self.tokenizer.batch_decode(*_a	,**_a )






  def        _a							(							self :       int	,*_a :       int	,**_a :       Optional[int] ):

        '''simple docstring'''




        return self.tokenizer.decode(*_a	,**_a )






  @property
  def        _a							(							self :       List[Any] ):

        '''simple docstring'''




        A_						:  Optional[int]          =				self.tokenizer.model_input_names
        A_						:  str          =				self.image_processor.model_input_names
        return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names ) )






  @property
  def        _a							(							self :       str ):

        '''simple docstring'''




        warnings.warn(
            """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""	,_a	,)
        return self.image_processor_class






  @property
  def        _a							(							self :       int ):

        '''simple docstring'''




        warnings.warn(
            """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""	,_a	,)
        return self.image_processor
