



'''simple docstring'''





import copy
import os
from typing import TYPE_CHECKING, List, Union


if TYPE_CHECKING:
 pass

from ...configuration_utils import PretrainedConfig
from ...utils import logging


SCREAMING_SNAKE_CASE__      					=     logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__      					=     {
    'kakaobrain/align-base': 'https://huggingface.co/kakaobrain/align-base/resolve/main/config.json',
}


class 				snake_case (UpperCamelCase   ):
       lowerCAmelCase__  :str						   =					"align_text_model"



       def __init__(				self     ,UpperCAmelCase_=30_522     ,UpperCAmelCase_=768     ,UpperCAmelCase_=12     ,UpperCAmelCase_=12     ,UpperCAmelCase_=3_072     ,UpperCAmelCase_="gelu"     ,UpperCAmelCase_=0.1     ,UpperCAmelCase_=0.1     ,UpperCAmelCase_=512     ,UpperCAmelCase_=2     ,UpperCAmelCase_=0.02     ,UpperCAmelCase_=1E-1_2     ,UpperCAmelCase_=0     ,UpperCAmelCase_="absolute"     ,UpperCAmelCase_=True     ,**UpperCAmelCase_     ,)  ->							int:
           super().__init__(**_lowerCAmelCase      )

           lowercase__		       =						vocab_size
           lowercase__		       =						hidden_size
           lowercase__		       =						num_hidden_layers
           lowercase__		       =						num_attention_heads
           lowercase__		       =						hidden_act
           lowercase__		       =						intermediate_size
           lowercase__		       =						hidden_dropout_prob
           lowercase__		       =						attention_probs_dropout_prob
           lowercase__		       =						max_position_embeddings
           lowercase__		       =						type_vocab_size
           lowercase__		       =						initializer_range
           lowercase__		       =						layer_norm_eps
           lowercase__		       =						position_embedding_type
           lowercase__		       =						use_cache
           lowercase__		       =						pad_token_id







       @classmethod
       def 							_a      (				cls     ,UpperCAmelCase_     ,**UpperCAmelCase_      )  ->							List[Any]:
           cls._set_token_in_kwargs(_lowerCAmelCase      )

           lowercase__       ,       lowercase__		       =						cls.get_config_dict(_lowerCAmelCase     ,**_lowerCAmelCase      )

           # get the text config dict if we are loading from AlignConfig
           if config_dict.get("model_type"      ) == "align":
               lowercase__		       =						config_dict["text_config"]

           if "model_type" in config_dict and hasattr(cls     ,"model_type"      ) and config_dict["model_type"] != cls.model_type:
               logger.warning(
                   F'''You are using a model of type {config_dict['model_type']} to instantiate a model of type '''
                   F'''{cls.model_type}. This is not supported for all configurations of models and can yield errors.'''      )

           return cls.from_dict(_lowerCAmelCase     ,**_lowerCAmelCase      )





class 				snake_case (UpperCamelCase   ):
       lowerCAmelCase__  :Optional[Any]						   =					"align_vision_model"



       def __init__(				self     ,UpperCAmelCase_ = 3     ,UpperCAmelCase_ = 600     ,UpperCAmelCase_ = 2.0     ,UpperCAmelCase_ = 3.1     ,UpperCAmelCase_ = 8     ,UpperCAmelCase_ = [3, 3, 5, 3, 5, 5, 3]     ,UpperCAmelCase_ = [32, 16, 24, 40, 80, 112, 192]     ,UpperCAmelCase_ = [16, 24, 40, 80, 112, 192, 320]     ,UpperCAmelCase_ = []     ,UpperCAmelCase_ = [1, 2, 2, 2, 1, 2, 1]     ,UpperCAmelCase_ = [1, 2, 2, 3, 3, 4, 1]     ,UpperCAmelCase_ = [1, 6, 6, 6, 6, 6, 6]     ,UpperCAmelCase_ = 0.25     ,UpperCAmelCase_ = "swish"     ,UpperCAmelCase_ = 2_560     ,UpperCAmelCase_ = "mean"     ,UpperCAmelCase_ = 0.02     ,UpperCAmelCase_ = 0.0_01     ,UpperCAmelCase_ = 0.99     ,UpperCAmelCase_ = 0.2     ,**UpperCAmelCase_     ,)  ->							Dict:
           super().__init__(**_lowerCAmelCase      )

           lowercase__		       =						num_channels
           lowercase__		       =						image_size
           lowercase__		       =						width_coefficient
           lowercase__		       =						depth_coefficient
           lowercase__		       =						depth_divisor
           lowercase__		       =						kernel_sizes
           lowercase__		       =						in_channels
           lowercase__		       =						out_channels
           lowercase__		       =						depthwise_padding
           lowercase__		       =						strides
           lowercase__		       =						num_block_repeats
           lowercase__		       =						expand_ratios
           lowercase__		       =						squeeze_expansion_ratio
           lowercase__		       =						hidden_act
           lowercase__		       =						hidden_dim
           lowercase__		       =						pooling_type
           lowercase__		       =						initializer_range
           lowercase__		       =						batch_norm_eps
           lowercase__		       =						batch_norm_momentum
           lowercase__		       =						drop_connect_rate
           lowercase__		       =						sum(_lowerCAmelCase      ) * 4







       @classmethod
       def 							_a      (				cls     ,UpperCAmelCase_     ,**UpperCAmelCase_      )  ->							int:
           cls._set_token_in_kwargs(_lowerCAmelCase      )

           lowercase__       ,       lowercase__		       =						cls.get_config_dict(_lowerCAmelCase     ,**_lowerCAmelCase      )

           # get the vision config dict if we are loading from AlignConfig
           if config_dict.get("model_type"      ) == "align":
               lowercase__		       =						config_dict["vision_config"]

           if "model_type" in config_dict and hasattr(cls     ,"model_type"      ) and config_dict["model_type"] != cls.model_type:
               logger.warning(
                   F'''You are using a model of type {config_dict['model_type']} to instantiate a model of type '''
                   F'''{cls.model_type}. This is not supported for all configurations of models and can yield errors.'''      )

           return cls.from_dict(_lowerCAmelCase     ,**_lowerCAmelCase      )





class 				snake_case (UpperCamelCase   ):
       lowerCAmelCase__  :str						   =					"align"
       lowerCAmelCase__  :Any						   =					True



       def __init__(				self     ,UpperCAmelCase_=None     ,UpperCAmelCase_=None     ,UpperCAmelCase_=640     ,UpperCAmelCase_=1.0     ,UpperCAmelCase_=0.02     ,**UpperCAmelCase_     ,)  ->							Dict:
           super().__init__(**_lowerCAmelCase      )

           if text_config is None:
               lowercase__		       =						{}
               logger.info("text_config is None. Initializing the AlignTextConfig with default values."      )

           if vision_config is None:
               lowercase__		       =						{}
               logger.info("vision_config is None. Initializing the AlignVisionConfig with default values."      )

           lowercase__		       =						AlignTextConfig(**_lowerCAmelCase      )
           lowercase__		       =						AlignVisionConfig(**_lowerCAmelCase      )

           lowercase__		       =						projection_dim
           lowercase__		       =						temperature_init_value
           lowercase__		       =						initializer_range



       @classmethod
       def 							_a      (				cls     ,UpperCAmelCase_     ,UpperCAmelCase_     ,**UpperCAmelCase_      )  ->							Optional[int]:
           return cls(text_config=text_config.to_dict()     ,vision_config=vision_config.to_dict()     ,**_lowerCAmelCase      )







       def 							_a      (				self      )  ->							Any:
           lowercase__		       =						copy.deepcopy(self.__dict__      )
           lowercase__		       =						self.text_config.to_dict()
           lowercase__		       =						self.vision_config.to_dict()
           lowercase__		       =						self.__class__.model_type
           return output









'''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 (UpperCamelCase		,      unittest.TestCase   ):
       lowerCAmelCase__  :int						   =					TextToVideoSDPipeline
       lowerCAmelCase__  :Union[str, Any]						   =					TEXT_TO_IMAGE_PARAMS
       lowerCAmelCase__  :List[str]						   =					TEXT_TO_IMAGE_BATCH_PARAMS
       # No `output_type`.
       lowerCAmelCase__  :Optional[Any]						   =					frozenset(
           [
               "num_inference_steps",
               "generator",
               "latents",
               "return_dict",
               "callback",
               "callback_steps",
           ]   )



       def 							_a      (				self      )  ->							Optional[int]:
           torch.manual_seed(0      )
           lowercase__		       =						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     ,)
           lowercase__		       =						DDIMScheduler(
               beta_start=0.0_00_85     ,beta_end=0.0_12     ,beta_schedule="scaled_linear"     ,clip_sample=UpperCAmelCase_     ,set_alpha_to_one=UpperCAmelCase_     ,)
           torch.manual_seed(0      )
           lowercase__		       =						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      )
           lowercase__		       =						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=1_000     ,hidden_act="gelu"     ,projection_dim=512     ,)
           lowercase__		       =						CLIPTextModel(UpperCAmelCase_      )
           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 							_a      (				self     ,UpperCAmelCase_     ,UpperCAmelCase_=0      )  ->							int:
           if str(UpperCAmelCase_      ).startswith("mps"      ):
               lowercase__		       =						torch.manual_seed(UpperCAmelCase_      )
           else:
               lowercase__		       =						torch.Generator(device=UpperCAmelCase_      ).manual_seed(UpperCAmelCase_      )
           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 							_a      (				self      )  ->							Tuple:
           lowercase__		       =						"cpu"  # ensure determinism for the device-dependent torch.Generator
           lowercase__		       =						self.get_dummy_components()
           lowercase__		       =						TextToVideoSDPipeline(**UpperCAmelCase_      )
           lowercase__		       =						sd_pipe.to(UpperCAmelCase_      )
           sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_      )

           lowercase__		       =						self.get_dummy_inputs(UpperCAmelCase_      )
           lowercase__		       =						"np"
           lowercase__		       =						sd_pipe(**UpperCAmelCase_      ).frames
           lowercase__		       =						frames[0][-3:, -3:, -1]

           assert frames[0].shape == (64, 64, 3)
           lowercase__		       =						np.array([1_58.0, 1_60.0, 1_53.0, 1_25.0, 1_00.0, 1_21.0, 1_11.0, 93.0, 1_13.0]      )

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



       def 							_a      (				self      )  ->							List[Any]:
           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 							_a      (				self      )  ->							Dict:
           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 							_a      (				self      )  ->							Union[str, Any]:
           pass



       @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."      )
       def 							_a      (				self      )  ->							Union[str, Any]:
           pass



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







       def 							_a      (				self      )  ->							str:
           return super().test_progress_bar()





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



       def 							_a      (				self      )  ->							Optional[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(UpperCAmelCase_     ,generator=UpperCAmelCase_     ,num_inference_steps=25     ,output_type="pt"      ).frames
           lowercase__		       =						video_frames.cpu().numpy()

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







       def 							_a      (				self      )  ->							Optional[int]:
           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(UpperCAmelCase_     ,generator=UpperCAmelCase_     ,num_inference_steps=2     ,output_type="pt"      ).frames
           lowercase__		       =						video_frames.cpu().numpy()

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




