

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:
  lowercase_     =       False





class 			__lowerCAmelCase					:




  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=512							,       lowerCAmelCase=16							,       lowerCAmelCase=2							,       lowerCAmelCase=3							,       lowerCAmelCase=4							,       lowerCAmelCase=None							,       )  ->			List[Any]:
        '''simple docstring'''






        _lowercase     	=parent
        _lowercase     	=batch_size
        _lowercase     	=seq_length
        _lowercase     	=is_training
        _lowercase     	=use_token_type_ids
        _lowercase     	=use_input_mask
        _lowercase     	=use_labels
        _lowercase     	=use_mc_token_ids
        _lowercase     	=vocab_size
        _lowercase     	=hidden_size
        _lowercase     	=num_hidden_layers
        _lowercase     	=intermediate_size
        _lowercase     	=hidden_act
        _lowercase     	=hidden_dropout_prob
        _lowercase     	=attention_probs_dropout_prob
        _lowercase     	=max_position_embeddings
        _lowercase     	=type_vocab_size
        _lowercase     	=type_sequence_label_size
        _lowercase     	=num_labels
        _lowercase     	=num_choices
        _lowercase     	=scope
        _lowercase     	=vocab_size - 1
        _lowercase     	=vocab_size - 1
        _lowercase     	=vocab_size - 1




  def   A__					(    self )  ->			str:
        '''simple docstring'''






        return RwkvConfig.from_pretrained('sgugger/rwkv-4-pile-7b' )




  def   A__					(    self							,       lowerCAmelCase=False							,       lowerCAmelCase=False							,       lowerCAmelCase=False )  ->			Optional[int]:
        '''simple docstring'''






        _lowercase     	=ids_tensor([self.batch_size, self.seq_length]							,       self.vocab_size )

        _lowercase     	=None
        if self.use_input_mask:
              _lowercase     	=random_attention_mask([self.batch_size, self.seq_length] )

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

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

        _lowercase     	=None
        _lowercase     	=None
        _lowercase     	=None
        if self.use_labels:
              _lowercase     	=ids_tensor([self.batch_size]							,       self.type_sequence_label_size )
              _lowercase     	=ids_tensor([self.batch_size, self.seq_length]							,       self.num_labels )
              _lowercase     	=ids_tensor([self.batch_size]							,       self.num_choices )

        _lowercase     	=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   A__					(    self							,       lowerCAmelCase=False							,       lowerCAmelCase=False							,       lowerCAmelCase=False )  ->			List[Any]:
        '''simple docstring'''






        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   A__					(    self )  ->			Tuple:
        '''simple docstring'''






        _lowercase     	=self.get_config()
        _lowercase     	=300
        return config




  def   A__					(    self )  ->			Union[str, Any]:
        '''simple docstring'''






        (
            _lowercase			
        )     	=self.prepare_config_and_inputs()

        _lowercase     	=floats_tensor([self.batch_size, self.seq_length, self.hidden_size] )
        _lowercase     	=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   A__					(    self							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       *lowerCAmelCase )  ->			str:
        '''simple docstring'''






        _lowercase     	=True
        _lowercase     	=RwkvModel(config=lowerCAmelCase )
        model.to(lowerCAmelCase )
        model.eval()

        _lowercase     	=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   A__					(    self							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       *lowerCAmelCase )  ->			List[Any]:
        '''simple docstring'''






        _lowercase     	=RwkvForCausalLM(lowerCAmelCase )
        model.to(lowerCAmelCase )
        model.eval()

        _lowercase     	=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   A__					(    self							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       *lowerCAmelCase )  ->			Union[str, Any]:
        '''simple docstring'''






        _lowercase     	=RwkvModel(config=lowerCAmelCase )
        model.to(lowerCAmelCase )
        model.eval()

        _lowercase     	=model(lowerCAmelCase )
        _lowercase     	=outputs.last_hidden_state

        _lowercase     	=model(input_ids[:, :2] )
        _lowercase     	=outputs.last_hidden_state

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

        self.parent.assertTrue(torch.allclose(torch.cat([output_one, output_two]							,       dim=1 )							,       lowerCAmelCase							,       atol=1e-5 ) )




  def   A__					(    self							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase							,       *lowerCAmelCase							,       lowerCAmelCase=False )  ->			int:
        '''simple docstring'''






        _lowercase     	=RwkvForCausalLM(lowerCAmelCase )
        model.to(lowerCAmelCase )
        if gradient_checkpointing:
              model.gradient_checkpointing_enable()

        _lowercase     	=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   A__					(    self )  ->			Tuple:
        '''simple docstring'''






        _lowercase     	=self.prepare_config_and_inputs()

        (
            _lowercase			
        )     	=config_and_inputs

        _lowercase     	={"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 			__lowerCAmelCase					( lowerCamelCase_      ,      lowerCamelCase_      ,      lowerCamelCase_      ,      unittest.TestCase							):
  _a						      =  (RwkvModel, RwkvForCausalLM) if is_torch_available() else ()
  _a						      =  (
      {"""feature-extraction""": RwkvModel, """text-generation""": RwkvForCausalLM} if is_torch_available() else {}
  )
  # all_generative_model_classes = (RwkvForCausalLM,) if is_torch_available() else ()
  _a						      =  False
  _a						      =  False
  _a						      =  False
  _a						      =  False
  _a						      =  False  # Rwkv does not support head masking




  def   A__					(    self )  ->			List[Any]:
        '''simple docstring'''






        _lowercase     	=RwkvModelTester(self )
        _lowercase     	=ConfigTester(
            self							,       config_class=lowerCAmelCase							,       n_embd=37							,       common_properties=['hidden_size', 'num_hidden_layers'] )




  def   A__					(    self							,       lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase=None )  ->			str:
        '''simple docstring'''






        if isinstance(lowerCAmelCase							,       torch.Tensor ):
              _lowercase     	=member.max().item(), member.min().item()
        elif isinstance(lowerCAmelCase							,       lowerCAmelCase ) or isinstance(lowerCAmelCase							,       lowerCAmelCase ):
              _lowercase     	=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' )

        _lowercase     	=container

        _lowercase     	=(min_value >= expected_min) and (max_value <= expected_max)

        if not is_inside_interval:
              _lowercase     	="%s not found in %s" % (safe_repr(lowerCAmelCase ), safe_repr(lowerCAmelCase ))
              self.fail(self._formatMessage(lowerCAmelCase							,       lowerCAmelCase ) )




  def   A__					(    self )  ->			int:
        '''simple docstring'''






        self.config_tester.run_common_tests()




  def   A__					(    self )  ->			Dict:
        '''simple docstring'''






        _lowercase     	=self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_rwkv_model(*lowerCAmelCase )




  def   A__					(    self )  ->			int:
        '''simple docstring'''






        _lowercase     	=self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_causl_lm(*lowerCAmelCase )




  def   A__					(    self )  ->			Dict:
        '''simple docstring'''






        _lowercase     	=self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_state_equivalency(*lowerCAmelCase )




  def   A__					(    self )  ->			Optional[Any]:
        '''simple docstring'''






        _lowercase     	=self.model_tester.prepare_config_and_inputs_for_common()

        for model_class in self.all_model_classes:
              _lowercase     	=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   A__					(    self )  ->			Optional[Any]:
        '''simple docstring'''






        _lowercase     	=self.model_tester.prepare_config_and_inputs_for_common()
        _lowercase     	=True

        _lowercase     	=getattr(self.model_tester							,       'seq_length'							,       lowerCAmelCase )

        for model_class in self.all_model_classes:
              _lowercase     	=True
              _lowercase     	=False
              _lowercase     	=True
              _lowercase     	=model_class(lowerCAmelCase )
              model.to(lowerCAmelCase )
              model.eval()

              _lowercase     	=self._prepare_for_class(lowerCAmelCase							,       lowerCAmelCase )
              _lowercase     	=inputs["input_ids"].shape[0]
              with torch.no_grad():
                    _lowercase     	=model(**lowerCAmelCase )
              _lowercase     	=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"]
              _lowercase     	=True
              _lowercase     	=model_class(lowerCAmelCase )
              model.to(lowerCAmelCase )
              model.eval()

              _lowercase     	=self._prepare_for_class(lowerCAmelCase							,       lowerCAmelCase )
              _lowercase     	=inputs["input_ids"].shape[0]
              with torch.no_grad():
                    _lowercase     	=model(**lowerCAmelCase )
              _lowercase     	=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]							,       )
              _lowercase     	=len(lowerCAmelCase )

              # Check attention is always last and order is fine
              _lowercase     	=True
              _lowercase     	=True
              _lowercase     	=model_class(lowerCAmelCase )
              model.to(lowerCAmelCase )
              model.eval()

              _lowercase     	=self._prepare_for_class(lowerCAmelCase							,       lowerCAmelCase )
              _lowercase     	=inputs["input_ids"].shape[0]
              with torch.no_grad():
                    _lowercase     	=model(**lowerCAmelCase )

              _lowercase     	=1
              self.assertEqual(out_len + added_hidden_states							,       len(lowerCAmelCase ) )

              _lowercase     	=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   A__					(    self )  ->			Optional[Any]:
        '''simple docstring'''






        for model_name in RWKV_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
              _lowercase     	=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 			__lowerCAmelCase					( unittest.TestCase							):




  def   A__					(    self )  ->			Any:
        '''simple docstring'''






        _lowercase     	="RWKV/rwkv-4-169m-pile"
        _lowercase     	=AutoTokenizer.from_pretrained(self.model_id )




  def   A__					(    self )  ->			Tuple:
        '''simple docstring'''






        _lowercase     	="Hello my name is Jasmine and I am a newbie to the"
        _lowercase     	=RwkvForCausalLM.from_pretrained(self.model_id ).to(lowerCAmelCase )

        _lowercase     	=self.tokenizer('Hello my name is'							,       return_tensors='pt' ).input_ids.to(lowerCAmelCase )
        _lowercase     	=model.generate(lowerCAmelCase							,       max_new_tokens=10 )
        _lowercase     	=self.tokenizer.decode(output[0].tolist() )

        self.assertEqual(lowerCAmelCase							,       lowerCAmelCase )




  def   A__					(    self )  ->			Optional[int]:
        '''simple docstring'''






        _lowercase     	="Hello my name is Jasmine and I am a newbie to the"

        _lowercase     	=self.tokenizer('Hello my name is'							,       return_tensors='pt' ).input_ids.to(lowerCAmelCase )
        _lowercase     	=RwkvForCausalLM.from_pretrained(self.model_id							,       torch_dtype=torch.bfloataa ).to(lowerCAmelCase )

        _lowercase     	=model.generate(lowerCAmelCase							,       max_new_tokens=10 )
        _lowercase     	=self.tokenizer.decode(output[0].tolist() )

        self.assertEqual(lowerCAmelCase							,       lowerCAmelCase )





from __future__ import annotations

import random
import string





class 			__lowerCAmelCase					:




  def __init__(    self							,       lowerCAmelCase = None )  ->			None:
        '''simple docstring'''






        _lowercase     	=passcode or self.__passcode_creator()
        _lowercase     	=self.__make_key_list()
        _lowercase     	=self.__make_shift_key()




  def __str__(    self )  ->			str:
        '''simple docstring'''






        return "".join(self.__passcode )




  def   A__					(    self							,       lowerCAmelCase )  ->			list[int]:
        '''simple docstring'''






        for i in range(1							,       len(lowerCAmelCase )							,       2 ):
              iterlist[i] *= -1
        return iterlist




  def   A__					(    self )  ->			list[str]:
        '''simple docstring'''






        _lowercase     	=string.ascii_letters + string.digits
        _lowercase     	=[random.choice(lowerCAmelCase ) for _ in range(random.randint(10							,       20 ) )]
        return password




  def   A__					(    self )  ->			list[str]:
        '''simple docstring'''






        _lowercase     	=(
            string.ascii_letters + string.digits + string.punctuation + ' \t\n'
        )

        _lowercase     	=[]

        # creates points known as breakpoints to break the key_list_options at those
        # points and pivot each substring
        _lowercase     	=sorted(set(self.__passcode ) )
        _lowercase     	=[]

        # algorithm for creating a new shuffled list, keys_l, out of key_list_options
        for i in key_list_options:
              temp_list.extend(lowerCAmelCase )

              # checking breakpoints at which to pivot temporary sublist and add it into
              # keys_l
              if i in breakpoints or i == key_list_options[-1]:
                    keys_l.extend(temp_list[::-1] )
                    temp_list.clear()

        # returning a shuffled keys_l to prevent brute force guessing of shift key
        return keys_l




  def   A__					(    self )  ->			int:
        '''simple docstring'''






        _lowercase     	=sum(self.__neg_pos([ord(lowerCAmelCase ) for x in self.__passcode] ) )
        return num if num > 0 else len(self.__passcode )




  def   A__					(    self							,       lowerCAmelCase )  ->			str:
        '''simple docstring'''






        _lowercase     	=''

        # decoding shift like Caesar cipher algorithm implementing negative shift or
        # reverse shift or left shift
        for i in encoded_message:
              _lowercase     	=self.__key_list.index(lowerCAmelCase )
              decoded_message += self.__key_list[
                  (position - self.__shift_key) % -len(self.__key_list )
              ]

        return decoded_message




  def   A__					(    self							,       lowerCAmelCase )  ->			str:
        '''simple docstring'''






        _lowercase     	=''

        # encoding shift like Caesar cipher algorithm implementing positive shift or
        # forward shift or right shift
        for i in plaintext:
              _lowercase     	=self.__key_list.index(lowerCAmelCase )
              encoded_message += self.__key_list[
                  (position + self.__shift_key) % len(self.__key_list )
              ]

        return encoded_message
def 					a				(      A__     :       str = "Hello, this is a modified Caesar cipher"					)   ->				str:





      """simple docstring"""



      _lowercase     	=ShuffledShiftCipher()
      return cipa.decrypt(cipa.encrypt(A__					)					)


if __name__ == "__main__":
  import doctest

  doctest.testmod()


