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





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		   =  VideoToVideoSDPipeline
     SCREAMING_SNAKE_CASE_		   =  TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'''video'''} ) - {'''image''', '''width''', '''height'''}
     SCREAMING_SNAKE_CASE_		   =  TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''video'''} ) - {'''image'''}
     SCREAMING_SNAKE_CASE_		   =  PipelineTesterMixin.required_optional_params - {'''latents'''}
     SCREAMING_SNAKE_CASE_		   =  False

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






     def    __SCREAMING_SNAKE_CASE			(							self  )  ->   Dict:




           """simple docstring"""







           torch.manual_seed(0  )
           UpperCamelCase__    :   Any					=				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						,						)
           UpperCamelCase__    :   Any					=				DDIMScheduler(
               beta_start=0.00085						,						beta_end=0.012						,						beta_schedule='''scaled_linear'''						,						clip_sample=__SCREAMING_SNAKE_CASE						,						set_alpha_to_one=__SCREAMING_SNAKE_CASE						,						)
           torch.manual_seed(0  )
           UpperCamelCase__    :   List[str]					=				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  )
           UpperCamelCase__    :   List[Any]					=				CLIPTextConfig(
               bos_token_id=0						,						eos_token_id=2						,						hidden_size=3_2						,						intermediate_size=3_7						,						layer_norm_eps=1e-05						,						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						,						)
           UpperCamelCase__    :   Optional[int]					=				CLIPTextModel(__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   Any					=				CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''  )

           UpperCamelCase__    :   List[str]					=				{
               '''unet''': unet,
               '''scheduler''': scheduler,
               '''vae''': vae,
               '''text_encoder''': text_encoder,
               '''tokenizer''': tokenizer,
           }
           return components






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE=0  )  ->   Dict:




           """simple docstring"""







           UpperCamelCase__    :   List[str]					=				floats_tensor((1, 3, 3, 3_2, 3_2)						,						rng=random.Random(__SCREAMING_SNAKE_CASE  )  ).to(__SCREAMING_SNAKE_CASE  )

           if str(__SCREAMING_SNAKE_CASE  ).startswith('''mps'''  ):
                 UpperCamelCase__    :   List[str]					=				torch.manual_seed(__SCREAMING_SNAKE_CASE  )
           else:
                 UpperCamelCase__    :   Any					=				torch.Generator(device=__SCREAMING_SNAKE_CASE  ).manual_seed(__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   Any					=				{
               '''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    __SCREAMING_SNAKE_CASE			(							self  )  ->   Tuple:




           """simple docstring"""







           UpperCamelCase__    :   Union[str, Any]					=				'''cpu'''  # ensure determinism for the device-dependent torch.Generator
           UpperCamelCase__    :   Union[str, Any]					=				self.get_dummy_components()
           UpperCamelCase__    :   Optional[int]					=				VideoToVideoSDPipeline(**__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   int					=				sd_pipe.to(__SCREAMING_SNAKE_CASE  )
           sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE  )

           UpperCamelCase__    :   int					=				self.get_dummy_inputs(__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   str					=				'''np'''
           UpperCamelCase__    :   Union[str, Any]					=				sd_pipe(**__SCREAMING_SNAKE_CASE  ).frames
           UpperCamelCase__    :   Union[str, Any]					=				frames[0][-3:, -3:, -1]

           assert frames[0].shape == (3_2, 3_2, 3)
           UpperCamelCase__    :   str					=				np.array([1_0_6, 1_1_7, 1_1_3, 1_7_4, 1_3_7, 1_1_2, 1_4_8, 1_5_1, 1_3_1]  )

           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    __SCREAMING_SNAKE_CASE			(							self  )  ->   Tuple:




           """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    __SCREAMING_SNAKE_CASE			(							self  )  ->   List[Any]:




           """simple docstring"""







           pass






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




           """simple docstring"""







           pass






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




           """simple docstring"""







           pass






     def    __SCREAMING_SNAKE_CASE			(							self  )  ->   Any:




           """simple docstring"""







           return super().test_progress_bar()




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





     """simple docstring"""







     def    __SCREAMING_SNAKE_CASE			(							self  )  ->   Any:




           """simple docstring"""







           UpperCamelCase__    :   Optional[Any]					=				VideoToVideoSDPipeline.from_pretrained('''cerspense/zeroscope_v2_XL'''						,						torch_dtype=torch.floataa  )
           pipe.enable_model_cpu_offload()

           # 10 frames
           UpperCamelCase__    :   int					=				torch.Generator(device='''cpu'''  ).manual_seed(0  )
           UpperCamelCase__    :   Dict					=				torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6)						,						generator=__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   Any					=				video.to('''cuda'''  )

           UpperCamelCase__    :   str					=				'''Spiderman is surfing'''

           UpperCamelCase__    :   Dict					=				pipe(__SCREAMING_SNAKE_CASE						,						video=__SCREAMING_SNAKE_CASE						,						generator=__SCREAMING_SNAKE_CASE						,						num_inference_steps=3						,						output_type='''pt'''  ).frames

           UpperCamelCase__    :   Optional[int]					=				np.array([-1.0458984, -1.1279297, -0.9663086, -0.91503906, -0.75097656]  )
           assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array  ).sum() < 1e-2




from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


lowerCamelCase      ={
    "configuration_mobilebert": [
        "MOBILEBERT_PRETRAINED_CONFIG_ARCHIVE_MAP",
        "MobileBertConfig",
        "MobileBertOnnxConfig",
    ],
    "tokenization_mobilebert": ["MobileBertTokenizer"],
}

try:
    if not is_tokenizers_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCamelCase      =["MobileBertTokenizerFast"]

try:
    if not is_torch_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCamelCase      =[
        "MOBILEBERT_PRETRAINED_MODEL_ARCHIVE_LIST",
        "MobileBertForMaskedLM",
        "MobileBertForMultipleChoice",
        "MobileBertForNextSentencePrediction",
        "MobileBertForPreTraining",
        "MobileBertForQuestionAnswering",
        "MobileBertForSequenceClassification",
        "MobileBertForTokenClassification",
        "MobileBertLayer",
        "MobileBertModel",
        "MobileBertPreTrainedModel",
        "load_tf_weights_in_mobilebert",
    ]

try:
    if not is_tf_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCamelCase      =[
        "TF_MOBILEBERT_PRETRAINED_MODEL_ARCHIVE_LIST",
        "TFMobileBertForMaskedLM",
        "TFMobileBertForMultipleChoice",
        "TFMobileBertForNextSentencePrediction",
        "TFMobileBertForPreTraining",
        "TFMobileBertForQuestionAnswering",
        "TFMobileBertForSequenceClassification",
        "TFMobileBertForTokenClassification",
        "TFMobileBertMainLayer",
        "TFMobileBertModel",
        "TFMobileBertPreTrainedModel",
    ]


if TYPE_CHECKING:
    from .configuration_mobilebert import (
        MOBILEBERT_PRETRAINED_CONFIG_ARCHIVE_MAP,
        MobileBertConfig,
        MobileBertOnnxConfig,
    )
    from .tokenization_mobilebert import MobileBertTokenizer

    try:
        if not is_tokenizers_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .tokenization_mobilebert_fast import MobileBertTokenizerFast

    try:
        if not is_torch_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_mobilebert import (
            MOBILEBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
            MobileBertForMaskedLM,
            MobileBertForMultipleChoice,
            MobileBertForNextSentencePrediction,
            MobileBertForPreTraining,
            MobileBertForQuestionAnswering,
            MobileBertForSequenceClassification,
            MobileBertForTokenClassification,
            MobileBertLayer,
            MobileBertModel,
            MobileBertPreTrainedModel,
            load_tf_weights_in_mobilebert,
        )

    try:
        if not is_tf_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_tf_mobilebert import (
            TF_MOBILEBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
            TFMobileBertForMaskedLM,
            TFMobileBertForMultipleChoice,
            TFMobileBertForNextSentencePrediction,
            TFMobileBertForPreTraining,
            TFMobileBertForQuestionAnswering,
            TFMobileBertForSequenceClassification,
            TFMobileBertForTokenClassification,
            TFMobileBertMainLayer,
            TFMobileBertModel,
            TFMobileBertPreTrainedModel,
        )

else:
    import sys

    lowerCamelCase      =_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)


