

from __future__ import annotations

import os
import tempfile
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import is_tensorflow_text_available, is_tf_available
from transformers.testing_utils import require_tensorflow_text, require_tf, slow

from ..test_modeling_tf_common import floats_tensor
from .test_framework_agnostic import GenerationIntegrationTestsMixin


if is_tf_available():
   import tensorflow as tf

   from transformers import (
       AutoTokenizer,
       TFAutoModelForCausalLM,
       TFAutoModelForSeqaSeqLM,
       TFAutoModelForSpeechSeqaSeq,
       TFAutoModelForVisionaSeq,
       TFBartForConditionalGeneration,
       TFLogitsProcessorList,
       TFMinLengthLogitsProcessor,
       tf_top_k_top_p_filtering,
   )

if is_tensorflow_text_available():
   import tensorflow_text as text



@require_tf
class SCREAMING_SNAKE_CASE      (    unittest.TestCase					):
  """simple docstring"""






  def 			SCREAMING_SNAKE_CASE       (     self					:       Dict       )				->      Optional[Any]:






     """simple docstring"""







     __lowerCAmelCase    :     List[str]          =		tf.convert_to_tensor(
         [
             [
                 8.222_0991,  # 3rd highest value; idx. 0
                 -0.562_0044,
                 5.2322_9752,
                 4.038_6393,
                 -6.879_8378,
                 -0.5478_5802,
                 -3.201_2153,
                 2.9277_7176,
                 1.8817_1953,
                 7.3534_1276,  # 5th highest value; idx. 9
                 8.4320_7833,  # 2nd highest value; idx. 10
                 -9.8571_1836,
                 -5.9620_9236,
                 -1.1303_9161,
                 -7.111_5294,
                 -0.836_9633,
                 -5.318_6408,
                 7.0642_7407,
                 0.8136_9344,
                 -0.8202_3817,
                 -5.917_9796,
                 0.5881_3443,
                 -6.9977_8438,
                 4.7155_1189,
                 -0.1877_1637,
                 7.4402_0759,  # 4th highest value; idx. 25
                 9.3845_0987,  # 1st highest value; idx. 26
                 2.1266_2941,
                 -9.3256_2038,
                 2.3565_2522,
             ],  # cummulative prob of 5 highest values <= 0.6
             [
                 0.5842_5518,
                 4.5313_9238,
                 -5.5751_0464,
                 -6.2803_0699,
                 -7.1952_9503,
                 -4.0212_2551,
                 1.3933_7037,
                 -6.0670_7057,
                 1.5948_0517,
                 -9.64_3119,
                 0.0390_7799,
                 0.6723_1762,
                 -8.8820_6726,
                 6.2711_5922,  # 4th highest value; idx. 13
                 2.2852_0723,
                 4.8276_7506,
                 4.3042_1368,
                 8.827_5313,  # 2nd highest value; idx. 17
                 5.4402_9958,  # 5th highest value; idx. 18
                 -4.473_5794,
                 7.3857_9536,  # 3rd highest value; idx. 20
                 -2.9105_1663,
                 2.6194_6077,
                 -2.567_4762,
                 -9.4895_9302,
                 -4.0292_2645,
                 -1.3541_6918,
                 9.6770_2323,  # 1st highest value; idx. 27
                 -5.8947_8553,
                 1.8537_0467,
             ],  # cummulative prob of 5 highest values <= 0.6
         ]			,				dtype=tf.floataa			,				)

     __lowerCAmelCase    :     Optional[int]          =		tf.convert_to_tensor(
         [[0, 0], [0, 9], [0, 10], [0, 25], [0, 26], [1, 13], [1, 17], [1, 18], [1, 20], [1, 27]]			,				dtype=tf.intaa			,				)  # expected non filtered idx as noted above

     __lowerCAmelCase    :     List[Any]          =		tf.convert_to_tensor(
         [8.22_2099, 7.353_4126, 8.43_2078, 7.440_2075, 9.3_8451, 6.27_1159, 8.82_7531, 5.440_2995, 7.385_7956, 9.67_7023]			,				dtype=tf.floataa			,				)  # expected non filtered values as noted above

     __lowerCAmelCase    :     Union[str, Any]          =		tf_top_k_top_p_filtering(lowerCAmelCase			,				top_k=10			,				top_p=0.6			,				min_tokens_to_keep=4       )

     __lowerCAmelCase    :     List[Any]          =		output[output != -float("""inf"""       )]
     __lowerCAmelCase    :     Tuple          =		tf.cast(
         tf.where(tf.not_equal(lowerCAmelCase			,				tf.constant(-float("""inf"""       )			,				dtype=tf.floataa       )       )       )			,				dtype=tf.intaa			,				)

     tf.debugging.assert_near(lowerCAmelCase			,				lowerCAmelCase			,				rtol=1e-12       )
     tf.debugging.assert_equal(lowerCAmelCase			,				lowerCAmelCase       )






@require_tf
class SCREAMING_SNAKE_CASE      (    unittest.TestCase  ,				__A					):
  """simple docstring"""






  if is_tf_available():
    lowerCamelCase       :     List[Any]		 ={
        "AutoModelForCausalLM": TFAutoModelForCausalLM,
        "AutoModelForSpeechSeq2Seq": TFAutoModelForSpeechSeqaSeq,
        "AutoModelForSeq2SeqLM": TFAutoModelForSeqaSeqLM,
        "AutoModelForVision2Seq": TFAutoModelForVisionaSeq,
        "LogitsProcessorList": TFLogitsProcessorList,
        "MinLengthLogitsProcessor": TFMinLengthLogitsProcessor,
        "create_tensor_fn": tf.convert_to_tensor,
        "floats_tensor": floats_tensor,
        "return_tensors": "tf",
    }







  @slow
  def 			SCREAMING_SNAKE_CASE       (     self					:       Optional[int]       )				->      Dict:






     """simple docstring"""







     __lowerCAmelCase    :     Tuple          =		TFAutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""       )
     __lowerCAmelCase    :     str          =		2
     __lowerCAmelCase    :     List[str]          =		2



     class SCREAMING_SNAKE_CASE      (    tf.Module					):
       """simple docstring"""






       def __init__(     self					:       List[str]			,				lowerCAmelCase					:       Optional[int]       )				->      str:






          """simple docstring"""







          super(lowerCAmelCase			,				self       ).__init__()
          __lowerCAmelCase    :     Dict          =		model







       @tf.function(
           input_signature=(
               tf.TensorSpec((None, input_length)			,				tf.intaa			,				name="""input_ids"""       ),
               tf.TensorSpec((None, input_length)			,				tf.intaa			,				name="""attention_mask"""       ),
           )			,				jit_compile=lowerCAmelCase			,				)
       def 			SCREAMING_SNAKE_CASE       (     self					:       Union[str, Any]			,				lowerCAmelCase					:       List[str]			,				lowerCAmelCase					:       Any       )				->      Tuple:






          """simple docstring"""







          __lowerCAmelCase    :     List[Any]          =		self.model.generate(
              input_ids=lowerCAmelCase			,				attention_mask=lowerCAmelCase			,				max_new_tokens=lowerCAmelCase			,				return_dict_in_generate=lowerCAmelCase			,				)
          return {"sequences": outputs["sequences"]}






     __lowerCAmelCase    :     Dict          =		[[2, 0], [1_02, 1_03]]
     __lowerCAmelCase    :     Optional[int]          =		[[1, 0], [1, 1]]
     __lowerCAmelCase    :     Any          =		DummyModel(model=lowerCAmelCase       )
     with tempfile.TemporaryDirectory() as tmp_dir:
        tf.saved_model.save(lowerCAmelCase			,				lowerCAmelCase			,				signatures={"""serving_default""": dummy_model.serving}       )
        __lowerCAmelCase    :     Tuple          =		tf.saved_model.load(lowerCAmelCase       ).signatures["serving_default"]
        for batch_size in range(1			,				len(lowerCAmelCase       ) + 1       ):
           __lowerCAmelCase    :     Optional[int]          =		{
               "input_ids": tf.constant(dummy_input_ids[:batch_size]       ),
               "attention_mask": tf.constant(dummy_attention_masks[:batch_size]       ),
           }
           __lowerCAmelCase    :     Dict          =		serving_func(**lowerCAmelCase       )["sequences"]
           __lowerCAmelCase    :     Union[str, Any]          =		test_model.generate(**lowerCAmelCase			,				max_new_tokens=lowerCAmelCase       )
           tf.debugging.assert_equal(lowerCAmelCase			,				lowerCAmelCase       )







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






     """simple docstring"""







     __lowerCAmelCase    :     List[Any]          =		TFAutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""       )
     __lowerCAmelCase    :     int          =		1
     __lowerCAmelCase    :     Union[str, Any]          =		2



     class SCREAMING_SNAKE_CASE      (    tf.Module					):
       """simple docstring"""






       def __init__(     self					:       List[str]			,				lowerCAmelCase					:       List[Any]       )				->      Optional[int]:






          """simple docstring"""







          super(lowerCAmelCase			,				self       ).__init__()
          __lowerCAmelCase    :     List[Any]          =		model







       @tf.function(
           input_signature=(
               tf.TensorSpec((batch_size, None)			,				tf.intaa			,				name="""input_ids"""       ),
               tf.TensorSpec((batch_size, None)			,				tf.intaa			,				name="""attention_mask"""       ),
           )			,				jit_compile=lowerCAmelCase			,				)
       def 			SCREAMING_SNAKE_CASE       (     self					:       str			,				lowerCAmelCase					:       List[str]			,				lowerCAmelCase					:       str       )				->      Any:






          """simple docstring"""







          __lowerCAmelCase    :     Any          =		self.model.generate(
              input_ids=lowerCAmelCase			,				attention_mask=lowerCAmelCase			,				max_new_tokens=lowerCAmelCase			,				return_dict_in_generate=lowerCAmelCase			,				)
          return {"sequences": outputs["sequences"]}






     __lowerCAmelCase    :     str          =		[[2], [1_02, 1_03]]
     __lowerCAmelCase    :     str          =		[[1], [1, 1]]
     __lowerCAmelCase    :     Optional[Any]          =		DummyModel(model=lowerCAmelCase       )
     with tempfile.TemporaryDirectory() as tmp_dir:
        tf.saved_model.save(lowerCAmelCase			,				lowerCAmelCase			,				signatures={"""serving_default""": dummy_model.serving}       )
        __lowerCAmelCase    :     List[Any]          =		tf.saved_model.load(lowerCAmelCase       ).signatures["serving_default"]
        for input_row in range(len(lowerCAmelCase       )       ):
           __lowerCAmelCase    :     List[str]          =		{
               "input_ids": tf.constant([dummy_input_ids[input_row]]       ),
               "attention_mask": tf.constant([dummy_attention_masks[input_row]]       ),
           }
           __lowerCAmelCase    :     Union[str, Any]          =		serving_func(**lowerCAmelCase       )["sequences"]
           __lowerCAmelCase    :     str          =		test_model.generate(**lowerCAmelCase			,				max_new_tokens=lowerCAmelCase       )
           tf.debugging.assert_equal(lowerCAmelCase			,				lowerCAmelCase       )







  @slow
  @require_tensorflow_text
  def 			SCREAMING_SNAKE_CASE       (     self					:       Tuple       )				->      str:






     """simple docstring"""







     with tempfile.TemporaryDirectory() as tmp_dir:
        # file needed to load the TF tokenizer
        hf_hub_download(repo_id="""google/flan-t5-small"""			,				filename="""spiece.model"""			,				local_dir=lowerCAmelCase       )



        class SCREAMING_SNAKE_CASE      (    tf.keras.layers.Layer					):
          """simple docstring"""






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






             """simple docstring"""







             super().__init__()
             __lowerCAmelCase    :     List[str]          =		text.SentencepieceTokenizer(
                 model=tf.io.gfile.GFile(os.path.join(lowerCAmelCase			,				"""spiece.model"""       )			,				"""rb"""       ).read()       )
             __lowerCAmelCase    :     str          =		TFAutoModelForSeqaSeqLM.from_pretrained("""hf-internal-testing/tiny-random-t5"""       )







          def 			SCREAMING_SNAKE_CASE       (     self					:       Union[str, Any]			,				lowerCAmelCase					:       Optional[Any]			,				*lowerCAmelCase					:       Union[str, Any]			,				**lowerCAmelCase					:       int       )				->      List[str]:






             """simple docstring"""







             __lowerCAmelCase    :     List[Any]          =		self.tokenizer.tokenize(lowerCAmelCase       )
             __lowerCAmelCase    :     Optional[int]          =		text.pad_model_inputs(
                 lowerCAmelCase			,				max_seq_length=64			,				pad_value=self.model.config.pad_token_id       )
             __lowerCAmelCase    :     str          =		self.model.generate(input_ids=lowerCAmelCase			,				attention_mask=lowerCAmelCase       )
             return self.tokenizer.detokenize(lowerCAmelCase       )






        __lowerCAmelCase    :     Union[str, Any]          =		CompleteSentenceTransformer()
        __lowerCAmelCase    :     Optional[Any]          =		tf.keras.layers.Input(shape=(1,)			,				dtype=tf.string			,				name="""inputs"""       )
        __lowerCAmelCase    :     Dict          =		complete_model(lowerCAmelCase       )
        __lowerCAmelCase    :     Union[str, Any]          =		tf.keras.Model(lowerCAmelCase			,				lowerCAmelCase       )
        keras_model.save(lowerCAmelCase       )







  def 			SCREAMING_SNAKE_CASE       (     self					:       int       )				->      Dict:






     """simple docstring"""







     __lowerCAmelCase    :     List[str]          =		{
         "do_sample": True,
         "num_beams": 1,
         "top_p": 0.7,
         "top_k": 10,
         "temperature": 0.7,
     }
     __lowerCAmelCase    :     Dict          =		14

     __lowerCAmelCase    :     int          =		AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""       )
     __lowerCAmelCase    :     int          =		"Hello, my dog is cute and"
     __lowerCAmelCase    :     Optional[Any]          =		tokenizer(lowerCAmelCase			,				return_tensors="""tf"""       )
     __lowerCAmelCase    :     str          =		TFAutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""       )

     __lowerCAmelCase    :     Dict          =		6_38
     # forces the generation to happen on CPU, to avoid GPU-related quirks
     with tf.device(""":/CPU:0"""       ):
        tf.random.set_seed(0       )
        __lowerCAmelCase    :     Union[str, Any]          =		model.generate(**lowerCAmelCase			,				eos_token_id=lowerCAmelCase			,				**lowerCAmelCase       )
     self.assertTrue(expectation == len(generated_tokens[0]       )       )

     __lowerCAmelCase    :     Union[str, Any]          =		[6_38, 1_98]
     with tf.device(""":/CPU:0"""       ):
        tf.random.set_seed(0       )
        __lowerCAmelCase    :     Optional[Any]          =		model.generate(**lowerCAmelCase			,				eos_token_id=lowerCAmelCase			,				**lowerCAmelCase       )
     self.assertTrue(expectation == len(generated_tokens[0]       )       )







  def 			SCREAMING_SNAKE_CASE       (     self					:       Tuple       )				->      str:






     """simple docstring"""







     __lowerCAmelCase    :     Tuple          =		AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-bart"""       )
     __lowerCAmelCase    :     Optional[int]          =		"Hugging Face is a technology company based in New York and Paris."
     __lowerCAmelCase    :     List[Any]          =		bart_tokenizer(lowerCAmelCase			,				return_tensors="""tf"""       ).input_ids
     __lowerCAmelCase    :     Dict          =		TFBartForConditionalGeneration.from_pretrained("""hf-internal-testing/tiny-random-bart"""       )
     __lowerCAmelCase    :     Dict          =		bart_model.generate(lowerCAmelCase       ).numpy()



     class SCREAMING_SNAKE_CASE      (    __A					):
       """simple docstring"""






       def 			SCREAMING_SNAKE_CASE       (     self					:       Optional[int]			,				lowerCAmelCase					:       int			,				lowerCAmelCase					:       List[str]=None			,				**lowerCAmelCase					:       List[str]       )				->      Optional[int]:






          """simple docstring"""







          return super().call(lowerCAmelCase			,				**lowerCAmelCase       )






     __lowerCAmelCase    :     Optional[int]          =		FakeBart.from_pretrained("""hf-internal-testing/tiny-random-bart"""       )
     __lowerCAmelCase    :     Union[str, Any]          =		bart_model.generate(lowerCAmelCase			,				foo="""bar"""       ).numpy()
     self.assertTrue(np.array_equal(lowerCAmelCase			,				lowerCAmelCase       )       )



     class SCREAMING_SNAKE_CASE      (    bart_model.model.encoder.__class__					):
       """simple docstring"""






       def 			SCREAMING_SNAKE_CASE       (     self					:       Tuple			,				lowerCAmelCase					:       Any			,				**lowerCAmelCase					:       List[Any]       )				->      Optional[Any]:






          """simple docstring"""







          return super().call(lowerCAmelCase			,				**lowerCAmelCase       )






     __lowerCAmelCase    :     int          =		FakeEncoder(bart_model.config			,				bart_model.model.shared       )
     __lowerCAmelCase    :     List[str]          =		fake_encoder

     # Normal generation still works (the output will be different because the encoder weights are different)
     __lowerCAmelCase    :     Union[str, Any]          =		bart_model.generate(lowerCAmelCase       ).numpy()
     with self.assertRaises(lowerCAmelCase       ):
        # FakeEncoder.call() accepts **kwargs -> no filtering -> value error due to unexpected input "foo"
        bart_model.generate(lowerCAmelCase			,				foo="""bar"""       )





from operator import delitem, getitem, setitem

import pytest

from data_structures.hashing.hash_map import HashMap

def 			lowerCamelCase__			(       _a):
	return getitem, k

def 			lowerCamelCase__			(       _a				,      _a):
	return setitem, k, v

def 			lowerCamelCase__			(       _a):
	return delitem, k

def 			lowerCamelCase__			(       _a				,      _a				,      *_a):
	try:
		return fun(_a				,      *_a), None
	except Exception as e:
		return None, e


a_      =     (
    _set('key_a', 'val_a'),
    _set('key_b', 'val_b'),
)

a_      =     [
    _set('key_a', 'val_a'),
    _set('key_a', 'val_b'),
]

a_      =     [
    _set('key_a', 'val_a'),
    _set('key_b', 'val_b'),
    _del('key_a'),
    _del('key_b'),
    _set('key_a', 'val_a'),
    _del('key_a'),
]

a_      =     [
    _get('key_a'),
    _del('key_a'),
    _set('key_a', 'val_a'),
    _del('key_a'),
    _del('key_a'),
    _get('key_a'),
]

a_      =     [
    *[_set(x, x) for x in range(5)],  # guaranteed upsize
]

a_      =     [
    *[_set(x, x) for x in range(5)],  # guaranteed upsize
    *[_del(x) for x in range(5)],
    _set('key_a', 'val_b'),
]

@pytest.mark.parametrize(
    "operations"				,      (
        pytest.param(_add_items				,      id="add items"),
        pytest.param(_overwrite_items				,      id="overwrite items"),
        pytest.param(_delete_items				,      id="delete items"),
        pytest.param(_access_absent_items				,      id="access absent items"),
        pytest.param(_add_with_resize_up				,      id="add with resize up"),
        pytest.param(_add_with_resize_down				,      id="add with resize down"),
    )				,      )
def 			lowerCamelCase__			(       _a):
	SCREAMING_SNAKE_CASE :   Dict					  =							HashMap(initial_block_size=4)
	SCREAMING_SNAKE_CASE :   List[str]					  =							{}
	for _, (fun, *args) in enumerate(_a):
		SCREAMING_SNAKE_CASE  ,SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							_run_operation(_a				,      _a				,      *_a)
		SCREAMING_SNAKE_CASE  ,SCREAMING_SNAKE_CASE :   List[Any]					  =							_run_operation(_a				,      _a				,      *_a)
		assert my_res == py_res
		assert str(_a) == str(_a)
		assert set(_a) == set(_a)
		assert len(_a) == len(_a)
		assert set(my.items()) == set(py.items())


def 			lowerCamelCase__			(       ):
	def is_public(_a) -> bool:
		return not name.startswith("_")

	SCREAMING_SNAKE_CASE :   List[str]					  =							{name for name in dir({}) if is_public(_a)}
	SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							{name for name in dir(HashMap()) if is_public(_a)}

	assert dict_public_names > hash_public_names