


'''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 		_snake_case       (  a__							, unittest.TestCase						):
					lowerCAmelCase   :Dict				=       TextToVideoSDPipeline
					lowerCAmelCase   :Any				=       TEXT_TO_IMAGE_PARAMS
					lowerCAmelCase   :Any				=       TEXT_TO_IMAGE_BATCH_PARAMS
					# No `output_type`.
					lowerCAmelCase   :Union[str, Any]				=       frozenset(
					    [
					        '''num_inference_steps''',
					        '''generator''',
					        '''latents''',
					        '''return_dict''',
					        '''callback''',
					        '''callback_steps''',
					    ]						)




					def 					snake_case__    (   self):
								torch.manual_seed(0)
								UpperCAmelCase__	:       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				,				)
								UpperCAmelCase__	:       List[str]   =  DDIMScheduler(
								    beta_start=0.00085				,				beta_end=0.012				,				beta_schedule="""scaled_linear"""				,				clip_sample=__UpperCAmelCase				,				set_alpha_to_one=__UpperCAmelCase				,				)
								torch.manual_seed(0)
								UpperCAmelCase__	:       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)
								UpperCAmelCase__	:       Optional[int]   =  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				,				)
								UpperCAmelCase__	:       Tuple   =  CLIPTextModel(__UpperCAmelCase)
								UpperCAmelCase__	:       Tuple   =  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

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




					def 					snake_case__    (   self				,				_lowerCamelCase				,				_lowerCamelCase=0):

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

								UpperCAmelCase__	:       Optional[Any]   =  self.get_dummy_inputs(__UpperCAmelCase)
								UpperCAmelCase__	:       Optional[int]   =  'np'
								UpperCAmelCase__	:       int   =  sd_pipe(**__UpperCAmelCase).frames
								UpperCAmelCase__	:       int   =  frames[0][-3:, -3:, -1]

								assert frames[0].shape == (64, 64, 3)
								UpperCAmelCase__	:       int   =  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):
								self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__UpperCAmelCase				,				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):
								self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__UpperCAmelCase				,				expected_max_diff=1e-2)




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




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




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




					def 					snake_case__    (   self):
								return super().test_progress_bar()


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




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

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

								UpperCAmelCase__	:       int   =  'Spiderman is surfing'
								UpperCAmelCase__	:       List[str]   =  torch.Generator(device="""cpu""").manual_seed(0)

								UpperCAmelCase__	:       str   =  pipe(__UpperCAmelCase				,				generator=__UpperCAmelCase				,				num_inference_steps=25				,				output_type="""pt""").frames
								UpperCAmelCase__	:       int   =  video_frames.cpu().numpy()

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




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

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

								UpperCAmelCase__	:       Dict   =  'Spiderman is surfing'
								UpperCAmelCase__	:       List[Any]   =  torch.Generator(device="""cpu""").manual_seed(0)

								UpperCAmelCase__	:       Tuple   =  pipe(__UpperCAmelCase				,				generator=__UpperCAmelCase				,				num_inference_steps=2				,				output_type="""pt""").frames
								UpperCAmelCase__	:       List[str]   =  video_frames.cpu().numpy()

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






"""simple docstring"""





import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers.testing_utils import require_vision
from transformers.utils import is_vision_available


if is_vision_available():
  from PIL import Image

  from transformers import (
      AutoProcessor,
      BertTokenizerFast,
      BlipImageProcessor,
      GPTaTokenizer,
      InstructBlipProcessor,
      PreTrainedTokenizerFast,
  )






@require_vision
class    _lowerCAmelCase  (	unittest.TestCase			):
       """simple docstring"""




       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Any											=				tempfile.mkdtemp()

         lowerCAmelCase__     :List[Any]											=				BlipImageProcessor()
         lowerCAmelCase__     :Union[str, Any]											=				GPTaTokenizer.from_pretrained('hf-internal-testing/tiny-random-GPT2Model' )
         lowerCAmelCase__     :Union[str, Any]											=				BertTokenizerFast.from_pretrained('hf-internal-testing/tiny-random-bert' )

         lowerCAmelCase__     :List[str]											=				InstructBlipProcessor(__UpperCAmelCase       ,					__UpperCAmelCase       ,					__UpperCAmelCase )

         processor.save_pretrained(self.tmpdirname )






       def 	snake_case		(       self       ,					**__UpperCAmelCase ):



         '''simple docstring'''

         return AutoProcessor.from_pretrained(self.tmpdirname       ,					**__UpperCAmelCase ).tokenizer






       def 	snake_case		(       self       ,					**__UpperCAmelCase ):



         '''simple docstring'''

         return AutoProcessor.from_pretrained(self.tmpdirname       ,					**__UpperCAmelCase ).image_processor






       def 	snake_case		(       self       ,					**__UpperCAmelCase ):



         '''simple docstring'''

         return AutoProcessor.from_pretrained(self.tmpdirname       ,					**__UpperCAmelCase ).qformer_tokenizer






       def 	snake_case		(       self ):



         '''simple docstring'''

         shutil.rmtree(self.tmpdirname )






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[Any]											=				[np.random.randint(2_5_5       ,					size=(3, 3_0, 4_0_0)       ,					dtype=np.uinta )]

         lowerCAmelCase__     :Dict											=				[Image.fromarray(np.moveaxis(__UpperCAmelCase       ,					0       ,					-1 ) ) for x in image_inputs]

         return image_inputs






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :List[str]											=				InstructBlipProcessor(
             tokenizer=self.get_tokenizer()       ,					image_processor=self.get_image_processor()       ,					qformer_tokenizer=self.get_qformer_tokenizer()       ,					)
         processor.save_pretrained(self.tmpdirname )

         lowerCAmelCase__     :Tuple											=				self.get_tokenizer(bos_token='(BOS)'       ,					eos_token='(EOS)' )
         lowerCAmelCase__     :str											=				self.get_image_processor(do_normalize=__UpperCAmelCase       ,					padding_value=1.0 )

         lowerCAmelCase__     :Dict											=				InstructBlipProcessor.from_pretrained(
             self.tmpdirname       ,					bos_token='(BOS)'       ,					eos_token='(EOS)'       ,					do_normalize=__UpperCAmelCase       ,					padding_value=1.0 )

         self.assertEqual(processor.tokenizer.get_vocab()       ,					tokenizer_add_kwargs.get_vocab() )
         self.assertIsInstance(processor.tokenizer       ,					__UpperCAmelCase )

         self.assertEqual(processor.image_processor.to_json_string()       ,					image_processor_add_kwargs.to_json_string() )
         self.assertIsInstance(processor.image_processor       ,					__UpperCAmelCase )
         self.assertIsInstance(processor.qformer_tokenizer       ,					__UpperCAmelCase )






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[Any]											=				self.get_image_processor()
         lowerCAmelCase__     :List[Any]											=				self.get_tokenizer()
         lowerCAmelCase__     :Optional[int]											=				self.get_qformer_tokenizer()

         lowerCAmelCase__     :str											=				InstructBlipProcessor(
             tokenizer=__UpperCAmelCase       ,					image_processor=__UpperCAmelCase       ,					qformer_tokenizer=__UpperCAmelCase )

         lowerCAmelCase__     :Dict											=				self.prepare_image_inputs()

         lowerCAmelCase__     :List[str]											=				image_processor(__UpperCAmelCase       ,					return_tensors='np' )
         lowerCAmelCase__     :Optional[Any]											=				processor(images=__UpperCAmelCase       ,					return_tensors='np' )

         for key in input_feat_extract.keys():
           self.assertAlmostEqual(input_feat_extract[key].sum()       ,					input_processor[key].sum()       ,					delta=1E-2 )






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Union[str, Any]											=				self.get_image_processor()
         lowerCAmelCase__     :Optional[int]											=				self.get_tokenizer()
         lowerCAmelCase__     :Optional[int]											=				self.get_qformer_tokenizer()

         lowerCAmelCase__     :Any											=				InstructBlipProcessor(
             tokenizer=__UpperCAmelCase       ,					image_processor=__UpperCAmelCase       ,					qformer_tokenizer=__UpperCAmelCase )

         lowerCAmelCase__     :Optional[int]											=				'lower newer'

         lowerCAmelCase__     :Dict											=				processor(text=__UpperCAmelCase )

         lowerCAmelCase__     :Union[str, Any]											=				tokenizer(__UpperCAmelCase       ,					return_token_type_ids=__UpperCAmelCase )
         lowerCAmelCase__     :Tuple											=				qformer_tokenizer(__UpperCAmelCase       ,					return_token_type_ids=__UpperCAmelCase )

         for key in encoded_tokens.keys():
           self.assertListEqual(encoded_tokens[key]       ,					encoded_processor[key] )

         for key in encoded_tokens_qformer.keys():
           self.assertListEqual(encoded_tokens_qformer[key]       ,					encoded_processor['qformer_' + key] )






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[Any]											=				self.get_image_processor()
         lowerCAmelCase__     :Optional[int]											=				self.get_tokenizer()
         lowerCAmelCase__     :str											=				self.get_qformer_tokenizer()

         lowerCAmelCase__     :Dict											=				InstructBlipProcessor(
             tokenizer=__UpperCAmelCase       ,					image_processor=__UpperCAmelCase       ,					qformer_tokenizer=__UpperCAmelCase )

         lowerCAmelCase__     :Union[str, Any]											=				'lower newer'
         lowerCAmelCase__     :Dict											=				self.prepare_image_inputs()

         lowerCAmelCase__     :Tuple											=				processor(text=__UpperCAmelCase       ,					images=__UpperCAmelCase )

         self.assertListEqual(
             list(inputs.keys() )       ,					['input_ids', 'attention_mask', 'qformer_input_ids', 'qformer_attention_mask', 'pixel_values']       ,					)

         # test if it raises when no input is passed
         with pytest.raises(__UpperCAmelCase ):
           processor()






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Tuple											=				self.get_image_processor()
         lowerCAmelCase__     :Dict											=				self.get_tokenizer()
         lowerCAmelCase__     :Optional[int]											=				self.get_qformer_tokenizer()

         lowerCAmelCase__     :int											=				InstructBlipProcessor(
             tokenizer=__UpperCAmelCase       ,					image_processor=__UpperCAmelCase       ,					qformer_tokenizer=__UpperCAmelCase )

         lowerCAmelCase__     :List[str]											=				[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

         lowerCAmelCase__     :Tuple											=				processor.batch_decode(__UpperCAmelCase )
         lowerCAmelCase__     :Optional[int]											=				tokenizer.batch_decode(__UpperCAmelCase )

         self.assertListEqual(__UpperCAmelCase       ,					__UpperCAmelCase )






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Tuple											=				self.get_image_processor()
         lowerCAmelCase__     :Optional[int]											=				self.get_tokenizer()
         lowerCAmelCase__     :Dict											=				self.get_qformer_tokenizer()

         lowerCAmelCase__     :Optional[Any]											=				InstructBlipProcessor(
             tokenizer=__UpperCAmelCase       ,					image_processor=__UpperCAmelCase       ,					qformer_tokenizer=__UpperCAmelCase )

         lowerCAmelCase__     :Optional[int]											=				'lower newer'
         lowerCAmelCase__     :Optional[int]											=				self.prepare_image_inputs()

         lowerCAmelCase__     :int											=				processor(text=__UpperCAmelCase       ,					images=__UpperCAmelCase )

         self.assertListEqual(
             list(inputs.keys() )       ,					['input_ids', 'attention_mask', 'qformer_input_ids', 'qformer_attention_mask', 'pixel_values']       ,					)
