






'''simple docstring'''





import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment






def 				lowerCamelCase__				(_UpperCAmelCase							,       _UpperCAmelCase=()							,       _UpperCAmelCase=None							,       _UpperCAmelCase="no"							,       _UpperCAmelCase="29500"):
			SCREAMING_SNAKE_CASE						= False
			SCREAMING_SNAKE_CASE						= False
			if any(key.startswith('KAGGLE') for key in os.environ.keys()):
						SCREAMING_SNAKE_CASE						= True
			elif "IPython" in sys.modules:
						SCREAMING_SNAKE_CASE						= 'google.colab' in str(sys.modules['IPython'].get_ipython())

			try:
						SCREAMING_SNAKE_CASE						= PrecisionType(mixed_precision.lower())
			except ValueError:
						raise ValueError(
						    F'''Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.''')

			if (in_colab or in_kaggle) and (os.environ.get('TPU_NAME'							,       lowercase__) is not None):
						# TPU launch
						import torch_xla.distributed.xla_multiprocessing as xmp

						if len(AcceleratorState._shared_state) > 0:
									raise ValueError(
									    'To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '
									    'your training function. Restart your notebook and make sure no cells initializes an '
									    '`Accelerator`.')
						if num_processes is None:
									SCREAMING_SNAKE_CASE						= 8

						SCREAMING_SNAKE_CASE						= PrepareForLaunch(lowercase__							,       distributed_type='TPU')
						print(F'''Launching a training on {num_processes} TPU cores.''')
						xmp.spawn(lowercase__							,       args=lowercase__							,       nprocs=lowercase__							,       start_method='fork')
			elif in_colab:
						# No need for a distributed launch otherwise as it's either CPU or one GPU.
						if torch.cuda.is_available():
									print('Launching training on one GPU.')
						else:
									print('Launching training on one CPU.')
						function(*lowercase__)
			else:
						if num_processes is None:
									raise ValueError(
									    'You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.')

						if num_processes > 1:
									# Multi-GPU launch
									from torch.multiprocessing import start_processes
									from torch.multiprocessing.spawn import ProcessRaisedException

									if len(AcceleratorState._shared_state) > 0:
												raise ValueError(
												    'To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '
												    'inside your training function. Restart your notebook and make sure no cells initializes an '
												    '`Accelerator`.')

									if torch.cuda.is_initialized():
												raise ValueError(
												    'To launch a multi-GPU training from your notebook, you need to avoid running any instruction '
												    'using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '
												    'function.')

									# torch.distributed will expect a few environment variable to be here. We set the ones common to each
									# process here (the other ones will be set be the launcher).
									with patch_environment(
									    world_size=lowercase__							,       master_addr='127.0.01'							,       master_port=lowercase__							,       mixed_precision=lowercase__):
												SCREAMING_SNAKE_CASE						= PrepareForLaunch(lowercase__							,       distributed_type='MULTI_GPU')
												print(F'''Launching training on {num_processes} GPUs.''')
												try:
															start_processes(lowercase__							,       args=lowercase__							,       nprocs=lowercase__							,       start_method='fork')
												except ProcessRaisedException as e:
															if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
																		raise RuntimeError(
																		    'CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '
																		    'This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '
																		    'Please review your imports and test them when running the `notebook_launcher()` to identify '
																		    'which one is problematic.') from e

						else:
									# No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
									if is_mps_available():
												SCREAMING_SNAKE_CASE						= '1'
												print('Launching training on MPS.')
									elif torch.cuda.is_available():
												print('Launching training on one GPU.')
									else:
												print('Launching training on CPU.')
									function(*lowercase__)







def 				lowerCamelCase__				(_UpperCAmelCase							,       _UpperCAmelCase=()							,       _UpperCAmelCase=2):
			from torch.multiprocessing import start_processes

			with tempfile.NamedTemporaryFile() as tmp_file:
						# torch.distributed will expect a few environment variable to be here. We set the ones common to each
						# process here (the other ones will be set be the launcher).
						with patch_environment(
						    world_size=lowercase__							,       master_addr='127.0.01'							,       master_port='29500'							,       accelerate_mixed_precision='no'							,       accelerate_debug_rdv_file=tmp_file.name							,       accelerate_use_cpu='yes'							,       ):
									SCREAMING_SNAKE_CASE						= PrepareForLaunch(lowercase__							,       debug=lowercase__)
									start_processes(lowercase__							,       args=lowercase__							,       nprocs=lowercase__							,       start_method='fork')












from __future__ import annotations

import unittest

from transformers import EsmConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
				import numpy
				import tensorflow as tf

				from transformers.models.esm.modeling_tf_esm import (
				    TF_ESM_PRETRAINED_MODEL_ARCHIVE_LIST,
				    TFEsmForMaskedLM,
				    TFEsmForSequenceClassification,
				    TFEsmForTokenClassification,
				    TFEsmModel,
				)

class 		_snake_case  :






			def __init__(       self		,					a		,					)      ->       Tuple:
						SCREAMING_SNAKE_CASE						= parent
						SCREAMING_SNAKE_CASE						= 13
						SCREAMING_SNAKE_CASE						= 7
						SCREAMING_SNAKE_CASE						= True
						SCREAMING_SNAKE_CASE						= True
						SCREAMING_SNAKE_CASE						= True
						SCREAMING_SNAKE_CASE						= 99
						SCREAMING_SNAKE_CASE						= 32
						SCREAMING_SNAKE_CASE						= 2
						SCREAMING_SNAKE_CASE						= 4
						SCREAMING_SNAKE_CASE						= 37
						SCREAMING_SNAKE_CASE						= 'gelu'
						SCREAMING_SNAKE_CASE						= 0.1
						SCREAMING_SNAKE_CASE						= 0.1
						SCREAMING_SNAKE_CASE						= 512
						SCREAMING_SNAKE_CASE						= 16
						SCREAMING_SNAKE_CASE						= 2
						SCREAMING_SNAKE_CASE						= 0.02
						SCREAMING_SNAKE_CASE						= 3
						SCREAMING_SNAKE_CASE						= 4
						SCREAMING_SNAKE_CASE						= None






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       int:
						SCREAMING_SNAKE_CASE						= ids_tensor([self.batch_size, self.seq_length]		,					self.vocab_size)

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

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

						SCREAMING_SNAKE_CASE						= EsmConfig(
						    vocab_size=self.vocab_size		,					hidden_size=self.hidden_size		,					num_hidden_layers=self.num_hidden_layers		,					pad_token_id=1		,					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		,					)

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






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Tuple:
						(
						    (
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						
						)						= self.prepare_config_and_inputs()

						SCREAMING_SNAKE_CASE						= True
						SCREAMING_SNAKE_CASE						= floats_tensor([self.batch_size, self.seq_length, self.hidden_size])
						SCREAMING_SNAKE_CASE						= ids_tensor([self.batch_size, self.seq_length]		,					vocab_size=2)

						return (
						    config,
						    input_ids,
						    input_mask,
						    sequence_labels,
						    token_labels,
						    choice_labels,
						    encoder_hidden_states,
						    encoder_attention_mask,
						)






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a		,					a		,					a		,					a		,					a)      ->       List[str]:
						SCREAMING_SNAKE_CASE						= TFEsmModel(config=a)
						SCREAMING_SNAKE_CASE						= {'input_ids': input_ids, 'attention_mask': input_mask}
						SCREAMING_SNAKE_CASE						= model(a)

						SCREAMING_SNAKE_CASE						= [input_ids, input_mask]
						SCREAMING_SNAKE_CASE						= model(a)

						SCREAMING_SNAKE_CASE						= model(a)

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






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a		,					a		,					a		,					a		,					a		,					a		,					a		,					)      ->       str:
						SCREAMING_SNAKE_CASE						= True

						SCREAMING_SNAKE_CASE						= TFEsmModel(config=a)
						SCREAMING_SNAKE_CASE						= {
						    'input_ids': input_ids,
						    'attention_mask': input_mask,
						    'encoder_hidden_states': encoder_hidden_states,
						    'encoder_attention_mask': encoder_attention_mask,
						}
						SCREAMING_SNAKE_CASE						= model(a)

						SCREAMING_SNAKE_CASE						= [input_ids, input_mask]
						SCREAMING_SNAKE_CASE						= model(a		,					encoder_hidden_states=a)

						# Also check the case where encoder outputs are not passed
						SCREAMING_SNAKE_CASE						= model(a		,					attention_mask=a)

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






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a		,					a		,					a		,					a		,					a)      ->       Optional[int]:
						SCREAMING_SNAKE_CASE						= TFEsmForMaskedLM(config=a)
						SCREAMING_SNAKE_CASE						= model([input_ids, input_mask])
						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)      ->       Dict:
						SCREAMING_SNAKE_CASE						= self.num_labels
						SCREAMING_SNAKE_CASE						= TFEsmForTokenClassification(config=a)
						SCREAMING_SNAKE_CASE						= {'input_ids': input_ids, 'attention_mask': input_mask}
						SCREAMING_SNAKE_CASE						= model(a)
						self.parent.assertEqual(result.logits.shape		,					(self.batch_size, self.seq_length, self.num_labels))







			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Tuple:
						SCREAMING_SNAKE_CASE						= self.prepare_config_and_inputs()
						(
						    (
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						(
						    SCREAMING_SNAKE_CASE
						)				,						
						)						= config_and_inputs
						SCREAMING_SNAKE_CASE						= {'input_ids': input_ids, 'attention_mask': input_mask}
						return config, inputs_dict



@require_tf
class 		_snake_case  (  A__   ,    A__   ,    unittest.TestCase					):
			_lowercase		:    str			     =							(
			    (
			        TFEsmModel,
			        TFEsmForMaskedLM,
			        TFEsmForSequenceClassification,
			        TFEsmForTokenClassification,
			    )
			    if is_tf_available()
			    else ()
			)
			_lowercase		:    int			     =							(
			    {
			        '''feature-extraction''': TFEsmModel,
			        '''fill-mask''': TFEsmForMaskedLM,
			        '''text-classification''': TFEsmForSequenceClassification,
			        '''token-classification''': TFEsmForTokenClassification,
			        '''zero-shot''': TFEsmForSequenceClassification,
			    }
			    if is_tf_available()
			    else {}
			)
			_lowercase		:    str			     =							False
			_lowercase		:    Tuple			     =							False






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Tuple:
						SCREAMING_SNAKE_CASE						= TFEsmModelTester(self)
						SCREAMING_SNAKE_CASE						= ConfigTester(self		,					config_class=a		,					hidden_size=37)






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       str:
						self.config_tester.run_common_tests()






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Optional[int]:
						SCREAMING_SNAKE_CASE						= self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_model(*a)






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Tuple:
						SCREAMING_SNAKE_CASE						= self.model_tester.prepare_config_and_inputs_for_decoder()
						self.model_tester.create_and_check_model_as_decoder(*a)






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Optional[Any]:
						SCREAMING_SNAKE_CASE						= self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_masked_lm(*a)






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Optional[int]:
						SCREAMING_SNAKE_CASE						= self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_token_classification(*a)






			@slow
			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Optional[Any]:
						for model_name in TF_ESM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
									SCREAMING_SNAKE_CASE						= TFEsmModel.from_pretrained(a)
									self.assertIsNotNone(a)






			@unittest.skip('Protein models do not support embedding resizing.')
			def    SCREAMING_SNAKE_CASE__    (       self)      ->       List[Any]:
						pass






			@unittest.skip('Protein models do not support embedding resizing.')
			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Dict:
						pass







			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Any:
						SCREAMING_SNAKE_CASE				,						SCREAMING_SNAKE_CASE						= self.model_tester.prepare_config_and_inputs_for_common()

						for model_class in self.all_model_classes:
									SCREAMING_SNAKE_CASE						= model_class(a)
									assert isinstance(model.get_input_embeddings()		,					tf.keras.layers.Layer)
									if model_class is TFEsmForMaskedLM:
												# Output embedding test differs from the main test because they're a matrix, not a layer
												SCREAMING_SNAKE_CASE						= model.get_bias()
												assert isinstance(a		,					a)
												for k, v in name.items():
															assert isinstance(a		,					tf.Variable)
									else:
												SCREAMING_SNAKE_CASE						= model.get_output_embeddings()
												assert x is None
												SCREAMING_SNAKE_CASE						= model.get_bias()
												assert name is None



@require_tf
class 		_snake_case  (  unittest.TestCase					):






			@slow
			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Optional[Any]:
						SCREAMING_SNAKE_CASE						= TFEsmForMaskedLM.from_pretrained('facebook/esm2_t6_8M_UR50D')

						SCREAMING_SNAKE_CASE						= tf.constant([[0, 1, 2, 3, 4, 5]])
						SCREAMING_SNAKE_CASE						= model(a)[0]
						SCREAMING_SNAKE_CASE						= [1, 6, 33]
						self.assertEqual(list(output.numpy().shape)		,					a)
						# compare the actual values for a slice.
						SCREAMING_SNAKE_CASE						= tf.constant(
						    [
						        [
						            [8.92_15_18, -10.58_98_14, -6.4_67_13_07],
						            [-6.3_96_71_56, -13.91_13_77, -1.1_21_19_15],
						            [-7.78_12_47, -13.95_15_57, -3.74_05_92],
						        ]
						    ])
						self.assertTrue(numpy.allclose(output[:, :3, :3].numpy()		,					expected_slice.numpy()		,					atol=1E-2))







			@slow
			def    SCREAMING_SNAKE_CASE__    (       self)      ->       str:
						SCREAMING_SNAKE_CASE						= TFEsmModel.from_pretrained('facebook/esm2_t6_8M_UR50D')

						SCREAMING_SNAKE_CASE						= tf.constant([[0, 6, 4, 13, 5, 4, 16, 12, 11, 7, 2]])
						SCREAMING_SNAKE_CASE						= model(a)[0]
						# compare the actual values for a slice.
						SCREAMING_SNAKE_CASE						= tf.constant(
						    [
						        [
						            [0.14_44_30_92, 0.54_12_53_27, 0.3_24_77_39],
						            [0.30_34_04_84, 0.00_52_66_76, 0.31_07_77_22],
						            [0.32_27_80_43, -0.24_98_70_96, 0.3_41_46_28],
						        ]
						    ])
						self.assertTrue(numpy.allclose(output[:, :3, :3].numpy()		,					expected_slice.numpy()		,					atol=1E-4))




