





import unittest

from transformers import DebertaConfig, is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device

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


if is_torch_available():
     import torch

     from transformers import (
         DebertaForMaskedLM,
         DebertaForQuestionAnswering,
         DebertaForSequenceClassification,
         DebertaForTokenClassification,
         DebertaModel,
     )
     from transformers.models.deberta.modeling_deberta import DEBERTA_PRETRAINED_MODEL_ARCHIVE_LIST



class 							_UpperCamelCase					(	lowerCAmelCase_     ):

       def __init__(		self: Union[str, Any]					,   _SCREAMING_SNAKE_CASE: List[str]					,   _SCREAMING_SNAKE_CASE: str=13					,   _SCREAMING_SNAKE_CASE: int=7					,   _SCREAMING_SNAKE_CASE: Optional[int]=True					,   _SCREAMING_SNAKE_CASE: str=True					,   _SCREAMING_SNAKE_CASE: Any=True					,   _SCREAMING_SNAKE_CASE: Union[str, Any]=True					,   _SCREAMING_SNAKE_CASE: Optional[Any]=99					,   _SCREAMING_SNAKE_CASE: str=32					,   _SCREAMING_SNAKE_CASE: List[str]=5					,   _SCREAMING_SNAKE_CASE: List[str]=4					,   _SCREAMING_SNAKE_CASE: Optional[Any]=37					,   _SCREAMING_SNAKE_CASE: str="gelu"					,   _SCREAMING_SNAKE_CASE: List[Any]=0.1					,   _SCREAMING_SNAKE_CASE: List[Any]=0.1					,   _SCREAMING_SNAKE_CASE: Tuple=512					,   _SCREAMING_SNAKE_CASE: Tuple=16					,   _SCREAMING_SNAKE_CASE: str=2					,   _SCREAMING_SNAKE_CASE: List[str]=0.02					,   _SCREAMING_SNAKE_CASE: List[Any]=False					,   _SCREAMING_SNAKE_CASE: Dict=True					,   _SCREAMING_SNAKE_CASE: List[Any]="None"					,   _SCREAMING_SNAKE_CASE: List[Any]=3					,   _SCREAMING_SNAKE_CASE: Any=4					,   _SCREAMING_SNAKE_CASE: Union[str, Any]=None					,   )       -> str:




              """simple docstring"""


              UpperCamelCase_          =  parent
              UpperCamelCase_          =  batch_size
              UpperCamelCase_          =  seq_length
              UpperCamelCase_          =  is_training
              UpperCamelCase_          =  use_input_mask
              UpperCamelCase_          =  use_token_type_ids
              UpperCamelCase_          =  use_labels
              UpperCamelCase_          =  vocab_size
              UpperCamelCase_          =  hidden_size
              UpperCamelCase_          =  num_hidden_layers
              UpperCamelCase_          =  num_attention_heads
              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_          =  initializer_range
              UpperCamelCase_          =  num_labels
              UpperCamelCase_          =  num_choices
              UpperCamelCase_          =  relative_attention
              UpperCamelCase_          =  position_biased_input
              UpperCamelCase_          =  pos_att_type
              UpperCamelCase_          =  scope

       def lowercase (		self: Optional[int]		)       -> List[Any]:




              """simple docstring"""


              UpperCamelCase_          =  ids_tensor([self.batch_size, self.seq_length]					,   self.vocab_size		)

              UpperCamelCase_          =  None
              if self.use_input_mask:
                     UpperCamelCase_          =  ids_tensor([self.batch_size, self.seq_length]					,   vocab_size=2		)

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

              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()

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

       def lowercase (		self: Union[str, Any]		)       -> Optional[Any]:




              """simple docstring"""


              return DebertaConfig(
                  vocab_size=self.vocab_size					,   hidden_size=self.hidden_size					,   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					,   relative_attention=self.relative_attention					,   position_biased_input=self.position_biased_input					,   pos_att_type=self.pos_att_type					,   )

       def lowercase (		self: Optional[int]		)       -> Optional[int]:




              """simple docstring"""


              UpperCamelCase_          =  self.get_config()
              UpperCamelCase_          =  300
              return config

       def lowercase (		self: List[Any]					,   _SCREAMING_SNAKE_CASE: str		)       -> List[Any]:




              """simple docstring"""


              self.parent.assertListEqual(list(result.loss.size()		)					,   []		)

       def lowercase (		self: Optional[Any]					,   _SCREAMING_SNAKE_CASE: Optional[Any]					,   _SCREAMING_SNAKE_CASE: Any					,   _SCREAMING_SNAKE_CASE: List[str]					,   _SCREAMING_SNAKE_CASE: Tuple					,   _SCREAMING_SNAKE_CASE: Optional[Any]					,   _SCREAMING_SNAKE_CASE: List[Any]					,   _SCREAMING_SNAKE_CASE: str		)       -> Tuple:




              """simple docstring"""


              UpperCamelCase_          =  DebertaModel(config=_SCREAMING_SNAKE_CASE		)
              model.to(_SCREAMING_SNAKE_CASE		)
              model.eval()
              UpperCamelCase_          =  model(_SCREAMING_SNAKE_CASE					,   attention_mask=_SCREAMING_SNAKE_CASE					,   token_type_ids=_SCREAMING_SNAKE_CASE		)[0]
              UpperCamelCase_          =  model(_SCREAMING_SNAKE_CASE					,   token_type_ids=_SCREAMING_SNAKE_CASE		)[0]
              UpperCamelCase_          =  model(_SCREAMING_SNAKE_CASE		)[0]

              self.parent.assertListEqual(list(sequence_output.size()		)					,   [self.batch_size, self.seq_length, self.hidden_size]		)

       def lowercase (		self: List[str]					,   _SCREAMING_SNAKE_CASE: List[str]					,   _SCREAMING_SNAKE_CASE: str					,   _SCREAMING_SNAKE_CASE: Any					,   _SCREAMING_SNAKE_CASE: Tuple					,   _SCREAMING_SNAKE_CASE: List[Any]					,   _SCREAMING_SNAKE_CASE: Dict					,   _SCREAMING_SNAKE_CASE: Any		)       -> str:




              """simple docstring"""


              UpperCamelCase_          =  DebertaForMaskedLM(config=_SCREAMING_SNAKE_CASE		)
              model.to(_SCREAMING_SNAKE_CASE		)
              model.eval()
              UpperCamelCase_          =  model(_SCREAMING_SNAKE_CASE					,   attention_mask=_SCREAMING_SNAKE_CASE					,   token_type_ids=_SCREAMING_SNAKE_CASE					,   labels=_SCREAMING_SNAKE_CASE		)

              self.parent.assertEqual(result.logits.shape					,   (self.batch_size, self.seq_length, self.vocab_size)		)

       def lowercase (		self: Tuple					,   _SCREAMING_SNAKE_CASE: Optional[int]					,   _SCREAMING_SNAKE_CASE: int					,   _SCREAMING_SNAKE_CASE: Any					,   _SCREAMING_SNAKE_CASE: str					,   _SCREAMING_SNAKE_CASE: Dict					,   _SCREAMING_SNAKE_CASE: int					,   _SCREAMING_SNAKE_CASE: int		)       -> Any:




              """simple docstring"""


              UpperCamelCase_          =  self.num_labels
              UpperCamelCase_          =  DebertaForSequenceClassification(_SCREAMING_SNAKE_CASE		)
              model.to(_SCREAMING_SNAKE_CASE		)
              model.eval()
              UpperCamelCase_          =  model(_SCREAMING_SNAKE_CASE					,   attention_mask=_SCREAMING_SNAKE_CASE					,   token_type_ids=_SCREAMING_SNAKE_CASE					,   labels=_SCREAMING_SNAKE_CASE		)
              self.parent.assertListEqual(list(result.logits.size()		)					,   [self.batch_size, self.num_labels]		)
              self.check_loss_output(_SCREAMING_SNAKE_CASE		)

       def lowercase (		self: str					,   _SCREAMING_SNAKE_CASE: Optional[Any]					,   _SCREAMING_SNAKE_CASE: Optional[Any]					,   _SCREAMING_SNAKE_CASE: Optional[Any]					,   _SCREAMING_SNAKE_CASE: Dict					,   _SCREAMING_SNAKE_CASE: Any					,   _SCREAMING_SNAKE_CASE: List[Any]					,   _SCREAMING_SNAKE_CASE: List[Any]		)       -> Any:




              """simple docstring"""


              UpperCamelCase_          =  self.num_labels
              UpperCamelCase_          =  DebertaForTokenClassification(config=_SCREAMING_SNAKE_CASE		)
              model.to(_SCREAMING_SNAKE_CASE		)
              model.eval()
              UpperCamelCase_          =  model(_SCREAMING_SNAKE_CASE					,   attention_mask=_SCREAMING_SNAKE_CASE					,   token_type_ids=_SCREAMING_SNAKE_CASE					,   labels=_SCREAMING_SNAKE_CASE		)
              self.parent.assertEqual(result.logits.shape					,   (self.batch_size, self.seq_length, self.num_labels)		)

       def lowercase (		self: List[str]					,   _SCREAMING_SNAKE_CASE: Tuple					,   _SCREAMING_SNAKE_CASE: Dict					,   _SCREAMING_SNAKE_CASE: Any					,   _SCREAMING_SNAKE_CASE: List[str]					,   _SCREAMING_SNAKE_CASE: Union[str, Any]					,   _SCREAMING_SNAKE_CASE: Union[str, Any]					,   _SCREAMING_SNAKE_CASE: int		)       -> List[Any]:




              """simple docstring"""


              UpperCamelCase_          =  DebertaForQuestionAnswering(config=_SCREAMING_SNAKE_CASE		)
              model.to(_SCREAMING_SNAKE_CASE		)
              model.eval()
              UpperCamelCase_          =  model(
                  _SCREAMING_SNAKE_CASE					,   attention_mask=_SCREAMING_SNAKE_CASE					,   token_type_ids=_SCREAMING_SNAKE_CASE					,   start_positions=_SCREAMING_SNAKE_CASE					,   end_positions=_SCREAMING_SNAKE_CASE					,   )
              self.parent.assertEqual(result.start_logits.shape					,   (self.batch_size, self.seq_length)		)
              self.parent.assertEqual(result.end_logits.shape					,   (self.batch_size, self.seq_length)		)



       def lowercase (		self: Optional[int]		)       -> Optional[int]:




              """simple docstring"""


              UpperCamelCase_          =  self.prepare_config_and_inputs()
              (
                  (
                  UpperCamelCase_
              )       , (
                  UpperCamelCase_
              )       , (
                  UpperCamelCase_
              )       , (
                  UpperCamelCase_
              )       , (
                  UpperCamelCase_
              )       , (
                  UpperCamelCase_
              )       , (
                  UpperCamelCase_
              )       , 
              )          =  config_and_inputs
              UpperCamelCase_          =  {"input_ids": input_ids, "token_type_ids": token_type_ids, "attention_mask": input_mask}
              return config, inputs_dict



@require_torch
class 							_UpperCamelCase					(	lowerCAmelCase_  ,						lowerCAmelCase_  ,						unittest.TestCase     ):
       _UpperCamelCase				:					Any         =   (
           (
               DebertaModel,
               DebertaForMaskedLM,
               DebertaForSequenceClassification,
               DebertaForTokenClassification,
               DebertaForQuestionAnswering,
           )
           if is_torch_available()
           else ()
       )
       _UpperCamelCase				:					Optional[Any]         =   (
           {
               '''feature-extraction''': DebertaModel,
               '''fill-mask''': DebertaForMaskedLM,
               '''question-answering''': DebertaForQuestionAnswering,
               '''text-classification''': DebertaForSequenceClassification,
               '''token-classification''': DebertaForTokenClassification,
               '''zero-shot''': DebertaForSequenceClassification,
           }
           if is_torch_available()
           else {}
       )

       _UpperCamelCase				:					str         =   True
       _UpperCamelCase				:					Dict         =   False
       _UpperCamelCase				:					Optional[int]         =   False
       _UpperCamelCase				:					Any         =   False
       _UpperCamelCase				:					Optional[int]         =   False

       def lowercase (		self: Optional[Any]		)       -> Tuple:




              """simple docstring"""


              UpperCamelCase_          =  DebertaModelTester(self		)
              UpperCamelCase_          =  ConfigTester(self					,   config_class=_SCREAMING_SNAKE_CASE					,   hidden_size=37		)

       def lowercase (		self: List[str]		)       -> Tuple:




              """simple docstring"""


              self.config_tester.run_common_tests()

       def lowercase (		self: Union[str, Any]		)       -> int:




              """simple docstring"""


              UpperCamelCase_          =  self.model_tester.prepare_config_and_inputs()
              self.model_tester.create_and_check_deberta_model(*_SCREAMING_SNAKE_CASE		)

       def lowercase (		self: Union[str, Any]		)       -> Union[str, Any]:




              """simple docstring"""


              UpperCamelCase_          =  self.model_tester.prepare_config_and_inputs()
              self.model_tester.create_and_check_deberta_for_sequence_classification(*_SCREAMING_SNAKE_CASE		)

       def lowercase (		self: Tuple		)       -> Union[str, Any]:




              """simple docstring"""


              UpperCamelCase_          =  self.model_tester.prepare_config_and_inputs()
              self.model_tester.create_and_check_deberta_for_masked_lm(*_SCREAMING_SNAKE_CASE		)

       def lowercase (		self: int		)       -> List[Any]:




              """simple docstring"""


              UpperCamelCase_          =  self.model_tester.prepare_config_and_inputs()
              self.model_tester.create_and_check_deberta_for_question_answering(*_SCREAMING_SNAKE_CASE		)

       def lowercase (		self: List[Any]		)       -> Dict:




              """simple docstring"""


              UpperCamelCase_          =  self.model_tester.prepare_config_and_inputs()
              self.model_tester.create_and_check_deberta_for_token_classification(*_SCREAMING_SNAKE_CASE		)



       @slow
       def lowercase (		self: int		)       -> List[str]:




              """simple docstring"""


              for model_name in DEBERTA_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                     UpperCamelCase_          =  DebertaModel.from_pretrained(_SCREAMING_SNAKE_CASE		)
                     self.assertIsNotNone(_SCREAMING_SNAKE_CASE		)



@require_torch
@require_sentencepiece
@require_tokenizers
class 							_UpperCamelCase					(	unittest.TestCase     ):

       @unittest.skip(reason="Model not available yet"		)
       def lowercase (		self: Optional[Any]		)       -> Optional[int]:




              """simple docstring"""


              pass



       @slow
       def lowercase (		self: List[str]		)       -> List[Any]:




              """simple docstring"""


              UpperCamelCase_          =  DebertaModel.from_pretrained("microsoft/deberta-base"		)

              UpperCamelCase_          =  torch.tensor([[0, 31414, 232, 328, 740, 1140, 12695, 69, 46078, 1588, 2]]		)
              UpperCamelCase_          =  torch.tensor([[0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]		)
              with torch.no_grad():
                     UpperCamelCase_          =  model(_SCREAMING_SNAKE_CASE					,   attention_mask=_SCREAMING_SNAKE_CASE		)[0]
              # compare the actual values for a slice.
              UpperCamelCase_          =  torch.tensor(
                  [[[-0.59_86, -0.80_55, -0.84_62], [1.44_84, -0.93_48, -0.80_59], [0.31_23, 0.00_32, -1.41_31]]]		)
              self.assertTrue(torch.allclose(output[:, 1:4, 1:4]					,   _SCREAMING_SNAKE_CASE					,   atol=1e-4		)					,   f'''{output[:, 1:4, 1:4]}'''		)








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"  )     ->		Optional[Any]:
       UpperCamelCase_          =  False
       UpperCamelCase_          =  False
       if any(key.startswith("KAGGLE"  ) for key in os.environ.keys()  ):
              UpperCamelCase_          =  True
       elif "IPython" in sys.modules:
              UpperCamelCase_          =  "google.colab" in str(sys.modules["IPython"].get_ipython()  )

       try:
              UpperCamelCase_          =  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"					,		UpperCamelCase_  ) 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:
                     UpperCamelCase_          =  8

              UpperCamelCase_          =  PrepareForLaunch(UpperCamelCase_					,		distributed_type="TPU"  )
              print(F'''Launching a training on {num_processes} TPU cores.'''  )
              xmp.spawn(UpperCamelCase_					,		args=UpperCamelCase_					,		nprocs=UpperCamelCase_					,		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(*UpperCamelCase_  )
       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=UpperCamelCase_					,		master_addr="127.0.01"					,		master_port=UpperCamelCase_					,		mixed_precision=UpperCamelCase_  ):
                            UpperCamelCase_          =  PrepareForLaunch(UpperCamelCase_					,		distributed_type="MULTI_GPU"  )
                            print(F'''Launching training on {num_processes} GPUs.'''  )
                            try:
                                   start_processes(UpperCamelCase_					,		args=UpperCamelCase_					,		nprocs=UpperCamelCase_					,		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():
                            UpperCamelCase_          =  "1"
                            print("Launching training on MPS."  )
                     elif torch.cuda.is_available():
                            print("Launching training on one GPU."  )
                     else:
                            print("Launching training on CPU."  )
                     function(*UpperCamelCase_  )







def  lowerCAmelCase_ (							UpperCamelCase_					,		UpperCamelCase_=()					,		UpperCamelCase_=2  )     ->		Optional[Any]:
       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=UpperCamelCase_					,		master_addr="127.0.01"					,		master_port="29500"					,		accelerate_mixed_precision="no"					,		accelerate_debug_rdv_file=tmp_file.name					,		accelerate_use_cpu="yes"					,		):
                     UpperCamelCase_          =  PrepareForLaunch(UpperCamelCase_					,		debug=UpperCamelCase_  )
                     start_processes(UpperCamelCase_					,		args=UpperCamelCase_					,		nprocs=UpperCamelCase_					,		start_method="fork"  )

