





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    _UpperCAmelCase							(		_lowerCamelCase			,       unittest.TestCase     ):
      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 		_lowerCamelCase      (		self       ):
         torch.manual_seed(0       )
         A_     :				Tuple			      =	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_     :				int			      =	DDIMScheduler(
             beta_start=0.0_0085      ,       beta_end=0.012      ,       beta_schedule="""scaled_linear"""      ,       clip_sample=a__      ,       set_alpha_to_one=a__      ,       )
         torch.manual_seed(0       )
         A_     :				List[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_     :				Dict			      =	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_     :				Any			      =	CLIPTextModel(a__       )
         A_     :				List[Any]			      =	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

         A_     :				int			      =	{
             """unet""": unet,
             """scheduler""": scheduler,
             """vae""": vae,
             """text_encoder""": text_encoder,
             """tokenizer""": tokenizer,
         }
         return components
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__=0       ):
         if str(a__       ).startswith("""mps"""       ):
            A_     :				List[Any]			      =	torch.manual_seed(a__       )
         else:
            A_     :				List[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 		_lowerCamelCase      (		self       ):
         A_     :				int			      =	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         A_     :				List[Any]			      =	self.get_dummy_components()
         A_     :				Optional[int]			      =	TextToVideoSDPipeline(**a__       )
         A_     :				int			      =	sd_pipe.to(a__       )
         sd_pipe.set_progress_bar_config(disable=a__       )

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

         assert frames[0].shape == (64, 64, 3)
         A_     :				Union[str, 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 		_lowerCamelCase      (		self       ):
         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 		_lowerCamelCase      (		self       ):
         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 		_lowerCamelCase      (		self       ):
         pass
      @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""       )
      def 		_lowerCamelCase      (		self       ):
         pass
      @unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""       )
      def 		_lowerCamelCase      (		self       ):
         pass






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






@slow
@skip_mps
class    _UpperCAmelCase							(		unittest.TestCase     ):
      def 		_lowerCamelCase      (		self       ):
         A_     :				Dict			      =	load_numpy(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy"""       )

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

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

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

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






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

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

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

         A_     :				str			      =	pipe(a__      ,       generator=a__      ,       num_inference_steps=2      ,       output_type="""pt"""       ).frames
         A_     :				Tuple			      =	video_frames.cpu().numpy()

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













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


_lowerCAmelCase 				=     logging.get_logger(__name__)

# TODO: upload to AWS
_lowerCAmelCase 				=     {
    """yjernite/retribert-base-uncased""": (
        """https://huggingface.co/yjernite/retribert-base-uncased/resolve/main/config.json"""
    ),
}

class    _UpperCAmelCase							(		_lowerCamelCase     ):
      a						       =							'''retribert'''
      def __init__(		self      ,       a__=30522      ,       a__=768      ,       a__=8      ,       a__=12      ,       a__=3072      ,       a__="gelu"      ,       a__=0.1      ,       a__=0.1      ,       a__=512      ,       a__=2      ,       a__=0.02      ,       a__=1E-12      ,       a__=True      ,       a__=128      ,       a__=0      ,       **a__      ,       ):
         super().__init__(pad_token_id=a__      ,       **a__       )

         A_     :				Union[str, Any]			      =	vocab_size
         A_     :				Optional[Any]			      =	hidden_size
         A_     :				Optional[Any]			      =	num_hidden_layers
         A_     :				List[str]			      =	num_attention_heads
         A_     :				Optional[int]			      =	hidden_act
         A_     :				Any			      =	intermediate_size
         A_     :				Tuple			      =	hidden_dropout_prob
         A_     :				Optional[Any]			      =	attention_probs_dropout_prob
         A_     :				Dict			      =	max_position_embeddings
         A_     :				Dict			      =	type_vocab_size
         A_     :				List[str]			      =	initializer_range
         A_     :				Optional[int]			      =	layer_norm_eps
         A_     :				Optional[Any]			      =	share_encoders
         A_     :				Dict			      =	projection_dim






