


"""simple docstring"""



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 					__magic_name__		(			__UpperCAmelCase			,						unittest.TestCase     ):
       __A			:      List[str]        =   TextToVideoSDPipeline
       __A			:      Dict        =   TEXT_TO_IMAGE_PARAMS
       __A			:      List[Any]        =   TEXT_TO_IMAGE_BATCH_PARAMS
       # No `output_type`.
       __A			:      Optional[int]        =   frozenset(
           [
               "num_inference_steps",
               "generator",
               "latents",
               "return_dict",
               "callback",
               "callback_steps",
           ]     )






       def  __snake_case						(   self  :    Tuple				):

              '''simple docstring'''






              torch.manual_seed(0				)
              lowercase		:List[Any]       							=				UNetaDConditionModel(
                  block_out_channels=(3_2, 6_4, 6_4, 6_4)  ,     layers_per_block=2  ,     sample_size=3_2  ,     in_channels=4  ,     out_channels=4  ,     down_block_types=('''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''DownBlock3D''')  ,     up_block_types=('''UpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''')  ,     cross_attention_dim=3_2  ,     attention_head_dim=4  ,     )
              lowercase		:str       							=				DDIMScheduler(
                  beta_start=0.0_00_85  ,     beta_end=0.0_12  ,     beta_schedule='''scaled_linear'''  ,     clip_sample=snake_case__  ,     set_alpha_to_one=snake_case__  ,     )
              torch.manual_seed(0				)
              lowercase		:Tuple       							=				AutoencoderKL(
                  block_out_channels=[3_2, 6_4]  ,     in_channels=3  ,     out_channels=3  ,     down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']  ,     up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']  ,     latent_channels=4  ,     sample_size=1_2_8  ,     )
              torch.manual_seed(0				)
              lowercase		:List[str]       							=				CLIPTextConfig(
                  bos_token_id=0  ,     eos_token_id=2  ,     hidden_size=3_2  ,     intermediate_size=3_7  ,     layer_norm_eps=1e-0_5  ,     num_attention_heads=4  ,     num_hidden_layers=5  ,     pad_token_id=1  ,     vocab_size=1_0_0_0  ,     hidden_act='''gelu'''  ,     projection_dim=5_1_2  ,     )
              lowercase		:str       							=				CLIPTextModel(snake_case__				)
              lowercase		:Optional[int]       							=				CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''				)

              lowercase		:Optional[int]       							=				{
                  '''unet''': unet,
                  '''scheduler''': scheduler,
                  '''vae''': vae,
                  '''text_encoder''': text_encoder,
                  '''tokenizer''': tokenizer,
              }
              return components






       def  __snake_case						(   self  :    str  ,     snake_case__  :    Optional[Any]  ,     snake_case__  :    Optional[int]=0				):

              '''simple docstring'''






              if str(snake_case__				).startswith('''mps'''				):
                     lowercase		:Dict       							=				torch.manual_seed(snake_case__				)
              else:
                     lowercase		:int       							=				torch.Generator(device=snake_case__				).manual_seed(snake_case__				)
              lowercase		:str       							=				{
                  '''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  __snake_case						(   self  :    Dict				):

              '''simple docstring'''






              lowercase		:Optional[int]       							=				'''cpu'''  # ensure determinism for the device-dependent torch.Generator
              lowercase		:Union[str, Any]       							=				self.get_dummy_components()
              lowercase		:Any       							=				TextToVideoSDPipeline(**snake_case__				)
              lowercase		:str       							=				sd_pipe.to(snake_case__				)
              sd_pipe.set_progress_bar_config(disable=snake_case__				)

              lowercase		:Union[str, Any]       							=				self.get_dummy_inputs(snake_case__				)
              lowercase		:Any       							=				'''np'''
              lowercase		:List[str]       							=				sd_pipe(**snake_case__				).frames
              lowercase		:List[str]       							=				frames[0][-3:, -3:, -1]

              assert frames[0].shape == (6_4, 6_4, 3)
              lowercase		:Optional[int]       							=				np.array([1_58.0, 1_60.0, 1_53.0, 1_25.0, 1_00.0, 1_21.0, 1_11.0, 93.0, 1_13.0]				)

              assert np.abs(image_slice.flatten() - expected_slice				).max() < 1e-2






       def  __snake_case						(   self  :    Tuple				):

              '''simple docstring'''






              self._test_attention_slicing_forward_pass(test_mean_pixel_difference=snake_case__  ,     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  __snake_case						(   self  :    int				):

              '''simple docstring'''






              self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=snake_case__  ,     expected_max_diff=1e-2				)






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

              '''simple docstring'''






              pass






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

              '''simple docstring'''






              pass






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

              '''simple docstring'''






              pass







       def  __snake_case						(   self  :    Tuple				):

              '''simple docstring'''






              return super().test_progress_bar()







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






       def  __snake_case						(   self  :    List[Any]				):

              '''simple docstring'''






              lowercase		:Optional[Any]       							=				load_numpy(
                  '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'''				)

              lowercase		:Union[str, Any]       							=				TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''				)
              lowercase		:Any       							=				DPMSolverMultistepScheduler.from_config(pipe.scheduler.config				)
              lowercase		:Any       							=				pipe.to('''cuda'''				)

              lowercase		:Tuple       							=				'''Spiderman is surfing'''
              lowercase		:int       							=				torch.Generator(device='''cpu'''				).manual_seed(0				)

              lowercase		:int       							=				pipe(snake_case__  ,     generator=snake_case__  ,     num_inference_steps=2_5  ,     output_type='''pt'''				).frames
              lowercase		:Any       							=				video_frames.cpu().numpy()

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







       def  __snake_case						(   self  :    Any				):

              '''simple docstring'''






              lowercase		:int       							=				load_numpy(
                  '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'''				)

              lowercase		:int       							=				TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''				)
              lowercase		:Optional[int]       							=				pipe.to('''cuda'''				)

              lowercase		:Any       							=				'''Spiderman is surfing'''
              lowercase		:List[Any]       							=				torch.Generator(device='''cpu'''				).manual_seed(0				)

              lowercase		:Optional[int]       							=				pipe(snake_case__  ,     generator=snake_case__  ,     num_inference_steps=2  ,     output_type='''pt'''				).frames
              lowercase		:Optional[Any]       							=				video_frames.cpu().numpy()

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





"""simple docstring"""



import argparse
import os
from pathlib import Path
from typing import Dict

import tensorflow as tf
import torch
from tqdm import tqdm

from transformers import PegasusConfig, PegasusForConditionalGeneration, PegasusTokenizer
from transformers.models.pegasus.configuration_pegasus import DEFAULTS, task_specific_params


UpperCAmelCase      		=			[
    # replace left string with right string to get the relevant state_dict key (identical state dict to bart)
    ['''memory_attention''', '''encoder_attn'''],
    ['''attention''', '''attn'''],
    ['''/''', '''.'''],
    ['''.LayerNorm.gamma''', '''_layer_norm.weight'''],
    ['''.LayerNorm.beta''', '''_layer_norm.bias'''],
    ['''r.layer_''', '''r.layers.'''],
    ['''output_proj''', '''out_proj'''],
    ['''ffn.dense_1.''', '''fc2.'''],
    ['''ffn.dense.''', '''fc1.'''],
    ['''ffn_layer_norm''', '''final_layer_norm'''],
    ['''kernel''', '''weight'''],
    ['''encoder_layer_norm.''', '''encoder.layer_norm.'''],
    ['''decoder_layer_norm.''', '''decoder.layer_norm.'''],
    ['''embeddings.weights''', '''shared.weight'''],
]






def       lowerCamelCase					(a_						:Union[str, Any])     ->							Optional[Any]:
       for pegasus_name, hf_name in PATTERNS:
              lowercase		:List[Any]       							=				k.replace(a_				,				a_)
       return k






def       lowerCamelCase					(a_						:dict				,				a_						:dict)     ->							PegasusForConditionalGeneration:
       lowercase		:Union[str, Any]       							=				DEFAULTS.copy()
       cfg_kwargs.update(a_)
       lowercase		:Optional[Any]       							=				PegasusConfig(**a_)
       lowercase		:int       							=				PegasusForConditionalGeneration(a_)
       lowercase		:Optional[int]       							=				torch_model.model.state_dict()
       lowercase		:Tuple       							=				{}
       for k, v in tf_weights.items():
              lowercase		:Dict       							=				rename_state_dict_key(a_)
              if new_k not in sd:
                     raise ValueError(F"""could not find new key {new_k} in state dict. (converted from {k})""")

              if "dense" in k or "proj" in new_k:
                     lowercase		:Union[str, Any]       							=				v.T
              lowercase		:List[str]       							=				torch.tensor(a_				,				dtype=sd[new_k].dtype)
              assert v.shape == sd[new_k].shape, F"""{new_k}, {k}, {v.shape}, {sd[new_k].shape}"""
       # make sure embedding.padding_idx is respected
       lowercase		:Optional[int]       							=				torch.zeros_like(mapping['''shared.weight'''][cfg.pad_token_id + 1])
       lowercase		:Union[str, Any]       							=				mapping['''shared.weight''']
       lowercase		:Dict       							=				mapping['''shared.weight''']
       lowercase		:int       							=				{k: torch.zeros_like(a_) for k, v in sd.items() if k.endswith('''bias''') and k not in mapping}
       mapping.update(**a_)
       lowercase							,  lowercase		:Tuple       							=				torch_model.model.load_state_dict(a_				,				strict=a_)
       lowercase		:Dict       							=				[
           k for k in missing if k not in ['''encoder.embed_positions.weight''', '''decoder.embed_positions.weight''']
       ]
       assert unexpected_missing == [], F"""no matches found for the following torch keys {unexpected_missing}"""
       assert extra == [], F"""no matches found for the following tf keys {extra}"""
       return torch_model






def       lowerCamelCase					(a_						:Union[str, Any]="./ckpt/aeslc/model.ckpt-32000")     ->							Dict:
       lowercase		:Any       							=				tf.train.list_variables(a_)
       lowercase		:str       							=				{}
       lowercase		:Optional[Any]       							=				['''Adafactor''', '''global_step''']
       for name, shape in tqdm(a_				,				desc='''converting tf checkpoint to dict'''):
              lowercase		:str       							=				any(pat in name for pat in ignore_name)
              if skip_key:
                     continue
              lowercase		:str       							=				tf.train.load_variable(a_				,				a_)
              lowercase		:Dict       							=				array
       return tf_weights






def       lowerCamelCase					(a_						:str				,				a_						:str)     ->							List[str]:
       # save tokenizer first
       lowercase		:List[str]       							=				Path(a_).parent.name
       lowercase		:List[Any]       							=				task_specific_params[F"""summarization_{dataset}"""]['''max_position_embeddings''']
       lowercase		:List[Any]       							=				PegasusTokenizer.from_pretrained('''sshleifer/pegasus'''				,				model_max_length=a_)
       assert tok.model_max_length == desired_max_model_length
       tok.save_pretrained(a_)

       # convert model
       lowercase		:Any       							=				get_tf_weights_as_numpy(a_)
       lowercase		:str       							=				task_specific_params[F"""summarization_{dataset}"""]
       if dataset == "large":
              lowercase		:int       							=				task_specific_params
       lowercase		:List[Any]       							=				convert_pegasus(a_				,				a_)
       torch_model.save_pretrained(a_)
       lowercase		:Union[str, Any]       							=				torch_model.state_dict()
       sd.pop('''model.decoder.embed_positions.weight''')
       sd.pop('''model.encoder.embed_positions.weight''')
       torch.save(a_				,				Path(a_) / '''pytorch_model.bin''')


if __name__ == "__main__":
 UpperCAmelCase      		=			argparse.ArgumentParser()
 # Required parameters
 parser.add_argument('''tf_ckpt_path''', type=str, help='''passed to tf.train.list_variables''')
 parser.add_argument('''save_dir''', default=None, type=str, help='''Path to the output PyTorch model.''')
 UpperCAmelCase      		=			parser.parse_args()
 if args.save_dir is None:
  UpperCAmelCase      		=			Path(args.tf_ckpt_path).parent.name
  UpperCAmelCase      		=			os.path.join('''pegasus''', dataset)
 convert_pegasus_ckpt_to_pytorch(args.tf_ckpt_path, args.save_dir)

