



"""simple docstring"""




from __future__ import annotations

import copy
import os
import tempfile
from importlib import import_module
from math import isnan

from transformers import is_tf_available
from transformers.models.auto import get_values
from transformers.testing_utils import _tf_gpu_memory_limit, require_tf, slow

from ..test_modeling_tf_common import ids_tensor


if is_tf_available():
 import numpy as np
 import tensorflow as tf

 from transformers import (
     TF_MODEL_FOR_CAUSAL_LM_MAPPING,
     TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
     TF_MODEL_FOR_MASKED_LM_MAPPING,
     TF_MODEL_FOR_MULTIPLE_CHOICE_MAPPING,
     TF_MODEL_FOR_NEXT_SENTENCE_PREDICTION_MAPPING,
     TF_MODEL_FOR_PRETRAINING_MAPPING,
     TF_MODEL_FOR_QUESTION_ANSWERING_MAPPING,
     TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
     TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
     TF_MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING,
     TFSharedEmbeddings,
 )

 if _tf_gpu_memory_limit is not None:
  _lowerCAmelCase				:					Any							=							tf.config.list_physical_devices("GPU")
  for gpu in gpus:
   # Restrict TensorFlow to only allocate x GB of memory on the GPUs
   try:
    tf.config.set_logical_device_configuration(
        gpu, [tf.config.LogicalDeviceConfiguration(memory_limit=_tf_gpu_memory_limit)]
    )
    _lowerCAmelCase				:					List[str]							=							tf.config.list_logical_devices("GPU")
    print("Logical GPUs", logical_gpus)
   except RuntimeError as e:
    # Virtual devices must be set before GPUs have been initialized
    print(e)






@require_tf
class 	UpperCAmelCase_  :
      __SCREAMING_SNAKE_CASE						:   Dict      = None
      __SCREAMING_SNAKE_CASE						:   Any      = ()
      __SCREAMING_SNAKE_CASE						:   List[Any]      = ()
      __SCREAMING_SNAKE_CASE						:   str      = True
      __SCREAMING_SNAKE_CASE						:   List[Any]      = True
      __SCREAMING_SNAKE_CASE						:   Tuple      = True
      __SCREAMING_SNAKE_CASE						:   int      = False



      def 						snake_case_					(     self     :	Tuple     ,	A     :	Any     ,	A     :	Dict     ,	A     :	Union[str, Any]=False						):
           _UpperCAmelCase			:	str								=    copy.deepcopy(A						)

           if model_class in get_values(A						):
                _UpperCAmelCase			:	int								=    {
                    k: tf.tile(tf.expand_dims(A     ,	1						)     ,	(1, self.model_tester.num_choices) + (1,) * (v.ndim - 1)						)
                    if isinstance(A     ,	tf.Tensor						) and v.ndim > 0
                    else v
                    for k, v in inputs_dict.items()
                }

           if return_labels:
                if model_class in get_values(A						):
                     _UpperCAmelCase			:	Any								=    tf.ones(self.model_tester.batch_size     ,	dtype=tf.intaa						)
                elif model_class in get_values(A						):
                     _UpperCAmelCase			:	Union[str, Any]								=    tf.zeros(self.model_tester.batch_size     ,	dtype=tf.intaa						)
                     _UpperCAmelCase			:	Optional[int]								=    tf.zeros(self.model_tester.batch_size     ,	dtype=tf.intaa						)
                elif model_class in [
                    *get_values(A						),
                    *get_values(A						),
                ]:
                     _UpperCAmelCase			:	List[str]								=    tf.zeros(self.model_tester.batch_size     ,	dtype=tf.intaa						)
                elif model_class in get_values(A						):
                     _UpperCAmelCase			:	int								=    tf.zeros(self.model_tester.batch_size     ,	dtype=tf.intaa						)
                elif model_class in [
                    *get_values(A						),
                    *get_values(A						),
                    *get_values(A						),
                    *get_values(A						),
                    *get_values(A						),
                ]:
                     _UpperCAmelCase			:	Union[str, Any]								=    tf.zeros(
                         (self.model_tester.batch_size, self.model_tester.seq_length)     ,	dtype=tf.intaa						)
           return inputs_dict



      @slow
      def 						snake_case_					(     self     :	str						):
           _UpperCAmelCase			:	Dict								=    self.model_tester.prepare_config_and_inputs_for_common()
           for model_class in self.all_model_classes[:2]:
                _UpperCAmelCase			:	Optional[Any]								=    self._prepare_for_class(A     ,	A						)
                _UpperCAmelCase			:	List[Any]								=    model_class(A						)

                @tf.function
                def run_in_graph_mode():
                     return model(A						)

                _UpperCAmelCase			:	Optional[int]								=    run_in_graph_mode()
                self.assertIsNotNone(A						)



      @slow
      def 						snake_case_					(     self     :	Optional[Any]						):
           _UpperCAmelCase			:	Optional[int]								=    self.model_tester.prepare_config_and_inputs_for_common()
           for model_class in self.all_model_classes[:2]:
                _UpperCAmelCase			:	str								=    self._prepare_for_class(A     ,	A						)
                _UpperCAmelCase			:	List[str]								=    model_class(A						)

                @tf.function(experimental_compile=A						)
                def run_in_graph_mode():
                     return model(A						)

                _UpperCAmelCase			:	Union[str, Any]								=    run_in_graph_mode()
                self.assertIsNotNone(A						)



      @slow
      def 						snake_case_					(     self     :	Optional[Any]						):
           # This is a copy of the test_keras_fit method, but we use XLA compilation instead of eager
           _UpperCAmelCase			:	Union[str, Any]								=    self.model_tester.prepare_config_and_inputs_for_common()
           for model_class in self.all_model_classes[:2]:
                _UpperCAmelCase			:	List[str]								=    model_class(A						)
                if getattr(A     ,	"hf_compute_loss"     ,	A						):
                     # Test that model correctly compute the loss with kwargs
                     _UpperCAmelCase			:	Dict								=    self._prepare_for_class(inputs_dict.copy()     ,	A     ,	return_labels=A						)
                     # Is there a better way to remove these decoder inputs?
                     _UpperCAmelCase			:	str								=    {
                         key: val
                         for key, val in prepared_for_class.items()
                         if key not in ('head_mask', 'decoder_head_mask', 'cross_attn_head_mask', 'decoder_input_ids')
                     }

                     _UpperCAmelCase			:	Optional[int]								=    {
                         'labels',
                         'label',
                         'label_ids',
                         'start_positions',
                         'start_position',
                         'end_positions',
                         'end_position',
                         'next_sentence_label',
                     }
                     _UpperCAmelCase			:	Union[str, Any]								=    possible_label_cols.intersection(set(A						)						)
                     self.assertGreater(len(A						)     ,	0     ,	msg="No matching label names found!"						)
                     _UpperCAmelCase			:	Optional[Any]								=    {key: val for key, val in prepared_for_class.items() if key in label_names}
                     _UpperCAmelCase			:	Optional[Any]								=    {key: val for key, val in prepared_for_class.items() if key not in label_names}
                     self.assertGreater(len(A						)     ,	0						)

                     # Make sure it works with XLA!
                     model.compile(optimizer=tf.keras.optimizers.SGD(0.0						)     ,	jit_compile=A						)
                     # Make sure the model fits without crashing regardless of where we pass the labels
                     _UpperCAmelCase			:	Optional[int]								=    model.fit(
                         A     ,	validation_data=A     ,	steps_per_epoch=1     ,	validation_steps=1     ,	shuffle=A     ,	verbose=0     ,	)
                     _UpperCAmelCase			:	int								=    history.history['loss'][0]
                     self.assertTrue(not isnan(A						)						)
                     _UpperCAmelCase			:	Optional[int]								=    history.history['val_loss'][0]
                     self.assertTrue(not isnan(A						)						)

                     # Now test it with separate labels, to make sure that path works in XLA too.
                     _UpperCAmelCase			:	List[Any]								=    model_class(A						)
                     model.compile(optimizer=tf.keras.optimizers.SGD(0.0						)     ,	jit_compile=A						)
                     _UpperCAmelCase			:	Any								=    model.fit(
                         A     ,	A     ,	validation_data=(inputs_minus_labels, labels)     ,	steps_per_epoch=1     ,	validation_steps=1     ,	shuffle=A     ,	verbose=0     ,	)

                     _UpperCAmelCase			:	Optional[Any]								=    history.history['loss'][0]
                     self.assertTrue(not isnan(A						)						)
                     _UpperCAmelCase			:	Optional[int]								=    history.history['val_loss'][0]
                     self.assertTrue(not isnan(A						)						)



      @slow
      def 						snake_case_					(     self     :	Any						):
           _UpperCAmelCase			:	Dict								=    self.model_tester.prepare_config_and_inputs_for_common()
           _UpperCAmelCase			:	Any								=    True
           _UpperCAmelCase			:	str								=    True

           if hasattr(A     ,	"use_cache"						):
                _UpperCAmelCase			:	Dict								=    True

           _UpperCAmelCase			:	Tuple								=    getattr(self.model_tester     ,	"encoder_seq_length"     ,	self.model_tester.seq_length						)
           _UpperCAmelCase			:	Optional[Any]								=    getattr(self.model_tester     ,	"key_length"     ,	A						)

           for model_class in self.all_model_classes[:2]:
                _UpperCAmelCase			:	List[Any]								=    self._prepare_for_class(A     ,	A						)
                _UpperCAmelCase			:	List[Any]								=    model_class(A						)
                model.build()
                _UpperCAmelCase			:	Any								=    len(model(A						)						)

                for key in list(class_inputs_dict.keys()						):
                     # Remove keys not in the serving signature, as the SavedModel will not be compiled to deal with them
                     if key not in model.input_signature:
                          del class_inputs_dict[key]
                     # Check it's a tensor, in case the inputs dict has some bools in it too
                     elif isinstance(class_inputs_dict[key]     ,	tf.Tensor						) and class_inputs_dict[key].dtype.is_integer:
                          _UpperCAmelCase			:	List[Any]								=    tf.cast(class_inputs_dict[key]     ,	tf.intaa						)

                if set(class_inputs_dict.keys()						) != set(model.input_signature.keys()						):
                     continue  # Some models have inputs that the preparation functions don't create, we skip those

                with tempfile.TemporaryDirectory() as tmpdirname:
                     model.save_pretrained(A     ,	saved_model=A						)
                     _UpperCAmelCase			:	Optional[Any]								=    os.path.join(A     ,	"saved_model"     ,	"1"						)
                     _UpperCAmelCase			:	Optional[int]								=    tf.keras.models.load_model(A						)
                     _UpperCAmelCase			:	Any								=    model(A						)

                     if self.is_encoder_decoder:
                          _UpperCAmelCase			:	Dict								=    outputs['encoder_hidden_states']
                          _UpperCAmelCase			:	int								=    outputs['encoder_attentions']
                     else:
                          _UpperCAmelCase			:	int								=    outputs['hidden_states']
                          _UpperCAmelCase			:	Dict								=    outputs['attentions']

                     self.assertEqual(len(A						)     ,	A						)

                     _UpperCAmelCase			:	List[str]								=    getattr(
                         self.model_tester     ,	"expected_num_hidden_layers"     ,	self.model_tester.num_hidden_layers + 1						)

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

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



      @slow
      def 						snake_case_					(     self     :	Tuple						):
           tf.keras.mixed_precision.set_global_policy("mixed_float16"						)

           # try/finally block to ensure subsequent tests run in float32
           try:
                _UpperCAmelCase			:	Any								=    self.model_tester.prepare_config_and_inputs_for_common()
                for model_class in self.all_model_classes[:2]:
                     _UpperCAmelCase			:	List[str]								=    self._prepare_for_class(A     ,	A						)
                     _UpperCAmelCase			:	List[str]								=    model_class(A						)
                     _UpperCAmelCase			:	Union[str, Any]								=    model(A						)

                     self.assertIsNotNone(A						)
           finally:
                tf.keras.mixed_precision.set_global_policy("float32"						)



      @slow
      def 						snake_case_					(     self     :	List[Any]						):
           _UpperCAmelCase			:	List[Any]								=    self.model_tester.prepare_config_and_inputs_for_common()
           # head_mask and decoder_head_mask has different shapes than other input args
           if "head_mask" in inputs_dict:
                del inputs_dict["head_mask"]
           if "decoder_head_mask" in inputs_dict:
                del inputs_dict["decoder_head_mask"]
           if "cross_attn_head_mask" in inputs_dict:
                del inputs_dict["cross_attn_head_mask"]
           _UpperCAmelCase			:	Dict								=    {
               module_member
               for model_class in self.all_model_classes
               for module in (import_module(model_class.__module__						),)
               for module_member_name in dir(A						)
               if module_member_name.endswith("MainLayer"						)
               for module_member in (getattr(A     ,	A						),)
               if isinstance(A     ,	A						)
               and tf.keras.layers.Layer in module_member.__bases__
               and getattr(A     ,	"_keras_serializable"     ,	A						)
           }

           for main_layer_class in tf_main_layer_classes:
                # T5MainLayer needs an embed_tokens parameter when called without the inputs_embeds parameter
                if "T5" in main_layer_class.__name__:
                     # Take the same values than in TFT5ModelTester for this shared layer
                     _UpperCAmelCase			:	List[str]								=    TFSharedEmbeddings(self.model_tester.vocab_size     ,	self.model_tester.hidden_size     ,	name="shared"						)
                     _UpperCAmelCase			:	Optional[int]								=    False
                     _UpperCAmelCase			:	Union[str, Any]								=    main_layer_class(A     ,	embed_tokens=A						)
                else:
                     _UpperCAmelCase			:	Optional[int]								=    main_layer_class(A						)

                _UpperCAmelCase			:	Optional[int]								=    {
                    name: tf.keras.Input(tensor.shape[1:]     ,	dtype=tensor.dtype						) for name, tensor in inputs_dict.items()
                }

                if hasattr(self.model_tester     ,	"num_labels"						):
                     _UpperCAmelCase			:	Any								=    self.model_tester.num_labels
                else:
                     _UpperCAmelCase			:	Dict								=    2

                _UpperCAmelCase			:	int								=    tf.data.Dataset.from_tensor_slices(
                    (inputs_dict, np.ones((self.model_tester.batch_size, self.model_tester.seq_length, num_labels, 1)						))						).batch(1						)

                _UpperCAmelCase			:	Union[str, Any]								=    main_layer(A						)[0]
                _UpperCAmelCase			:	Optional[int]								=    tf.keras.layers.Dense(A     ,	activation="softmax"     ,	name="outputs"						)(A						)
                _UpperCAmelCase			:	Dict								=    tf.keras.models.Model(inputs=A     ,	outputs=[outputs]						)

                model.compile(loss="binary_crossentropy"     ,	optimizer="adam"     ,	metrics=["binary_accuracy"]						)
                model.fit(A     ,	epochs=1						)

                with tempfile.TemporaryDirectory() as tmpdirname:
                     _UpperCAmelCase			:	Optional[int]								=    os.path.join(A     ,	"keras_model.h5"						)
                     model.save(A						)
                     if "T5" in main_layer_class.__name__:
                          _UpperCAmelCase			:	Optional[int]								=    tf.keras.models.load_model(
                              A     ,	custom_objects={
                                  main_layer_class.__name__: main_layer_class,
                                  "TFSharedEmbeddings": TFSharedEmbeddings,
                              }     ,	)
                     else:
                          _UpperCAmelCase			:	int								=    tf.keras.models.load_model(
                              A     ,	custom_objects={main_layer_class.__name__: main_layer_class}						)
                     assert isinstance(A     ,	tf.keras.Model						)
                     model(A						)



      @slow
      def 						snake_case_					(     self     :	int						):
           _UpperCAmelCase			:	List[Any]								=    self.model_tester.prepare_config_and_inputs_for_common()

           for model_class in self.all_model_classes[:2]:
                _UpperCAmelCase			:	Tuple								=    model_class(A						)

                _UpperCAmelCase			:	List[str]								=    copy.deepcopy(A						)

                if not self.is_encoder_decoder:
                     _UpperCAmelCase			:	Any								=    inputs['input_ids']
                     del inputs["input_ids"]
                else:
                     _UpperCAmelCase			:	Optional[Any]								=    inputs['input_ids']
                     _UpperCAmelCase			:	Tuple								=    inputs.get("decoder_input_ids"     ,	A						)
                     del inputs["input_ids"]
                     inputs.pop("decoder_input_ids"     ,	A						)

                if not self.is_encoder_decoder:
                     _UpperCAmelCase			:	Dict								=    model.get_input_embeddings()(A						)
                else:
                     _UpperCAmelCase			:	List[str]								=    model.get_input_embeddings()(A						)
                     _UpperCAmelCase			:	Union[str, Any]								=    model.get_input_embeddings()(A						)

                _UpperCAmelCase			:	int								=    self._prepare_for_class(A     ,	A						)

                @tf.function
                def run_in_graph_mode():
                     return model(A						)

                _UpperCAmelCase			:	Union[str, Any]								=    run_in_graph_mode()
                self.assertIsNotNone(A						)



      def 						snake_case_					(     self     :	Tuple     ,	A     :	List[str]     ,	A     :	Tuple						):
           # special tokens cannot be bad tokens
           _UpperCAmelCase			:	str								=    []
           if model.config.bos_token_id is not None:
                special_tokens.append(model.config.bos_token_id						)
           if model.config.pad_token_id is not None:
                special_tokens.append(model.config.pad_token_id						)
           if model.config.eos_token_id is not None:
                special_tokens.append(model.config.eos_token_id						)

           # create random bad tokens that are not special tokens
           _UpperCAmelCase			:	Optional[int]								=    []
           while len(A						) < num_bad_tokens:
                _UpperCAmelCase			:	Any								=    tf.squeeze(ids_tensor((1, 1)     ,	self.model_tester.vocab_size						)     ,	0						).numpy()[0]
                if token not in special_tokens:
                     bad_tokens.append(A						)
           return bad_tokens



      def 						snake_case_					(     self     :	Tuple     ,	A     :	Tuple						):
           for token_id in output_ids[0].numpy().tolist():
                self.assertGreaterEqual(A     ,	0						)
                self.assertLess(A     ,	self.model_tester.vocab_size						)



      def 						snake_case_					(     self     :	Tuple     ,	A     :	List[Any]     ,	A     :	str						):
           # for all bad word tokens
           for bad_word_ids in bad_words_ids:
                # for all slices in batch
                for generated_ids_slice in generated_ids:
                     # for all word idx
                     for i in range(len(A						)     ,	len(A						)						):
                          # if tokens match
                          if generated_ids_slice[i - len(A						) : i] == bad_word_ids:
                               return True
           return False




import unittest

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


if is_torch_available():
       import torch

       from transformers import CamembertModel


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




      '''simple docstring'''


      @slow
      def      _snake_case	(				self:       Any				):
       __lowerCamelCase		:							Dict							=						CamembertModel.from_pretrained('camembert-base'				)
       model.to(a				)

       __lowerCamelCase		:							List[Any]							=						torch.tensor(
           [[5, 121, 11, 660, 16, 730, 2_5543, 110, 83, 6]]						,      device=a						,      dtype=torch.long						,      )  # J'aime le camembert !
       with torch.no_grad():
        __lowerCamelCase		:							Tuple							=						model(a				)['last_hidden_state']
       __lowerCamelCase		:							Union[str, Any]							=						torch.Size((1, 10, 768)				)
       self.assertEqual(output.shape						,      a				)
       # compare the actual values for a slice.
       __lowerCamelCase		:							Tuple							=						torch.tensor(
           [[[-0.0_2_5_4, 0.0_2_3_5, 0.1_0_2_7], [0.0_6_0_6, -0.1_8_1_1, -0.0_4_1_8], [-0.1_5_6_1, -0.1_1_2_7, 0.2_6_8_7]]]						,      device=a						,      dtype=torch.float						,      )
       # camembert = torch.hub.load('pytorch/fairseq', 'camembert.v0')
       # camembert.eval()
       # expected_slice = roberta.model.forward(input_ids)[0][:, :3, :3].detach()

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

