





"""simple docstring"""






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







					'''simple docstring'''




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

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







					def  _lowerCamelCase		(						self     ):
									torch.manual_seed(0     )
									__a :						List[str]			       =  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 :						Optional[int]			       =  DDIMScheduler(
									    beta_start=0.0_0_0_8_5	,			beta_end=0.0_1_2	,			beta_schedule='''scaled_linear'''	,			clip_sample=_UpperCAmelCase	,			set_alpha_to_one=_UpperCAmelCase	,			)
									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 :						List[str]			       =  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=1000	,			hidden_act='''gelu'''	,			projection_dim=512	,			)
									__a :						Optional[int]			       =  CLIPTextModel(_UpperCAmelCase     )
									__a :						Optional[Any]			       =  CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''     )

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







					def  _lowerCamelCase		(						self	,			_UpperCAmelCase	,			_UpperCAmelCase=0     ):
									# 3 frames
									__a :						List[Any]			       =  floats_tensor((1, 3, 3, 32, 32)	,			rng=random.Random(_UpperCAmelCase     )     ).to(_UpperCAmelCase     )

									if str(_UpperCAmelCase     ).startswith('''mps'''     ):
													__a :						Union[str, Any]			       =  torch.manual_seed(_UpperCAmelCase     )
									else:
													__a :						Optional[Any]			       =  torch.Generator(device=_UpperCAmelCase     ).manual_seed(_UpperCAmelCase     )
									__a :						Dict			       =  {
									    '''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  _lowerCamelCase		(						self     ):
									__a :						Any			       =  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
									__a :						Optional[Any]			       =  self.get_dummy_components()
									__a :						int			       =  VideoToVideoSDPipeline(**_UpperCAmelCase     )
									__a :						Any			       =  sd_pipe.to(_UpperCAmelCase     )
									sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

									__a :						Tuple			       =  self.get_dummy_inputs(_UpperCAmelCase     )
									__a :						Optional[Any]			       =  '''np'''
									__a :						int			       =  sd_pipe(**_UpperCAmelCase     ).frames
									__a :						Union[str, Any]			       =  frames[0][-3:, -3:, -1]

									assert frames[0].shape == (32, 32, 3)
									__a :						Any			       =  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  _lowerCamelCase		(						self     ):
									self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=_UpperCAmelCase	,			expected_max_diff=5e-3     )







					@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 				__lowercase			(			unittest.TestCase							):







					'''simple docstring'''




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

									# 10 frames
									__a :						List[str]			       =  torch.Generator(device='''cpu'''     ).manual_seed(0     )
									__a :						Union[str, Any]			       =  torch.randn((1, 10, 3, 1024, 576)	,			generator=_UpperCAmelCase     )
									__a :						Tuple			       =  video.to('''cuda'''     )

									__a :						str			       =  '''Spiderman is surfing'''

									__a :						List[str]			       =  pipe(_UpperCAmelCase	,			video=_UpperCAmelCase	,			generator=_UpperCAmelCase	,			num_inference_steps=3	,			output_type='''pt'''     ).frames

									__a :						Any			       =  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






"""simple docstring"""






import copy
import os
from typing import Union

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


A 			=  logging.get_logger(__name__)

A 			=  {
    '''microsoft/git-base''': '''https://huggingface.co/microsoft/git-base/resolve/main/config.json''',
}




class 				__lowercase			(			_UpperCamelCase							):







					'''simple docstring'''




					__lowerCAmelCase  =  '''git_vision_model'''







					def __init__(						self	,			_UpperCAmelCase=768	,			_UpperCAmelCase=3072	,			_UpperCAmelCase=12	,			_UpperCAmelCase=12	,			_UpperCAmelCase=3	,			_UpperCAmelCase=224	,			_UpperCAmelCase=16	,			_UpperCAmelCase="quick_gelu"	,			_UpperCAmelCase=1e-5	,			_UpperCAmelCase=0.0	,			_UpperCAmelCase=0.0_2	,			**_UpperCAmelCase	,			):
									super().__init__(**_UpperCAmelCase     )

									__a :						Union[str, Any]			       =  hidden_size
									__a :						Optional[int]			       =  intermediate_size
									__a :						Tuple			       =  num_hidden_layers
									__a :						Tuple			       =  num_attention_heads
									__a :						Optional[int]			       =  num_channels
									__a :						Tuple			       =  patch_size
									__a :						Any			       =  image_size
									__a :						Optional[Any]			       =  initializer_range
									__a :						Optional[int]			       =  attention_dropout
									__a :						Any			       =  layer_norm_eps
									__a :						Dict			       =  hidden_act







					@classmethod
					def  _lowerCamelCase		(						cls	,			_UpperCAmelCase	,			**_UpperCAmelCase     ):
									cls._set_token_in_kwargs(_UpperCAmelCase     )

									__a       ,						__a :						Dict			       =  cls.get_config_dict(_UpperCAmelCase	,			**_UpperCAmelCase     )

									# get the vision config dict if we are loading from GITConfig
									if config_dict.get('''model_type'''     ) == "git":
													__a :						int			       =  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(_UpperCAmelCase	,			**_UpperCAmelCase     )






class 				__lowercase			(			_UpperCamelCase							):







					'''simple docstring'''




					__lowerCAmelCase  =  '''git'''







					def __init__(						self	,			_UpperCAmelCase=None	,			_UpperCAmelCase=30522	,			_UpperCAmelCase=768	,			_UpperCAmelCase=6	,			_UpperCAmelCase=12	,			_UpperCAmelCase=3072	,			_UpperCAmelCase="gelu"	,			_UpperCAmelCase=0.1	,			_UpperCAmelCase=0.1	,			_UpperCAmelCase=1024	,			_UpperCAmelCase=0.0_2	,			_UpperCAmelCase=1e-1_2	,			_UpperCAmelCase=0	,			_UpperCAmelCase="absolute"	,			_UpperCAmelCase=True	,			_UpperCAmelCase=False	,			_UpperCAmelCase=101	,			_UpperCAmelCase=102	,			_UpperCAmelCase=None	,			**_UpperCAmelCase	,			):
									super().__init__(bos_token_id=_UpperCAmelCase	,			eos_token_id=_UpperCAmelCase	,			pad_token_id=_UpperCAmelCase	,			**_UpperCAmelCase     )

									if vision_config is None:
													__a :						Union[str, Any]			       =  {}
													logger.info('''vision_config is None. initializing the GitVisionConfig with default values.'''     )

									__a :						Union[str, Any]			       =  GitVisionConfig(**_UpperCAmelCase     )
									__a :						List[str]			       =  vocab_size
									__a :						Any			       =  hidden_size
									__a :						Dict			       =  num_hidden_layers
									__a :						Tuple			       =  num_attention_heads
									__a :						List[str]			       =  hidden_act
									__a :						Any			       =  intermediate_size
									__a :						int			       =  hidden_dropout_prob
									__a :						Dict			       =  attention_probs_dropout_prob
									__a :						Tuple			       =  max_position_embeddings
									__a :						int			       =  initializer_range
									__a :						List[str]			       =  layer_norm_eps
									__a :						Tuple			       =  position_embedding_type
									__a :						Optional[int]			       =  use_cache
									__a :						str			       =  tie_word_embeddings
									__a :						Optional[Any]			       =  num_image_with_embedding

									__a :						str			       =  bos_token_id
									__a :						Dict			       =  eos_token_id







					def  _lowerCamelCase		(						self     ):
									__a :						Optional[int]			       =  copy.deepcopy(self.__dict__     )
									__a :						int			       =  self.vision_config.to_dict()
									__a :						Tuple			       =  self.__class__.model_type
									return output