

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






class     __magic_name__     :


 def __init__(    self		:			Optional[int]      ,    lowerCAmelCase__		:			Optional[Any]      ,    lowerCAmelCase__		:			Tuple=1_4      ,    lowerCAmelCase__		:			Optional[Any]=7      ,    lowerCAmelCase__		:			Dict=True      ,    lowerCAmelCase__		:			List[Any]=False      ,    lowerCAmelCase__		:			Union[str, Any]=True      ,    lowerCAmelCase__		:			Dict=True      ,    lowerCAmelCase__		:			Dict=True      ,    lowerCAmelCase__		:			Tuple=9_9      ,    lowerCAmelCase__		:			Any=3_2      ,    lowerCAmelCase__		:			Union[str, Any]=5      ,    lowerCAmelCase__		:			Any=3_7      ,    lowerCAmelCase__		:			Any="gelu"      ,    lowerCAmelCase__		:			List[Any]=0.1      ,    lowerCAmelCase__		:			List[str]=0.1      ,    lowerCAmelCase__		:			Any=5_1_2      ,    lowerCAmelCase__		:			Union[str, Any]=1_6      ,    lowerCAmelCase__		:			Any=2      ,    lowerCAmelCase__		:			Union[str, Any]=3      ,    lowerCAmelCase__		:			Optional[int]=4      ,    lowerCAmelCase__		:			List[str]=None      ,    )     ->   Dict:
   UpperCAmelCase          =	parent
   UpperCAmelCase          =	batch_size
   UpperCAmelCase          =	seq_length
   UpperCAmelCase          =	is_training
   UpperCAmelCase          =	use_token_type_ids
   UpperCAmelCase          =	use_input_mask
   UpperCAmelCase          =	use_labels
   UpperCAmelCase          =	use_mc_token_ids
   UpperCAmelCase          =	vocab_size
   UpperCAmelCase          =	hidden_size
   UpperCAmelCase          =	num_hidden_layers
   UpperCAmelCase          =	intermediate_size
   UpperCAmelCase          =	hidden_act
   UpperCAmelCase          =	hidden_dropout_prob
   UpperCAmelCase          =	attention_probs_dropout_prob
   UpperCAmelCase          =	max_position_embeddings
   UpperCAmelCase          =	type_vocab_size
   UpperCAmelCase          =	type_sequence_label_size
   UpperCAmelCase          =	num_labels
   UpperCAmelCase          =	num_choices
   UpperCAmelCase          =	scope
   UpperCAmelCase          =	vocab_size - 1
   UpperCAmelCase          =	vocab_size - 1
   UpperCAmelCase          =	vocab_size - 1


 def _UpperCamelCase						(    self		:			Union[str, Any]     )     ->   Optional[int]:
   return RwkvConfig.from_pretrained("sgugger/rwkv-4-pile-7b"     )


 def _UpperCamelCase						(    self		:			Union[str, Any]      ,    lowerCAmelCase__		:			List[str]=False      ,    lowerCAmelCase__		:			Any=False      ,    lowerCAmelCase__		:			str=False     )     ->   Any:
   UpperCAmelCase          =	ids_tensor([self.batch_size, self.seq_length]      ,    self.vocab_size     )

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

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

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

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

   UpperCAmelCase          =	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 _UpperCamelCase						(    self		:			Dict      ,    lowerCAmelCase__		:			Optional[int]=False      ,    lowerCAmelCase__		:			Any=False      ,    lowerCAmelCase__		:			Dict=False     )     ->   int:
   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 _UpperCamelCase						(    self		:			Optional[Any]     )     ->   int:
   UpperCAmelCase          =	self.get_config()
   UpperCAmelCase          =	3_0_0
   return config


 def _UpperCamelCase						(    self		:			List[str]     )     ->   Dict:
   (
       (
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						
   )          =	self.prepare_config_and_inputs()

   UpperCAmelCase          =	floats_tensor([self.batch_size, self.seq_length, self.hidden_size]     )
   UpperCAmelCase          =	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 _UpperCamelCase						(    self		:			Tuple      ,    lowerCAmelCase__		:			Tuple      ,    lowerCAmelCase__		:			List[str]      ,    lowerCAmelCase__		:			Optional[int]      ,    lowerCAmelCase__		:			Union[str, Any]      ,    lowerCAmelCase__		:			int      ,    *lowerCAmelCase__		:			Dict     )     ->   List[Any]:
   UpperCAmelCase          =	True
   UpperCAmelCase          =	RwkvModel(config=lowerCAmelCase__     )
   model.to(lowerCAmelCase__     )
   model.eval()

   UpperCAmelCase          =	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 _UpperCamelCase						(    self		:			List[str]      ,    lowerCAmelCase__		:			Union[str, Any]      ,    lowerCAmelCase__		:			Optional[int]      ,    lowerCAmelCase__		:			List[Any]      ,    lowerCAmelCase__		:			Union[str, Any]      ,    lowerCAmelCase__		:			Tuple      ,    *lowerCAmelCase__		:			Any     )     ->   Tuple:
   UpperCAmelCase          =	RwkvForCausalLM(lowerCAmelCase__     )
   model.to(lowerCAmelCase__     )
   model.eval()

   UpperCAmelCase          =	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 _UpperCamelCase						(    self		:			List[Any]      ,    lowerCAmelCase__		:			str      ,    lowerCAmelCase__		:			int      ,    lowerCAmelCase__		:			List[str]      ,    lowerCAmelCase__		:			Any      ,    lowerCAmelCase__		:			str      ,    *lowerCAmelCase__		:			List[Any]     )     ->   List[Any]:
   UpperCAmelCase          =	RwkvModel(config=lowerCAmelCase__     )
   model.to(lowerCAmelCase__     )
   model.eval()

   UpperCAmelCase          =	model(lowerCAmelCase__     )
   UpperCAmelCase          =	outputs.last_hidden_state

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

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

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


 def _UpperCamelCase						(    self		:			List[Any]      ,    lowerCAmelCase__		:			str      ,    lowerCAmelCase__		:			Dict      ,    lowerCAmelCase__		:			Any      ,    lowerCAmelCase__		:			str      ,    lowerCAmelCase__		:			Any      ,    *lowerCAmelCase__		:			Optional[int]      ,    lowerCAmelCase__		:			Union[str, Any]=False     )     ->   Any:
   UpperCAmelCase          =	RwkvForCausalLM(lowerCAmelCase__     )
   model.to(lowerCAmelCase__     )
   if gradient_checkpointing:
     model.gradient_checkpointing_enable()

   UpperCAmelCase          =	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 _UpperCamelCase						(    self		:			int     )     ->   List[str]:
   UpperCAmelCase          =	self.prepare_config_and_inputs()

   (
       (
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						(
       UpperCAmelCase
   )   ,						
   )          =	config_and_inputs

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


 def _UpperCamelCase						(    self		:			Optional[int]     )     ->   Union[str, Any]:
   UpperCAmelCase          =	RwkvModelTester(self     )
   UpperCAmelCase          =	ConfigTester(
       self      ,    config_class=lowerCAmelCase__      ,    n_embd=3_7      ,    common_properties=["hidden_size", "num_hidden_layers"]     )


 def _UpperCamelCase						(    self		:			Optional[int]      ,    lowerCAmelCase__		:			Any      ,    lowerCAmelCase__		:			Dict      ,    lowerCAmelCase__		:			List[Any]=None     )     ->   Dict:
   if isinstance(lowerCAmelCase__      ,    torch.Tensor     ):
     UpperCAmelCase   ,						UpperCAmelCase          =	member.max().item(), member.min().item()
   elif isinstance(lowerCAmelCase__      ,    lowerCAmelCase__     ) or isinstance(lowerCAmelCase__      ,    lowerCAmelCase__     ):
     UpperCAmelCase   ,						UpperCAmelCase          =	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"     )

   UpperCAmelCase   ,						UpperCAmelCase          =	container

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

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


 def _UpperCamelCase						(    self		:			int     )     ->   Tuple:
   self.config_tester.run_common_tests()


 def _UpperCamelCase						(    self		:			Optional[int]     )     ->   str:
   UpperCAmelCase          =	self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_rwkv_model(*lowerCAmelCase__     )


 def _UpperCamelCase						(    self		:			int     )     ->   int:
   UpperCAmelCase          =	self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_causl_lm(*lowerCAmelCase__     )


 def _UpperCamelCase						(    self		:			Dict     )     ->   List[str]:
   UpperCAmelCase          =	self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_state_equivalency(*lowerCAmelCase__     )


 def _UpperCamelCase						(    self		:			Dict     )     ->   Union[str, Any]:
   UpperCAmelCase   ,						UpperCAmelCase          =	self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
     UpperCAmelCase          =	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 _UpperCamelCase						(    self		:			Optional[Any]     )     ->   Union[str, Any]:
   UpperCAmelCase   ,						UpperCAmelCase          =	self.model_tester.prepare_config_and_inputs_for_common()
   UpperCAmelCase          =	True

   UpperCAmelCase          =	getattr(self.model_tester      ,    "seq_length"      ,    lowerCAmelCase__     )

   for model_class in self.all_model_classes:
     UpperCAmelCase          =	True
     UpperCAmelCase          =	False
     UpperCAmelCase          =	True
     UpperCAmelCase          =	model_class(lowerCAmelCase__     )
     model.to(lowerCAmelCase__     )
     model.eval()

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

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

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

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

     UpperCAmelCase          =	1
     self.assertEqual(out_len + added_hidden_states      ,    len(lowerCAmelCase__     )     )

     UpperCAmelCase          =	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 _UpperCamelCase						(    self		:			Any     )     ->   Optional[int]:
   for model_name in RWKV_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
     UpperCAmelCase          =	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     __magic_name__     (							unittest.TestCase     ):


 def _UpperCamelCase						(    self		:			int     )     ->   str:
   UpperCAmelCase          =	"RWKV/rwkv-4-169m-pile"
   UpperCAmelCase          =	AutoTokenizer.from_pretrained(self.model_id     )


 def _UpperCamelCase						(    self		:			str     )     ->   Any:
   UpperCAmelCase          =	"Hello my name is Jasmine and I am a newbie to the"
   UpperCAmelCase          =	RwkvForCausalLM.from_pretrained(self.model_id     ).to(lowerCAmelCase__     )

   UpperCAmelCase          =	self.tokenizer("Hello my name is"      ,    return_tensors="pt"     ).input_ids.to(lowerCAmelCase__     )
   UpperCAmelCase          =	model.generate(lowerCAmelCase__      ,    max_new_tokens=1_0     )
   UpperCAmelCase          =	self.tokenizer.decode(output[0].tolist()     )

   self.assertEqual(lowerCAmelCase__      ,    lowerCAmelCase__     )


 def _UpperCamelCase						(    self		:			Optional[int]     )     ->   Any:
   UpperCAmelCase          =	"Hello my name is Jasmine and I am a newbie to the"

   UpperCAmelCase          =	self.tokenizer("Hello my name is"      ,    return_tensors="pt"     ).input_ids.to(lowerCAmelCase__     )
   UpperCAmelCase          =	RwkvForCausalLM.from_pretrained(self.model_id      ,    torch_dtype=torch.bfloataa     ).to(lowerCAmelCase__     )

   UpperCAmelCase          =	model.generate(lowerCAmelCase__      ,    max_new_tokens=1_0     )
   UpperCAmelCase          =	self.tokenizer.decode(output[0].tolist()     )

   self.assertEqual(lowerCAmelCase__      ,    lowerCAmelCase__     )






import matplotlib.pyplot as plt
import numpy as np
def   _lowerCAmelCase(						__A ,							__A ,							__A      ):
  UpperCAmelCase          =	len(__A      )  # Number of training samples
  UpperCAmelCase          =	np.eye(__A      )  # Initialize weights as identity matrix
  for j in range(__A      ):
    UpperCAmelCase          =	point - x_train[j]
    UpperCAmelCase          =	np.exp(diff @ diff.T / (-2.0 * tau**2)      )

  return weights
def   _lowerCAmelCase(						__A ,							__A ,							__A ,							__A      ):
  UpperCAmelCase          =	weight_matrix(__A ,							__A ,							__A      )
  UpperCAmelCase          =	np.linalg.inv(x_train.T @ weight_mat @ x_train      ) @ (
      x_train.T @ weight_mat @ y_train.T
  )

  return weight
def   _lowerCAmelCase(						__A ,							__A ,							__A      ):
  UpperCAmelCase          =	np.zeros(len(__A      )      )  # Initialize array of predictions
  for i, item in enumerate(__A      ):
    UpperCAmelCase          =	item @ local_weight(__A ,							__A ,							__A ,							__A      )

  return y_pred
def   _lowerCAmelCase(						__A ,							__A ,							__A      ):
  import seaborn as sns

  UpperCAmelCase          =	sns.load_dataset(__A      )
  UpperCAmelCase          =	np.array(data[x_name]      )
  UpperCAmelCase          =	np.array(data[y_name]      )

  UpperCAmelCase          =	np.ones(len(__A      )      )

  # pairing elements of one and x_data
  UpperCAmelCase          =	np.column_stack((one, x_data)      )

  return x_train, x_data, y_data
def   _lowerCAmelCase(						__A ,							__A ,							__A ,							__A ,							__A ,							__A ,							):
  UpperCAmelCase          =	np.sort(__A ,							axis=0      )
  plt.scatter(__A ,							__A ,							color="blue"      )
  plt.plot(
      x_train_sorted[:, 1] ,							preds[x_train[:, 1].argsort(0      )] ,							color="yellow" ,							linewidth=5 ,							)
  plt.title("Local Weighted Regression"      )
  plt.xlabel(__A      )
  plt.ylabel(__A      )
  plt.show()


if __name__ == "__main__":
      import doctest

      doctest.testmod()

      # Demo with a dataset from the seaborn module
      lowerCAmelCase__,			lowerCAmelCase__,			lowerCAmelCase__							=			load_data("tips", "total_bill", "tip")
      lowerCAmelCase__							=			local_weight_regression(training_data_x, tip, 5)
      plot_preds(training_data_x, predictions, total_bill, tip, "total_bill", "tip")



