





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()
lowercase_					     =	logging.get_logger(__name__)




def     __UpperCamelCase						(_SCREAMING_SNAKE_CASE	,		_SCREAMING_SNAKE_CASE	,		_SCREAMING_SNAKE_CASE	)       ->     Union[str, Any]:
  lowercase__			    =						os.path.abspath(_SCREAMING_SNAKE_CASE	)
  logger.info(F"""Converting TensorFlow checkpoint from {tf_path}"""	)
  # Load weights from TF model
  lowercase__			    =						tf.train.list_variables(_SCREAMING_SNAKE_CASE	)
  lowercase__			    =						[]
  lowercase__			    =						[]
  lowercase__			    =						[]
  for full_name, shape in init_vars:
    # logger.info(f"Loading TF weight {name} with shape {shape}")
    lowercase__			    =						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'
      lowercase__			    =						name[1:]
    # figure out how many levels deep the name is
    lowercase__			    =						0
    for _name in name:
      if _name.startswith('layer_with_weights'	):
        depth += 1
      else:
        break
    layer_depth.append(_SCREAMING_SNAKE_CASE	)
    # read data
    lowercase__			    =						tf.train.load_variable(_SCREAMING_SNAKE_CASE	,		_SCREAMING_SNAKE_CASE	)
    names.append('/'.join(_SCREAMING_SNAKE_CASE	)	)
    arrays.append(_SCREAMING_SNAKE_CASE	)
  logger.info(F"""Read a total of {len(_SCREAMING_SNAKE_CASE	):,} layers"""	)

  # Sanity check
  if len(set(_SCREAMING_SNAKE_CASE	)	) != 1:
    raise ValueError(F"""Found layer names with different depths (layer depth {list(set(_SCREAMING_SNAKE_CASE	)	)})"""	)
  lowercase__			    =						list(set(_SCREAMING_SNAKE_CASE	)	)[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(_SCREAMING_SNAKE_CASE	,		_SCREAMING_SNAKE_CASE	):
    lowercase__			    =						full_name.split('/'	)
    lowercase__			    =						model
    lowercase__			    =						[]
    for i, m_name in enumerate(_SCREAMING_SNAKE_CASE	):
      if m_name == ".ATTRIBUTES":
        # variable names end with .ATTRIBUTES/VARIABLE_VALUE
        break
      if m_name.startswith('layer_with_weights'	):
        lowercase__			    =						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']	)
          lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'embeddings'	)
          lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'LayerNorm'	)
        elif layer_num > 3 and layer_num < config.num_hidden_layers + 4:
          # encoder layers
          trace.extend(['encoder', 'layer', str(layer_num - 4	)]	)
          lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'encoder'	)
          lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'layer'	)
          lowercase__			    =						pointer[layer_num - 4]
        elif layer_num == config.num_hidden_layers + 4:
          # pooler layer
          trace.extend(['pooler', 'dense']	)
          lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'pooler'	)
          lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'dense'	)
      elif m_name == "embeddings":
        trace.append('embeddings'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'embeddings'	)
        if layer_num == 0:
          trace.append('word_embeddings'	)
          lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'word_embeddings'	)
        elif layer_num == 1:
          trace.append('position_embeddings'	)
          lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'position_embeddings'	)
        elif layer_num == 2:
          trace.append('token_type_embeddings'	)
          lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'token_type_embeddings'	)
        else:
          raise ValueError(F"""Unknown embedding layer with name {full_name}"""	)
        trace.append('weight'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'weight'	)
      elif m_name == "_attention_layer":
        # self-attention layer
        trace.extend(['attention', 'self']	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'attention'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'self'	)
      elif m_name == "_attention_layer_norm":
        # output attention norm
        trace.extend(['attention', 'output', 'LayerNorm']	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'attention'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'output'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'LayerNorm'	)
      elif m_name == "_attention_output_dense":
        # output attention dense
        trace.extend(['attention', 'output', 'dense']	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'attention'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'output'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'dense'	)
      elif m_name == "_output_dense":
        # output dense
        trace.extend(['output', 'dense']	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'output'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'dense'	)
      elif m_name == "_output_layer_norm":
        # output dense
        trace.extend(['output', 'LayerNorm']	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'output'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'LayerNorm'	)
      elif m_name == "_key_dense":
        # attention key
        trace.append('key'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'key'	)
      elif m_name == "_query_dense":
        # attention query
        trace.append('query'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'query'	)
      elif m_name == "_value_dense":
        # attention value
        trace.append('value'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'value'	)
      elif m_name == "_intermediate_dense":
        # attention intermediate dense
        trace.extend(['intermediate', 'dense']	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'intermediate'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'dense'	)
      elif m_name == "_output_layer_norm":
        # output layer norm
        trace.append('output'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'output'	)
      # weights & biases
      elif m_name in ["bias", "beta"]:
        trace.append('bias'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'bias'	)
      elif m_name in ["kernel", "gamma"]:
        trace.append('weight'	)
        lowercase__			    =						getattr(_SCREAMING_SNAKE_CASE	,		'weight'	)
      else:
        logger.warning(F"""Ignored {m_name}"""	)
        # for certain layers reshape is necessary
    lowercase__			    =						'.'.join(_SCREAMING_SNAKE_CASE	)
    if re.match(R'(\S+)\.attention\.self\.(key|value|query)\.(bias|weight)'	,		_SCREAMING_SNAKE_CASE	) or re.match(
        R'(\S+)\.attention\.output\.dense\.weight'	,		_SCREAMING_SNAKE_CASE	):
      lowercase__			    =						array.reshape(pointer.data.shape	)
    if "kernel" in full_name:
      lowercase__			    =						array.transpose()
    if pointer.shape == array.shape:
      lowercase__			    =						torch.from_numpy(_SCREAMING_SNAKE_CASE	)
    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     __UpperCamelCase						(_SCREAMING_SNAKE_CASE	,		_SCREAMING_SNAKE_CASE	,		_SCREAMING_SNAKE_CASE	)       ->     Dict:
  # Instantiate model
  logger.info(F"""Loading model based on config from {config_path}..."""	)
  lowercase__			    =						BertConfig.from_json_file(_SCREAMING_SNAKE_CASE	)
  lowercase__			    =						BertModel(_SCREAMING_SNAKE_CASE	)

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

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


if __name__ == "__main__":
 lowercase_					     =	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).""",
 )
 lowercase_					     =	parser.parse_args()
 convert_tfa_checkpoint_to_pytorch(args.tf_checkpoint_path, args.bert_config_file, args.pytorch_dump_path)









import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class        SCREAMING_SNAKE_CASE     (UpperCAmelCase    ,			UpperCAmelCase    ,			unittest.TestCase   ):
   _UpperCamelCase					:	List[Any]				      = StableDiffusionSAGPipeline
   _UpperCamelCase					:	str				      = TEXT_TO_IMAGE_PARAMS
   _UpperCamelCase					:	Dict				      = TEXT_TO_IMAGE_BATCH_PARAMS
   _UpperCamelCase					:	str				      = TEXT_TO_IMAGE_IMAGE_PARAMS
   _UpperCamelCase					:	Optional[int]				      = TEXT_TO_IMAGE_IMAGE_PARAMS
   _UpperCamelCase					:	Union[str, Any]				      = False


   def  SCREAMING_SNAKE_CASE_							(					self    :       List[str] )->     Dict:





     """simple docstring"""







     torch.manual_seed(0 )
     lowercase__			    =						UNetaDConditionModel(
         block_out_channels=(32, 64)    ,  layers_per_block=2    ,  sample_size=32    ,  in_channels=4    ,  out_channels=4    ,  down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')    ,  up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')    ,  cross_attention_dim=32    ,  )
     lowercase__			    =						DDIMScheduler(
         beta_start=0.00085    ,  beta_end=0.012    ,  beta_schedule='scaled_linear'    ,  clip_sample=a    ,  set_alpha_to_one=a    ,  )
     torch.manual_seed(0 )
     lowercase__			    =						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    ,  )
     torch.manual_seed(0 )
     lowercase__			    =						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=1_000    ,  )
     lowercase__			    =						CLIPTextModel(a )
     lowercase__			    =						CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

     lowercase__			    =						{
         'unet': unet,
         'scheduler': scheduler,
         'vae': vae,
         'text_encoder': text_encoder,
         'tokenizer': tokenizer,
         'safety_checker': None,
         'feature_extractor': None,
     }
     return components


   def  SCREAMING_SNAKE_CASE_							(					self    :       Tuple    ,  a    :       Optional[int]    ,  a    :       Any=0 )->     Union[str, Any]:





     """simple docstring"""







     if str(a ).startswith('mps' ):
       lowercase__			    =						torch.manual_seed(a )
     else:
       lowercase__			    =						torch.Generator(device=a ).manual_seed(a )
     lowercase__			    =						{
         'prompt': '.',
         'generator': generator,
         'num_inference_steps': 2,
         'guidance_scale': 1.0,
         'sag_scale': 1.0,
         'output_type': 'numpy',
     }
     return inputs


   def  SCREAMING_SNAKE_CASE_							(					self    :       Any )->     Any:





     """simple docstring"""







     super().test_inference_batch_single_identical(expected_max_diff=3E-3 )






@slow
@require_torch_gpu
class        SCREAMING_SNAKE_CASE     (unittest.TestCase   ):


   def  SCREAMING_SNAKE_CASE_							(					self    :       str )->     int:





     """simple docstring"""







     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()


   def  SCREAMING_SNAKE_CASE_							(					self    :       Tuple )->     Optional[Any]:





     """simple docstring"""







     lowercase__			    =						StableDiffusionSAGPipeline.from_pretrained('CompVis/stable-diffusion-v1-4' )
     lowercase__			    =						sag_pipe.to(a )
     sag_pipe.set_progress_bar_config(disable=a )

     lowercase__			    =						'.'
     lowercase__			    =						torch.manual_seed(0 )
     lowercase__			    =						sag_pipe(
         [prompt]    ,  generator=a    ,  guidance_scale=7.5    ,  sag_scale=1.0    ,  num_inference_steps=20    ,  output_type='np' )

     lowercase__			    =						output.images

     lowercase__			    =						image[0, -3:, -3:, -1]

     assert image.shape == (1, 512, 512, 3)
     lowercase__			    =						np.array([0.1568, 0.1738, 0.1695, 0.1693, 0.1507, 0.1705, 0.1547, 0.1751, 0.1949] )

     assert np.abs(image_slice.flatten() - expected_slice ).max() < 5E-2


   def  SCREAMING_SNAKE_CASE_							(					self    :       Tuple )->     Optional[Any]:





     """simple docstring"""







     lowercase__			    =						StableDiffusionSAGPipeline.from_pretrained('stabilityai/stable-diffusion-2-1-base' )
     lowercase__			    =						sag_pipe.to(a )
     sag_pipe.set_progress_bar_config(disable=a )

     lowercase__			    =						'.'
     lowercase__			    =						torch.manual_seed(0 )
     lowercase__			    =						sag_pipe(
         [prompt]    ,  generator=a    ,  guidance_scale=7.5    ,  sag_scale=1.0    ,  num_inference_steps=20    ,  output_type='np' )

     lowercase__			    =						output.images

     lowercase__			    =						image[0, -3:, -3:, -1]

     assert image.shape == (1, 512, 512, 3)
     lowercase__			    =						np.array([0.3459, 0.2876, 0.2537, 0.3002, 0.2671, 0.2160, 0.3026, 0.2262, 0.2371] )

     assert np.abs(image_slice.flatten() - expected_slice ).max() < 5E-2


   def  SCREAMING_SNAKE_CASE_							(					self    :       Optional[int] )->     List[Any]:





     """simple docstring"""







     lowercase__			    =						StableDiffusionSAGPipeline.from_pretrained('stabilityai/stable-diffusion-2-1-base' )
     lowercase__			    =						sag_pipe.to(a )
     sag_pipe.set_progress_bar_config(disable=a )

     lowercase__			    =						'.'
     lowercase__			    =						torch.manual_seed(0 )
     lowercase__			    =						sag_pipe(
         [prompt]    ,  width=768    ,  height=512    ,  generator=a    ,  guidance_scale=7.5    ,  sag_scale=1.0    ,  num_inference_steps=20    ,  output_type='np'    ,  )

     lowercase__			    =						output.images

     assert image.shape == (1, 512, 768, 3)


