

"""simple docstring"""






import unittest

from transformers import EsmConfig, is_torch_available
from transformers.testing_utils import TestCasePlus, require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
	import torch

	from transformers import EsmForMaskedLM, EsmForSequenceClassification, EsmForTokenClassification, EsmModel
	from transformers.models.esm.modeling_esm import (
	    ESM_PRETRAINED_MODEL_ARCHIVE_LIST,
	    EsmEmbeddings,
	    create_position_ids_from_input_ids,
	)







class       snake_case__  :







			def __init__(			self			: str			,  lowercase			: Optional[int]			,  lowercase			: Union[str, Any]=13			,  lowercase			: Union[str, Any]=7			,  lowercase			: List[str]=False			,  lowercase			: Optional[Any]=True			,  lowercase			: List[str]=False			,  lowercase			: Any=True			,  lowercase			: Optional[int]=33			,  lowercase			: str=32			,  lowercase			: Any=5			,  lowercase			: Optional[int]=4			,  lowercase			: str=37			,  lowercase			: Optional[Any]="gelu"			,  lowercase			: List[str]=0.1			,  lowercase			: List[Any]=0.1			,  lowercase			: str=5_12			,  lowercase			: List[str]=16			,  lowercase			: int=2			,  lowercase			: Optional[int]=0.0_2			,  lowercase			: Optional[Any]=3			,  lowercase			: Any=4			,  lowercase			: int=None			,  ):





								'''simple docstring'''



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







			def 			__lowerCAmelCase				(			self			: Optional[Any]  ):





								'''simple docstring'''



								UpperCAmelCase			:   Optional[Any]      =				ids_tensor([self.batch_size, self.seq_length]			,  self.vocab_size  )

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

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

								UpperCAmelCase			:   Dict      =				self.get_config()

								return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels







			def 			__lowerCAmelCase				(			self			: Any  ):





								'''simple docstring'''



								return EsmConfig(
								    vocab_size=self.vocab_size			,  hidden_size=self.hidden_size			,  pad_token_id=1			,  num_hidden_layers=self.num_hidden_layers			,  num_attention_heads=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			,  max_position_embeddings=self.max_position_embeddings			,  type_vocab_size=self.type_vocab_size			,  initializer_range=self.initializer_range			,  )







			def 			__lowerCAmelCase				(			self			: Optional[int]			,  lowercase			: Union[str, Any]			,  lowercase			: Union[str, Any]			,  lowercase			: List[str]			,  lowercase			: List[str]			,  lowercase			: List[str]			,  lowercase			: Optional[int]  ):





								'''simple docstring'''



								UpperCAmelCase			:   Tuple      =				EsmModel(config=lowercase  )
								model.to(lowercase  )
								model.eval()
								UpperCAmelCase			:   List[Any]      =				model(lowercase			,  attention_mask=lowercase  )
								UpperCAmelCase			:   List[str]      =				model(lowercase  )
								UpperCAmelCase			:   Optional[int]      =				model(lowercase  )

								self.parent.assertEqual(result.last_hidden_state.shape			,  (self.batch_size, self.seq_length, self.hidden_size)  )
								self.parent.assertEqual(result.pooler_output.shape			,  (self.batch_size, self.hidden_size)  )







			def 			__lowerCAmelCase				(			self			: Tuple			,  lowercase			: Union[str, Any]			,  lowercase			: List[str]			,  lowercase			: Tuple			,  lowercase			: Any			,  lowercase			: Any			,  lowercase			: Optional[Any]  ):





								'''simple docstring'''



								UpperCAmelCase			:   Any      =				EsmForMaskedLM(config=lowercase  )
								model.to(lowercase  )
								model.eval()
								UpperCAmelCase			:   int      =				model(lowercase			,  attention_mask=lowercase			,  labels=lowercase  )
								self.parent.assertEqual(result.logits.shape			,  (self.batch_size, self.seq_length, self.vocab_size)  )







			def 			__lowerCAmelCase				(			self			: Dict			,  lowercase			: List[str]			,  lowercase			: List[Any]			,  lowercase			: Optional[int]			,  lowercase			: Tuple			,  lowercase			: Union[str, Any]			,  lowercase			: Tuple  ):





								'''simple docstring'''



								UpperCAmelCase			:   Tuple      =				self.num_labels
								UpperCAmelCase			:   str      =				EsmForTokenClassification(config=lowercase  )
								model.to(lowercase  )
								model.eval()
								UpperCAmelCase			:   str      =				model(lowercase			,  attention_mask=lowercase			,  labels=lowercase  )
								self.parent.assertEqual(result.logits.shape			,  (self.batch_size, self.seq_length, self.num_labels)  )







			def 			__lowerCAmelCase				(			self			: Any  ):





								'''simple docstring'''



								UpperCAmelCase			:   int      =				self.prepare_config_and_inputs()
								(
								    (
								    UpperCAmelCase
								)      ,							(
								    UpperCAmelCase
								)      ,							(
								    UpperCAmelCase
								)      ,							(
								    UpperCAmelCase
								)      ,							(
								    UpperCAmelCase
								)      ,							(
								    UpperCAmelCase
								)      ,							
								)			:   List[str]      =				config_and_inputs
								UpperCAmelCase			:   List[str]      =				{"input_ids": input_ids, "attention_mask": input_mask}
								return config, inputs_dict







@require_torch
class       snake_case__  ( lowerCAmelCase_					,  lowerCAmelCase_					,  unittest.TestCase     ):
			SCREAMING_SNAKE_CASE__     				=						False

			SCREAMING_SNAKE_CASE__     				=						(
			    (
			        EsmForMaskedLM,
			        EsmModel,
			        EsmForSequenceClassification,
			        EsmForTokenClassification,
			    )
			    if is_torch_available()
			    else ()
			)
			SCREAMING_SNAKE_CASE__     				=						()
			SCREAMING_SNAKE_CASE__     				=						(
			    {
			        '''feature-extraction''': EsmModel,
			        '''fill-mask''': EsmForMaskedLM,
			        '''text-classification''': EsmForSequenceClassification,
			        '''token-classification''': EsmForTokenClassification,
			        '''zero-shot''': EsmForSequenceClassification,
			    }
			    if is_torch_available()
			    else {}
			)
			SCREAMING_SNAKE_CASE__     				=						True







			def 			__lowerCAmelCase				(			self			: str  ):





								'''simple docstring'''



								UpperCAmelCase			:   Tuple      =				EsmModelTester(self  )
								UpperCAmelCase			:   List[str]      =				ConfigTester(self			,  config_class=lowercase			,  hidden_size=37  )







			def 			__lowerCAmelCase				(			self			: Tuple  ):





								'''simple docstring'''



								self.config_tester.run_common_tests()







			def 			__lowerCAmelCase				(			self			: Union[str, Any]  ):





								'''simple docstring'''



								UpperCAmelCase			:   Any      =				self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*lowercase  )







			def 			__lowerCAmelCase				(			self			: str  ):





								'''simple docstring'''



								UpperCAmelCase			:   int      =				self.model_tester.prepare_config_and_inputs()
								for type in ["absolute", "relative_key", "relative_key_query"]:
													UpperCAmelCase			:   int      =				type
													self.model_tester.create_and_check_model(*lowercase  )







			def 			__lowerCAmelCase				(			self			: Optional[Any]  ):





								'''simple docstring'''



								UpperCAmelCase			:   Dict      =				self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_masked_lm(*lowercase  )







			def 			__lowerCAmelCase				(			self			: List[Any]  ):





								'''simple docstring'''



								UpperCAmelCase			:   List[str]      =				self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_token_classification(*lowercase  )







			@slow
			def 			__lowerCAmelCase				(			self			: List[Any]  ):





								'''simple docstring'''



								for model_name in ESM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
													UpperCAmelCase			:   Dict      =				EsmModel.from_pretrained(lowercase  )
													self.assertIsNotNone(lowercase  )







			def 			__lowerCAmelCase				(			self			: Optional[Any]  ):





								'''simple docstring'''



								UpperCAmelCase			:   Optional[Any]      =				self.model_tester.prepare_config_and_inputs()[0]
								UpperCAmelCase			:   int      =				EsmEmbeddings(config=lowercase  )

								UpperCAmelCase			:   Optional[Any]      =				torch.as_tensor([[12, 31, 13, model.padding_idx]]  )
								UpperCAmelCase			:   Optional[int]      =				torch.as_tensor(
								    [
								        [
								            0 + model.padding_idx + 1,
								            1 + model.padding_idx + 1,
								            2 + model.padding_idx + 1,
								            model.padding_idx,
								        ]
								    ]  )
								UpperCAmelCase			:   List[Any]      =				create_position_ids_from_input_ids(lowercase			,  model.padding_idx  )
								self.assertEqual(position_ids.shape			,  expected_positions.shape  )
								self.assertTrue(torch.all(torch.eq(lowercase			,  lowercase  )  )  )







			def 			__lowerCAmelCase				(			self			: List[str]  ):





								'''simple docstring'''



								UpperCAmelCase			:   str      =				self.model_tester.prepare_config_and_inputs()[0]
								UpperCAmelCase			:   Union[str, Any]      =				EsmEmbeddings(config=lowercase  )

								UpperCAmelCase			:   Optional[int]      =				torch.empty(2			,  4			,  30  )
								UpperCAmelCase			:   Optional[int]      =				[
								    0 + embeddings.padding_idx + 1,
								    1 + embeddings.padding_idx + 1,
								    2 + embeddings.padding_idx + 1,
								    3 + embeddings.padding_idx + 1,
								]
								UpperCAmelCase			:   Union[str, Any]      =				torch.as_tensor([expected_single_positions, expected_single_positions]  )
								UpperCAmelCase			:   int      =				embeddings.create_position_ids_from_inputs_embeds(lowercase  )
								self.assertEqual(position_ids.shape			,  expected_positions.shape  )
								self.assertTrue(torch.all(torch.eq(lowercase			,  lowercase  )  )  )







			@unittest.skip("Esm does not support embedding resizing"  )
			def 			__lowerCAmelCase				(			self			: Dict  ):





								'''simple docstring'''



								pass







			@unittest.skip("Esm does not support embedding resizing"  )
			def 			__lowerCAmelCase				(			self			: Optional[Any]  ):





								'''simple docstring'''



								pass







			@unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."  )
			def 			__lowerCAmelCase				(			self			: Tuple  ):





								'''simple docstring'''



								pass







@require_torch
class       snake_case__  ( lowerCAmelCase_     ):







			@slow
			def 			__lowerCAmelCase				(			self			: Optional[Any]  ):





								'''simple docstring'''



								with torch.no_grad():
													UpperCAmelCase			:   Tuple      =				EsmForMaskedLM.from_pretrained("facebook/esm2_t6_8M_UR50D"  )
													model.eval()
													UpperCAmelCase			:   Any      =				torch.tensor([[0, 1, 2, 3, 4, 5]]  )
													UpperCAmelCase			:   List[str]      =				model(lowercase  )[0]

													UpperCAmelCase			:   List[Any]      =				33

													UpperCAmelCase			:   Tuple      =				torch.Size((1, 6, vocab_size)  )
													self.assertEqual(output.shape			,  lowercase  )

													UpperCAmelCase			:   Dict      =				torch.tensor(
													    [[[8.9_2_1_5, -1_0.5_8_9_8, -6.4_6_7_1], [-6.3_9_6_7, -1_3.9_1_1_4, -1.1_2_1_2], [-7.7_8_1_2, -1_3.9_5_1_6, -3.7_4_0_6]]]  )
													self.assertTrue(torch.allclose(output[:, :3, :3]			,  lowercase			,  atol=1E-4  )  )







			@slow
			def 			__lowerCAmelCase				(			self			: int  ):





								'''simple docstring'''



								with torch.no_grad():
													UpperCAmelCase			:   Tuple      =				EsmModel.from_pretrained("facebook/esm2_t6_8M_UR50D"  )
													model.eval()

													UpperCAmelCase			:   Union[str, Any]      =				torch.tensor([[0, 6, 4, 13, 5, 4, 16, 12, 11, 7, 2]]  )
													UpperCAmelCase			:   List[str]      =				model(lowercase  )[0]
													# compare the actual values for a slice.
													UpperCAmelCase			:   List[Any]      =				torch.tensor(
													    [[[0.1_4_4_4, 0.5_4_1_3, 0.3_2_4_8], [0.3_0_3_4, 0.0_0_5_3, 0.3_1_0_8], [0.3_2_2_8, -0.2_4_9_9, 0.3_4_1_5]]]  )
													self.assertTrue(torch.allclose(output[:, :3, :3]			,  lowercase			,  atol=1E-4  )  )





"""simple docstring"""






import pytest

from datasets import Dataset, DatasetDict, Features, NamedSplit, Value
from datasets.io.text import TextDatasetReader

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases



def   lowercase_	(    _lowercase       :		Optional[int]							,      _lowercase       :		int	):




					'''simple docstring'''






					assert isinstance(_lowercase							,      _lowercase	)
					assert dataset.num_rows == 4
					assert dataset.num_columns == 1
					assert dataset.column_names == ["text"]
					for feature, expected_dtype in expected_features.items():
										assert dataset.features[feature].dtype == expected_dtype



@pytest.mark.parametrize("keep_in_memory"							,      [False, True]	)
def   lowercase_	(    _lowercase       :		Union[str, Any]							,      _lowercase       :		Optional[int]							,      _lowercase       :		Dict	):




					'''simple docstring'''






					UpperCAmelCase			:   Dict      =				tmp_path / "cache"
					UpperCAmelCase			:   str      =				{"text": "string"}
					with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
										UpperCAmelCase			:   Tuple      =				TextDatasetReader(_lowercase							,      cache_dir=_lowercase							,      keep_in_memory=_lowercase	).read()
					_check_text_dataset(_lowercase							,      _lowercase	)



@pytest.mark.parametrize(
    "features"							,      [
        None,
        {"text": "string"},
        {"text": "int32"},
        {"text": "float32"},
    ]							,      )
def   lowercase_	(    _lowercase       :		Dict							,      _lowercase       :		Union[str, Any]							,      _lowercase       :		Optional[int]	):




					'''simple docstring'''






					UpperCAmelCase			:   int      =				tmp_path / "cache"
					UpperCAmelCase			:   Any      =				{"text": "string"}
					UpperCAmelCase			:   List[str]      =				features.copy() if features else default_expected_features
					UpperCAmelCase			:   Union[str, Any]      =				(
					    Features({feature: Value(_lowercase	) for feature, dtype in features.items()}	) if features is not None else None
					)
					UpperCAmelCase			:   str      =				TextDatasetReader(_lowercase							,      features=_lowercase							,      cache_dir=_lowercase	).read()
					_check_text_dataset(_lowercase							,      _lowercase	)



@pytest.mark.parametrize("split"							,      [None, NamedSplit("train"	), "train", "test"]	)
def   lowercase_	(    _lowercase       :		int							,      _lowercase       :		Tuple							,      _lowercase       :		List[str]	):




					'''simple docstring'''






					UpperCAmelCase			:   str      =				tmp_path / "cache"
					UpperCAmelCase			:   Any      =				{"text": "string"}
					UpperCAmelCase			:   List[Any]      =				TextDatasetReader(_lowercase							,      cache_dir=_lowercase							,      split=_lowercase	).read()
					_check_text_dataset(_lowercase							,      _lowercase	)
					assert dataset.split == split if split else "train"



@pytest.mark.parametrize("path_type"							,      [str, list]	)
def   lowercase_	(    _lowercase       :		Optional[int]							,      _lowercase       :		Any							,      _lowercase       :		Union[str, Any]	):




					'''simple docstring'''






					if issubclass(_lowercase							,      _lowercase	):
										UpperCAmelCase			:   List[str]      =				text_path
					elif issubclass(_lowercase							,      _lowercase	):
										UpperCAmelCase			:   List[Any]      =				[text_path]
					UpperCAmelCase			:   Union[str, Any]      =				tmp_path / "cache"
					UpperCAmelCase			:   List[Any]      =				{"text": "string"}
					UpperCAmelCase			:   Tuple      =				TextDatasetReader(_lowercase							,      cache_dir=_lowercase	).read()
					_check_text_dataset(_lowercase							,      _lowercase	)



def   lowercase_	(    _lowercase       :		Dict							,      _lowercase       :		int							,      _lowercase       :		Optional[Any]=("train",)	):




					'''simple docstring'''






					assert isinstance(_lowercase							,      _lowercase	)
					for split in splits:
										UpperCAmelCase			:   str      =				dataset_dict[split]
										assert dataset.num_rows == 4
										assert dataset.num_columns == 1
										assert dataset.column_names == ["text"]
										for feature, expected_dtype in expected_features.items():
															assert dataset.features[feature].dtype == expected_dtype



@pytest.mark.parametrize("keep_in_memory"							,      [False, True]	)
def   lowercase_	(    _lowercase       :		Dict							,      _lowercase       :		str							,      _lowercase       :		Dict	):




					'''simple docstring'''






					UpperCAmelCase			:   str      =				tmp_path / "cache"
					UpperCAmelCase			:   Union[str, Any]      =				{"text": "string"}
					with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
										UpperCAmelCase			:   Tuple      =				TextDatasetReader({"train": text_path}							,      cache_dir=_lowercase							,      keep_in_memory=_lowercase	).read()
					_check_text_datasetdict(_lowercase							,      _lowercase	)



@pytest.mark.parametrize(
    "features"							,      [
        None,
        {"text": "string"},
        {"text": "int32"},
        {"text": "float32"},
    ]							,      )
def   lowercase_	(    _lowercase       :		List[Any]							,      _lowercase       :		Union[str, Any]							,      _lowercase       :		int	):




					'''simple docstring'''






					UpperCAmelCase			:   int      =				tmp_path / "cache"
					# CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
					UpperCAmelCase			:   List[str]      =				{"text": "string"}
					UpperCAmelCase			:   str      =				features.copy() if features else default_expected_features
					UpperCAmelCase			:   Optional[int]      =				(
					    Features({feature: Value(_lowercase	) for feature, dtype in features.items()}	) if features is not None else None
					)
					UpperCAmelCase			:   Optional[int]      =				TextDatasetReader({"train": text_path}							,      features=_lowercase							,      cache_dir=_lowercase	).read()
					_check_text_datasetdict(_lowercase							,      _lowercase	)




@pytest.mark.parametrize("split"							,      [None, NamedSplit("train"	), "train", "test"]	)
def   lowercase_	(    _lowercase       :		int							,      _lowercase       :		Tuple							,      _lowercase       :		Optional[Any]	):




					'''simple docstring'''






					if split:
										UpperCAmelCase			:   Optional[Any]      =				{split: text_path}
					else:
										UpperCAmelCase			:   str      =				"train"
										UpperCAmelCase			:   Union[str, Any]      =				{"train": text_path, "test": text_path}
					UpperCAmelCase			:   Any      =				tmp_path / "cache"
					UpperCAmelCase			:   List[Any]      =				{"text": "string"}
					UpperCAmelCase			:   Any      =				TextDatasetReader(_lowercase							,      cache_dir=_lowercase	).read()
					_check_text_datasetdict(_lowercase							,      _lowercase							,      splits=list(path.keys()	)	)
					assert all(dataset[split].split == split for split in path.keys()	)


