



'''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 						lowerCamelCase_				(       snake_case_     ,					unittest.TestCase      ):
						_lowerCAmelCase  : Any						  =					VideoToVideoSDPipeline
						_lowerCAmelCase  : Any						  =					TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'video'}      ) - {'image', 'width', 'height'}
						_lowerCAmelCase  : Optional[Any]						  =					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'video'}      ) - {'image'}
						_lowerCAmelCase  : Tuple						  =					PipelineTesterMixin.required_optional_params - {'latents'}
						_lowerCAmelCase  : Optional[Any]						  =					False

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







						def        __lowercase     (	self :   Optional[int]					):

													"""simple docstring"""
													torch.manual_seed(0					)
													SCREAMING_SNAKE_CASE       :   List[Any]         =      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					,     )
													SCREAMING_SNAKE_CASE       :   int         =      DDIMScheduler(
													    beta_start=0.0_0085					,     beta_end=0.012					,     beta_schedule='''scaled_linear'''					,     clip_sample=lowerCAmelCase__					,     set_alpha_to_one=lowerCAmelCase__					,     )
													torch.manual_seed(0					)
													SCREAMING_SNAKE_CASE       :   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=1_28					,     )
													torch.manual_seed(0					)
													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      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=10_00					,     hidden_act='''gelu'''					,     projection_dim=5_12					,     )
													SCREAMING_SNAKE_CASE       :   Any         =      CLIPTextModel(lowerCAmelCase__					)
													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      {
													    '''unet''': unet,
													    '''scheduler''': scheduler,
													    '''vae''': vae,
													    '''text_encoder''': text_encoder,
													    '''tokenizer''': tokenizer,
													}
													return components







						def        __lowercase     (	self :   Optional[Any]					,     lowerCAmelCase__ :   Dict					,     lowerCAmelCase__ :   List[Any]=0					):

													"""simple docstring"""
													# 3 frames
													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      floats_tensor((1, 3, 3, 32, 32)					,     rng=random.Random(lowerCAmelCase__					)					).to(lowerCAmelCase__					)

													if str(lowerCAmelCase__					).startswith('''mps'''					):
																				SCREAMING_SNAKE_CASE       :   Tuple         =      torch.manual_seed(lowerCAmelCase__					)
													else:
																				SCREAMING_SNAKE_CASE       :   Dict         =      torch.Generator(device=lowerCAmelCase__					).manual_seed(lowerCAmelCase__					)
													SCREAMING_SNAKE_CASE       :   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        __lowercase     (	self :   int					):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE       :   str         =      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
													SCREAMING_SNAKE_CASE       :   List[str]         =      self.get_dummy_components()
													SCREAMING_SNAKE_CASE       :   str         =      VideoToVideoSDPipeline(**lowerCAmelCase__					)
													SCREAMING_SNAKE_CASE       :   Optional[int]         =      sd_pipe.to(lowerCAmelCase__					)
													sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__					)

													SCREAMING_SNAKE_CASE       :   Optional[int]         =      self.get_dummy_inputs(lowerCAmelCase__					)
													SCREAMING_SNAKE_CASE       :   Any         =      '''np'''
													SCREAMING_SNAKE_CASE       :   str         =      sd_pipe(**lowerCAmelCase__					).frames
													SCREAMING_SNAKE_CASE       :   str         =      frames[0][-3:, -3:, -1]

													assert frames[0].shape == (32, 32, 3)
													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      np.array([1_06, 1_17, 1_13, 1_74, 1_37, 1_12, 1_48, 1_51, 1_31]					)

													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        __lowercase     (	self :   Dict					):

													"""simple docstring"""
													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        __lowercase     (	self :   Optional[Any]					):

													"""simple docstring"""
													pass







						@unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''					)
						def        __lowercase     (	self :   str					):

													"""simple docstring"""
													pass







						@unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''					)
						def        __lowercase     (	self :   Union[str, Any]					):

													"""simple docstring"""
													pass







						def        __lowercase     (	self :   Optional[int]					):

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


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







						def        __lowercase     (	self :   List[Any]					):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE       :   Optional[Any]         =      VideoToVideoSDPipeline.from_pretrained('''cerspense/zeroscope_v2_XL'''					,     torch_dtype=torch.floataa					)
													pipe.enable_model_cpu_offload()

													# 10 frames
													SCREAMING_SNAKE_CASE       :   Optional[Any]         =      torch.Generator(device='''cpu'''					).manual_seed(0					)
													SCREAMING_SNAKE_CASE       :   Optional[int]         =      torch.randn((1, 10, 3, 10_24, 5_76)					,     generator=lowerCAmelCase__					)
													SCREAMING_SNAKE_CASE       :   str         =      video.to('''cuda'''					)

													SCREAMING_SNAKE_CASE       :   Dict         =      '''Spiderman is surfing'''

													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      pipe(lowerCAmelCase__					,     video=lowerCAmelCase__					,     generator=lowerCAmelCase__					,     num_inference_steps=3					,     output_type='''pt'''					).frames

													SCREAMING_SNAKE_CASE       :   Optional[Any]         =      np.array([-1.045_8984, -1.127_9297, -0.966_3086, -0.9150_3906, -0.7509_7656]					)
													assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array					).sum() < 1e-2









'''simple docstring'''



import argparse
from pathlib import Path

import torch

from transformers import OPTConfig, OPTModel
from transformers.utils import logging


logging.set_verbosity_info()
lowerCAmelCase_  :							List[Any]             =    logging.get_logger(__name__)
def        UpperCAmelCase (    A	:					Dict					):
							SCREAMING_SNAKE_CASE       :   List[str]         =      torch.load(A       ,		map_location='''cpu'''					)
							if "model" in sd.keys():
														SCREAMING_SNAKE_CASE       :   List[Any]         =      torch.load(A       ,		map_location='''cpu'''					)['''model''']

							# pop unnecessary weights
							SCREAMING_SNAKE_CASE       :   Optional[int]         =      [
							    '''decoder.version''',
							    '''decoder.output_projection.weight''',
							]
							for key in keys_to_delete:
														if key in sd:
																					sd.pop(A					)

							SCREAMING_SNAKE_CASE       :   int         =      {
							    '''decoder.project_in_dim.weight''': '''decoder.project_in.weight''',
							    '''decoder.project_out_dim.weight''': '''decoder.project_out.weight''',
							    '''decoder.layer_norm.weight''': '''decoder.final_layer_norm.weight''',
							    '''decoder.layer_norm.bias''': '''decoder.final_layer_norm.bias''',
							}
							for old_key, new_key in keys_to_rename.items():
														if old_key in sd:
																					SCREAMING_SNAKE_CASE       :   Optional[int]         =      sd.pop(A					)

							SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      list(sd.keys()					)
							for key in keys:
														if ".qkv_proj." in key:
																					SCREAMING_SNAKE_CASE       :   str         =      sd[key]
																					# We split QKV in separate Q,K,V

																					SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      key.replace('''.qkv_proj.'''       ,		'''.q_proj.'''					)
																					SCREAMING_SNAKE_CASE       :   str         =      key.replace('''.qkv_proj.'''       ,		'''.k_proj.'''					)
																					SCREAMING_SNAKE_CASE       :   Any         =      key.replace('''.qkv_proj.'''       ,		'''.v_proj.'''					)

																					SCREAMING_SNAKE_CASE       :   Tuple         =      value.shape[0]
																					assert depth % 3 == 0
																					# `SequeuceParallelTransformerBlock` has QKV weight is separated in K,V,Q despite the naming:
																					# https://cs.github.com/facebookresearch/metaseq/blob/51871bd73cd04c038f239ea2a26db1d7f6b37927/metaseq/modules/sequence_parallel_transformer_layer.py#L97
																					SCREAMING_SNAKE_CASE       ,SCREAMING_SNAKE_CASE       ,SCREAMING_SNAKE_CASE       :   Tuple         =      torch.split(A       ,		depth // 3       ,		dim=0					)

																					SCREAMING_SNAKE_CASE       :   List[str]         =      q
																					SCREAMING_SNAKE_CASE       :   List[str]         =      k
																					SCREAMING_SNAKE_CASE       :   List[Any]         =      v
																					del sd[key]

							return sd
@torch.no_grad()
def        UpperCAmelCase (    A	:					int       ,		A	:					Union[str, Any]       ,		A	:					Dict=None					):
							SCREAMING_SNAKE_CASE       :   Optional[Any]         =      load_checkpoint(A					)

							if config is not None:
														SCREAMING_SNAKE_CASE       :   Tuple         =      OPTConfig.from_pretrained(A					)
							else:
														SCREAMING_SNAKE_CASE       :   int         =      OPTConfig()

							SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      OPTModel(A					).half().eval()
							model.load_state_dict(A					)

							# Check results
							Path(A					).mkdir(exist_ok=A					)
							model.save_pretrained(A					)


if __name__ == "__main__":
							lowerCAmelCase_  :							List[str]             =    argparse.ArgumentParser()
							# Required parameters
							parser.add_argument(
							    '--fairseq_path',
							    type=str,
							    help=(
							        'path to fairseq checkpoint in correct format. You can find all checkpoints in the correct format here:'
							        ' https://huggingface.co/models?other=opt_metasq'
							    ),
							)
							parser.add_argument('--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model.')
							parser.add_argument('--hf_config', default=None, type=str, help='Define HF config.')
							lowerCAmelCase_  :							str             =    parser.parse_args()
							convert_opt_checkpoint(args.fairseq_path, args.pytorch_dump_folder_path, config=args.hf_config)




