





'''simple docstring'''


import argparse
import os.path as osp
import re

import torch
from safetensors.torch import load_file, save_file


# =================#
# UNet Conversion #
# =================#

snake_case__					:      Any				      =  [
    # (stable-diffusion, HF Diffusers)
    ('''time_embed.0.weight''', '''time_embedding.linear_1.weight'''),
    ('''time_embed.0.bias''', '''time_embedding.linear_1.bias'''),
    ('''time_embed.2.weight''', '''time_embedding.linear_2.weight'''),
    ('''time_embed.2.bias''', '''time_embedding.linear_2.bias'''),
    ('''input_blocks.0.0.weight''', '''conv_in.weight'''),
    ('''input_blocks.0.0.bias''', '''conv_in.bias'''),
    ('''out.0.weight''', '''conv_norm_out.weight'''),
    ('''out.0.bias''', '''conv_norm_out.bias'''),
    ('''out.2.weight''', '''conv_out.weight'''),
    ('''out.2.bias''', '''conv_out.bias'''),
]

snake_case__					:      Optional[Any]				      =  [
    # (stable-diffusion, HF Diffusers)
    ('''in_layers.0''', '''norm1'''),
    ('''in_layers.2''', '''conv1'''),
    ('''out_layers.0''', '''norm2'''),
    ('''out_layers.3''', '''conv2'''),
    ('''emb_layers.1''', '''time_emb_proj'''),
    ('''skip_connection''', '''conv_shortcut'''),
]

snake_case__					:      int				      =  []
# hardcoded number of downblocks and resnets/attentions...
# would need smarter logic for other networks.
for i in range(4):
 # loop over downblocks/upblocks

 for j in range(2):
  # loop over resnets/attentions for downblocks
  snake_case__					:      Union[str, Any]				      =  f'''down_blocks.{i}.resnets.{j}.'''
  snake_case__					:      Dict				      =  f'''input_blocks.{3*i + j + 1}.0.'''
  unet_conversion_map_layer.append((sd_down_res_prefix, hf_down_res_prefix))

  if i < 3:
   # no attention layers in down_blocks.3
   snake_case__					:      Optional[Any]				      =  f'''down_blocks.{i}.attentions.{j}.'''
   snake_case__					:      Tuple				      =  f'''input_blocks.{3*i + j + 1}.1.'''
   unet_conversion_map_layer.append((sd_down_atn_prefix, hf_down_atn_prefix))

 for j in range(3):
  # loop over resnets/attentions for upblocks
  snake_case__					:      Any				      =  f'''up_blocks.{i}.resnets.{j}.'''
  snake_case__					:      Dict				      =  f'''output_blocks.{3*i + j}.0.'''
  unet_conversion_map_layer.append((sd_up_res_prefix, hf_up_res_prefix))

  if i > 0:
   # no attention layers in up_blocks.0
   snake_case__					:      Tuple				      =  f'''up_blocks.{i}.attentions.{j}.'''
   snake_case__					:      Optional[Any]				      =  f'''output_blocks.{3*i + j}.1.'''
   unet_conversion_map_layer.append((sd_up_atn_prefix, hf_up_atn_prefix))

 if i < 3:
  # no downsample in down_blocks.3
  snake_case__					:      List[str]				      =  f'''down_blocks.{i}.downsamplers.0.conv.'''
  snake_case__					:      str				      =  f'''input_blocks.{3*(i+1)}.0.op.'''
  unet_conversion_map_layer.append((sd_downsample_prefix, hf_downsample_prefix))

  # no upsample in up_blocks.3
  snake_case__					:      Union[str, Any]				      =  f'''up_blocks.{i}.upsamplers.0.'''
  snake_case__					:      List[str]				      =  f'''output_blocks.{3*i + 2}.{1 if i == 0 else 2}.'''
  unet_conversion_map_layer.append((sd_upsample_prefix, hf_upsample_prefix))

snake_case__					:      str				      =  '''mid_block.attentions.0.'''
snake_case__					:      Union[str, Any]				      =  '''middle_block.1.'''
unet_conversion_map_layer.append((sd_mid_atn_prefix, hf_mid_atn_prefix))

for j in range(2):
 snake_case__					:      str				      =  f'''mid_block.resnets.{j}.'''
 snake_case__					:      List[Any]				      =  f'''middle_block.{2*j}.'''
 unet_conversion_map_layer.append((sd_mid_res_prefix, hf_mid_res_prefix))


def  _lowerCamelCase   (				lowerCamelCase_ :							Optional[Any]       ):



     """simple docstring"""





     UpperCAmelCase_							:       str       =				{k: k for k in unet_state_dict.keys()}
     for sd_name, hf_name in unet_conversion_map:
          UpperCAmelCase_							:       Dict       =				sd_name
     for k, v in mapping.items():
          if "resnets" in k:
               for sd_part, hf_part in unet_conversion_map_resnet:
                    UpperCAmelCase_							:       str       =				v.replace(a__     ,					a__       )
               UpperCAmelCase_							:       Tuple       =				v
     for k, v in mapping.items():
          for sd_part, hf_part in unet_conversion_map_layer:
               UpperCAmelCase_							:       Optional[Any]       =				v.replace(a__     ,					a__       )
          UpperCAmelCase_							:       Optional[Any]       =				v
     UpperCAmelCase_							:       Tuple       =				{v: unet_state_dict[k] for k, v in mapping.items()}
     return new_state_dict


# ================#
# VAE Conversion #
# ================#

snake_case__					:      Dict				      =  [
    # (stable-diffusion, HF Diffusers)
    ('''nin_shortcut''', '''conv_shortcut'''),
    ('''norm_out''', '''conv_norm_out'''),
    ('''mid.attn_1.''', '''mid_block.attentions.0.'''),
]

for i in range(4):
 # down_blocks have two resnets
 for j in range(2):
  snake_case__					:      Optional[Any]				      =  f'''encoder.down_blocks.{i}.resnets.{j}.'''
  snake_case__					:      Optional[Any]				      =  f'''encoder.down.{i}.block.{j}.'''
  vae_conversion_map.append((sd_down_prefix, hf_down_prefix))

 if i < 3:
  snake_case__					:      Optional[Any]				      =  f'''down_blocks.{i}.downsamplers.0.'''
  snake_case__					:      Any				      =  f'''down.{i}.downsample.'''
  vae_conversion_map.append((sd_downsample_prefix, hf_downsample_prefix))

  snake_case__					:      Optional[Any]				      =  f'''up_blocks.{i}.upsamplers.0.'''
  snake_case__					:      Tuple				      =  f'''up.{3-i}.upsample.'''
  vae_conversion_map.append((sd_upsample_prefix, hf_upsample_prefix))

 # up_blocks have three resnets
 # also, up blocks in hf are numbered in reverse from sd
 for j in range(3):
  snake_case__					:      Any				      =  f'''decoder.up_blocks.{i}.resnets.{j}.'''
  snake_case__					:      Any				      =  f'''decoder.up.{3-i}.block.{j}.'''
  vae_conversion_map.append((sd_up_prefix, hf_up_prefix))

# this part accounts for mid blocks in both the encoder and the decoder
for i in range(2):
 snake_case__					:      Any				      =  f'''mid_block.resnets.{i}.'''
 snake_case__					:      Optional[int]				      =  f'''mid.block_{i+1}.'''
 vae_conversion_map.append((sd_mid_res_prefix, hf_mid_res_prefix))


snake_case__					:      Dict				      =  [
    # (stable-diffusion, HF Diffusers)
    ('''norm.''', '''group_norm.'''),
    ('''q.''', '''query.'''),
    ('''k.''', '''key.'''),
    ('''v.''', '''value.'''),
    ('''proj_out.''', '''proj_attn.'''),
]


def  _lowerCamelCase   (				lowerCamelCase_ :							str       ):



     """simple docstring"""





     return w.reshape(*w.shape     ,					1     ,					1       )


def  _lowerCamelCase   (				lowerCamelCase_ :							Optional[Any]       ):



     """simple docstring"""





     UpperCAmelCase_							:       Union[str, Any]       =				{k: k for k in vae_state_dict.keys()}
     for k, v in mapping.items():
          for sd_part, hf_part in vae_conversion_map:
               UpperCAmelCase_							:       Union[str, Any]       =				v.replace(a__     ,					a__       )
          UpperCAmelCase_							:       List[Any]       =				v
     for k, v in mapping.items():
          if "attentions" in k:
               for sd_part, hf_part in vae_conversion_map_attn:
                    UpperCAmelCase_							:       Dict       =				v.replace(a__     ,					a__       )
               UpperCAmelCase_							:       int       =				v
     UpperCAmelCase_							:       List[Any]       =				{v: vae_state_dict[k] for k, v in mapping.items()}
     UpperCAmelCase_							:       Dict       =				['q', 'k', 'v', 'proj_out']
     for k, v in new_state_dict.items():
          for weight_name in weights_to_convert:
               if F'''mid.attn_1.{weight_name}.weight''' in k:
                    print(F'''Reshaping {k} for SD format'''       )
                    UpperCAmelCase_							:       List[Any]       =				reshape_weight_for_sd(a__       )
     return new_state_dict


# =========================#
# Text Encoder Conversion #
# =========================#


snake_case__					:      Optional[Any]				      =  [
    # (stable-diffusion, HF Diffusers)
    ('''resblocks.''', '''text_model.encoder.layers.'''),
    ('''ln_1''', '''layer_norm1'''),
    ('''ln_2''', '''layer_norm2'''),
    ('''.c_fc.''', '''.fc1.'''),
    ('''.c_proj.''', '''.fc2.'''),
    ('''.attn''', '''.self_attn'''),
    ('''ln_final.''', '''transformer.text_model.final_layer_norm.'''),
    ('''token_embedding.weight''', '''transformer.text_model.embeddings.token_embedding.weight'''),
    ('''positional_embedding''', '''transformer.text_model.embeddings.position_embedding.weight'''),
]
snake_case__					:      List[Any]				      =  {re.escape(x[1]): x[0] for x in textenc_conversion_lst}
snake_case__					:      Tuple				      =  re.compile('''|'''.join(protected.keys()))

# Ordering is from https://github.com/pytorch/pytorch/blob/master/test/cpp/api/modules.cpp
snake_case__					:      Union[str, Any]				      =  {'''q''': 0, '''k''': 1, '''v''': 2}


def  _lowerCamelCase   (				lowerCamelCase_ :							List[str]       ):



     """simple docstring"""





     UpperCAmelCase_							:       Dict       =				{}
     UpperCAmelCase_							:       int       =				{}
     UpperCAmelCase_							:       List[str]       =				{}
     for k, v in text_enc_dict.items():
          if (
              k.endswith('.self_attn.q_proj.weight'       )
              or k.endswith('.self_attn.k_proj.weight'       )
              or k.endswith('.self_attn.v_proj.weight'       )
          ):
               UpperCAmelCase_							:       str       =				k[: -len('.q_proj.weight'       )]
               UpperCAmelCase_							:       List[str]       =				k[-len('q_proj.weight'       )]
               if k_pre not in capture_qkv_weight:
                    UpperCAmelCase_							:       Optional[Any]       =				[None, None, None]
               UpperCAmelCase_							:       Union[str, Any]       =				v
               continue

          if (
              k.endswith('.self_attn.q_proj.bias'       )
              or k.endswith('.self_attn.k_proj.bias'       )
              or k.endswith('.self_attn.v_proj.bias'       )
          ):
               UpperCAmelCase_							:       Union[str, Any]       =				k[: -len('.q_proj.bias'       )]
               UpperCAmelCase_							:       Tuple       =				k[-len('q_proj.bias'       )]
               if k_pre not in capture_qkv_bias:
                    UpperCAmelCase_							:       Dict       =				[None, None, None]
               UpperCAmelCase_							:       List[str]       =				v
               continue

          UpperCAmelCase_							:       List[str]       =				textenc_pattern.sub(lambda lowerCamelCase_       : protected[re.escape(m.group(0       )       )]     ,					a__       )
          UpperCAmelCase_							:       List[Any]       =				v

     for k_pre, tensors in capture_qkv_weight.items():
          if None in tensors:
               raise Exception('CORRUPTED MODEL: one of the q-k-v values for the text encoder was missing'       )
          UpperCAmelCase_							:       Tuple       =				textenc_pattern.sub(lambda lowerCamelCase_       : protected[re.escape(m.group(0       )       )]     ,					a__       )
          UpperCAmelCase_							:       Optional[Any]       =				torch.cat(a__       )

     for k_pre, tensors in capture_qkv_bias.items():
          if None in tensors:
               raise Exception('CORRUPTED MODEL: one of the q-k-v values for the text encoder was missing'       )
          UpperCAmelCase_							:       List[str]       =				textenc_pattern.sub(lambda lowerCamelCase_       : protected[re.escape(m.group(0       )       )]     ,					a__       )
          UpperCAmelCase_							:       Optional[int]       =				torch.cat(a__       )

     return new_state_dict


def  _lowerCamelCase   (				lowerCamelCase_ :							Union[str, Any]       ):



     """simple docstring"""





     return text_enc_dict


if __name__ == "__main__":
 snake_case__					:      Optional[Any]				      =  argparse.ArgumentParser()

 parser.add_argument('''--model_path''', default=None, type=str, required=True, help='''Path to the model to convert.''')
 parser.add_argument('''--checkpoint_path''', default=None, type=str, required=True, help='''Path to the output model.''')
 parser.add_argument('''--half''', action='''store_true''', help='''Save weights in half precision.''')
 parser.add_argument(
     '''--use_safetensors''', action='''store_true''', help='''Save weights use safetensors, default is ckpt.'''
 )

 snake_case__					:      List[Any]				      =  parser.parse_args()

 assert args.model_path is not None, "Must provide a model path!"

 assert args.checkpoint_path is not None, "Must provide a checkpoint path!"

 # Path for safetensors
 snake_case__					:      Dict				      =  osp.join(args.model_path, '''unet''', '''diffusion_pytorch_model.safetensors''')
 snake_case__					:      List[Any]				      =  osp.join(args.model_path, '''vae''', '''diffusion_pytorch_model.safetensors''')
 snake_case__					:      Dict				      =  osp.join(args.model_path, '''text_encoder''', '''model.safetensors''')

 # Load models from safetensors if it exists, if it doesn't pytorch
 if osp.exists(unet_path):
  snake_case__					:      Any				      =  load_file(unet_path, device='''cpu''')
 else:
  snake_case__					:      Any				      =  osp.join(args.model_path, '''unet''', '''diffusion_pytorch_model.bin''')
  snake_case__					:      Tuple				      =  torch.load(unet_path, map_location='''cpu''')

 if osp.exists(vae_path):
  snake_case__					:      Tuple				      =  load_file(vae_path, device='''cpu''')
 else:
  snake_case__					:      Optional[Any]				      =  osp.join(args.model_path, '''vae''', '''diffusion_pytorch_model.bin''')
  snake_case__					:      List[str]				      =  torch.load(vae_path, map_location='''cpu''')

 if osp.exists(text_enc_path):
  snake_case__					:      List[Any]				      =  load_file(text_enc_path, device='''cpu''')
 else:
  snake_case__					:      Dict				      =  osp.join(args.model_path, '''text_encoder''', '''pytorch_model.bin''')
  snake_case__					:      Union[str, Any]				      =  torch.load(text_enc_path, map_location='''cpu''')

 # Convert the UNet model
 snake_case__					:      str				      =  convert_unet_state_dict(unet_state_dict)
 snake_case__					:      Union[str, Any]				      =  {'''model.diffusion_model.''' + k: v for k, v in unet_state_dict.items()}

 # Convert the VAE model
 snake_case__					:      Dict				      =  convert_vae_state_dict(vae_state_dict)
 snake_case__					:      Dict				      =  {'''first_stage_model.''' + k: v for k, v in vae_state_dict.items()}

 # Easiest way to identify v2.0 model seems to be that the text encoder (OpenCLIP) is deeper
 snake_case__					:      Union[str, Any]				      =  '''text_model.encoder.layers.22.layer_norm2.bias''' in text_enc_dict

 if is_vaa_model:
  # Need to add the tag 'transformer' in advance so we can knock it out from the final layer-norm
  snake_case__					:      Optional[int]				      =  {'''transformer.''' + k: v for k, v in text_enc_dict.items()}
  snake_case__					:      Optional[Any]				      =  convert_text_enc_state_dict_vaa(text_enc_dict)
  snake_case__					:      Union[str, Any]				      =  {'''cond_stage_model.model.''' + k: v for k, v in text_enc_dict.items()}
 else:
  snake_case__					:      Optional[Any]				      =  convert_text_enc_state_dict(text_enc_dict)
  snake_case__					:      List[str]				      =  {'''cond_stage_model.transformer.''' + k: v for k, v in text_enc_dict.items()}

 # Put together new checkpoint
 snake_case__					:      List[str]				      =  {**unet_state_dict, **vae_state_dict, **text_enc_dict}
 if args.half:
  snake_case__					:      str				      =  {k: v.half() for k, v in state_dict.items()}

 if args.use_safetensors:
  save_file(state_dict, args.checkpoint_path)
 else:
  snake_case__					:      Optional[int]				      =  {'''state_dict''': state_dict}
  torch.save(state_dict, args.checkpoint_path)







'''simple docstring'''


import unittest

from parameterized import parameterized

from transformers import OpenLlamaConfig, is_torch_available, set_seed
from transformers.testing_utils import require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
 import torch

 from transformers import OpenLlamaForCausalLM, OpenLlamaForSequenceClassification, OpenLlamaModel






class 					__SCREAMING_SNAKE_CASE						:







  '''simple docstring'''






  def __init__(			self				,		snake_case_				,		snake_case_=1_3				,		snake_case_=7				,		snake_case_=True				,		snake_case_=True				,		snake_case_=False				,		snake_case_=True				,		snake_case_=9_9				,		snake_case_=3_2				,		snake_case_=5				,		snake_case_=4				,		snake_case_=3_7				,		snake_case_="gelu"				,		snake_case_=0.1				,		snake_case_=0.1				,		snake_case_=5_1_2				,		snake_case_=1_6				,		snake_case_=2				,		snake_case_=0.02				,		snake_case_=3				,		snake_case_=4				,		snake_case_=None				,		):




       '''simple docstring'''

       UpperCAmelCase_							:       Union[str, Any]       =				parent
       UpperCAmelCase_							:       List[str]       =				batch_size
       UpperCAmelCase_							:       Dict       =				seq_length
       UpperCAmelCase_							:       Any       =				is_training
       UpperCAmelCase_							:       List[Any]       =				use_input_mask
       UpperCAmelCase_							:       Tuple       =				use_token_type_ids
       UpperCAmelCase_							:       Tuple       =				use_labels
       UpperCAmelCase_							:       Dict       =				vocab_size
       UpperCAmelCase_							:       Union[str, Any]       =				hidden_size
       UpperCAmelCase_							:       Optional[Any]       =				num_hidden_layers
       UpperCAmelCase_							:       Union[str, Any]       =				num_attention_heads
       UpperCAmelCase_							:       Optional[int]       =				intermediate_size
       UpperCAmelCase_							:       Dict       =				hidden_act
       UpperCAmelCase_							:       Any       =				hidden_dropout_prob
       UpperCAmelCase_							:       Optional[Any]       =				attention_probs_dropout_prob
       UpperCAmelCase_							:       Optional[int]       =				max_position_embeddings
       UpperCAmelCase_							:       List[Any]       =				type_vocab_size
       UpperCAmelCase_							:       Optional[Any]       =				type_sequence_label_size
       UpperCAmelCase_							:       List[Any]       =				initializer_range
       UpperCAmelCase_							:       List[Any]       =				num_labels
       UpperCAmelCase_							:       Any       =				num_choices
       UpperCAmelCase_							:       List[str]       =				scope


  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       UpperCAmelCase_							:       List[str]       =				ids_tensor([self.batch_size, self.seq_length]				,		self.vocab_size						)

       UpperCAmelCase_							:       List[str]       =				None
       if self.use_input_mask:
            UpperCAmelCase_							:       Optional[int]       =				random_attention_mask([self.batch_size, self.seq_length]						)

       UpperCAmelCase_							:       int       =				None
       if self.use_token_type_ids:
            UpperCAmelCase_							:       Dict       =				ids_tensor([self.batch_size, self.seq_length]				,		self.type_vocab_size						)

       UpperCAmelCase_							:       Tuple       =				None
       UpperCAmelCase_							:       Any       =				None
       UpperCAmelCase_							:       str       =				None
       if self.use_labels:
            UpperCAmelCase_							:       str       =				ids_tensor([self.batch_size]				,		self.type_sequence_label_size						)
            UpperCAmelCase_							:       Optional[int]       =				ids_tensor([self.batch_size, self.seq_length]				,		self.num_labels						)
            UpperCAmelCase_							:       Any       =				ids_tensor([self.batch_size]				,		self.num_choices						)

       UpperCAmelCase_							:       int       =				self.get_config()

       return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels


  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       return OpenLlamaConfig(
           vocab_size=self.vocab_size				,		hidden_size=self.hidden_size				,		num_hidden_layers=self.num_hidden_layers				,		num_attention_heads=self.num_attention_heads				,		intermediate_size=self.intermediate_size				,		hidden_act=self.hidden_act				,		hidden_dropout_prob=self.hidden_dropout_prob				,		attention_probs_dropout_prob=self.attention_probs_dropout_prob				,		max_position_embeddings=self.max_position_embeddings				,		type_vocab_size=self.type_vocab_size				,		is_decoder=snake_case_				,		initializer_range=self.initializer_range				,		use_stable_embedding=snake_case_				,		)


  def _UpperCamelCase					(			self				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_						):




       '''simple docstring'''

       UpperCAmelCase_							:       str       =				OpenLlamaModel(config=snake_case_						)
       model.to(snake_case_						)
       model.eval()
       UpperCAmelCase_							:       str       =				model(snake_case_				,		attention_mask=snake_case_						)
       UpperCAmelCase_							:       Dict       =				model(snake_case_						)
       self.parent.assertEqual(result.last_hidden_state.shape				,		(self.batch_size, self.seq_length, self.hidden_size)						)


  def _UpperCamelCase					(			self				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		):




       '''simple docstring'''

       UpperCAmelCase_							:       List[Any]       =				True
       UpperCAmelCase_							:       Dict       =				OpenLlamaModel(snake_case_						)
       model.to(snake_case_						)
       model.eval()
       UpperCAmelCase_							:       Optional[Any]       =				model(
           snake_case_				,		attention_mask=snake_case_				,		encoder_hidden_states=snake_case_				,		encoder_attention_mask=snake_case_				,		)
       UpperCAmelCase_							:       Any       =				model(
           snake_case_				,		attention_mask=snake_case_				,		encoder_hidden_states=snake_case_				,		)
       UpperCAmelCase_							:       Optional[Any]       =				model(snake_case_				,		attention_mask=snake_case_						)
       self.parent.assertEqual(result.last_hidden_state.shape				,		(self.batch_size, self.seq_length, self.hidden_size)						)


  def _UpperCamelCase					(			self				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		):




       '''simple docstring'''

       UpperCAmelCase_							:       Optional[Any]       =				OpenLlamaForCausalLM(config=snake_case_						)
       model.to(snake_case_						)
       model.eval()
       UpperCAmelCase_							:       Union[str, Any]       =				model(snake_case_				,		attention_mask=snake_case_				,		labels=snake_case_						)
       self.parent.assertEqual(result.logits.shape				,		(self.batch_size, self.seq_length, self.vocab_size)						)


  def _UpperCamelCase					(			self				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		):




       '''simple docstring'''

       UpperCAmelCase_							:       Optional[Any]       =				True
       UpperCAmelCase_							:       List[Any]       =				True
       UpperCAmelCase_							:       Tuple       =				OpenLlamaForCausalLM(config=snake_case_						)
       model.to(snake_case_						)
       model.eval()

       # first forward pass
       UpperCAmelCase_							:       Tuple       =				model(
           snake_case_				,		attention_mask=snake_case_				,		encoder_hidden_states=snake_case_				,		encoder_attention_mask=snake_case_				,		use_cache=snake_case_				,		)
       UpperCAmelCase_							:       int       =				outputs.past_key_values

       # create hypothetical multiple next token and extent to next_input_ids
       UpperCAmelCase_							:       Optional[int]       =				ids_tensor((self.batch_size, 3)				,		config.vocab_size						)
       UpperCAmelCase_							:       Optional[int]       =				ids_tensor((self.batch_size, 3)				,		vocab_size=2						)

       # append to next input_ids and
       UpperCAmelCase_							:       Union[str, Any]       =				torch.cat([input_ids, next_tokens]				,		dim=-1						)
       UpperCAmelCase_							:       Tuple       =				torch.cat([input_mask, next_mask]				,		dim=-1						)

       UpperCAmelCase_							:       List[Any]       =				model(
           snake_case_				,		attention_mask=snake_case_				,		encoder_hidden_states=snake_case_				,		encoder_attention_mask=snake_case_				,		output_hidden_states=snake_case_				,		)['hidden_states'][0]
       UpperCAmelCase_							:       List[str]       =				model(
           snake_case_				,		attention_mask=snake_case_				,		encoder_hidden_states=snake_case_				,		encoder_attention_mask=snake_case_				,		past_key_values=snake_case_				,		output_hidden_states=snake_case_				,		)['hidden_states'][0]

       # select random slice
       UpperCAmelCase_							:       Union[str, Any]       =				ids_tensor((1,)				,		output_from_past.shape[-1]						).item()
       UpperCAmelCase_							:       List[Any]       =				output_from_no_past[:, -3:, random_slice_idx].detach()
       UpperCAmelCase_							:       Optional[Any]       =				output_from_past[:, :, random_slice_idx].detach()

       self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]						)

       # test that outputs are equal for slice
       self.parent.assertTrue(torch.allclose(snake_case_				,		snake_case_				,		atol=1E-3						)						)






  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       UpperCAmelCase_							:       int       =				self.prepare_config_and_inputs()
       (
           (
           UpperCAmelCase_
       )       ,      (
           UpperCAmelCase_
       )       ,      (
           UpperCAmelCase_
       )       ,      (
           UpperCAmelCase_
       )       ,      (
           UpperCAmelCase_
       )       ,      (
           UpperCAmelCase_
       )       ,      (
           UpperCAmelCase_
       )       ,      
       )							:       Tuple       =				config_and_inputs
       UpperCAmelCase_							:       List[Any]       =				{'input_ids': input_ids, 'attention_mask': input_mask}
       return config, inputs_dict







@require_torch
class 					__SCREAMING_SNAKE_CASE						(	lowerCamelCase_						,     lowerCamelCase_						,     lowerCamelCase_						,     unittest.TestCase	):







  '''simple docstring'''






  lowerCamelCase_		:Tuple						   =		(
      (OpenLlamaModel, OpenLlamaForCausalLM, OpenLlamaForSequenceClassification) if is_torch_available() else ()
  )
  lowerCamelCase_		:Tuple						   =		(OpenLlamaForCausalLM,) if is_torch_available() else ()
  lowerCamelCase_		:Union[str, Any]						   =		(
      {
          '''feature-extraction''': OpenLlamaModel,
          '''text-classification''': OpenLlamaForSequenceClassification,
          '''text-generation''': OpenLlamaForCausalLM,
          '''zero-shot''': OpenLlamaForSequenceClassification,
      }
      if is_torch_available()
      else {}
  )
  lowerCamelCase_		:str						   =		False
  lowerCamelCase_		:Optional[int]						   =		False


  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       UpperCAmelCase_							:       List[str]       =				OpenLlamaModelTester(self						)
       UpperCAmelCase_							:       List[Any]       =				ConfigTester(self				,		config_class=snake_case_				,		hidden_size=3_7						)


  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       self.config_tester.run_common_tests()


  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       UpperCAmelCase_							:       Any       =				self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*snake_case_						)


  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       UpperCAmelCase_							:       Optional[int]       =				self.model_tester.prepare_config_and_inputs()
       for type in ["absolute", "relative_key", "relative_key_query"]:
            UpperCAmelCase_							:       int       =				type
            self.model_tester.create_and_check_model(*snake_case_						)


  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       UpperCAmelCase_       ,      UpperCAmelCase_							:       List[Any]       =				self.model_tester.prepare_config_and_inputs_for_common()
       UpperCAmelCase_							:       Optional[Any]       =				3
       UpperCAmelCase_							:       Union[str, Any]       =				input_dict['input_ids']
       UpperCAmelCase_							:       int       =				input_ids.ne(1						).to(snake_case_						)
       UpperCAmelCase_							:       str       =				ids_tensor([self.model_tester.batch_size]				,		self.model_tester.type_sequence_label_size						)
       UpperCAmelCase_							:       Any       =				OpenLlamaForSequenceClassification(snake_case_						)
       model.to(snake_case_						)
       model.eval()
       UpperCAmelCase_							:       Optional[Any]       =				model(snake_case_				,		attention_mask=snake_case_				,		labels=snake_case_						)
       self.assertEqual(result.logits.shape				,		(self.model_tester.batch_size, self.model_tester.num_labels)						)


  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       UpperCAmelCase_       ,      UpperCAmelCase_							:       Any       =				self.model_tester.prepare_config_and_inputs_for_common()
       UpperCAmelCase_							:       List[str]       =				3
       UpperCAmelCase_							:       str       =				'single_label_classification'
       UpperCAmelCase_							:       List[str]       =				input_dict['input_ids']
       UpperCAmelCase_							:       Optional[Any]       =				input_ids.ne(1						).to(snake_case_						)
       UpperCAmelCase_							:       List[str]       =				ids_tensor([self.model_tester.batch_size]				,		self.model_tester.type_sequence_label_size						)
       UpperCAmelCase_							:       Optional[int]       =				OpenLlamaForSequenceClassification(snake_case_						)
       model.to(snake_case_						)
       model.eval()
       UpperCAmelCase_							:       List[Any]       =				model(snake_case_				,		attention_mask=snake_case_				,		labels=snake_case_						)
       self.assertEqual(result.logits.shape				,		(self.model_tester.batch_size, self.model_tester.num_labels)						)


  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       UpperCAmelCase_       ,      UpperCAmelCase_							:       Optional[int]       =				self.model_tester.prepare_config_and_inputs_for_common()
       UpperCAmelCase_							:       List[str]       =				3
       UpperCAmelCase_							:       int       =				'multi_label_classification'
       UpperCAmelCase_							:       Dict       =				input_dict['input_ids']
       UpperCAmelCase_							:       int       =				input_ids.ne(1						).to(snake_case_						)
       UpperCAmelCase_							:       str       =				ids_tensor(
           [self.model_tester.batch_size, config.num_labels]				,		self.model_tester.type_sequence_label_size						).to(torch.float						)
       UpperCAmelCase_							:       Union[str, Any]       =				OpenLlamaForSequenceClassification(snake_case_						)
       model.to(snake_case_						)
       model.eval()
       UpperCAmelCase_							:       Optional[Any]       =				model(snake_case_				,		attention_mask=snake_case_				,		labels=snake_case_						)
       self.assertEqual(result.logits.shape				,		(self.model_tester.batch_size, self.model_tester.num_labels)						)


  @unittest.skip('Open-Llama buffers include complex numbers, which breaks this test'						)
  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       pass






  @parameterized.expand([('linear',), ('dynamic',)]						)
  def _UpperCamelCase					(			self				,		snake_case_						):




       '''simple docstring'''

       UpperCAmelCase_       ,      UpperCAmelCase_							:       Optional[Any]       =				self.model_tester.prepare_config_and_inputs_for_common()
       UpperCAmelCase_							:       int       =				ids_tensor([1, 1_0]				,		config.vocab_size						)
       UpperCAmelCase_							:       Tuple       =				ids_tensor([1, int(config.max_position_embeddings * 1.5						)]				,		config.vocab_size						)

       set_seed(4_2						)  # Fixed seed at init time so the two models get the same random weights
       UpperCAmelCase_							:       Tuple       =				OpenLlamaModel(snake_case_						)
       original_model.to(snake_case_						)
       original_model.eval()
       UpperCAmelCase_							:       List[str]       =				original_model(snake_case_						).last_hidden_state
       UpperCAmelCase_							:       Any       =				original_model(snake_case_						).last_hidden_state

       set_seed(4_2						)  # Fixed seed at init time so the two models get the same random weights
       UpperCAmelCase_							:       Tuple       =				{'type': scaling_type, 'factor': 10.0}
       UpperCAmelCase_							:       List[str]       =				OpenLlamaModel(snake_case_						)
       scaled_model.to(snake_case_						)
       scaled_model.eval()
       UpperCAmelCase_							:       Tuple       =				scaled_model(snake_case_						).last_hidden_state
       UpperCAmelCase_							:       Optional[int]       =				scaled_model(snake_case_						).last_hidden_state

       # Dynamic scaling does not change the RoPE embeddings until it receives an input longer than the original
       # maximum sequence length, so the outputs for the short input should match.
       if scaling_type == "dynamic":
            self.assertTrue(torch.allclose(snake_case_				,		snake_case_				,		atol=1E-5						)						)
       else:
            self.assertFalse(torch.allclose(snake_case_				,		snake_case_				,		atol=1E-5						)						)

       # The output should be different for long inputs
       self.assertFalse(torch.allclose(snake_case_				,		snake_case_				,		atol=1E-5						)						)






