






'''simple docstring'''






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




class a_     :







   def __init__(self       , __a       , __a=1_4       , __a=7       , __a=True       , __a=False       , __a=True       , __a=True       , __a=True       , __a=9_9       , __a=3_2       , __a=5       , __a=3_7       , __a="gelu"       , __a=0.1       , __a=0.1       , __a=5_1_2       , __a=1_6       , __a=2       , __a=3       , __a=4       , __a=None       , )      ->   str:
      """simple docstring"""






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







   def      SCREAMING_SNAKE_CASE__       (self)      ->   Union[str, Any]:
      """simple docstring"""






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







   def      SCREAMING_SNAKE_CASE__       (self       , __a=False       , __a=False       , __a=False)      ->   Any:
      """simple docstring"""






      __snake_case					:   List[Any]  				=			ids_tensor([self.batch_size, self.seq_length]       , self.vocab_size)

      __snake_case					:   int  				=			None
      if self.use_input_mask:
         __snake_case					:   Tuple  				=			random_attention_mask([self.batch_size, self.seq_length])

      __snake_case					:   Optional[Any]  				=			None
      if self.use_token_type_ids:
         __snake_case					:   List[Any]  				=			ids_tensor([self.batch_size, self.seq_length]       , self.type_vocab_size)

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

      __snake_case					:   str  				=			None
      __snake_case					:   Tuple  				=			None
      __snake_case					:   Tuple  				=			None
      if self.use_labels:
         __snake_case					:   Tuple  				=			ids_tensor([self.batch_size]       , self.type_sequence_label_size)
         __snake_case					:   str  				=			ids_tensor([self.batch_size, self.seq_length]       , self.num_labels)
         __snake_case					:   List[str]  				=			ids_tensor([self.batch_size]       , self.num_choices)

      __snake_case					:   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      SCREAMING_SNAKE_CASE__       (self       , __a=False       , __a=False       , __a=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      SCREAMING_SNAKE_CASE__       (self)      ->   Optional[Any]:
      """simple docstring"""






      __snake_case					:   Tuple  				=			self.get_config()
      __snake_case					:   Optional[Any]  				=			3_0_0
      return config







   def      SCREAMING_SNAKE_CASE__       (self)      ->   Tuple:
      """simple docstring"""






      (
          __snake_case     
      )					:   Dict  				=			self.prepare_config_and_inputs()

      __snake_case					:   Optional[int]  				=			floats_tensor([self.batch_size, self.seq_length, self.hidden_size])
      __snake_case					:   Union[str, Any]  				=			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      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a       , __a       , __a       , *__a)      ->   Union[str, Any]:
      """simple docstring"""






      __snake_case					:   Tuple  				=			True
      __snake_case					:   List[str]  				=			RwkvModel(config=_lowerCAmelCase)
      model.to(_lowerCAmelCase)
      model.eval()

      __snake_case					:   List[Any]  				=			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      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a       , __a       , __a       , *__a)      ->   Optional[Any]:
      """simple docstring"""






      __snake_case					:   Optional[int]  				=			RwkvForCausalLM(_lowerCAmelCase)
      model.to(_lowerCAmelCase)
      model.eval()

      __snake_case					:   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      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a       , __a       , __a       , *__a)      ->   List[Any]:
      """simple docstring"""






      __snake_case					:   Any  				=			RwkvModel(config=_lowerCAmelCase)
      model.to(_lowerCAmelCase)
      model.eval()

      __snake_case					:   List[Any]  				=			model(_lowerCAmelCase)
      __snake_case					:   Union[str, Any]  				=			outputs.last_hidden_state

      __snake_case					:   Optional[int]  				=			model(input_ids[:, :2])
      __snake_case					:   Dict  				=			outputs.last_hidden_state

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

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







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a       , __a       , __a       , *__a       , __a=False)      ->   Any:
      """simple docstring"""






      __snake_case					:   Union[str, Any]  				=			RwkvForCausalLM(_lowerCAmelCase)
      model.to(_lowerCAmelCase)
      if gradient_checkpointing:
         model.gradient_checkpointing_enable()

      __snake_case					:   Tuple  				=			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      SCREAMING_SNAKE_CASE__       (self)      ->   Optional[Any]:
      """simple docstring"""






      __snake_case					:   List[Any]  				=			self.prepare_config_and_inputs()

      (
          __snake_case     
      )					:   str  				=			config_and_inputs

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







   def      SCREAMING_SNAKE_CASE__       (self)      ->   Union[str, Any]:
      """simple docstring"""






      __snake_case					:   Tuple  				=			RwkvModelTester(self)
      __snake_case					:   Union[str, Any]  				=			ConfigTester(
          self       , config_class=_lowerCAmelCase       , n_embd=3_7       , common_properties=['hidden_size', 'num_hidden_layers'])







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a=None)      ->   Tuple:
      """simple docstring"""






      if isinstance(_lowerCAmelCase       , torch.Tensor):
         __snake_case					:   Dict  				=			member.max().item(), member.min().item()
      elif isinstance(_lowerCAmelCase       , _lowerCAmelCase) or isinstance(_lowerCAmelCase       , _lowerCAmelCase):
         __snake_case					:   List[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')

      __snake_case					:   List[str]  				=			container

      __snake_case					:   Optional[Any]  				=			(min_value >= expected_min) and (max_value <= expected_max)

      if not is_inside_interval:
         __snake_case					:   int  				=			"%s not found in %s" % (safe_repr(_lowerCAmelCase), safe_repr(_lowerCAmelCase))
         self.fail(self._formatMessage(_lowerCAmelCase       , _lowerCAmelCase))







   def      SCREAMING_SNAKE_CASE__       (self)      ->   Optional[Any]:
      """simple docstring"""






      self.config_tester.run_common_tests()







   def      SCREAMING_SNAKE_CASE__       (self)      ->   str:
      """simple docstring"""






      __snake_case					:   int  				=			self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_rwkv_model(*_lowerCAmelCase)







   def      SCREAMING_SNAKE_CASE__       (self)      ->   Union[str, Any]:
      """simple docstring"""






      __snake_case					:   Optional[int]  				=			self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_causl_lm(*_lowerCAmelCase)







   def      SCREAMING_SNAKE_CASE__       (self)      ->   Optional[Any]:
      """simple docstring"""






      __snake_case					:   str  				=			self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_state_equivalency(*_lowerCAmelCase)







   def      SCREAMING_SNAKE_CASE__       (self)      ->   Tuple:
      """simple docstring"""






      __snake_case					:   Any  				=			self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
         __snake_case					:   Optional[int]  				=			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      SCREAMING_SNAKE_CASE__       (self)      ->   Optional[Any]:
      """simple docstring"""






      __snake_case					:   int  				=			self.model_tester.prepare_config_and_inputs_for_common()
      __snake_case					:   Optional[int]  				=			True

      __snake_case					:   Optional[int]  				=			getattr(self.model_tester       , 'seq_length'       , _lowerCAmelCase)

      for model_class in self.all_model_classes:
         __snake_case					:   int  				=			True
         __snake_case					:   Tuple  				=			False
         __snake_case					:   int  				=			True
         __snake_case					:   List[Any]  				=			model_class(_lowerCAmelCase)
         model.to(_lowerCAmelCase)
         model.eval()

         __snake_case					:   Any  				=			self._prepare_for_class(_lowerCAmelCase       , _lowerCAmelCase)
         __snake_case					:   Tuple  				=			inputs["input_ids"].shape[0]
         with torch.no_grad():
            __snake_case					:   List[str]  				=			model(**_lowerCAmelCase)
         __snake_case					:   Optional[Any]  				=			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"]
         __snake_case					:   List[Any]  				=			True
         __snake_case					:   str  				=			model_class(_lowerCAmelCase)
         model.to(_lowerCAmelCase)
         model.eval()

         __snake_case					:   Tuple  				=			self._prepare_for_class(_lowerCAmelCase       , _lowerCAmelCase)
         __snake_case					:   Optional[Any]  				=			inputs["input_ids"].shape[0]
         with torch.no_grad():
            __snake_case					:   Dict  				=			model(**_lowerCAmelCase)
         __snake_case					:   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]       , )
         __snake_case					:   str  				=			len(_lowerCAmelCase)

         # Check attention is always last and order is fine
         __snake_case					:   Union[str, Any]  				=			True
         __snake_case					:   Optional[int]  				=			True
         __snake_case					:   str  				=			model_class(_lowerCAmelCase)
         model.to(_lowerCAmelCase)
         model.eval()

         __snake_case					:   Optional[int]  				=			self._prepare_for_class(_lowerCAmelCase       , _lowerCAmelCase)
         __snake_case					:   Union[str, Any]  				=			inputs["input_ids"].shape[0]
         with torch.no_grad():
            __snake_case					:   int  				=			model(**_lowerCAmelCase)

         __snake_case					:   Optional[int]  				=			1
         self.assertEqual(out_len + added_hidden_states       , len(_lowerCAmelCase))

         __snake_case					:   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      SCREAMING_SNAKE_CASE__       (self)      ->   Optional[int]:
      """simple docstring"""






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







   def      SCREAMING_SNAKE_CASE__       (self)      ->   str:
      """simple docstring"""






      __snake_case					:   List[str]  				=			"RWKV/rwkv-4-169m-pile"
      __snake_case					:   Union[str, Any]  				=			AutoTokenizer.from_pretrained(self.model_id)







   def      SCREAMING_SNAKE_CASE__       (self)      ->   str:
      """simple docstring"""






      __snake_case					:   Optional[Any]  				=			"Hello my name is Jasmine and I am a newbie to the"
      __snake_case					:   int  				=			RwkvForCausalLM.from_pretrained(self.model_id).to(_lowerCAmelCase)

      __snake_case					:   List[str]  				=			self.tokenizer('Hello my name is'       , return_tensors='pt').input_ids.to(_lowerCAmelCase)
      __snake_case					:   Dict  				=			model.generate(_lowerCAmelCase       , max_new_tokens=1_0)
      __snake_case					:   List[Any]  				=			self.tokenizer.decode(output[0].tolist())

      self.assertEqual(_lowerCAmelCase       , _lowerCAmelCase)







   def      SCREAMING_SNAKE_CASE__       (self)      ->   Tuple:
      """simple docstring"""






      __snake_case					:   List[str]  				=			"Hello my name is Jasmine and I am a newbie to the"

      __snake_case					:   Optional[int]  				=			self.tokenizer('Hello my name is'       , return_tensors='pt').input_ids.to(_lowerCAmelCase)
      __snake_case					:   Optional[int]  				=			RwkvForCausalLM.from_pretrained(self.model_id       , torch_dtype=torch.bfloataa).to(_lowerCAmelCase)

      __snake_case					:   Dict  				=			model.generate(_lowerCAmelCase       , max_new_tokens=1_0)
      __snake_case					:   List[Any]  				=			self.tokenizer.decode(output[0].tolist())

      self.assertEqual(_lowerCAmelCase       , _lowerCAmelCase)





import os

from apache_beam.io.filesystems import FileSystems
from apache_beam.pipeline import Pipeline

from .logging import get_logger


_UpperCAmelCase						:    List[str]      =  2 << 20  # 2mb
_UpperCAmelCase						:    Optional[int]      =  get_logger(__name__)




class  __lowerCAmelCase							(				lowerCAmelCase):






       def        SCREAMING_SNAKE_CASE     ( self:       str					):
         lowercase     :str   						=    self._options.get_all_options().get("runner"					)
         return runner in [None, "DirectRunner", "PortableRunner"]







def    UpperCAmelCase__	(  lowerCamelCase,      lowerCamelCase,      lowerCamelCase=False ):
  lowercase     :str   						=    FileSystems
  if fs.exists(lowerCamelCase ):
    if force_upload:
      logger.info(F"Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True." )
    else:
      logger.info(F"Remote path already exist: {remote_file_path}. Skipping it as force_upload=False." )
      return
  with fs.create(lowerCamelCase ) as remote_file:
    with open(lowerCamelCase,      "rb" ) as local_file:
      lowercase     :int   						=    local_file.read(lowerCamelCase )
      while chunk:
        remote_file.write(lowerCamelCase )
        lowercase     :List[Any]   						=    local_file.read(lowerCamelCase )







def    UpperCAmelCase__	(  lowerCamelCase,      lowerCamelCase,      lowerCamelCase=False ):
  lowercase     :Union[str, Any]   						=    FileSystems
  if os.path.exists(lowerCamelCase ):
    if force_download:
      logger.info(F"Local path already exist: {remote_file_path}. Overwriting it as force_upload=True." )
    else:
      logger.info(F"Local path already exist: {remote_file_path}. Skipping it as force_upload=False." )
      return
  with fs.open(lowerCamelCase ) as remote_file:
    with open(lowerCamelCase,      "wb" ) as local_file:
      lowercase     :Optional[Any]   						=    remote_file.read(lowerCamelCase )
      while chunk:
        local_file.write(lowerCamelCase )
        lowercase     :Dict   						=    remote_file.read(lowerCamelCase )



