
import argparse
import os
import re

import tensorflow as tf
import torch

from transformers import BertConfig, BertModel
from transformers.utils import logging


logging.set_verbosity_info()
A     :	Union[str, Any]    = logging.get_logger(__name__)


def     __lowerCAmelCase			(			a__						,    a__						,    a__  )       ->  Tuple:
				__a										=    os.path.abspath(__A  )
				logger.info(F"""Converting TensorFlow checkpoint from {tf_path}"""  )
				# Load weights from TF model
				__a										=    tf.train.list_variables(__A  )
				__a										=    []
				__a										=    []
				__a										=    []
				for full_name, shape in init_vars:
								# logger.info(f"Loading TF weight {name} with shape {shape}")
								__a										=    full_name.split('''/'''  )
								if full_name == "_CHECKPOINTABLE_OBJECT_GRAPH" or name[0] in ["global_step", "save_counter"]:
												logger.info(F"""Skipping non-model layer {full_name}"""  )
												continue
								if "optimizer" in full_name:
												logger.info(F"""Skipping optimization layer {full_name}"""  )
												continue
								if name[0] == "model":
												# ignore initial 'model'
												__a										=    name[1:]
								# figure out how many levels deep the name is
								__a										=    0
								for _name in name:
												if _name.startswith('''layer_with_weights'''  ):
																depth += 1
												else:
																break
								layer_depth.append(__A  )
								# read data
								__a										=    tf.train.load_variable(__A						,    __A  )
								names.append('''/'''.join(__A  )  )
								arrays.append(__A  )
				logger.info(F"""Read a total of {len(__A  ):,} layers"""  )

				# Sanity check
				if len(set(__A  )  ) != 1:
								raise ValueError(F"""Found layer names with different depths (layer depth {list(set(__A  )  )})"""  )
				__a										=    list(set(__A  )  )[0]
				if layer_depth != 1:
								raise ValueError(
								    '''The model contains more than just the embedding/encoder layers. This script does not handle MLM/NSP'''
								    ''' heads.'''  )

				# convert layers
				logger.info('''Converting weights...'''  )
				for full_name, array in zip(__A						,    __A  ):
								__a										=    full_name.split('''/'''  )
								__a										=    model
								__a										=    []
								for i, m_name in enumerate(__A  ):
												if m_name == ".ATTRIBUTES":
																# variable names end with .ATTRIBUTES/VARIABLE_VALUE
																break
												if m_name.startswith('''layer_with_weights'''  ):
																__a										=    int(m_name.split('''-'''  )[-1]  )
																if layer_num <= 2:
																				# embedding layers
																				# layer_num 0: word_embeddings
																				# layer_num 1: position_embeddings
																				# layer_num 2: token_type_embeddings
																				continue
																elif layer_num == 3:
																				# embedding LayerNorm
																				trace.extend(['''embeddings''', '''LayerNorm''']  )
																				__a										=    getattr(__A						,    '''embeddings'''  )
																				__a										=    getattr(__A						,    '''LayerNorm'''  )
																elif layer_num > 3 and layer_num < config.num_hidden_layers + 4:
																				# encoder layers
																				trace.extend(['''encoder''', '''layer''', str(layer_num - 4  )]  )
																				__a										=    getattr(__A						,    '''encoder'''  )
																				__a										=    getattr(__A						,    '''layer'''  )
																				__a										=    pointer[layer_num - 4]
																elif layer_num == config.num_hidden_layers + 4:
																				# pooler layer
																				trace.extend(['''pooler''', '''dense''']  )
																				__a										=    getattr(__A						,    '''pooler'''  )
																				__a										=    getattr(__A						,    '''dense'''  )
												elif m_name == "embeddings":
																trace.append('''embeddings'''  )
																__a										=    getattr(__A						,    '''embeddings'''  )
																if layer_num == 0:
																				trace.append('''word_embeddings'''  )
																				__a										=    getattr(__A						,    '''word_embeddings'''  )
																elif layer_num == 1:
																				trace.append('''position_embeddings'''  )
																				__a										=    getattr(__A						,    '''position_embeddings'''  )
																elif layer_num == 2:
																				trace.append('''token_type_embeddings'''  )
																				__a										=    getattr(__A						,    '''token_type_embeddings'''  )
																else:
																				raise ValueError(F"""Unknown embedding layer with name {full_name}"""  )
																trace.append('''weight'''  )
																__a										=    getattr(__A						,    '''weight'''  )
												elif m_name == "_attention_layer":
																# self-attention layer
																trace.extend(['''attention''', '''self''']  )
																__a										=    getattr(__A						,    '''attention'''  )
																__a										=    getattr(__A						,    '''self'''  )
												elif m_name == "_attention_layer_norm":
																# output attention norm
																trace.extend(['''attention''', '''output''', '''LayerNorm''']  )
																__a										=    getattr(__A						,    '''attention'''  )
																__a										=    getattr(__A						,    '''output'''  )
																__a										=    getattr(__A						,    '''LayerNorm'''  )
												elif m_name == "_attention_output_dense":
																# output attention dense
																trace.extend(['''attention''', '''output''', '''dense''']  )
																__a										=    getattr(__A						,    '''attention'''  )
																__a										=    getattr(__A						,    '''output'''  )
																__a										=    getattr(__A						,    '''dense'''  )
												elif m_name == "_output_dense":
																# output dense
																trace.extend(['''output''', '''dense''']  )
																__a										=    getattr(__A						,    '''output'''  )
																__a										=    getattr(__A						,    '''dense'''  )
												elif m_name == "_output_layer_norm":
																# output dense
																trace.extend(['''output''', '''LayerNorm''']  )
																__a										=    getattr(__A						,    '''output'''  )
																__a										=    getattr(__A						,    '''LayerNorm'''  )
												elif m_name == "_key_dense":
																# attention key
																trace.append('''key'''  )
																__a										=    getattr(__A						,    '''key'''  )
												elif m_name == "_query_dense":
																# attention query
																trace.append('''query'''  )
																__a										=    getattr(__A						,    '''query'''  )
												elif m_name == "_value_dense":
																# attention value
																trace.append('''value'''  )
																__a										=    getattr(__A						,    '''value'''  )
												elif m_name == "_intermediate_dense":
																# attention intermediate dense
																trace.extend(['''intermediate''', '''dense''']  )
																__a										=    getattr(__A						,    '''intermediate'''  )
																__a										=    getattr(__A						,    '''dense'''  )
												elif m_name == "_output_layer_norm":
																# output layer norm
																trace.append('''output'''  )
																__a										=    getattr(__A						,    '''output'''  )
												# weights & biases
												elif m_name in ["bias", "beta"]:
																trace.append('''bias'''  )
																__a										=    getattr(__A						,    '''bias'''  )
												elif m_name in ["kernel", "gamma"]:
																trace.append('''weight'''  )
																__a										=    getattr(__A						,    '''weight'''  )
												else:
																logger.warning(F"""Ignored {m_name}"""  )
        # for certain layers reshape is necessary
								__a										=    '.'.join(__A  )
								if re.match(R'''(\S+)\.attention\.self\.(key|value|query)\.(bias|weight)'''						,    __A  ) or re.match(
								    R'''(\S+)\.attention\.output\.dense\.weight'''						,    __A  ):
												__a										=    array.reshape(pointer.data.shape  )
								if "kernel" in full_name:
												__a										=    array.transpose()
								if pointer.shape == array.shape:
												__a										=    torch.from_numpy(__A  )
								else:
												raise ValueError(
												    F"""Shape mismatch in layer {full_name}: Model expects shape {pointer.shape} but layer contains shape:"""
												    F""" {array.shape}"""  )
								logger.info(F"""Successfully set variable {full_name} to PyTorch layer {trace}"""  )
				return model


def     __lowerCAmelCase			(			a__						,    a__						,    a__  )       ->  List[Any]:
				logger.info(F"""Loading model based on config from {config_path}..."""  )
				__a										=    BertConfig.from_json_file(__A  )
				__a										=    BertModel(__A  )

				# Load weights from checkpoint
				logger.info(F"""Loading weights from checkpoint {tf_checkpoint_path}..."""  )
				load_tfa_weights_in_bert(__A						,    __A						,    __A  )

				# Save pytorch-model
				logger.info(F"""Saving PyTorch model to {pytorch_dump_path}..."""  )
				torch.save(model.state_dict()						,    __A  )


if __name__ == "__main__":
					A     :	Any    = argparse.ArgumentParser()
					parser.add_argument(
					    '--tf_checkpoint_path', type=str, required=True, help='Path to the TensorFlow 2.x checkpoint path.'
					)
					parser.add_argument(
					    '--bert_config_file',
					    type=str,
					    required=True,
					    help='The config json file corresponding to the BERT model. This specifies the model architecture.',
					)
					parser.add_argument(
					    '--pytorch_dump_path',
					    type=str,
					    required=True,
					    help='Path to the output PyTorch model (must include filename).',
					)
					A     :	Tuple    = parser.parse_args()
					convert_tfa_checkpoint_to_pytorch(args.tf_checkpoint_path, args.bert_config_file, args.pytorch_dump_path)



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  SCREAMING_SNAKE_CASE__  (				lowercase__						,    unittest.TestCase			):
      snake_case__				:     Optional[Any]        =						TextToVideoSDPipeline
      snake_case__				:     Optional[int]        =						TEXT_TO_IMAGE_PARAMS
      snake_case__				:     str        =						TEXT_TO_IMAGE_BATCH_PARAMS
      # No `output_type`.
      snake_case__				:     Optional[Any]        =						frozenset(
          [
              '''num_inference_steps''',
              '''generator''',
              '''latents''',
              '''return_dict''',
              '''callback''',
              '''callback_steps''',
          ]			)
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Union[str, Any]:
             torch.manual_seed(0     )
             a_					:		Optional[int]							=  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     ,							)
             a_					:		int							=  DDIMScheduler(
                 beta_start=0.00085     ,							beta_end=0.012     ,							beta_schedule='scaled_linear'     ,							clip_sample=SCREAMING_SNAKE_CASE__     ,							set_alpha_to_one=SCREAMING_SNAKE_CASE__     ,							)
             torch.manual_seed(0     )
             a_					:		int							=  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     )
             a_					:		Optional[int]							=  CLIPTextConfig(
                 bos_token_id=0     ,							eos_token_id=2     ,							hidden_size=3_2     ,							intermediate_size=3_7     ,							layer_norm_eps=1E-05     ,							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     ,							)
             a_					:		Dict							=  CLIPTextModel(SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

             a_					:		Union[str, Any]							=  {
                 'unet': unet,
                 'scheduler': scheduler,
                 'vae': vae,
                 'text_encoder': text_encoder,
                 'tokenizer': tokenizer,
             }
             return components
      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       str     ,							SCREAMING_SNAKE_CASE__				:       List[Any]=0     )		->			List[str]:
             if str(SCREAMING_SNAKE_CASE__     ).startswith('mps'     ):
                    a_					:		Dict							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             else:
                    a_					:		Optional[Any]							=  torch.Generator(device=SCREAMING_SNAKE_CASE__     ).manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		int							=  {
                 '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        SCREAMING_SNAKE_CASE							(       self				:       Union[str, Any]     )		->			Tuple:
             a_					:		List[Any]							=  'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_					:		Dict							=  self.get_dummy_components()
             a_					:		str							=  TextToVideoSDPipeline(**SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Tuple							=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__     )
             a_					:		Dict							=  'np'
             a_					:		Dict							=  sd_pipe(**SCREAMING_SNAKE_CASE__     ).frames
             a_					:		int							=  frames[0][-3:, -3:, -1]

             assert frames[0].shape == (6_4, 6_4, 3)
             a_					:		Union[str, Any]							=  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        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			List[Any]:
             self._test_attention_slicing_forward_pass(test_mean_pixel_difference=SCREAMING_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        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			str:
             self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__     ,							expected_max_diff=1E-2     )
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             pass
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			List[str]:
             pass
      @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       int     )		->			List[Any]:
             pass

      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             return super().test_progress_bar()
@slow
@skip_mps
class  SCREAMING_SNAKE_CASE__  (				unittest.TestCase			):
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     )		->			List[Any]:
             a_					:		str							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'     )

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

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

             a_					:		Optional[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2_5     ,							output_type='pt'     ).frames
             a_					:		str							=  video_frames.cpu().numpy()

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

      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			Any:
             a_					:		Dict							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'     )

             a_					:		Tuple							=  TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'     )
             a_					:		Tuple							=  pipe.to('cuda'     )

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

             a_					:		List[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2     ,							output_type='pt'     ).frames
             a_					:		List[str]							=  video_frames.cpu().numpy()

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


