





from __future__ import annotations

import os
import tempfile
import unittest

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

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


if is_tf_available():
  import tensorflow as tf

  from transformers import (
      TFConvBertForMaskedLM,
      TFConvBertForMultipleChoice,
      TFConvBertForQuestionAnswering,
      TFConvBertForSequenceClassification,
      TFConvBertForTokenClassification,
      TFConvBertModel,
  )





class  lowerCAmelCase							:







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






     '''simple docstring'''





     lowercase__			       =  parent
     lowercase__			       =  13
     lowercase__			       =  7
     lowercase__			       =  True
     lowercase__			       =  True
     lowercase__			       =  True
     lowercase__			       =  True
     lowercase__			       =  99
     lowercase__			       =  3_84
     lowercase__			       =  2
     lowercase__			       =  4
     lowercase__			       =  37
     lowercase__			       =  "gelu"
     lowercase__			       =  0.1
     lowercase__			       =  0.1
     lowercase__			       =  5_12
     lowercase__			       =  16
     lowercase__			       =  2
     lowercase__			       =  0.02
     lowercase__			       =  3
     lowercase__			       =  4
     lowercase__			       =  1_28
     lowercase__			       =  2
     lowercase__			       =  9
     lowercase__			       =  1
     lowercase__			       =  None







  def   UpperCAmelCase					(      self  :Tuple ):






     '''simple docstring'''





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

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

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

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

     lowercase__			       =  ConvBertConfig(
         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   ,							return_dict=_lowercase   ,							)

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







  def   UpperCAmelCase					(      self  :List[Any]   ,							_lowercase  :Optional[Any]   ,							_lowercase  :Optional[int]   ,							_lowercase  :int   ,							_lowercase  :Any   ,							_lowercase  :Tuple   ,							_lowercase  :List[Any]   ,							_lowercase  :Dict ):






     '''simple docstring'''





     lowercase__			       =  TFConvBertModel(config=_lowercase )
     lowercase__			       =  {"input_ids": input_ids, "attention_mask": input_mask, "token_type_ids": token_type_ids}

     lowercase__			       =  [input_ids, input_mask]
     lowercase__			       =  model(_lowercase )

     lowercase__			       =  model(_lowercase )

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







  def   UpperCAmelCase					(      self  :List[Any]   ,							_lowercase  :Dict   ,							_lowercase  :int   ,							_lowercase  :Optional[Any]   ,							_lowercase  :Dict   ,							_lowercase  :Any   ,							_lowercase  :Union[str, Any]   ,							_lowercase  :Any ):






     '''simple docstring'''





     lowercase__			       =  TFConvBertForMaskedLM(config=_lowercase )
     lowercase__			       =  {
         "input_ids": input_ids,
         "attention_mask": input_mask,
         "token_type_ids": token_type_ids,
     }
     lowercase__			       =  model(_lowercase )
     self.parent.assertEqual(result.logits.shape   ,							(self.batch_size, self.seq_length, self.vocab_size) )







  def   UpperCAmelCase					(      self  :List[Any]   ,							_lowercase  :Optional[Any]   ,							_lowercase  :Any   ,							_lowercase  :str   ,							_lowercase  :int   ,							_lowercase  :List[str]   ,							_lowercase  :Dict   ,							_lowercase  :Optional[Any] ):






     '''simple docstring'''





     lowercase__			       =  self.num_labels
     lowercase__			       =  TFConvBertForSequenceClassification(config=_lowercase )
     lowercase__			       =  {
         "input_ids": input_ids,
         "attention_mask": input_mask,
         "token_type_ids": token_type_ids,
     }

     lowercase__			       =  model(_lowercase )
     self.parent.assertEqual(result.logits.shape   ,							(self.batch_size, self.num_labels) )







  def   UpperCAmelCase					(      self  :Any   ,							_lowercase  :Optional[int]   ,							_lowercase  :Tuple   ,							_lowercase  :Dict   ,							_lowercase  :Optional[Any]   ,							_lowercase  :Union[str, Any]   ,							_lowercase  :Any   ,							_lowercase  :Tuple ):






     '''simple docstring'''





     lowercase__			       =  self.num_choices
     lowercase__			       =  TFConvBertForMultipleChoice(config=_lowercase )
     lowercase__			       =  tf.tile(tf.expand_dims(_lowercase   ,							1 )   ,							(1, self.num_choices, 1) )
     lowercase__			       =  tf.tile(tf.expand_dims(_lowercase   ,							1 )   ,							(1, self.num_choices, 1) )
     lowercase__			       =  tf.tile(tf.expand_dims(_lowercase   ,							1 )   ,							(1, self.num_choices, 1) )
     lowercase__			       =  {
         "input_ids": multiple_choice_inputs_ids,
         "attention_mask": multiple_choice_input_mask,
         "token_type_ids": multiple_choice_token_type_ids,
     }
     lowercase__			       =  model(_lowercase )
     self.parent.assertEqual(result.logits.shape   ,							(self.batch_size, self.num_choices) )







  def   UpperCAmelCase					(      self  :Any   ,							_lowercase  :Tuple   ,							_lowercase  :Dict   ,							_lowercase  :Any   ,							_lowercase  :Any   ,							_lowercase  :Dict   ,							_lowercase  :Tuple   ,							_lowercase  :int ):






     '''simple docstring'''





     lowercase__			       =  self.num_labels
     lowercase__			       =  TFConvBertForTokenClassification(config=_lowercase )
     lowercase__			       =  {
         "input_ids": input_ids,
         "attention_mask": input_mask,
         "token_type_ids": token_type_ids,
     }
     lowercase__			       =  model(_lowercase )
     self.parent.assertEqual(result.logits.shape   ,							(self.batch_size, self.seq_length, self.num_labels) )







  def   UpperCAmelCase					(      self  :List[Any]   ,							_lowercase  :Dict   ,							_lowercase  :str   ,							_lowercase  :Any   ,							_lowercase  :int   ,							_lowercase  :int   ,							_lowercase  :List[str]   ,							_lowercase  :Optional[Any] ):






     '''simple docstring'''





     lowercase__			       =  TFConvBertForQuestionAnswering(config=_lowercase )
     lowercase__			       =  {
         "input_ids": input_ids,
         "attention_mask": input_mask,
         "token_type_ids": token_type_ids,
     }

     lowercase__			       =  model(_lowercase )
     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   UpperCAmelCase					(      self  :Optional[Any] ):






     '''simple docstring'''





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





@require_tf
class  lowerCAmelCase							(       lowercase_							,							lowercase_							,							unittest.TestCase ):
  __lowerCamelCase		=      (
      (
          TFConvBertModel,
          TFConvBertForMaskedLM,
          TFConvBertForQuestionAnswering,
          TFConvBertForSequenceClassification,
          TFConvBertForTokenClassification,
          TFConvBertForMultipleChoice,
      )
      if is_tf_available()
      else ()
  )
  __lowerCamelCase		=      (
      {
          'feature-extraction': TFConvBertModel,
          'fill-mask': TFConvBertForMaskedLM,
          'question-answering': TFConvBertForQuestionAnswering,
          'text-classification': TFConvBertForSequenceClassification,
          'token-classification': TFConvBertForTokenClassification,
          'zero-shot': TFConvBertForSequenceClassification,
      }
      if is_tf_available()
      else {}
  )
  __lowerCamelCase		=      False
  __lowerCamelCase		=      False
  __lowerCamelCase		=      False







  def   UpperCAmelCase					(      self  :List[str] ):






     '''simple docstring'''





     lowercase__			       =  TFConvBertModelTester(self )
     lowercase__			       =  ConfigTester(self   ,							config_class=_lowercase   ,							hidden_size=37 )







  def   UpperCAmelCase					(      self  :Tuple ):






     '''simple docstring'''





     self.config_tester.run_common_tests()







  def   UpperCAmelCase					(      self  :Optional[int] ):






     '''simple docstring'''





     lowercase__			       =  self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_model(*_lowercase )







  def   UpperCAmelCase					(      self  :Any ):






     '''simple docstring'''





     lowercase__			       =  self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_masked_lm(*_lowercase )







  def   UpperCAmelCase					(      self  :Dict ):






     '''simple docstring'''





     lowercase__			       =  self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_multiple_choice(*_lowercase )







  def   UpperCAmelCase					(      self  :str ):






     '''simple docstring'''





     lowercase__			       =  self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_question_answering(*_lowercase )







  def   UpperCAmelCase					(      self  :Optional[int] ):






     '''simple docstring'''





     lowercase__			       =  self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_sequence_classification(*_lowercase )







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






     '''simple docstring'''





     lowercase__			       =  self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_token_classification(*_lowercase )







  @slow
  def   UpperCAmelCase					(      self  :List[str] ):






     '''simple docstring'''





     lowercase__       ,							lowercase__			       =  self.model_tester.prepare_config_and_inputs_for_common()
     lowercase__			       =  True
     lowercase__			       =  True

     if hasattr(_lowercase   ,							"use_cache" ):
        lowercase__			       =  True

     lowercase__			       =  getattr(self.model_tester   ,							"encoder_seq_length"   ,							self.model_tester.seq_length )
     lowercase__			       =  getattr(self.model_tester   ,							"key_length"   ,							_lowercase )

     for model_class in self.all_model_classes:
        lowercase__			       =  self._prepare_for_class(_lowercase   ,							_lowercase )
        lowercase__			       =  model_class(_lowercase )
        lowercase__			       =  len(model(_lowercase ) )

        with tempfile.TemporaryDirectory() as tmpdirname:
           model.save_pretrained(_lowercase   ,							saved_model=_lowercase )
           lowercase__			       =  os.path.join(_lowercase   ,							"saved_model"   ,							"1" )
           lowercase__			       =  tf.keras.models.load_model(_lowercase )
           lowercase__			       =  model(_lowercase )

           if self.is_encoder_decoder:
              lowercase__			       =  outputs["encoder_hidden_states"]
              lowercase__			       =  outputs["encoder_attentions"]
           else:
              lowercase__			       =  outputs["hidden_states"]
              lowercase__			       =  outputs["attentions"]

           self.assertEqual(len(_lowercase )   ,							_lowercase )

           lowercase__			       =  getattr(
               self.model_tester   ,							"expected_num_hidden_layers"   ,							self.model_tester.num_hidden_layers + 1 )

           self.assertEqual(len(_lowercase )   ,							_lowercase )
           self.assertListEqual(
               list(output_hidden_states[0].shape[-2:] )   ,							[self.model_tester.seq_length, self.model_tester.hidden_size]   ,							)

           self.assertEqual(len(_lowercase )   ,							self.model_tester.num_hidden_layers )
           self.assertListEqual(
               list(output_attentions[0].shape[-3:] )   ,							[self.model_tester.num_attention_heads / 2, encoder_seq_length, encoder_key_length]   ,							)







  @slow
  def   UpperCAmelCase					(      self  :int ):






     '''simple docstring'''





     lowercase__			       =  TFConvBertModel.from_pretrained("YituTech/conv-bert-base" )
     self.assertIsNotNone(_lowercase )







  def   UpperCAmelCase					(      self  :Tuple ):






     '''simple docstring'''





     lowercase__       ,							lowercase__			       =  self.model_tester.prepare_config_and_inputs_for_common()
     lowercase__			       =  True
     lowercase__			       =  getattr(self.model_tester   ,							"decoder_seq_length"   ,							self.model_tester.seq_length )
     lowercase__			       =  getattr(self.model_tester   ,							"encoder_seq_length"   ,							self.model_tester.seq_length )
     lowercase__			       =  getattr(self.model_tester   ,							"key_length"   ,							_lowercase )
     lowercase__			       =  getattr(self.model_tester   ,							"key_length"   ,							_lowercase )

     def check_decoder_attentions_output(_lowercase  :int ):
        lowercase__			       =  len(_lowercase )
        self.assertEqual(out_len % 2   ,							0 )
        lowercase__			       =  outputs.decoder_attentions
        self.assertEqual(len(_lowercase )   ,							self.model_tester.num_hidden_layers )
        self.assertListEqual(
            list(decoder_attentions[0].shape[-3:] )   ,							[self.model_tester.num_attention_heads / 2, decoder_seq_length, decoder_key_length]   ,							)

     def check_encoder_attentions_output(_lowercase  :Union[str, Any] ):
        lowercase__			       =  [
            t.numpy() for t in (outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions)
        ]
        self.assertEqual(len(_lowercase )   ,							self.model_tester.num_hidden_layers )
        self.assertListEqual(
            list(attentions[0].shape[-3:] )   ,							[self.model_tester.num_attention_heads / 2, encoder_seq_length, encoder_key_length]   ,							)

     for model_class in self.all_model_classes:
        lowercase__			       =  True
        lowercase__			       =  False
        lowercase__			       =  model_class(_lowercase )
        lowercase__			       =  model(self._prepare_for_class(_lowercase   ,							_lowercase ) )
        lowercase__			       =  len(_lowercase )
        self.assertEqual(config.output_hidden_states   ,							_lowercase )
        check_encoder_attentions_output(_lowercase )

        if self.is_encoder_decoder:
           lowercase__			       =  model_class(_lowercase )
           lowercase__			       =  model(self._prepare_for_class(_lowercase   ,							_lowercase ) )
           self.assertEqual(config.output_hidden_states   ,							_lowercase )
           check_decoder_attentions_output(_lowercase )

        # Check that output attentions can also be changed via the config
        del inputs_dict["output_attentions"]
        lowercase__			       =  True
        lowercase__			       =  model_class(_lowercase )
        lowercase__			       =  model(self._prepare_for_class(_lowercase   ,							_lowercase ) )
        self.assertEqual(config.output_hidden_states   ,							_lowercase )
        check_encoder_attentions_output(_lowercase )

        # Check attention is always last and order is fine
        lowercase__			       =  True
        lowercase__			       =  True
        lowercase__			       =  model_class(_lowercase )
        lowercase__			       =  model(self._prepare_for_class(_lowercase   ,							_lowercase ) )

        self.assertEqual(out_len + (2 if self.is_encoder_decoder else 1)   ,							len(_lowercase ) )
        self.assertEqual(model.config.output_hidden_states   ,							_lowercase )
        check_encoder_attentions_output(_lowercase )





@require_tf
class  lowerCAmelCase							(       unittest.TestCase ):







  @slow
  def   UpperCAmelCase					(      self  :Optional[int] ):






     '''simple docstring'''





     lowercase__			       =  TFConvBertModel.from_pretrained("YituTech/conv-bert-base" )
     lowercase__			       =  tf.constant([[0, 1, 2, 3, 4, 5]] )
     lowercase__			       =  model(_lowercase )[0]

     lowercase__			       =  [1, 6, 7_68]
     self.assertEqual(output.shape   ,							_lowercase )

     lowercase__			       =  tf.constant(
         [
             [
                 [-0.03475493, -0.4686034, -0.30638832],
                 [0.22637248, -0.26988646, -0.7423424],
                 [0.10324868, -0.45013508, -0.58280784],
             ]
         ] )
     tf.debugging.assert_near(output[:, :3, :3]   ,							_lowercase   ,							atol=1e-4 )













import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
  from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
  import torch_xla.core.xla_model as xm


def _A       (   __magic_name__	):
   if is_torch_version("<"  , "2.0.0"	) or not hasattr(__magic_name__  , "_dynamo"	):
      return False
   return isinstance(__magic_name__  , torch._dynamo.eval_frame.OptimizedModule	)


def _A       (   __magic_name__  , __magic_name__ = True	):
   lowercase__			       =  (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

   lowercase__			       =  is_compiled_module(__magic_name__	)
   if is_compiled:
      lowercase__			       =  model
      lowercase__			       =  model._orig_mod

   if is_deepspeed_available():
      options += (DeepSpeedEngine,)

   while isinstance(__magic_name__  , __magic_name__	):
      lowercase__			       =  model.module

   if not keep_fpaa_wrapper:
      lowercase__			       =  getattr(__magic_name__  , "forward"	)
      lowercase__			       =  model.__dict__.pop("_original_forward"  , __magic_name__	)
      if original_forward is not None:
         while hasattr(__magic_name__  , "__wrapped__"	):
            lowercase__			       =  forward.__wrapped__
            if forward == original_forward:
               break
         lowercase__			       =  forward
      if getattr(__magic_name__  , "_converted_to_transformer_engine"  , __magic_name__	):
         convert_model(__magic_name__  , to_transformer_engine=__magic_name__	)

   if is_compiled:
      lowercase__			       =  model
      lowercase__			       =  compiled_model

   return model


def _A       (   ):
   PartialState().wait_for_everyone()


def _A       (   __magic_name__  , __magic_name__	):
   if PartialState().distributed_type == DistributedType.TPU:
      xm.save(__magic_name__  , __magic_name__	)
   elif PartialState().local_process_index == 0:
      torch.save(__magic_name__  , __magic_name__	)


@contextmanager
def _A       (   **__magic_name__	):
   for key, value in kwargs.items():
      lowercase__			       =  str(__magic_name__	)

   yield

   for key in kwargs:
      if key.upper() in os.environ:
         del os.environ[key.upper()]


def _A       (   __magic_name__	):
   if not hasattr(__magic_name__  , "__qualname__"	) and not hasattr(__magic_name__  , "__name__"	):
      lowercase__			       =  getattr(__magic_name__  , "__class__"  , __magic_name__	)
   if hasattr(__magic_name__  , "__qualname__"	):
      return obj.__qualname__
   if hasattr(__magic_name__  , "__name__"	):
      return obj.__name__
   return str(__magic_name__	)


def _A       (   __magic_name__  , __magic_name__	):
   for key, value in source.items():
      if isinstance(__magic_name__  , __magic_name__	):
         lowercase__			       =  destination.setdefault(__magic_name__  , {}	)
         merge_dicts(__magic_name__  , __magic_name__	)
      else:
         lowercase__			       =  value

   return destination







def _A       (   __magic_name__ = None	):
   if port is None:
      lowercase__			       =  2_9500
   with socket.socket(socket.AF_INET  , socket.SOCK_STREAM	) as s:
      return s.connect_ex(("localhost", port)	) == 0






