"""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        snake_case (				__UpperCAmelCase   ,					unittest.TestCase       ):




					"""simple docstring"""

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

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







					def 	__lowerCAmelCase		(       self     :			List[Any]    ):
									torch.manual_seed(0    )
									UpperCAmelCase__    =							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					,)
									UpperCAmelCase__    =							DDIMScheduler(
									    beta_start=0.0_0_0_8_5					,beta_end=0.0_1_2					,beta_schedule='scaled_linear'					,clip_sample=lowerCamelCase__					,set_alpha_to_one=lowerCamelCase__					,)
									torch.manual_seed(0    )
									UpperCAmelCase__    =							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    )
									UpperCAmelCase__    =							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					,)
									UpperCAmelCase__    =							CLIPTextModel(lowerCamelCase__    )
									UpperCAmelCase__    =							CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

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







					def 	__lowerCAmelCase		(       self     :			str					,lowerCamelCase__     :			Any					,lowerCamelCase__     :			Union[str, Any]=0    ):
									# 3 frames
									UpperCAmelCase__    =							floats_tensor((1, 3, 3, 32, 32)					,rng=random.Random(lowerCamelCase__    )    ).to(lowerCamelCase__    )

									if str(lowerCamelCase__    ).startswith('mps'    ):
													UpperCAmelCase__    =							torch.manual_seed(lowerCamelCase__    )
									else:
													UpperCAmelCase__    =							torch.Generator(device=lowerCamelCase__    ).manual_seed(lowerCamelCase__    )
									UpperCAmelCase__    =							{
									    '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     :			Dict    ):
									UpperCAmelCase__    =							'cpu'  # ensure determinism for the device-dependent torch.Generator
									UpperCAmelCase__    =							self.get_dummy_components()
									UpperCAmelCase__    =							VideoToVideoSDPipeline(**lowerCamelCase__    )
									UpperCAmelCase__    =							sd_pipe.to(lowerCamelCase__    )
									sd_pipe.set_progress_bar_config(disable=lowerCamelCase__    )

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

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







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







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







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







					def 	__lowerCAmelCase		(       self     :			int    ):
									return super().test_progress_bar()






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




					"""simple docstring"""

					def 	__lowerCAmelCase		(       self     :			List[Any]    ):
									UpperCAmelCase__    =							VideoToVideoSDPipeline.from_pretrained('cerspense/zeroscope_v2_XL'					,torch_dtype=torch.floataa    )
									pipe.enable_model_cpu_offload()

									# 10 frames
									UpperCAmelCase__    =							torch.Generator(device='cpu'    ).manual_seed(0    )
									UpperCAmelCase__    =							torch.randn((1, 10, 3, 1_024, 576)					,generator=lowerCamelCase__    )
									UpperCAmelCase__    =							video.to('cuda'    )

									UpperCAmelCase__    =							'Spiderman is surfing'

									UpperCAmelCase__    =							pipe(lowerCamelCase__					,video=lowerCamelCase__					,generator=lowerCamelCase__					,num_inference_steps=3					,output_type='pt'    ).frames

									UpperCAmelCase__    =							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"""






from typing import Optional

import numpy as np
import torch
from torch import nn
from transformers import GPTaConfig, GPTaLMHeadModel
from transformers.modeling_utils import ModuleUtilsMixin

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin
class        snake_case (				__UpperCAmelCase   ,					__UpperCAmelCase   ,					__UpperCAmelCase       ):




					"""simple docstring"""

					snake_case__       					=				[R"h\.\d+\.attn\.bias", R"h\.\d+\.attn\.masked_bias"]







					@register_to_config
					def __init__(       self     :			str					,lowerCamelCase__     :			int					,lowerCamelCase__     :			int					,lowerCamelCase__     :			Optional[int] = None					,lowerCamelCase__     :			int = 50_257					,lowerCamelCase__     :			int = 1_024					,lowerCamelCase__     :			int = 768					,lowerCamelCase__     :			int = 12					,lowerCamelCase__     :			int = 12					,lowerCamelCase__     :			Optional[int] = None					,lowerCamelCase__     :			str = "gelu_new"					,lowerCamelCase__     :			float = 0.1					,lowerCamelCase__     :			float = 0.1					,lowerCamelCase__     :			float = 0.1					,lowerCamelCase__     :			float = 1e-5					,lowerCamelCase__     :			float = 0.0_2					,lowerCamelCase__     :			bool = True					,lowerCamelCase__     :			bool = True					,lowerCamelCase__     :			bool = False					,lowerCamelCase__     :			bool = False					,):
									super().__init__()

									UpperCAmelCase__    =							prefix_length

									if prefix_inner_dim != n_embd and prefix_hidden_dim is None:
													raise ValueError(
													    f'''`prefix_hidden_dim` cannot be `None` when `prefix_inner_dim`: {prefix_hidden_dim} and'''
													    f''' `n_embd`: {n_embd} are not equal.'''    )

									UpperCAmelCase__    =							prefix_inner_dim
									UpperCAmelCase__    =							prefix_hidden_dim

									UpperCAmelCase__    =							(
									    nn.Linear(self.prefix_inner_dim					,self.prefix_hidden_dim    )
									    if self.prefix_hidden_dim is not None
									    else nn.Identity()
									)
									UpperCAmelCase__    =							(
									    nn.Linear(self.prefix_hidden_dim					,lowerCamelCase__    ) if self.prefix_hidden_dim is not None else nn.Identity()
									)

									UpperCAmelCase__    =							GPTaConfig(
									    vocab_size=lowerCamelCase__					,n_positions=lowerCamelCase__					,n_embd=lowerCamelCase__					,n_layer=lowerCamelCase__					,n_head=lowerCamelCase__					,n_inner=lowerCamelCase__					,activation_function=lowerCamelCase__					,resid_pdrop=lowerCamelCase__					,embd_pdrop=lowerCamelCase__					,attn_pdrop=lowerCamelCase__					,layer_norm_epsilon=lowerCamelCase__					,initializer_range=lowerCamelCase__					,scale_attn_weights=lowerCamelCase__					,use_cache=lowerCamelCase__					,scale_attn_by_inverse_layer_idx=lowerCamelCase__					,reorder_and_upcast_attn=lowerCamelCase__					,)
									UpperCAmelCase__    =							GPTaLMHeadModel(lowerCamelCase__    )







					def 	__lowerCAmelCase		(       self     :			str					,lowerCamelCase__     :			torch.Tensor					,lowerCamelCase__     :			torch.Tensor					,lowerCamelCase__     :			Optional[torch.Tensor] = None					,lowerCamelCase__     :			Optional[torch.Tensor] = None					,):
									UpperCAmelCase__    =							self.transformer.transformer.wte(lowerCamelCase__    )
									UpperCAmelCase__    =							self.encode_prefix(lowerCamelCase__    )
									UpperCAmelCase__    =							self.decode_prefix(lowerCamelCase__    )
									UpperCAmelCase__    =							torch.cat((prefix_embeds, embedding_text)					,dim=1    )

									if labels is not None:
													UpperCAmelCase__    =							self.get_dummy_token(input_ids.shape[0]					,input_ids.device    )
													UpperCAmelCase__    =							torch.cat((dummy_token, input_ids)					,dim=1    )
									UpperCAmelCase__    =							self.transformer(inputs_embeds=lowerCamelCase__					,labels=lowerCamelCase__					,attention_mask=lowerCamelCase__    )
									if self.prefix_hidden_dim is not None:
													return out, hidden
									else:
													return out







					def 	__lowerCAmelCase		(       self     :			Union[str, Any]					,lowerCamelCase__     :			int					,lowerCamelCase__     :			torch.device    ):
									return torch.zeros(lowerCamelCase__					,self.prefix_length					,dtype=torch.intaa					,device=lowerCamelCase__    )







					def 	__lowerCAmelCase		(       self     :			Tuple					,lowerCamelCase__     :			List[str]    ):
									return self.encode_prefix(lowerCamelCase__    )







					@torch.no_grad()
					def 	__lowerCAmelCase		(       self     :			Optional[Any]					,lowerCamelCase__     :			Union[str, Any]					,lowerCamelCase__     :			Optional[int]					,lowerCamelCase__     :			Optional[int]    ):
									UpperCAmelCase__    =							torch.split(lowerCamelCase__					,1					,dim=0    )
									UpperCAmelCase__    =							[]
									UpperCAmelCase__    =							[]
									for feature in features:
													UpperCAmelCase__    =							self.decode_prefix(feature.to(lowerCamelCase__    )    )  # back to the clip feature
													# Only support beam search for now
													UpperCAmelCase__   ,			UpperCAmelCase__    =							self.generate_beam(
													    input_embeds=lowerCamelCase__					,device=lowerCamelCase__					,eos_token_id=lowerCamelCase__    )
													generated_tokens.append(output_tokens[0]    )
													generated_seq_lengths.append(seq_lengths[0]    )
									UpperCAmelCase__    =							torch.stack(lowerCamelCase__    )
									UpperCAmelCase__    =							torch.stack(lowerCamelCase__    )
									return generated_tokens, generated_seq_lengths







					@torch.no_grad()
					def 	__lowerCAmelCase		(       self     :			Optional[Any]					,lowerCamelCase__     :			str=None					,lowerCamelCase__     :			List[Any]=None					,lowerCamelCase__     :			Tuple=None					,lowerCamelCase__     :			int = 5					,lowerCamelCase__     :			int = 67					,lowerCamelCase__     :			float = 1.0					,lowerCamelCase__     :			Optional[int] = None					,):
									UpperCAmelCase__    =							eos_token_id
									UpperCAmelCase__    =							None
									UpperCAmelCase__    =							None
									UpperCAmelCase__    =							torch.ones(lowerCamelCase__					,device=lowerCamelCase__					,dtype=torch.int    )
									UpperCAmelCase__    =							torch.zeros(lowerCamelCase__					,device=lowerCamelCase__					,dtype=torch.bool    )

									if input_embeds is not None:
													UpperCAmelCase__    =							input_embeds
									else:
													UpperCAmelCase__    =							self.transformer.transformer.wte(lowerCamelCase__    )

									for i in range(lowerCamelCase__    ):
													UpperCAmelCase__    =							self.transformer(inputs_embeds=lowerCamelCase__    )
													UpperCAmelCase__    =							outputs.logits
													UpperCAmelCase__    =							logits[:, -1, :] / (temperature if temperature > 0 else 1.0)
													UpperCAmelCase__    =							logits.softmax(-1    ).log()

													if scores is None:
																	UpperCAmelCase__   ,			UpperCAmelCase__    =							logits.topk(lowerCamelCase__					,-1    )
																	UpperCAmelCase__    =							generated.expand(lowerCamelCase__					,*generated.shape[1:]    )
																	UpperCAmelCase__   ,			UpperCAmelCase__    =							next_tokens.permute(1					,0    ), scores.squeeze(0    )
																	if tokens is None:
																					UpperCAmelCase__    =							next_tokens
																	else:
																					UpperCAmelCase__    =							tokens.expand(lowerCamelCase__					,*tokens.shape[1:]    )
																					UpperCAmelCase__    =							torch.cat((tokens, next_tokens)					,dim=1    )
													else:
																	UpperCAmelCase__    =							-float(np.inf    )
																	UpperCAmelCase__    =							0
																	UpperCAmelCase__    =							scores[:, None] + logits
																	seq_lengths[~is_stopped] += 1
																	UpperCAmelCase__    =							scores_sum / seq_lengths[:, None]
																	UpperCAmelCase__   ,			UpperCAmelCase__    =							scores_sum_average.view(-1    ).topk(lowerCamelCase__					,-1    )
																	UpperCAmelCase__    =							next_tokens // scores_sum.shape[1]
																	UpperCAmelCase__    =							seq_lengths[next_tokens_source]
																	UpperCAmelCase__    =							next_tokens % scores_sum.shape[1]
																	UpperCAmelCase__    =							next_tokens.unsqueeze(1    )
																	UpperCAmelCase__    =							tokens[next_tokens_source]
																	UpperCAmelCase__    =							torch.cat((tokens, next_tokens)					,dim=1    )
																	UpperCAmelCase__    =							generated[next_tokens_source]
																	UpperCAmelCase__    =							scores_sum_average * seq_lengths
																	UpperCAmelCase__    =							is_stopped[next_tokens_source]

													UpperCAmelCase__    =							self.transformer.transformer.wte(next_tokens.squeeze()    ).view(generated.shape[0]					,1					,-1    )
													UpperCAmelCase__    =							torch.cat((generated, next_token_embed)					,dim=1    )
													UpperCAmelCase__    =							is_stopped + next_tokens.eq(lowerCamelCase__    ).squeeze()
													if is_stopped.all():
																	break

									UpperCAmelCase__    =							scores / seq_lengths
									UpperCAmelCase__    =							scores.argsort(descending=lowerCamelCase__    )
									# tokens tensors are already padded to max_seq_length
									UpperCAmelCase__    =							[tokens[i] for i in order]
									UpperCAmelCase__    =							torch.stack(lowerCamelCase__					,dim=0    )
									UpperCAmelCase__    =							torch.tensor([seq_lengths[i] for i in order]					,dtype=seq_lengths.dtype    )
									return output_texts, seq_lengths



