from __future__ import annotations

import unittest

from transformers import AutoTokenizer, GPTJConfig, is_tf_available
from transformers.testing_utils import require_tf, slow, tooslow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin
from ...utils.test_modeling_tf_core import TFCoreModelTesterMixin


if is_tf_available():
  import tensorflow as tf

  from transformers.models.gptj.modeling_tf_gptj import (
      TFGPTJForCausalLM,
      TFGPTJForQuestionAnswering,
      TFGPTJForSequenceClassification,
      TFGPTJModel,
      shape_list,
  )






class 							_SCREAMING_SNAKE_CASE:
   def __init__(    self							,     lowerCAmelCase_    )	->Optional[int]:
        a  :    Any				=						parent
        a  :    int				=						13
        a  :    Any				=						7
        a  :    Dict				=						True
        a  :    Optional[Any]				=						True
        a  :    Any				=						True
        a  :    int				=						True
        a  :    Union[str, Any]				=						True
        a  :    Union[str, Any]				=						99
        a  :    List[Any]				=						32
        a  :    List[str]				=						4
        a  :    Optional[Any]				=						2
        a  :    List[Any]				=						4
        a  :    List[Any]				=						37
        a  :    List[Any]				=						'''gelu'''
        a  :    Optional[int]				=						0.1
        a  :    Any				=						0.1
        a  :    Union[str, Any]				=						5_12
        a  :    Union[str, Any]				=						16
        a  :    Any				=						2
        a  :    str				=						0.02
        a  :    str				=						3
        a  :    Any				=						4
        a  :    int				=						None
        a  :    str				=						self.vocab_size - 1
        a  :    int				=						self.vocab_size - 1
        a  :    str				=						self.vocab_size - 1
   def _lowercase(    self    )	->List[Any]:
        a  :    List[str]				=						ids_tensor([self.batch_size, self.seq_length]							,     self.vocab_size    )

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

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

        a  :    List[Any]				=						None
        if self.use_mc_token_ids:
             a  :    Union[str, Any]				=						ids_tensor([self.batch_size, self.num_choices]							,     self.seq_length    )

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

        a  :    Optional[int]				=						GPTJConfig(
            vocab_size=self.vocab_size							,     n_embd=self.hidden_size							,     n_layer=self.num_hidden_layers							,     n_head=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							,     n_positions=self.max_position_embeddings							,     type_vocab_size=self.type_vocab_size							,     initializer_range=self.initializer_range							,     bos_token_id=self.bos_token_id							,     eos_token_id=self.eos_token_id							,     pad_token_id=self.pad_token_id							,     rotary_dim=self.rotary_dim							,     return_dict=lowerCAmelCase_							,     )

        a  :    Dict				=						ids_tensor([self.num_hidden_layers, self.num_attention_heads]							,     2    )

        return (
            config,
            input_ids,
            input_mask,
            head_mask,
            token_type_ids,
            mc_token_ids,
            sequence_labels,
            token_labels,
            choice_labels,
        )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     *lowerCAmelCase_    )	->Dict:
        a  :    Dict				=						TFGPTJModel(config=lowerCAmelCase_    )
        a  :    Any				=						{
            '''input_ids''': input_ids,
            '''attention_mask''': input_mask,
            '''token_type_ids''': token_type_ids,
        }
        a  :    int				=						model(lowerCAmelCase_    )

        a  :    Any				=						[input_ids, None, input_mask]  # None is the input for 'past'
        a  :    Tuple				=						model(lowerCAmelCase_    )

        a  :    Dict				=						model(lowerCAmelCase_    )

        self.parent.assertEqual(result.last_hidden_state.shape							,     (self.batch_size, self.seq_length, self.hidden_size)    )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     *lowerCAmelCase_    )	->Optional[int]:
        a  :    Union[str, Any]				=						TFGPTJModel(config=lowerCAmelCase_    )

        # first forward pass
        a  :    Tuple				=						model(lowerCAmelCase_							,     token_type_ids=lowerCAmelCase_							,     use_cache=lowerCAmelCase_    )
        a  :    Any				=						model(lowerCAmelCase_							,     token_type_ids=lowerCAmelCase_    )
        a  :    Dict				=						model(lowerCAmelCase_							,     token_type_ids=lowerCAmelCase_							,     use_cache=lowerCAmelCase_    )

        self.parent.assertTrue(len(lowerCAmelCase_    ) == len(lowerCAmelCase_    )    )
        self.parent.assertTrue(len(lowerCAmelCase_    ) == len(lowerCAmelCase_    ) + 1    )

        a			,							a  :    List[str]				=						outputs.to_tuple()

        # create hypothetical next token and extent to next_input_ids
        a  :    Optional[int]				=						ids_tensor((self.batch_size, 1)							,     config.vocab_size    )
        a  :    Tuple				=						ids_tensor([self.batch_size, 1]							,     self.type_vocab_size    )

        # append to next input_ids and token_type_ids
        a  :    Optional[int]				=						tf.concat([input_ids, next_tokens]							,     axis=-1    )
        a  :    List[Any]				=						tf.concat([token_type_ids, next_token_types]							,     axis=-1    )

        a  :    Dict				=						model(lowerCAmelCase_							,     token_type_ids=lowerCAmelCase_    )['''last_hidden_state''']
        a  :    List[Any]				=						model(lowerCAmelCase_							,     token_type_ids=lowerCAmelCase_							,     past_key_values=lowerCAmelCase_    )[
            '''last_hidden_state'''
        ]

        # select random slice
        a  :    Optional[Any]				=						int(ids_tensor((1,)							,     shape_list(lowerCAmelCase_    )[-1]    )    )
        a  :    Union[str, Any]				=						output_from_no_past[:, -1, random_slice_idx]
        a  :    List[str]				=						output_from_past[:, 0, random_slice_idx]

        # test that outputs are equal for slice
        tf.debugging.assert_near(lowerCAmelCase_							,     lowerCAmelCase_							,     rtol=1e-6    )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     *lowerCAmelCase_    )	->int:
        a  :    Tuple				=						TFGPTJModel(config=lowerCAmelCase_    )

        # create attention mask
        a  :    Union[str, Any]				=						self.seq_length // 2
        a  :    List[Any]				=						tf.ones((self.batch_size, half_seq_length)							,     dtype=tf.intaa    )
        a  :    str				=						tf.zeros((self.batch_size, self.seq_length - half_seq_length)							,     dtype=tf.intaa    )
        a  :    List[str]				=						tf.concat([attn_mask_begin, attn_mask_end]							,     axis=1    )

        # first forward pass
        a			,							a  :    int				=						model(lowerCAmelCase_							,     attention_mask=lowerCAmelCase_    ).to_tuple()

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

        # change a random masked slice from input_ids
        a  :    Union[str, Any]				=						ids_tensor((1,)							,     lowerCAmelCase_    ).numpy() + 1
        a  :    Any				=						ids_tensor((self.batch_size, self.seq_length)							,     config.vocab_size    )
        a  :    str				=						tf.range(self.seq_length    ) == (self.seq_length - random_seq_idx_to_change)
        a  :    Dict				=						tf.transpose(
            tf.broadcast_to(tf.expand_dims(lowerCAmelCase_							,     -1    )							,     (self.seq_length, self.batch_size)    )    )
        a  :    Tuple				=						tf.where(lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_    )

        # append to next input_ids and attn_mask
        a  :    Any				=						tf.concat([input_ids, next_tokens]							,     axis=-1    )
        a  :    Union[str, Any]				=						tf.concat([attn_mask, tf.ones((shape_list(lowerCAmelCase_    )[0], 1)							,     dtype=tf.intaa    )]							,     axis=1    )

        # get two different outputs
        a  :    str				=						model(lowerCAmelCase_							,     attention_mask=lowerCAmelCase_    )['''last_hidden_state''']
        a  :    Optional[int]				=						model(lowerCAmelCase_							,     past_key_values=lowerCAmelCase_							,     attention_mask=lowerCAmelCase_    )[
            '''last_hidden_state'''
        ]

        # select random slice
        a  :    Any				=						int(ids_tensor((1,)							,     shape_list(lowerCAmelCase_    )[-1]    )    )
        a  :    Dict				=						output_from_no_past[:, -1, random_slice_idx]
        a  :    List[str]				=						output_from_past[:, 0, random_slice_idx]

        # test that outputs are equal for slice
        tf.debugging.assert_near(lowerCAmelCase_							,     lowerCAmelCase_							,     rtol=1e-12    )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     *lowerCAmelCase_    )	->int:
        a  :    List[Any]				=						TFGPTJModel(config=lowerCAmelCase_    )

        a  :    Optional[int]				=						input_ids[:1, :]
        a  :    Dict				=						input_mask[:1, :]
        a  :    Optional[Any]				=						token_type_ids[:1, :]
        a  :    List[str]				=						1

        # first forward pass
        a  :    Tuple				=						model(lowerCAmelCase_							,     attention_mask=lowerCAmelCase_							,     token_type_ids=lowerCAmelCase_							,     use_cache=lowerCAmelCase_    )

        a			,							a  :    Any				=						outputs.to_tuple()

        # create hypothetical next token and extent to next_input_ids
        a  :    Union[str, Any]				=						ids_tensor((self.batch_size, 3)							,     config.vocab_size    )
        a  :    Union[str, Any]				=						ids_tensor((self.batch_size, 3)							,     2    )
        a  :    Optional[int]				=						ids_tensor((self.batch_size, 3)							,     self.type_vocab_size    )

        # append to next input_ids and token_type_ids
        a  :    int				=						tf.concat([input_ids, next_tokens]							,     axis=-1    )
        a  :    Dict				=						tf.concat([input_mask, next_attn_mask]							,     axis=-1    )
        a  :    List[str]				=						tf.concat([token_type_ids, next_token_types]							,     axis=-1    )

        a  :    Union[str, Any]				=						model(
            lowerCAmelCase_							,     token_type_ids=lowerCAmelCase_							,     attention_mask=lowerCAmelCase_    )['''last_hidden_state''']
        a  :    Dict				=						model(
            lowerCAmelCase_							,     token_type_ids=lowerCAmelCase_							,     attention_mask=lowerCAmelCase_							,     past_key_values=lowerCAmelCase_							,     )['''last_hidden_state''']
        self.parent.assertTrue(output_from_past.shape[1] == next_tokens.shape[1]    )

        # select random slice
        a  :    List[Any]				=						int(ids_tensor((1,)							,     shape_list(lowerCAmelCase_    )[-1]    )    )
        a  :    Dict				=						output_from_no_past[:, -3:, random_slice_idx]
        a  :    Optional[int]				=						output_from_past[:, :, random_slice_idx]

        # test that outputs are equal for slice
        tf.debugging.assert_near(lowerCAmelCase_							,     lowerCAmelCase_							,     rtol=1e-3    )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     *lowerCAmelCase_    )	->Any:
        a  :    Optional[Any]				=						TFGPTJForCausalLM(config=lowerCAmelCase_    )
        a  :    Dict				=						{
            '''input_ids''': input_ids,
            '''attention_mask''': input_mask,
            '''token_type_ids''': token_type_ids,
        }
        a  :    Optional[Any]				=						model(lowerCAmelCase_    )
        self.parent.assertEqual(result.logits.shape							,     (self.batch_size, self.seq_length, self.vocab_size)    )


   def _lowercase(    self    )	->List[Any]:
        a  :    int				=						self.prepare_config_and_inputs()

        (
            (
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							
        )  :    Tuple				=						config_and_inputs

        a  :    List[Any]				=						{
            '''input_ids''': input_ids,
            '''token_type_ids''': token_type_ids,
            '''attention_mask''': input_mask,
        }
        return config, inputs_dict






@require_tf
class 							_SCREAMING_SNAKE_CASE(   snake_case__       ,      snake_case__       ,      snake_case__       ,      unittest.TestCase	):
   UpperCamelCase_				:						Optional[int] 			=   (
       (TFGPTJForCausalLM, TFGPTJForSequenceClassification, TFGPTJForQuestionAnswering, TFGPTJModel)
       if is_tf_available()
       else ()
   )

   UpperCamelCase_				:						str 			=   (TFGPTJForCausalLM,) if is_tf_available() else ()
   UpperCamelCase_				:						Dict 			=   (
       {
           'feature-extraction': TFGPTJModel,
           'question-answering': TFGPTJForQuestionAnswering,
           'text-classification': TFGPTJForSequenceClassification,
           'text-generation': TFGPTJForCausalLM,
           'zero-shot': TFGPTJForSequenceClassification,
       }
       if is_tf_available()
       else {}
   )
   UpperCamelCase_				:						Union[str, Any] 			=   False
   UpperCamelCase_				:						List[Any] 			=   False
   UpperCamelCase_				:						List[Any] 			=   False
   UpperCamelCase_				:						str 			=   False
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_    )	->List[str]:
        if (
            pipeline_test_casse_name == "QAPipelineTests"
            and tokenizer_name is not None
            and not tokenizer_name.endswith('''Fast'''    )
        ):
             # `QAPipelineTests` fails for a few models when the slower tokenizer are used.
             # (The slower tokenizers were never used for pipeline tests before the pipeline testing rework)
             # TODO: check (and possibly fix) the `QAPipelineTests` with slower tokenizer
             return True

        return False
   def _lowercase(    self    )	->str:
        a  :    Dict				=						TFGPTJModelTester(self    )
        a  :    List[str]				=						ConfigTester(self							,     config_class=lowerCAmelCase_							,     n_embd=37    )
   def _lowercase(    self    )	->str:
        self.config_tester.run_common_tests()
   def _lowercase(    self    )	->Optional[int]:
        a  :    Optional[Any]				=						self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_gptj_model(*lowerCAmelCase_    )
   def _lowercase(    self    )	->Optional[Any]:
        a  :    Optional[int]				=						self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_gptj_model_past(*lowerCAmelCase_    )
   def _lowercase(    self    )	->Optional[Any]:
        a  :    Union[str, Any]				=						self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_gptj_model_attention_mask_past(*lowerCAmelCase_    )
   def _lowercase(    self    )	->str:
        a  :    Any				=						self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_gptj_model_past_large_inputs(*lowerCAmelCase_    )
   def _lowercase(    self    )	->Dict:
        a  :    Optional[Any]				=						self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_gptj_lm_head_model(*lowerCAmelCase_    )
   @slow
   @unittest.skipIf(
       not is_tf_available() or len(tf.config.list_physical_devices('''GPU'''    )    ) > 0							,     '''skip testing on GPU for now to avoid GPU OOM.'''							,     )
   def _lowercase(    self    )	->Any:
        a  :    Optional[Any]				=						TFGPTJModel.from_pretrained('''EleutherAI/gpt-j-6B'''							,     from_pt=lowerCAmelCase_    )
        self.assertIsNotNone(lowerCAmelCase_    )


   @unittest.skip(reason='''Currently, model embeddings are going to undergo a major refactor.'''    )
   def _lowercase(    self    )	->List[Any]:
        super().test_resize_token_embeddings()






@require_tf
@tooslow
# Marked as @tooslow due to GPU OOM -- but still useful to run locally. Requires ~39GB of RAM.
class 							_SCREAMING_SNAKE_CASE(   unittest.TestCase	):
   def _lowercase(    self    )	->Any:
        a  :    Optional[int]				=						TFGPTJForCausalLM.from_pretrained('''EleutherAI/gpt-j-6B'''							,     from_pt=lowerCAmelCase_    )
        a  :    Optional[Any]				=						tf.convert_to_tensor([[4_64, 32_90]]							,     dtype=tf.intaa    )  # The dog
        # fmt: off
        # The dog is a man's best friend. It is a loyal companion, and it is a friend
        a  :    List[Any]				=						[4_64, 32_90, 3_18, 2_57, 5_82, 3_38, 12_66, 15_45, 13, 6_32, 3_18, 2_57, 91_12, 1_51_85, 11, 2_90, 3_40, 3_18, 2_57, 15_45]
        # fmt: on
        a  :    Any				=						model.generate(lowerCAmelCase_							,     do_sample=lowerCAmelCase_    )
        self.assertListEqual(output_ids[0].numpy().tolist()							,     lowerCAmelCase_    )
   def _lowercase(    self    )	->Optional[Any]:
        a  :    Tuple				=						AutoTokenizer.from_pretrained('''EleutherAI/gpt-j-6B'''							,     revision='''float16'''    )
        a  :    Union[str, Any]				=						TFGPTJForCausalLM.from_pretrained('''EleutherAI/gpt-j-6B'''							,     revision='''float16'''							,     from_pt=lowerCAmelCase_    )

        a  :    Optional[int]				=						tokenizer('''Today is a nice day and'''							,     return_tensors='''tf'''    )
        # forces the generation to happen on CPU, to avoid GPU-related quirks
        with tf.device(''':/CPU:0'''    ):
             a  :    Dict				=						model.generate(**lowerCAmelCase_							,     do_sample=lowerCAmelCase_							,     seed=[42, 0]    )
        a  :    List[str]				=						tokenizer.decode(output_ids[0]							,     skip_special_tokens=lowerCAmelCase_    )

        a  :    Union[str, Any]				=						'''Today is a nice day and I’m going to go for a walk. I’'''
        self.assertEqual(lowerCAmelCase_							,     lowerCAmelCase_    )
   def _lowercase(    self    )	->Optional[int]:
        a  :    Optional[Any]				=						TFGPTJForCausalLM.from_pretrained('''EleutherAI/gpt-j-6B'''							,     revision='''float16'''							,     from_pt=lowerCAmelCase_    )
        a  :    Optional[int]				=						AutoTokenizer.from_pretrained('''EleutherAI/gpt-j-6B'''							,     revision='''float16'''    )

        a  :    List[Any]				=						'''left'''

        # Define PAD Token = EOS Token = 50256
        a  :    Optional[Any]				=						tokenizer.eos_token
        a  :    List[str]				=						model.config.eos_token_id

        # use different length sentences to test batching
        a  :    Any				=						[
            '''Hello, my dog is a little''',
            '''Today, I''',
        ]
        a  :    Tuple				=						[
            '''Hello, my dog is a little over a year old and has been diagnosed with hip dysplasia''',
            '''Today, I’m going to be talking about a topic that’''',
        ]
        return model, tokenizer, sentences, expected_output_sentences
   def _lowercase(    self    )	->Tuple:
        # Confirms that we get the expected results with left-padded beam search
        a			,							a			,							a			,							a  :    Optional[Any]				=						self._get_beam_search_test_objects()

        a  :    Tuple				=						tokenizer(lowerCAmelCase_							,     return_tensors='''tf'''							,     padding=lowerCAmelCase_    )
        a  :    Dict				=						model.generate(**lowerCAmelCase_							,     do_sample=lowerCAmelCase_							,     num_beams=2    )
        a  :    Optional[Any]				=						tokenizer.batch_decode(lowerCAmelCase_							,     skip_special_tokens=lowerCAmelCase_    )
        self.assertListEqual(lowerCAmelCase_							,     lowerCAmelCase_    )
   def _lowercase(    self    )	->str:
        # Confirms that left-padding is working properly
        a			,							a			,							a			,							a  :    Optional[Any]				=						self._get_beam_search_test_objects()

        a  :    Optional[Any]				=						tokenizer(lowerCAmelCase_							,     return_tensors='''tf'''							,     padding=lowerCAmelCase_    )
        a  :    str				=						tokenizer(sentences[0]							,     return_tensors='''tf'''    )
        a  :    str				=						model.generate(**lowerCAmelCase_							,     do_sample=lowerCAmelCase_							,     num_beams=2    )
        a  :    List[Any]				=						(
            shape_list(inputs_non_padded['''input_ids''']    )[-1]
            - tf.reduce_sum(tf.cast(inputs['''attention_mask'''][-1]							,     tf.intaa    )    ).numpy()
        )
        a  :    List[str]				=						tokenizer(sentences[1]							,     return_tensors='''tf'''    )
        a  :    Union[str, Any]				=						model.generate(
            **lowerCAmelCase_							,     do_sample=lowerCAmelCase_							,     num_beams=2							,     max_length=model.config.max_length - num_paddings    )
        a  :    Optional[int]				=						tokenizer.decode(output_non_padded[0]							,     skip_special_tokens=lowerCAmelCase_    )
        a  :    str				=						tokenizer.decode(output_padded[0]							,     skip_special_tokens=lowerCAmelCase_    )
        self.assertListEqual(lowerCAmelCase_							,     [non_padded_sentence, padded_sentence]    )


   def _lowercase(    self    )	->Union[str, Any]:
        # Confirms that XLA is working properly
        a			,							a			,							a			,							a  :    Optional[Any]				=						self._get_beam_search_test_objects()

        a  :    int				=						tokenizer(lowerCAmelCase_							,     return_tensors='''tf'''							,     padding=lowerCAmelCase_    )
        a  :    Tuple				=						tf.function(model.generate							,     jit_compile=lowerCAmelCase_    )
        a  :    Tuple				=						xla_generate(**lowerCAmelCase_							,     do_sample=lowerCAmelCase_							,     num_beams=2    )
        a  :    List[Any]				=						tokenizer.batch_decode(lowerCAmelCase_							,     skip_special_tokens=lowerCAmelCase_    )
        self.assertListEqual(lowerCAmelCase_							,     lowerCAmelCase_    )
import unittest
from unittest.util import safe_repr

from transformers import AutoTokenizer, RwkvConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

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


if is_torch_available():
  import torch

  from transformers import (
      RWKV_PRETRAINED_MODEL_ARCHIVE_LIST,
      RwkvForCausalLM,
      RwkvModel,
  )
  from transformers.pytorch_utils import is_torch_greater_or_equal_than_2_0
else:
  UpperCAmelCase							   =      False






class 							_SCREAMING_SNAKE_CASE:
   def __init__(    self							,     lowerCAmelCase_							,     lowerCAmelCase_=14							,     lowerCAmelCase_=7							,     lowerCAmelCase_=True							,     lowerCAmelCase_=False							,     lowerCAmelCase_=True							,     lowerCAmelCase_=True							,     lowerCAmelCase_=True							,     lowerCAmelCase_=99							,     lowerCAmelCase_=32							,     lowerCAmelCase_=5							,     lowerCAmelCase_=37							,     lowerCAmelCase_="gelu"							,     lowerCAmelCase_=0.1							,     lowerCAmelCase_=0.1							,     lowerCAmelCase_=5_12							,     lowerCAmelCase_=16							,     lowerCAmelCase_=2							,     lowerCAmelCase_=3							,     lowerCAmelCase_=4							,     lowerCAmelCase_=None							,     )	->int:
        a  :    Optional[int]				=						parent
        a  :    Optional[Any]				=						batch_size
        a  :    Union[str, Any]				=						seq_length
        a  :    int				=						is_training
        a  :    List[Any]				=						use_token_type_ids
        a  :    Tuple				=						use_input_mask
        a  :    Optional[int]				=						use_labels
        a  :    List[str]				=						use_mc_token_ids
        a  :    str				=						vocab_size
        a  :    Any				=						hidden_size
        a  :    List[Any]				=						num_hidden_layers
        a  :    Optional[int]				=						intermediate_size
        a  :    Dict				=						hidden_act
        a  :    Tuple				=						hidden_dropout_prob
        a  :    str				=						attention_probs_dropout_prob
        a  :    Optional[Any]				=						max_position_embeddings
        a  :    Optional[int]				=						type_vocab_size
        a  :    Any				=						type_sequence_label_size
        a  :    Any				=						num_labels
        a  :    int				=						num_choices
        a  :    List[str]				=						scope
        a  :    List[Any]				=						vocab_size - 1
        a  :    Optional[Any]				=						vocab_size - 1
        a  :    Optional[Any]				=						vocab_size - 1
   def _lowercase(    self    )	->Optional[int]:
        return RwkvConfig.from_pretrained('''sgugger/rwkv-4-pile-7b'''    )
   def _lowercase(    self							,     lowerCAmelCase_=False							,     lowerCAmelCase_=False							,     lowerCAmelCase_=False    )	->Tuple:
        a  :    int				=						ids_tensor([self.batch_size, self.seq_length]							,     self.vocab_size    )

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

        a  :    Union[str, Any]				=						None
        if self.use_token_type_ids:
             a  :    str				=						ids_tensor([self.batch_size, self.seq_length]							,     self.type_vocab_size    )

        a  :    Optional[Any]				=						None
        if self.use_mc_token_ids:
             a  :    Any				=						ids_tensor([self.batch_size, self.num_choices]							,     self.seq_length    )

        a  :    List[Any]				=						None
        a  :    List[str]				=						None
        a  :    int				=						None
        if self.use_labels:
             a  :    Dict				=						ids_tensor([self.batch_size]							,     self.type_sequence_label_size    )
             a  :    List[str]				=						ids_tensor([self.batch_size, self.seq_length]							,     self.num_labels    )
             a  :    List[str]				=						ids_tensor([self.batch_size]							,     self.num_choices    )

        a  :    Optional[int]				=						self.get_config(
            gradient_checkpointing=lowerCAmelCase_							,     scale_attn_by_inverse_layer_idx=lowerCAmelCase_							,     reorder_and_upcast_attn=lowerCAmelCase_							,     )

        return (
            config,
            input_ids,
            input_mask,
            None,
            token_type_ids,
            mc_token_ids,
            sequence_labels,
            token_labels,
            choice_labels,
        )
   def _lowercase(    self							,     lowerCAmelCase_=False							,     lowerCAmelCase_=False							,     lowerCAmelCase_=False    )	->Dict:
        return RwkvConfig(
            vocab_size=self.vocab_size							,     hidden_size=self.hidden_size							,     num_hidden_layers=self.num_hidden_layers							,     intermediate_size=self.intermediate_size							,     activation_function=self.hidden_act							,     resid_pdrop=self.hidden_dropout_prob							,     attn_pdrop=self.attention_probs_dropout_prob							,     n_positions=self.max_position_embeddings							,     type_vocab_size=self.type_vocab_size							,     use_cache=lowerCAmelCase_							,     bos_token_id=self.bos_token_id							,     eos_token_id=self.eos_token_id							,     pad_token_id=self.pad_token_id							,     gradient_checkpointing=lowerCAmelCase_							,     scale_attn_by_inverse_layer_idx=lowerCAmelCase_							,     reorder_and_upcast_attn=lowerCAmelCase_							,     )
   def _lowercase(    self    )	->Optional[Any]:
        a  :    Optional[Any]				=						self.get_config()
        a  :    Any				=						3_00
        return config
   def _lowercase(    self    )	->int:
        (
            (
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							
        )  :    Any				=						self.prepare_config_and_inputs()

        a  :    str				=						floats_tensor([self.batch_size, self.seq_length, self.hidden_size]    )
        a  :    str				=						ids_tensor([self.batch_size, self.seq_length]							,     vocab_size=2    )

        return (
            config,
            input_ids,
            input_mask,
            head_mask,
            token_type_ids,
            sequence_labels,
            token_labels,
            choice_labels,
            encoder_hidden_states,
            encoder_attention_mask,
        )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     *lowerCAmelCase_    )	->Optional[int]:
        a  :    List[str]				=						True
        a  :    Union[str, Any]				=						RwkvModel(config=lowerCAmelCase_    )
        model.to(lowerCAmelCase_    )
        model.eval()

        a  :    Tuple				=						model(lowerCAmelCase_    )

        self.parent.assertEqual(result.last_hidden_state.shape							,     (self.batch_size, self.seq_length, self.hidden_size)    )
        self.parent.assertEqual(len(result.hidden_states    )							,     config.num_hidden_layers + 1    )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     *lowerCAmelCase_    )	->Optional[int]:
        a  :    List[Any]				=						RwkvForCausalLM(lowerCAmelCase_    )
        model.to(lowerCAmelCase_    )
        model.eval()

        a  :    List[Any]				=						model(lowerCAmelCase_							,     labels=lowerCAmelCase_    )
        self.parent.assertEqual(result.loss.shape							,     ()    )
        self.parent.assertEqual(result.logits.shape							,     (self.batch_size, self.seq_length, self.vocab_size)    )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     *lowerCAmelCase_    )	->Tuple:
        a  :    int				=						RwkvModel(config=lowerCAmelCase_    )
        model.to(lowerCAmelCase_    )
        model.eval()

        a  :    Dict				=						model(lowerCAmelCase_    )
        a  :    Union[str, Any]				=						outputs.last_hidden_state

        a  :    str				=						model(input_ids[:, :2]    )
        a  :    List[str]				=						outputs.last_hidden_state

        # Using the state computed on the first inputs, we will get the same output
        a  :    List[str]				=						model(input_ids[:, 2:]							,     state=outputs.state    )
        a  :    str				=						outputs.last_hidden_state

        self.parent.assertTrue(torch.allclose(torch.cat([output_one, output_two]							,     dim=1    )							,     lowerCAmelCase_							,     atol=1e-5    )    )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_							,     *lowerCAmelCase_							,     lowerCAmelCase_=False    )	->Dict:
        a  :    Dict				=						RwkvForCausalLM(lowerCAmelCase_    )
        model.to(lowerCAmelCase_    )
        if gradient_checkpointing:
             model.gradient_checkpointing_enable()

        a  :    List[Any]				=						model(lowerCAmelCase_							,     labels=lowerCAmelCase_    )
        self.parent.assertEqual(result.loss.shape							,     ()    )
        self.parent.assertEqual(result.logits.shape							,     (self.batch_size, self.seq_length, self.vocab_size)    )
        result.loss.backward()


   def _lowercase(    self    )	->Tuple:
        a  :    str				=						self.prepare_config_and_inputs()

        (
            (
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							(
            a
        )			,							
        )  :    Any				=						config_and_inputs

        a  :    List[str]				=						{'''input_ids''': input_ids}

        return config, inputs_dict






@unittest.skipIf(
    not is_torch_greater_or_equal_than_2_0       ,      reason='See https://github.com/huggingface/transformers/pull/24204'	)
@require_torch
class 							_SCREAMING_SNAKE_CASE(   snake_case__       ,      snake_case__       ,      snake_case__       ,      unittest.TestCase	):
   UpperCamelCase_				:						Dict 			=   (RwkvModel, RwkvForCausalLM) if is_torch_available() else ()
   UpperCamelCase_				:						List[Any] 			=   (
       {'feature-extraction': RwkvModel, 'text-generation': RwkvForCausalLM} if is_torch_available() else {}
   )
   # all_generative_model_classes = (RwkvForCausalLM,) if is_torch_available() else ()
   UpperCamelCase_				:						Optional[Any] 			=   False
   UpperCamelCase_				:						Dict 			=   False
   UpperCamelCase_				:						Optional[int] 			=   False
   UpperCamelCase_				:						int 			=   False
   UpperCamelCase_				:						Optional[int] 			=   False  # Rwkv does not support head masking
   def _lowercase(    self    )	->List[str]:
        a  :    Optional[int]				=						RwkvModelTester(self    )
        a  :    Dict				=						ConfigTester(
            self							,     config_class=lowerCAmelCase_							,     n_embd=37							,     common_properties=['''hidden_size''', '''num_hidden_layers''']    )
   def _lowercase(    self							,     lowerCAmelCase_							,     lowerCAmelCase_							,     lowerCAmelCase_=None    )	->Dict:
        if isinstance(lowerCAmelCase_							,     torch.Tensor    ):
             a			,							a  :    Union[str, Any]				=						member.max().item(), member.min().item()
        elif isinstance(lowerCAmelCase_							,     lowerCAmelCase_    ) or isinstance(lowerCAmelCase_							,     lowerCAmelCase_    ):
             a			,							a  :    Optional[Any]				=						max(lowerCAmelCase_    ), min(lowerCAmelCase_    )

        if not isinstance(lowerCAmelCase_							,     lowerCAmelCase_    ):
             raise TypeError('''container should be a list or tuple'''    )
        elif len(lowerCAmelCase_    ) != 2:
             raise ValueError('''container should have 2 elements'''    )

        a			,							a  :    Optional[int]				=						container

        a  :    Dict				=						(min_value >= expected_min) and (max_value <= expected_max)

        if not is_inside_interval:
             a  :    Tuple				=						'''%s not found in %s''' % (safe_repr(lowerCAmelCase_    ), safe_repr(lowerCAmelCase_    ))
             self.fail(self._formatMessage(lowerCAmelCase_							,     lowerCAmelCase_    )    )
   def _lowercase(    self    )	->str:
        self.config_tester.run_common_tests()
   def _lowercase(    self    )	->Tuple:
        a  :    int				=						self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_rwkv_model(*lowerCAmelCase_    )
   def _lowercase(    self    )	->Dict:
        a  :    Dict				=						self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_causl_lm(*lowerCAmelCase_    )
   def _lowercase(    self    )	->List[str]:
        a  :    str				=						self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_state_equivalency(*lowerCAmelCase_    )
   def _lowercase(    self    )	->Optional[Any]:
        a			,							a  :    str				=						self.model_tester.prepare_config_and_inputs_for_common()

        for model_class in self.all_model_classes:
             a  :    List[str]				=						model_class(config=lowerCAmelCase_    )
             for name, param in model.named_parameters():
                  if "time_decay" in name:
                       if param.requires_grad:
                            self.assertTrue(param.data.max().item() == 3.0    )
                            self.assertTrue(param.data.min().item() == -5.0    )
                  elif "time_first" in name:
                       if param.requires_grad:
                            # check if it's a ones like
                            self.assertTrue(torch.allclose(param.data							,     torch.ones_like(param.data    )							,     atol=1e-5							,     rtol=1e-5    )    )
                  elif any(x in name for x in ['''time_mix_key''', '''time_mix_receptance''']    ):
                       if param.requires_grad:
                            self.assertInterval(
                                param.data							,     [0.0, 1.0]							,     msg=F"""Parameter {name} of model {model_class} seems not properly initialized"""							,     )
                  elif "time_mix_value" in name:
                       if param.requires_grad:
                            self.assertInterval(
                                param.data							,     [0.0, 1.3]							,     msg=F"""Parameter {name} of model {model_class} seems not properly initialized"""							,     )
   def _lowercase(    self    )	->str:
        a			,							a  :    Optional[int]				=						self.model_tester.prepare_config_and_inputs_for_common()
        a  :    List[str]				=						True

        a  :    Optional[int]				=						getattr(self.model_tester							,     '''seq_length'''							,     lowerCAmelCase_    )

        for model_class in self.all_model_classes:
             a  :    Tuple				=						True
             a  :    str				=						False
             a  :    str				=						True
             a  :    Union[str, Any]				=						model_class(lowerCAmelCase_    )
             model.to(lowerCAmelCase_    )
             model.eval()

             a  :    Optional[int]				=						self._prepare_for_class(lowerCAmelCase_							,     lowerCAmelCase_    )
             a  :    Dict				=						inputs['''input_ids'''].shape[0]
             with torch.no_grad():
                  a  :    int				=						model(**lowerCAmelCase_    )
             a  :    str				=						outputs.attentions
             self.assertEqual(len(lowerCAmelCase_    )							,     self.model_tester.num_hidden_layers    )

             # check that output_attentions also work using config
             del inputs_dict["output_attentions"]
             a  :    str				=						True
             a  :    Any				=						model_class(lowerCAmelCase_    )
             model.to(lowerCAmelCase_    )
             model.eval()

             a  :    Tuple				=						self._prepare_for_class(lowerCAmelCase_							,     lowerCAmelCase_    )
             a  :    Tuple				=						inputs['''input_ids'''].shape[0]
             with torch.no_grad():
                  a  :    Any				=						model(**lowerCAmelCase_    )
             a  :    Union[str, Any]				=						outputs.attentions
             self.assertEqual(len(lowerCAmelCase_    )							,     self.model_tester.num_hidden_layers    )

             self.assertListEqual(
                 list(attentions[0].shape[-3:]    )							,     [batch_size, seq_len, config.hidden_size]							,     )
             a  :    Optional[Any]				=						len(lowerCAmelCase_    )

             # Check attention is always last and order is fine
             a  :    List[str]				=						True
             a  :    Union[str, Any]				=						True
             a  :    Tuple				=						model_class(lowerCAmelCase_    )
             model.to(lowerCAmelCase_    )
             model.eval()

             a  :    Optional[Any]				=						self._prepare_for_class(lowerCAmelCase_							,     lowerCAmelCase_    )
             a  :    int				=						inputs['''input_ids'''].shape[0]
             with torch.no_grad():
                  a  :    List[Any]				=						model(**lowerCAmelCase_    )

             a  :    List[str]				=						1
             self.assertEqual(out_len + added_hidden_states							,     len(lowerCAmelCase_    )    )

             a  :    str				=						outputs.attentions

             self.assertEqual(len(lowerCAmelCase_    )							,     self.model_tester.num_hidden_layers    )
             self.assertListEqual(
                 list(self_attentions[0].shape[-3:]    )							,     [batch_size, seq_len, config.hidden_size]							,     )


   @slow
   def _lowercase(    self    )	->str:
        for model_name in RWKV_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
             a  :    List[str]				=						RwkvModel.from_pretrained(lowerCAmelCase_    )
             self.assertIsNotNone(lowerCAmelCase_    )






@unittest.skipIf(
    not is_torch_greater_or_equal_than_2_0       ,      reason='See https://github.com/huggingface/transformers/pull/24204'	)
@slow
class 							_SCREAMING_SNAKE_CASE(   unittest.TestCase	):
   def _lowercase(    self    )	->List[Any]:
        a  :    List[Any]				=						'''RWKV/rwkv-4-169m-pile'''
        a  :    Union[str, Any]				=						AutoTokenizer.from_pretrained(self.model_id    )
   def _lowercase(    self    )	->Tuple:
        a  :    Tuple				=						'''Hello my name is Jasmine and I am a newbie to the'''
        a  :    List[Any]				=						RwkvForCausalLM.from_pretrained(self.model_id    ).to(lowerCAmelCase_    )

        a  :    Optional[int]				=						self.tokenizer('''Hello my name is'''							,     return_tensors='''pt'''    ).input_ids.to(lowerCAmelCase_    )
        a  :    Optional[int]				=						model.generate(lowerCAmelCase_							,     max_new_tokens=10    )
        a  :    Dict				=						self.tokenizer.decode(output[0].tolist()    )

        self.assertEqual(lowerCAmelCase_							,     lowerCAmelCase_    )


   def _lowercase(    self    )	->Union[str, Any]:
        a  :    Any				=						'''Hello my name is Jasmine and I am a newbie to the'''

        a  :    str				=						self.tokenizer('''Hello my name is'''							,     return_tensors='''pt'''    ).input_ids.to(lowerCAmelCase_    )
        a  :    Any				=						RwkvForCausalLM.from_pretrained(self.model_id							,     torch_dtype=torch.bfloataa    ).to(lowerCAmelCase_    )

        a  :    Optional[Any]				=						model.generate(lowerCAmelCase_							,     max_new_tokens=10    )
        a  :    Optional[Any]				=						self.tokenizer.decode(output[0].tolist()    )

        self.assertEqual(lowerCAmelCase_							,     lowerCAmelCase_    )