


import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()



@skip_mps
class 	__lowerCamelCase			(UpperCamelCase_				,      unittest.TestCase						):
       _lowercase											=     VideoToVideoSDPipeline
       _lowercase											=     TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"""video"""}						) - {"""image""", """width""", """height"""}
       _lowercase											=     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""video"""}						) - {"""image"""}
       _lowercase											=     PipelineTesterMixin.required_optional_params - {"""latents"""}
       _lowercase											=     False

       # No `output_type`.
       _lowercase											=     frozenset(
           [
               """num_inference_steps""",
               """generator""",
               """latents""",
               """return_dict""",
               """callback""",
               """callback_steps""",
           ]						)







       def 		snake_case_	(			self: List[str]					):






            '''simple docstring'''




            torch.manual_seed(0					)
            __UpperCamelCase =			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 =			DDIMScheduler(
                beta_start=0.0_0_0_8_5,beta_end=0.0_1_2,beta_schedule='scaled_linear',clip_sample=__SCREAMING_SNAKE_CASE,set_alpha_to_one=__SCREAMING_SNAKE_CASE,)
            torch.manual_seed(0					)
            __UpperCamelCase =			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 =			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,)
            __UpperCamelCase =			CLIPTextModel(__SCREAMING_SNAKE_CASE					)
            __UpperCamelCase =			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					)

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







       def 		snake_case_	(			self: List[str],A_: Dict,A_: List[str]=0					):






            '''simple docstring'''




            __UpperCamelCase =			floats_tensor((1, 3, 3, 32, 32),rng=random.Random(__SCREAMING_SNAKE_CASE					)					).to(__SCREAMING_SNAKE_CASE					)

            if str(__SCREAMING_SNAKE_CASE					).startswith('mps'					):
                 __UpperCamelCase =			torch.manual_seed(__SCREAMING_SNAKE_CASE					)
            else:
                 __UpperCamelCase =			torch.Generator(device=__SCREAMING_SNAKE_CASE					).manual_seed(__SCREAMING_SNAKE_CASE					)
            __UpperCamelCase =			{
                '''prompt''': '''A painting of a squirrel eating a burger''',
                '''video''': video,
                '''generator''': generator,
                '''num_inference_steps''': 2,
                '''guidance_scale''': 6.0,
                '''output_type''': '''pt''',
            }
            return inputs







       def 		snake_case_	(			self: Tuple					):






            '''simple docstring'''




            __UpperCamelCase =			'''cpu'''  # ensure determinism for the device-dependent torch.Generator
            __UpperCamelCase =			self.get_dummy_components()
            __UpperCamelCase =			VideoToVideoSDPipeline(**__SCREAMING_SNAKE_CASE					)
            __UpperCamelCase =			sd_pipe.to(__SCREAMING_SNAKE_CASE					)
            sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE					)

            __UpperCamelCase =			self.get_dummy_inputs(__SCREAMING_SNAKE_CASE					)
            __UpperCamelCase =			'''np'''
            __UpperCamelCase =			sd_pipe(**__SCREAMING_SNAKE_CASE					).frames
            __UpperCamelCase =			frames[0][-3:, -3:, -1]

            assert frames[0].shape == (32, 32, 3)
            __UpperCamelCase =			np.array([106, 117, 113, 174, 137, 112, 148, 151, 131]					)

            assert np.abs(image_slice.flatten() - expected_slice					).max() < 1E-2







       @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: int					):






            '''simple docstring'''




            self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE,expected_max_diff=5E-3					)







       @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'					)
       def 		snake_case_	(			self: Any					):






            '''simple docstring'''




            pass







       @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'					)
       def 		snake_case_	(			self: List[str]					):






            '''simple docstring'''




            pass







       @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'					)
       def 		snake_case_	(			self: int					):






            '''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: List[Any]					):






            '''simple docstring'''




            __UpperCamelCase =			VideoToVideoSDPipeline.from_pretrained('cerspense/zeroscope_v2_XL',torch_dtype=torch.floataa					)
            pipe.enable_model_cpu_offload()

            # 10 frames
            __UpperCamelCase =			torch.Generator(device='cpu'					).manual_seed(0					)
            __UpperCamelCase =			torch.randn((1, 10, 3, 1024, 576),generator=__SCREAMING_SNAKE_CASE					)
            __UpperCamelCase =			video.to('cuda'					)

            __UpperCamelCase =			'''Spiderman is surfing'''

            __UpperCamelCase =			pipe(__SCREAMING_SNAKE_CASE,video=__SCREAMING_SNAKE_CASE,generator=__SCREAMING_SNAKE_CASE,num_inference_steps=3,output_type='pt'					).frames

            __UpperCamelCase =			np.array([-1.0_4_5_8_9_8_4, -1.1_2_7_9_2_9_7, -0.9_6_6_3_0_8_6, -0.9_1_5_0_3_9_0_6, -0.7_5_0_9_7_6_5_6]					)
            assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array					).sum() < 1E-2




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, BlipaProcessor, BlipImageProcessor, GPTaTokenizer, PreTrainedTokenizerFast




@require_vision
class 					_lowerCamelCase			(   unittest.TestCase			):





     """simple docstring"""







     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







           UpperCamelCase__     : int											=					tempfile.mkdtemp()

           UpperCamelCase__     : str											=					BlipImageProcessor()
           UpperCamelCase__     : List[Any]											=					GPTaTokenizer.from_pretrained('''hf-internal-testing/tiny-random-GPT2Model'''	)

           UpperCamelCase__     : List[Any]											=					BlipaProcessor(__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)

           processor.save_pretrained(self.tmpdirname	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    **__SCREAMING_SNAKE_CASE	)	-> int:




           """simple docstring"""







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






     def  __SCREAMING_SNAKE_CASE      (				self    ,    **__SCREAMING_SNAKE_CASE	)	-> int:




           """simple docstring"""







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






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







           shutil.rmtree(self.tmpdirname	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







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

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

           return image_inputs






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[Any]:




           """simple docstring"""







           UpperCamelCase__     : str											=					BlipaProcessor(tokenizer=self.get_tokenizer()    ,    image_processor=self.get_image_processor()	)
           processor.save_pretrained(self.tmpdirname	)

           UpperCamelCase__     : Any											=					self.get_tokenizer(bos_token='''(BOS)'''    ,    eos_token='''(EOS)'''	)
           UpperCamelCase__     : str											=					self.get_image_processor(do_normalize=__SCREAMING_SNAKE_CASE    ,    padding_value=1.0	)

           UpperCamelCase__     : Optional[Any]											=					BlipaProcessor.from_pretrained(
               self.tmpdirname    ,    bos_token='''(BOS)'''    ,    eos_token='''(EOS)'''    ,    do_normalize=__SCREAMING_SNAKE_CASE    ,    padding_value=1.0	)

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

           self.assertEqual(processor.image_processor.to_json_string()    ,    image_processor_add_kwargs.to_json_string()	)
           self.assertIsInstance(processor.image_processor    ,    __SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           UpperCamelCase__     : Dict											=					self.get_image_processor()
           UpperCamelCase__     : Tuple											=					self.get_tokenizer()

           UpperCamelCase__     : int											=					BlipaProcessor(tokenizer=__SCREAMING_SNAKE_CASE    ,    image_processor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : Union[str, Any]											=					self.prepare_image_inputs()

           UpperCamelCase__     : Optional[Any]											=					image_processor(__SCREAMING_SNAKE_CASE    ,    return_tensors='''np'''	)
           UpperCamelCase__     : List[Any]											=					processor(images=__SCREAMING_SNAKE_CASE    ,    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  __SCREAMING_SNAKE_CASE      (				self	)	-> Dict:




           """simple docstring"""







           UpperCamelCase__     : str											=					self.get_image_processor()
           UpperCamelCase__     : Any											=					self.get_tokenizer()

           UpperCamelCase__     : Tuple											=					BlipaProcessor(tokenizer=__SCREAMING_SNAKE_CASE    ,    image_processor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : List[str]											=					'''lower newer'''

           UpperCamelCase__     : Tuple											=					processor(text=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					tokenizer(__SCREAMING_SNAKE_CASE    ,    return_token_type_ids=__SCREAMING_SNAKE_CASE	)

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






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







           UpperCamelCase__     : Any											=					self.get_image_processor()
           UpperCamelCase__     : Optional[int]											=					self.get_tokenizer()

           UpperCamelCase__     : Dict											=					BlipaProcessor(tokenizer=__SCREAMING_SNAKE_CASE    ,    image_processor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : Tuple											=					'''lower newer'''
           UpperCamelCase__     : Any											=					self.prepare_image_inputs()

           UpperCamelCase__     : Any											=					processor(text=__SCREAMING_SNAKE_CASE    ,    images=__SCREAMING_SNAKE_CASE	)

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

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






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







           UpperCamelCase__     : List[str]											=					self.get_image_processor()
           UpperCamelCase__     : Optional[int]											=					self.get_tokenizer()

           UpperCamelCase__     : Tuple											=					BlipaProcessor(tokenizer=__SCREAMING_SNAKE_CASE    ,    image_processor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : List[Any]											=					[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

           UpperCamelCase__     : Tuple											=					processor.batch_decode(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[str]											=					tokenizer.batch_decode(__SCREAMING_SNAKE_CASE	)

           self.assertListEqual(__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           UpperCamelCase__     : Optional[Any]											=					self.get_image_processor()
           UpperCamelCase__     : Dict											=					self.get_tokenizer()

           UpperCamelCase__     : int											=					BlipaProcessor(tokenizer=__SCREAMING_SNAKE_CASE    ,    image_processor=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : List[Any]											=					'''lower newer'''
           UpperCamelCase__     : List[str]											=					self.prepare_image_inputs()

           UpperCamelCase__     : List[Any]											=					processor(text=__SCREAMING_SNAKE_CASE    ,    images=__SCREAMING_SNAKE_CASE	)

           # For now the processor supports only ['pixel_values', 'input_ids', 'attention_mask']
           self.assertListEqual(list(inputs.keys()	)    ,    ['''pixel_values''', '''input_ids''', '''attention_mask''']	)


