

'''simple docstring'''




import os
import tempfile
import unittest

from transformers import FlaubertConfig, is_torch_available
from transformers.testing_utils import require_torch, require_torch_gpu, 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 (
          FlaubertForMultipleChoice,
          FlaubertForQuestionAnswering,
          FlaubertForQuestionAnsweringSimple,
          FlaubertForSequenceClassification,
          FlaubertForTokenClassification,
          FlaubertModel,
          FlaubertWithLMHeadModel,
      )
      from transformers.models.flaubert.modeling_flaubert import FLAUBERT_PRETRAINED_MODEL_ARCHIVE_LIST






class     _SCREAMING_SNAKE_CASE       (       __a      ):



 def __init__(       self						:       Optional[int]					, a__						:       Any					, a__						:       Any=13					, a__						:       List[str]=7					, a__						:       Any=True					, a__						:       str=True					, a__						:       List[Any]=True					, a__						:       Any=True					, a__						:       int=True					, a__						:       int=False					, a__						:       Union[str, Any]=False					, a__						:       List[Any]=False					, a__						:       Tuple=2					, a__						:       Optional[Any]=99					, a__						:       Any=0					, a__						:       Optional[Any]=32					, a__						:       str=5					, a__						:       int=4					, a__						:       int=0.1					, a__						:       Dict=0.1					, a__						:       List[str]=512					, a__						:       Any=12					, a__						:       List[Any]=2					, a__						:       Optional[int]=0.02					, a__						:       str=3					, a__						:       List[Any]=4					, a__						:       List[str]="last"					, a__						:       str=None					, a__						:       Union[str, Any]=None					, ):
      __magic_name__		  =							parent
      __magic_name__		  =							batch_size
      __magic_name__		  =							seq_length
      __magic_name__		  =							is_training
      __magic_name__		  =							use_input_lengths
      __magic_name__		  =							use_token_type_ids
      __magic_name__		  =							use_labels
      __magic_name__		  =							gelu_activation
      __magic_name__		  =							sinusoidal_embeddings
      __magic_name__		  =							causal
      __magic_name__		  =							asm
      __magic_name__		  =							n_langs
      __magic_name__		  =							vocab_size
      __magic_name__		  =							n_special
      __magic_name__		  =							hidden_size
      __magic_name__		  =							num_hidden_layers
      __magic_name__		  =							num_attention_heads
      __magic_name__		  =							hidden_dropout_prob
      __magic_name__		  =							attention_probs_dropout_prob
      __magic_name__		  =							max_position_embeddings
      __magic_name__		  =							type_vocab_size
      __magic_name__		  =							type_sequence_label_size
      __magic_name__		  =							initializer_range
      __magic_name__		  =							num_labels
      __magic_name__		  =							num_choices
      __magic_name__		  =							summary_type
      __magic_name__		  =							use_proj
      __magic_name__		  =							scope



 def 					snake_case__    (       self						:       Tuple     ):
      __magic_name__		  =							ids_tensor([self.batch_size, self.seq_length]					, self.vocab_size     )
      __magic_name__		  =							random_attention_mask([self.batch_size, self.seq_length]     )

      __magic_name__		  =							None
      if self.use_input_lengths:
           __magic_name__		  =							(
               ids_tensor([self.batch_size]					, vocab_size=2     ) + self.seq_length - 2
           )  # small variation of seq_length

      __magic_name__		  =							None
      if self.use_token_type_ids:
           __magic_name__		  =							ids_tensor([self.batch_size, self.seq_length]					, self.n_langs     )

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

      __magic_name__		  =							self.get_config()

      return (
          config,
          input_ids,
          token_type_ids,
          input_lengths,
          sequence_labels,
          token_labels,
          is_impossible_labels,
          choice_labels,
          input_mask,
      )



 def 					snake_case__    (       self						:       str     ):
      return FlaubertConfig(
          vocab_size=self.vocab_size					, n_special=self.n_special					, emb_dim=self.hidden_size					, n_layers=self.num_hidden_layers					, n_heads=self.num_attention_heads					, dropout=self.hidden_dropout_prob					, attention_dropout=self.attention_probs_dropout_prob					, gelu_activation=self.gelu_activation					, sinusoidal_embeddings=self.sinusoidal_embeddings					, asm=self.asm					, causal=self.causal					, n_langs=self.n_langs					, max_position_embeddings=self.max_position_embeddings					, initializer_range=self.initializer_range					, summary_type=self.summary_type					, use_proj=self.use_proj					, )



 def 					snake_case__    (       self						:       Union[str, Any]					, a__						:       int					, a__						:       Optional[Any]					, a__						:       List[str]					, a__						:       Dict					, a__						:       List[Any]					, a__						:       Dict					, a__						:       int					, a__						:       Optional[Any]					, a__						:       Any					, ):
      __magic_name__		  =							FlaubertModel(config=lowerCamelCase_     )
      model.to(lowerCamelCase_     )
      model.eval()
      __magic_name__		  =							model(lowerCamelCase_					, lengths=lowerCamelCase_					, langs=lowerCamelCase_     )
      __magic_name__		  =							model(lowerCamelCase_					, langs=lowerCamelCase_     )
      __magic_name__		  =							model(lowerCamelCase_     )
      self.parent.assertEqual(result.last_hidden_state.shape					, (self.batch_size, self.seq_length, self.hidden_size)     )



 def 					snake_case__    (       self						:       Optional[int]					, a__						:       Tuple					, a__						:       Optional[int]					, a__						:       Union[str, Any]					, a__						:       str					, a__						:       str					, a__						:       Optional[Any]					, a__						:       Any					, a__						:       List[str]					, a__						:       Tuple					, ):
      __magic_name__		  =							FlaubertWithLMHeadModel(lowerCamelCase_     )
      model.to(lowerCamelCase_     )
      model.eval()

      __magic_name__		  =							model(lowerCamelCase_					, token_type_ids=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 					snake_case__    (       self						:       int					, a__						:       Any					, a__						:       Any					, a__						:       Tuple					, a__						:       Any					, a__						:       Optional[int]					, a__						:       Tuple					, a__						:       str					, a__						:       Dict					, a__						:       List[Any]					, ):
      __magic_name__		  =							FlaubertForQuestionAnsweringSimple(lowerCamelCase_     )
      model.to(lowerCamelCase_     )
      model.eval()

      __magic_name__		  =							model(lowerCamelCase_     )

      __magic_name__		  =							model(lowerCamelCase_					, start_positions=lowerCamelCase_					, end_positions=lowerCamelCase_     )
      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 					snake_case__    (       self						:       Tuple					, a__						:       Optional[Any]					, a__						:       Tuple					, a__						:       List[Any]					, a__						:       str					, a__						:       Dict					, a__						:       Union[str, Any]					, a__						:       List[Any]					, a__						:       Any					, a__						:       Union[str, Any]					, ):
      __magic_name__		  =							FlaubertForQuestionAnswering(lowerCamelCase_     )
      model.to(lowerCamelCase_     )
      model.eval()

      __magic_name__		  =							model(lowerCamelCase_     )

      __magic_name__		  =							model(
          lowerCamelCase_					, start_positions=lowerCamelCase_					, end_positions=lowerCamelCase_					, cls_index=lowerCamelCase_					, is_impossible=lowerCamelCase_					, p_mask=lowerCamelCase_					, )

      __magic_name__		  =							model(
          lowerCamelCase_					, start_positions=lowerCamelCase_					, end_positions=lowerCamelCase_					, cls_index=lowerCamelCase_					, is_impossible=lowerCamelCase_					, )

      ((__magic_name__			)  ,    			)		  =							result_with_labels.to_tuple()

      __magic_name__		  =							model(lowerCamelCase_					, start_positions=lowerCamelCase_					, end_positions=lowerCamelCase_     )

      ((__magic_name__			)  ,    			)		  =							result_with_labels.to_tuple()

      self.parent.assertEqual(result_with_labels.loss.shape					, ()     )
      self.parent.assertEqual(result.start_top_log_probs.shape					, (self.batch_size, model.config.start_n_top)     )
      self.parent.assertEqual(result.start_top_index.shape					, (self.batch_size, model.config.start_n_top)     )
      self.parent.assertEqual(
          result.end_top_log_probs.shape					, (self.batch_size, model.config.start_n_top * model.config.end_n_top)     )
      self.parent.assertEqual(
          result.end_top_index.shape					, (self.batch_size, model.config.start_n_top * model.config.end_n_top)     )
      self.parent.assertEqual(result.cls_logits.shape					, (self.batch_size,)     )



 def 					snake_case__    (       self						:       Tuple					, a__						:       Dict					, a__						:       int					, a__						:       Optional[Any]					, a__						:       List[str]					, a__						:       Optional[Any]					, a__						:       Tuple					, a__						:       Union[str, Any]					, a__						:       Dict					, a__						:       List[str]					, ):
      __magic_name__		  =							FlaubertForSequenceClassification(lowerCamelCase_     )
      model.to(lowerCamelCase_     )
      model.eval()

      __magic_name__		  =							model(lowerCamelCase_     )
      __magic_name__		  =							model(lowerCamelCase_					, labels=lowerCamelCase_     )

      self.parent.assertEqual(result.loss.shape					, ()     )
      self.parent.assertEqual(result.logits.shape					, (self.batch_size, self.type_sequence_label_size)     )



 def 					snake_case__    (       self						:       Dict					, a__						:       List[str]					, a__						:       int					, a__						:       Tuple					, a__						:       int					, a__						:       Optional[Any]					, a__						:       Tuple					, a__						:       Optional[int]					, a__						:       Union[str, Any]					, a__						:       Any					, ):
      __magic_name__		  =							self.num_labels
      __magic_name__		  =							FlaubertForTokenClassification(lowerCamelCase_     )
      model.to(lowerCamelCase_     )
      model.eval()

      __magic_name__		  =							model(lowerCamelCase_					, attention_mask=lowerCamelCase_					, labels=lowerCamelCase_     )
      self.parent.assertEqual(result.logits.shape					, (self.batch_size, self.seq_length, self.num_labels)     )



 def 					snake_case__    (       self						:       Tuple					, a__						:       Tuple					, a__						:       Tuple					, a__						:       List[str]					, a__						:       List[str]					, a__						:       Optional[Any]					, a__						:       str					, a__						:       Tuple					, a__						:       Any					, a__						:       Optional[Any]					, ):
      __magic_name__		  =							self.num_choices
      __magic_name__		  =							FlaubertForMultipleChoice(config=lowerCamelCase_     )
      model.to(lowerCamelCase_     )
      model.eval()
      __magic_name__		  =							input_ids.unsqueeze(1     ).expand(-1					, self.num_choices					, -1     ).contiguous()
      __magic_name__		  =							token_type_ids.unsqueeze(1     ).expand(-1					, self.num_choices					, -1     ).contiguous()
      __magic_name__		  =							input_mask.unsqueeze(1     ).expand(-1					, self.num_choices					, -1     ).contiguous()
      __magic_name__		  =							model(
          lowerCamelCase_					, attention_mask=lowerCamelCase_					, token_type_ids=lowerCamelCase_					, labels=lowerCamelCase_					, )
      self.parent.assertEqual(result.logits.shape					, (self.batch_size, self.num_choices)     )



 def 					snake_case__    (       self						:       Any     ):
      __magic_name__		  =							self.prepare_config_and_inputs()
      (
          (
          __magic_name__			
      )  ,    (
          __magic_name__			
      )  ,    (
          __magic_name__			
      )  ,    (
          __magic_name__			
      )  ,    (
          __magic_name__			
      )  ,    (
          __magic_name__			
      )  ,    (
          __magic_name__			
      )  ,    (
          __magic_name__			
      )  ,    (
          __magic_name__			
      )  ,    			
      )		  =							config_and_inputs
      __magic_name__		  =							{
          '''input_ids''': input_ids,
          '''token_type_ids''': token_type_ids,
          '''lengths''': input_lengths,
          '''attention_mask''': input_mask,
      }
      return config, inputs_dict






@require_torch
class     _SCREAMING_SNAKE_CASE       (       __a						,__a						,unittest.TestCase      ):
 __SCREAMING_SNAKE_CASE	:str     			=	(
     (
         FlaubertModel,
         FlaubertWithLMHeadModel,
         FlaubertForQuestionAnswering,
         FlaubertForQuestionAnsweringSimple,
         FlaubertForSequenceClassification,
         FlaubertForTokenClassification,
         FlaubertForMultipleChoice,
     )
     if is_torch_available()
     else ()
 )
 __SCREAMING_SNAKE_CASE	:Any     			=	(
     {
         """feature-extraction""": FlaubertModel,
         """fill-mask""": FlaubertWithLMHeadModel,
         """question-answering""": FlaubertForQuestionAnsweringSimple,
         """text-classification""": FlaubertForSequenceClassification,
         """token-classification""": FlaubertForTokenClassification,
         """zero-shot""": FlaubertForSequenceClassification,
     }
     if is_torch_available()
     else {}
 )



 def 					snake_case__    (       self						:       int					, a__						:       Any					, a__						:       Optional[int]					, a__						:       Optional[Any]					, a__						:       str					, a__						:       str     ):
      if (
          pipeline_test_casse_name == "QAPipelineTests"
          and tokenizer_name is not None
          and not tokenizer_name.endswith('''Fast'''     )
      ):
           # `QAPipelineTests` fails for a few models when the slower tokenizer are used.
           # (The slower tokenizers were never used for pipeline tests before the pipeline testing rework)
           # TODO: check (and possibly fix) the `QAPipelineTests` with slower tokenizer
           return True

      return False



 def 					snake_case__    (       self						:       Optional[Any]					, a__						:       Any					, a__						:       Optional[Any]					, a__						:       int=False     ):
      __magic_name__		  =							super()._prepare_for_class(lowerCamelCase_					, lowerCamelCase_					, return_labels=lowerCamelCase_     )

      if return_labels:
           if model_class.__name__ == "FlaubertForQuestionAnswering":
                __magic_name__		  =							torch.zeros(
                    self.model_tester.batch_size					, dtype=torch.long					, device=lowerCamelCase_     )
                __magic_name__		  =							torch.zeros(
                    self.model_tester.batch_size					, dtype=torch.long					, device=lowerCamelCase_     )

      return inputs_dict



 def 					snake_case__    (       self						:       int     ):
      __magic_name__		  =							FlaubertModelTester(self     )
      __magic_name__		  =							ConfigTester(self					, config_class=lowerCamelCase_					, emb_dim=37     )



 def 					snake_case__    (       self						:       Tuple     ):
      self.config_tester.run_common_tests()



 def 					snake_case__    (       self						:       Union[str, Any]     ):
      __magic_name__		  =							self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_flaubert_model(*lowerCamelCase_     )



 def 					snake_case__    (       self						:       Optional[int]     ):
      __magic_name__		  =							self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_flaubert_lm_head(*lowerCamelCase_     )



 def 					snake_case__    (       self						:       Tuple     ):
      __magic_name__		  =							self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_flaubert_simple_qa(*lowerCamelCase_     )



 def 					snake_case__    (       self						:       Optional[int]     ):
      __magic_name__		  =							self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_flaubert_qa(*lowerCamelCase_     )



 def 					snake_case__    (       self						:       Union[str, Any]     ):
      __magic_name__		  =							self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_flaubert_sequence_classif(*lowerCamelCase_     )



 def 					snake_case__    (       self						:       List[Any]     ):
      __magic_name__		  =							self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_flaubert_token_classif(*lowerCamelCase_     )



 def 					snake_case__    (       self						:       int     ):
      __magic_name__		  =							self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_flaubert_multiple_choice(*lowerCamelCase_     )



 @slow
 def 					snake_case__    (       self						:       Optional[int]     ):
      for model_name in FLAUBERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           __magic_name__		  =							FlaubertModel.from_pretrained(lowerCamelCase_     )
           self.assertIsNotNone(lowerCamelCase_     )



 @slow
 @require_torch_gpu
 def 					snake_case__    (       self						:       Union[str, Any]     ):
      __magic_name__  ,    __magic_name__		  =							self.model_tester.prepare_config_and_inputs_for_common()
      for model_class in self.all_model_classes:
           # FlauBertForMultipleChoice behaves incorrectly in JIT environments.
           if model_class == FlaubertForMultipleChoice:
                return

           __magic_name__		  =							True
           __magic_name__		  =							model_class(config=lowerCamelCase_     )

           __magic_name__		  =							self._prepare_for_class(lowerCamelCase_					, lowerCamelCase_     )
           __magic_name__		  =							torch.jit.trace(
               lowerCamelCase_					, (inputs_dict['''input_ids'''].to('''cpu'''     ), inputs_dict['''attention_mask'''].to('''cpu'''     ))     )

           with tempfile.TemporaryDirectory() as tmp:
                torch.jit.save(lowerCamelCase_					, os.path.join(lowerCamelCase_					, '''traced_model.pt'''     )     )
                __magic_name__		  =							torch.jit.load(os.path.join(lowerCamelCase_					, '''traced_model.pt'''     )					, map_location=lowerCamelCase_     )
                loaded(inputs_dict['''input_ids'''].to(lowerCamelCase_     )					, inputs_dict['''attention_mask'''].to(lowerCamelCase_     )     )






@require_torch
class     _SCREAMING_SNAKE_CASE       (       unittest.TestCase      ):



 @slow
 def 					snake_case__    (       self						:       Dict     ):
      __magic_name__		  =							FlaubertModel.from_pretrained('''flaubert/flaubert_base_cased'''     )
      __magic_name__		  =							torch.tensor([[0, 345, 232, 328, 740, 140, 1695, 69, 6078, 1588, 2]]     )
      with torch.no_grad():
           __magic_name__		  =							model(lowerCamelCase_     )[0]
      __magic_name__		  =							torch.Size((1, 11, 768)     )
      self.assertEqual(output.shape					, lowerCamelCase_     )
      __magic_name__		  =							torch.tensor(
          [[[-2.6_251, -1.4_298, -0.0_227], [-2.8_510, -1.6_387, 0.2_258], [-2.8_114, -1.1_832, -0.3_066]]]     )

      self.assertTrue(torch.allclose(output[:, :3, :3]					, lowerCamelCase_					, atol=1E-4     )     )









"""simple docstring"""


import functools
import gc
import inspect

import torch

from .imports import is_npu_available, is_xpu_available



def   _A					(	*_a	:						int				):



      """simple docstring"""






      if not isinstance(_a     ,   _a				):
            A				  =		list(_a				)
      for i in range(len(_a				)				):
            A				  =		None
      gc.collect()
      if is_xpu_available():
            torch.xpu.empty_cache()
      elif is_npu_available():
            torch.npu.empty_cache()
      else:
            torch.cuda.empty_cache()
      return objects



def   _A					(	_a	:						Exception				):



      """simple docstring"""






      A				  =		[
          """CUDA out of memory.""",  # CUDA OOM
          """cuDNN error: CUDNN_STATUS_NOT_SUPPORTED.""",  # CUDNN SNAFU
          """DefaultCPUAllocator: can't allocate memory""",  # CPU OOM
      ]
      if isinstance(_a     ,   _a				) and len(exception.args				) == 1:
            return any(err in exception.args[0] for err in _statements				)
      return False





def   _A					(	_a	:						callable = None     ,   _a	:						int = 1_2_8				):



      """simple docstring"""






      if function is None:
            return functools.partial(_a     ,   starting_batch_size=_a				)

      A				  =		starting_batch_size

      def decorator(*_a	:						Union[str, Any]     ,   **_a	:						List[Any]				):
            nonlocal batch_size
            gc.collect()
            if is_xpu_available():
                  torch.xpu.empty_cache()
            elif is_npu_available():
                  torch.npu.empty_cache()
            else:
                  torch.cuda.empty_cache()
            A				  =		list(inspect.signature(_a				).parameters.keys()				)
            # Guard against user error
            if len(_a				) < (len(_a				) + 1):
                  A				  =		""", """.join([f'{arg}={value}' for arg, value in zip(params[1:]     ,   args[1:]				)]				)
                  raise TypeError(
                      f'Batch size was passed into `{function.__name__}` as the first argument when called.'
                      f'Remove this as the decorator already does so: `{function.__name__}({arg_str})`'				)
            while True:
                  if batch_size == 0:
                        raise RuntimeError("""No executable batch size found, reached zero."""				)
                  try:
                        return function(_a     ,   *_a     ,   **_a				)
                  except Exception as e:
                        if should_reduce_batch_size(_a				):
                              gc.collect()
                              if is_xpu_available():
                                    torch.xpu.empty_cache()
                              elif is_npu_available():
                                    torch.npu.empty_cache()
                              else:
                                    torch.cuda.empty_cache()
                              batch_size //= 2
                        else:
                              raise

      return decorator




