





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							:Optional[Any]							=	False



class  _a  :






							"""simple docstring"""

							def __init__(   self					:  int				,						a					:  Any				,						a					:  Optional[int]=14				,						a					:  Optional[Any]=7				,						a					:  Dict=True				,						a					:  Dict=False				,						a					:  str=True				,						a					:  Dict=True				,						a					:  Tuple=True				,						a					:  str=99				,						a					:  Union[str, Any]=32				,						a					:  Optional[int]=5				,						a					:  Optional[Any]=37				,						a					:  Union[str, Any]="gelu"				,						a					:  Dict=0.1				,						a					:  Any=0.1				,						a					:  Tuple=5_12				,						a					:  Optional[int]=16				,						a					:  str=2				,						a					:  Dict=3				,						a					:  List[Any]=4				,						a					:  str=None				,						)						->Tuple:
													SCREAMING_SNAKE_CASE__  :					Optional[int] =       parent
													SCREAMING_SNAKE_CASE__  :					Dict =       batch_size
													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       seq_length
													SCREAMING_SNAKE_CASE__  :					Dict =       is_training
													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       use_token_type_ids
													SCREAMING_SNAKE_CASE__  :					Dict =       use_input_mask
													SCREAMING_SNAKE_CASE__  :					Optional[int] =       use_labels
													SCREAMING_SNAKE_CASE__  :					List[Any] =       use_mc_token_ids
													SCREAMING_SNAKE_CASE__  :					List[Any] =       vocab_size
													SCREAMING_SNAKE_CASE__  :					str =       hidden_size
													SCREAMING_SNAKE_CASE__  :					Optional[int] =       num_hidden_layers
													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       intermediate_size
													SCREAMING_SNAKE_CASE__  :					Optional[Any] =       hidden_act
													SCREAMING_SNAKE_CASE__  :					Dict =       hidden_dropout_prob
													SCREAMING_SNAKE_CASE__  :					List[str] =       attention_probs_dropout_prob
													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       max_position_embeddings
													SCREAMING_SNAKE_CASE__  :					Optional[Any] =       type_vocab_size
													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       type_sequence_label_size
													SCREAMING_SNAKE_CASE__  :					List[Any] =       num_labels
													SCREAMING_SNAKE_CASE__  :					Tuple =       num_choices
													SCREAMING_SNAKE_CASE__  :					List[str] =       scope
													SCREAMING_SNAKE_CASE__  :					List[Any] =       vocab_size - 1
													SCREAMING_SNAKE_CASE__  :					int =       vocab_size - 1
													SCREAMING_SNAKE_CASE__  :					Any =       vocab_size - 1

							def   A_	(   self					:  int     )						->List[str]:
													return RwkvConfig.from_pretrained("sgugger/rwkv-4-pile-7b"     )

							def   A_	(   self					:  List[str]				,						a					:  Optional[Any]=False				,						a					:  List[str]=False				,						a					:  Optional[Any]=False     )						->List[Any]:
													SCREAMING_SNAKE_CASE__  :					Optional[Any] =       ids_tensor([self.batch_size, self.seq_length]				,						self.vocab_size     )

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

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

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

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

													SCREAMING_SNAKE_CASE__  :					Any =       self.get_config(
													    gradient_checkpointing=UpperCAmelCase__				,						scale_attn_by_inverse_layer_idx=UpperCAmelCase__				,						reorder_and_upcast_attn=UpperCAmelCase__				,						)

													return (
													    config,
													    input_ids,
													    input_mask,
													    None,
													    token_type_ids,
													    mc_token_ids,
													    sequence_labels,
													    token_labels,
													    choice_labels,
													)

							def   A_	(   self					:  List[str]				,						a					:  Any=False				,						a					:  Optional[Any]=False				,						a					:  Optional[Any]=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=UpperCAmelCase__				,						bos_token_id=self.bos_token_id				,						eos_token_id=self.eos_token_id				,						pad_token_id=self.pad_token_id				,						gradient_checkpointing=UpperCAmelCase__				,						scale_attn_by_inverse_layer_idx=UpperCAmelCase__				,						reorder_and_upcast_attn=UpperCAmelCase__				,						)

							def   A_	(   self					:  Dict     )						->str:
													SCREAMING_SNAKE_CASE__  :					Optional[Any] =       self.get_config()
													SCREAMING_SNAKE_CASE__  :					Optional[int] =       3_00
													return config

							def   A_	(   self					:  List[Any]     )						->List[str]:
													(
													    SCREAMING_SNAKE_CASE__       
													)  :					List[Any] =       self.prepare_config_and_inputs()

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

													SCREAMING_SNAKE_CASE__  :					int =       model(UpperCAmelCase__     )

													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					:  Dict				,						a					:  List[Any]				,						a					:  List[str]				,						a					:  int				,						a					:  str				,						a					:  List[str]				,						*a					:  Optional[Any]     )						->Any:
													SCREAMING_SNAKE_CASE__  :					str =       RwkvForCausalLM(UpperCAmelCase__     )
													model.to(UpperCAmelCase__     )
													model.eval()

													SCREAMING_SNAKE_CASE__  :					Dict =       model(UpperCAmelCase__				,						labels=UpperCAmelCase__     )
													self.parent.assertEqual(result.loss.shape				,						()     )
													self.parent.assertEqual(result.logits.shape				,						(self.batch_size, self.seq_length, self.vocab_size)     )

							def   A_	(   self					:  List[str]				,						a					:  Union[str, Any]				,						a					:  List[str]				,						a					:  str				,						a					:  Dict				,						a					:  Tuple				,						*a					:  List[str]     )						->Dict:
													SCREAMING_SNAKE_CASE__  :					Any =       RwkvModel(config=UpperCAmelCase__     )
													model.to(UpperCAmelCase__     )
													model.eval()

													SCREAMING_SNAKE_CASE__  :					Optional[int] =       model(UpperCAmelCase__     )
													SCREAMING_SNAKE_CASE__  :					Optional[int] =       outputs.last_hidden_state

													SCREAMING_SNAKE_CASE__  :					Optional[Any] =       model(input_ids[:, :2]     )
													SCREAMING_SNAKE_CASE__  :					List[str] =       outputs.last_hidden_state

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

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

							def   A_	(   self					:  Optional[int]				,						a					:  Optional[Any]				,						a					:  Optional[int]				,						a					:  Dict				,						a					:  int				,						a					:  Tuple				,						*a					:  Union[str, Any]				,						a					:  Tuple=False     )						->Union[str, Any]:
													SCREAMING_SNAKE_CASE__  :					str =       RwkvForCausalLM(UpperCAmelCase__     )
													model.to(UpperCAmelCase__     )
													if gradient_checkpointing:
																			model.gradient_checkpointing_enable()

													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       model(UpperCAmelCase__				,						labels=UpperCAmelCase__     )
													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					:  int     )						->Optional[int]:
													SCREAMING_SNAKE_CASE__  :					List[Any] =       self.prepare_config_and_inputs()

													(
													    SCREAMING_SNAKE_CASE__       
													)  :					Optional[Any] =       config_and_inputs

													SCREAMING_SNAKE_CASE__  :					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  _a  (					__UpperCamelCase       ,				__UpperCamelCase       ,				__UpperCamelCase       ,				unittest.TestCase				):






							"""simple docstring"""

							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   A_	(   self					:  Any     )						->Optional[Any]:
													SCREAMING_SNAKE_CASE__  :					Dict =       RwkvModelTester(self     )
													SCREAMING_SNAKE_CASE__  :					str =       ConfigTester(
													    self				,						config_class=UpperCAmelCase__				,						n_embd=37				,						common_properties=["hidden_size", "num_hidden_layers"]     )

							def   A_	(   self					:  Any				,						a					:  Any				,						a					:  Optional[Any]				,						a					:  Dict=None     )						->Union[str, Any]:






													if isinstance(UpperCAmelCase__				,						torch.Tensor     ):
																			SCREAMING_SNAKE_CASE__  :					List[str] =       member.max().item(), member.min().item()
													elif isinstance(UpperCAmelCase__				,						UpperCAmelCase__     ) or isinstance(UpperCAmelCase__				,						UpperCAmelCase__     ):
																			SCREAMING_SNAKE_CASE__  :					List[str] =       max(UpperCAmelCase__     ), min(UpperCAmelCase__     )

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

													SCREAMING_SNAKE_CASE__  :					Tuple =       container

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

													if not is_inside_interval:
																			SCREAMING_SNAKE_CASE__  :					List[Any] =       """%s not found in %s""" % (safe_repr(UpperCAmelCase__     ), safe_repr(UpperCAmelCase__     ))
																			self.fail(self._formatMessage(UpperCAmelCase__				,						UpperCAmelCase__     )     )

							def   A_	(   self					:  List[Any]     )						->Any:
													self.config_tester.run_common_tests()

							def   A_	(   self					:  Optional[int]     )						->Any:
													SCREAMING_SNAKE_CASE__  :					int =       self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_rwkv_model(*UpperCAmelCase__     )

							def   A_	(   self					:  Optional[int]     )						->Dict:
													SCREAMING_SNAKE_CASE__  :					Tuple =       self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_causl_lm(*UpperCAmelCase__     )

							def   A_	(   self					:  Union[str, Any]     )						->Optional[Any]:
													SCREAMING_SNAKE_CASE__  :					List[str] =       self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_state_equivalency(*UpperCAmelCase__     )

							def   A_	(   self					:  List[str]     )						->str:
													SCREAMING_SNAKE_CASE__  :					Optional[Any] =       self.model_tester.prepare_config_and_inputs_for_common()

													for model_class in self.all_model_classes:
																			SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       model_class(config=UpperCAmelCase__     )
																			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					:  str     )						->List[str]:
													SCREAMING_SNAKE_CASE__  :					Tuple =       self.model_tester.prepare_config_and_inputs_for_common()
													SCREAMING_SNAKE_CASE__  :					Any =       True

													SCREAMING_SNAKE_CASE__  :					Optional[int] =       getattr(self.model_tester				,						"seq_length"				,						UpperCAmelCase__     )

													for model_class in self.all_model_classes:
																			SCREAMING_SNAKE_CASE__  :					Dict =       True
																			SCREAMING_SNAKE_CASE__  :					int =       False
																			SCREAMING_SNAKE_CASE__  :					Any =       True
																			SCREAMING_SNAKE_CASE__  :					Optional[int] =       model_class(UpperCAmelCase__     )
																			model.to(UpperCAmelCase__     )
																			model.eval()

																			SCREAMING_SNAKE_CASE__  :					Optional[Any] =       self._prepare_for_class(UpperCAmelCase__				,						UpperCAmelCase__     )
																			SCREAMING_SNAKE_CASE__  :					Dict =       inputs["""input_ids"""].shape[0]
																			with torch.no_grad():
																									SCREAMING_SNAKE_CASE__  :					Optional[int] =       model(**UpperCAmelCase__     )
																			SCREAMING_SNAKE_CASE__  :					int =       outputs.attentions
																			self.assertEqual(len(UpperCAmelCase__     )				,						self.model_tester.num_hidden_layers     )

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

																			SCREAMING_SNAKE_CASE__  :					List[str] =       self._prepare_for_class(UpperCAmelCase__				,						UpperCAmelCase__     )
																			SCREAMING_SNAKE_CASE__  :					Optional[int] =       inputs["""input_ids"""].shape[0]
																			with torch.no_grad():
																									SCREAMING_SNAKE_CASE__  :					str =       model(**UpperCAmelCase__     )
																			SCREAMING_SNAKE_CASE__  :					str =       outputs.attentions
																			self.assertEqual(len(UpperCAmelCase__     )				,						self.model_tester.num_hidden_layers     )

																			self.assertListEqual(
																			    list(attentions[0].shape[-3:]     )				,						[batch_size, seq_len, config.hidden_size]				,						)
																			SCREAMING_SNAKE_CASE__  :					Tuple =       len(UpperCAmelCase__     )

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

																			SCREAMING_SNAKE_CASE__  :					Optional[int] =       self._prepare_for_class(UpperCAmelCase__				,						UpperCAmelCase__     )
																			SCREAMING_SNAKE_CASE__  :					List[Any] =       inputs["""input_ids"""].shape[0]
																			with torch.no_grad():
																									SCREAMING_SNAKE_CASE__  :					int =       model(**UpperCAmelCase__     )

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

																			SCREAMING_SNAKE_CASE__  :					List[str] =       outputs.attentions

																			self.assertEqual(len(UpperCAmelCase__     )				,						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					:  Dict     )						->Dict:






													for model_name in RWKV_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																			SCREAMING_SNAKE_CASE__  :					Any =       RwkvModel.from_pretrained(UpperCAmelCase__     )
																			self.assertIsNotNone(UpperCAmelCase__     )







@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				):






							"""simple docstring"""

							def   A_	(   self					:  Optional[Any]     )						->int:
													SCREAMING_SNAKE_CASE__  :					List[str] =       """RWKV/rwkv-4-169m-pile"""
													SCREAMING_SNAKE_CASE__  :					Optional[int] =       AutoTokenizer.from_pretrained(self.model_id     )

							def   A_	(   self					:  List[str]     )						->Optional[Any]:
													SCREAMING_SNAKE_CASE__  :					List[str] =       """Hello my name is Jasmine and I am a newbie to the"""
													SCREAMING_SNAKE_CASE__  :					Any =       RwkvForCausalLM.from_pretrained(self.model_id     ).to(UpperCAmelCase__     )

													SCREAMING_SNAKE_CASE__  :					List[Any] =       self.tokenizer("Hello my name is"				,						return_tensors="pt"     ).input_ids.to(UpperCAmelCase__     )
													SCREAMING_SNAKE_CASE__  :					Any =       model.generate(UpperCAmelCase__				,						max_new_tokens=10     )
													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       self.tokenizer.decode(output[0].tolist()     )

													self.assertEqual(UpperCAmelCase__				,						UpperCAmelCase__     )






							def   A_	(   self					:  List[str]     )						->str:
													SCREAMING_SNAKE_CASE__  :					List[Any] =       """Hello my name is Jasmine and I am a newbie to the"""

													SCREAMING_SNAKE_CASE__  :					Tuple =       self.tokenizer("Hello my name is"				,						return_tensors="pt"     ).input_ids.to(UpperCAmelCase__     )
													SCREAMING_SNAKE_CASE__  :					Optional[Any] =       RwkvForCausalLM.from_pretrained(self.model_id				,						torch_dtype=torch.bfloataa     ).to(UpperCAmelCase__     )

													SCREAMING_SNAKE_CASE__  :					Any =       model.generate(UpperCAmelCase__				,						max_new_tokens=10     )
													SCREAMING_SNAKE_CASE__  :					Tuple =       self.tokenizer.decode(output[0].tolist()     )

													self.assertEqual(UpperCAmelCase__				,						UpperCAmelCase__     )







from datetime import date

import requests

a 			=							'https://api.carbonintensity.org.uk/intensity'
def    a__    (	):
   lowercase__    :						Union[str, Any]       =			requests.get(A       ).json()["""data"""][0]
   return last_half_hour["intensity"]["actual"]
def    a__    (	A		,					A       ):
   return requests.get(F'''{BASE_URL}/{start}/{end}'''       ).json()["data"]


if __name__ == "__main__":
 for entry in fetch_from_to(start=date(2020, 10, 1), end=date(2020, 10, 3)):
  print('from {from} to {to}: {intensity[actual]}'.format(**entry))
 print(F'{fetch_last_half_hour() = }')


