






import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...models.auto.modeling_auto import MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
from ...utils import logging
from ..auto import CONFIG_MAPPING


A__	:			str			   = logging.get_logger(__name__)

A__	:			Optional[int]			   = {
    'salesforce/blip2-opt-2.7b': 'https://huggingface.co/salesforce/blip2-opt-2.7b/resolve/main/config.json',
}
class       _UpperCAmelCase					(       A__	):

				"""simple docstring"""




				lowercase__											=       """blip_2_vision_model"""




				def __init__(     self						:    Optional[int],  lowerCamelCase						:    Optional[int]=1_408,  lowerCamelCase						:    Optional[Any]=6_144,  lowerCamelCase						:    Any=39,  lowerCamelCase						:    Tuple=16,  lowerCamelCase						:    Tuple=224,  lowerCamelCase						:    Optional[Any]=14,  lowerCamelCase						:    Union[str, Any]="gelu",  lowerCamelCase						:    Tuple=0.00001,  lowerCamelCase						:    int=0.0,  lowerCamelCase						:    Tuple=1E-10,  lowerCamelCase						:    Dict=True,  **lowerCamelCase						:    str,  ):

								'''simple docstring'''
								super().__init__(**lowerCamelCase  )

								lowercase__    =			hidden_size
								lowercase__    =			intermediate_size
								lowercase__    =			num_hidden_layers
								lowercase__    =			num_attention_heads
								lowercase__    =			patch_size
								lowercase__    =			image_size
								lowercase__    =			initializer_range
								lowercase__    =			attention_dropout
								lowercase__    =			layer_norm_eps
								lowercase__    =			hidden_act
								lowercase__    =			qkv_bias




				@classmethod
				def 				lowercase__   (     cls						:    str,  lowerCamelCase						:    Union[str, os.PathLike],  **lowerCamelCase						:    int  ):

								'''simple docstring'''
								cls._set_token_in_kwargs(lowerCamelCase  )

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

								# get the vision config dict if we are loading from Blip2Config
								if config_dict.get('''model_type'''  ) == "blip-2":
												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       _UpperCAmelCase					(       A__	):

				"""simple docstring"""




				lowercase__											=       """blip_2_qformer"""




				def __init__(     self						:    Dict,  lowerCamelCase						:    Optional[int]=30_522,  lowerCamelCase						:    Union[str, Any]=768,  lowerCamelCase						:    Any=12,  lowerCamelCase						:    int=12,  lowerCamelCase						:    Dict=3_072,  lowerCamelCase						:    Optional[Any]="gelu",  lowerCamelCase						:    List[str]=0.1,  lowerCamelCase						:    Optional[Any]=0.1,  lowerCamelCase						:    Optional[int]=512,  lowerCamelCase						:    List[str]=0.02,  lowerCamelCase						:    Any=1E-12,  lowerCamelCase						:    Optional[Any]=0,  lowerCamelCase						:    int="absolute",  lowerCamelCase						:    Dict=2,  lowerCamelCase						:    int=1_408,  **lowerCamelCase						:    Union[str, Any],  ):

								'''simple docstring'''
								super().__init__(pad_token_id=lowerCamelCase,  **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__    =			initializer_range
								lowercase__    =			layer_norm_eps
								lowercase__    =			position_embedding_type
								lowercase__    =			cross_attention_frequency
								lowercase__    =			encoder_hidden_size




				@classmethod
				def 				lowercase__   (     cls						:    Union[str, Any],  lowerCamelCase						:    Union[str, os.PathLike],  **lowerCamelCase						:    List[str]  ):

								'''simple docstring'''
								cls._set_token_in_kwargs(lowerCamelCase  )

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

								# get the qformer config dict if we are loading from Blip2Config
								if config_dict.get('''model_type'''  ) == "blip-2":
												lowercase__    =			config_dict['''qformer_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       _UpperCAmelCase					(       A__	):

				"""simple docstring"""




				lowercase__											=       """blip-2"""
				lowercase__											=       True




				def __init__(     self						:    Optional[Any],  lowerCamelCase						:    Dict=None,  lowerCamelCase						:    Any=None,  lowerCamelCase						:    str=None,  lowerCamelCase						:    int=32,  **lowerCamelCase						:    Any  ):

								'''simple docstring'''
								super().__init__(**lowerCamelCase  )

								if vision_config is None:
												lowercase__    =			{}
												logger.info('''vision_config is None. initializing the Blip2VisionConfig with default values.'''  )

								if qformer_config is None:
												lowercase__    =			{}
												logger.info('''qformer_config is None. Initializing the Blip2QFormerConfig with default values.'''  )

								if text_config is None:
												lowercase__    =			{}
												logger.info('''text_config is None. Initializing the text config with default values (`OPTConfig`).'''  )

								lowercase__    =			BlipaVisionConfig(**lowerCamelCase  )
								lowercase__    =			BlipaQFormerConfig(**lowerCamelCase  )
								lowercase__    =			text_config['''model_type'''] if '''model_type''' in text_config else '''opt'''
								lowercase__    =			CONFIG_MAPPING[text_model_type](**lowerCamelCase  )

								lowercase__    =			self.text_config.tie_word_embeddings
								lowercase__    =			self.text_config.is_encoder_decoder

								lowercase__    =			num_query_tokens
								lowercase__    =			self.vision_config.hidden_size
								lowercase__    =			self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
								lowercase__    =			1.0
								lowercase__    =			0.02




				@classmethod
				def 				lowercase__   (     cls						:    int,  lowerCamelCase						:    BlipaVisionConfig,  lowerCamelCase						:    BlipaQFormerConfig,  lowerCamelCase						:    PretrainedConfig,  **lowerCamelCase						:    Any,  ):

								'''simple docstring'''
								return cls(
								    vision_config=vision_config.to_dict(),  qformer_config=qformer_config.to_dict(),  text_config=text_config.to_dict(),  **lowerCamelCase,  )




				def 				lowercase__   (     self						:    int  ):

								'''simple docstring'''
								lowercase__    =			copy.deepcopy(self.__dict__  )
								lowercase__    =			self.vision_config.to_dict()
								lowercase__    =			self.qformer_config.to_dict()
								lowercase__    =			self.text_config.to_dict()
								lowercase__    =			self.__class__.model_type
								return output












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					(       A__		,unittest.TestCase	):

				"""simple docstring"""




				lowercase__											=       TextToVideoSDPipeline
				lowercase__											=       TEXT_TO_IMAGE_PARAMS
				lowercase__											=       TEXT_TO_IMAGE_BATCH_PARAMS
				# No `output_type`.
				lowercase__											=       frozenset(
				    [
				        """num_inference_steps""",
				        """generator""",
				        """latents""",
				        """return_dict""",
				        """callback""",
				        """callback_steps""",
				    ]	)




				def 				lowercase__   (     self						:    str  ):

								'''simple docstring'''
								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.00085,  beta_end=0.012,  beta_schedule='''scaled_linear''',  clip_sample=lowerCamelCase,  set_alpha_to_one=lowerCamelCase,  )
								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-05,  num_attention_heads=4,  num_hidden_layers=5,  pad_token_id=1,  vocab_size=1_000,  hidden_act='''gelu''',  projection_dim=512,  )
								lowercase__    =			CLIPTextModel(lowerCamelCase  )
								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 				lowercase__   (     self						:    int,  lowerCamelCase						:    Union[str, Any],  lowerCamelCase						:    int=0  ):

								'''simple docstring'''
								if str(lowerCamelCase  ).startswith('''mps'''  ):
												lowercase__    =			torch.manual_seed(lowerCamelCase  )
								else:
												lowercase__    =			torch.Generator(device=lowerCamelCase  ).manual_seed(lowerCamelCase  )
								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 				lowercase__   (     self						:    Union[str, Any]  ):

								'''simple docstring'''
								lowercase__    =			'''cpu'''  # ensure determinism for the device-dependent torch.Generator
								lowercase__    =			self.get_dummy_components()
								lowercase__    =			TextToVideoSDPipeline(**lowerCamelCase  )
								lowercase__    =			sd_pipe.to(lowerCamelCase  )
								sd_pipe.set_progress_bar_config(disable=lowerCamelCase  )

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

								assert frames[0].shape == (64, 64, 3)
								lowercase__    =			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 				lowercase__   (     self						:    str  ):

								'''simple docstring'''
								self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowerCamelCase,  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 				lowercase__   (     self						:    Optional[int]  ):

								'''simple docstring'''
								self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCamelCase,  expected_max_diff=1E-2  )




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

								'''simple docstring'''
								pass




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

								'''simple docstring'''
								pass




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

								'''simple docstring'''
								pass




				def 				lowercase__   (     self						:    List[Any]  ):

								'''simple docstring'''
								return super().test_progress_bar()



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

				"""simple docstring"""




				def 				lowercase__   (     self						:    int  ):

								'''simple docstring'''
								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(lowerCamelCase,  generator=lowerCamelCase,  num_inference_steps=25,  output_type='''pt'''  ).frames
								lowercase__    =			video_frames.cpu().numpy()

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




				def 				lowercase__   (     self						:    int  ):

								'''simple docstring'''
								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(lowerCamelCase,  generator=lowerCamelCase,  num_inference_steps=2,  output_type='''pt'''  ).frames
								lowercase__    =			video_frames.cpu().numpy()

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




