



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  __snake_case      (   unittest.TestCase):







							'''simple docstring'''






							def _a      (							self		):
											a__				      =		tf.convert_to_tensor(
											    [
											        [
											            8.2_220_991,  # 3rd highest value; idx. 0
											            -0.5_620_044,
											            5.23_229_752,
											            4.0_386_393,
											            -6.8_798_378,
											            -0.54_785_802,
											            -3.2_012_153,
											            2.92_777_176,
											            1.88_171_953,
											            7.35_341_276,  # 5th highest value; idx. 9
											            8.43_207_833,  # 2nd highest value; idx. 10
											            -9.85_711_836,
											            -5.96_209_236,
											            -1.13_039_161,
											            -7.1_115_294,
											            -0.8_369_633,
											            -5.3_186_408,
											            7.06_427_407,
											            0.81_369_344,
											            -0.82_023_817,
											            -5.9_179_796,
											            0.58_813_443,
											            -6.99_778_438,
											            4.71_551_189,
											            -0.18_771_637,
											            7.44_020_759,  # 4th highest value; idx. 25
											            9.38_450_987,  # 1st highest value; idx. 26
											            2.12_662_941,
											            -9.32_562_038,
											            2.35_652_522,
											        ],  # cummulative prob of 5 highest values <= 0.6
											        [
											            0.58_425_518,
											            4.53_139_238,
											            -5.57_510_464,
											            -6.28_030_699,
											            -7.19_529_503,
											            -4.02_122_551,
											            1.39_337_037,
											            -6.06_707_057,
											            1.59_480_517,
											            -9.643_119,
											            0.03_907_799,
											            0.67_231_762,
											            -8.88_206_726,
											            6.27_115_922,  # 4th highest value; idx. 13
											            2.28_520_723,
											            4.82_767_506,
											            4.30_421_368,
											            8.8_275_313,  # 2nd highest value; idx. 17
											            5.44_029_958,  # 5th highest value; idx. 18
											            -4.4_735_794,
											            7.38_579_536,  # 3rd highest value; idx. 20
											            -2.91_051_663,
											            2.61_946_077,
											            -2.5_674_762,
											            -9.48_959_302,
											            -4.02_922_645,
											            -1.35_416_918,
											            9.67_702_323,  # 1st highest value; idx. 27
											            -5.89_478_553,
											            1.85_370_467,
											        ],  # cummulative prob of 5 highest values <= 0.6
											    ]  ,  dtype=tf.floataa  ,  )

											a__				      =		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

											a__				      =		tf.convert_to_tensor(
											    [8.222_099, 7.3_534_126, 8.432_078, 7.4_402_075, 9.38_451, 6.271_159, 8.827_531, 5.4_402_995, 7.3_857_956, 9.677_023]  ,  dtype=tf.floataa  ,  )  # expected non filtered values as noted above

											a__				      =		tf_top_k_top_p_filtering(a_  ,  top_k=10  ,  top_p=0.6  ,  min_tokens_to_keep=4		)

											a__				      =		output[output != -float("""inf"""		)]
											a__				      =		tf.cast(
											    tf.where(tf.not_equal(a_  ,  tf.constant(-float("""inf"""		)  ,  dtype=tf.floataa		)		)		)  ,  dtype=tf.intaa  ,  )

											tf.debugging.assert_near(a_  ,  a_  ,  rtol=1E-1_2		)
											tf.debugging.assert_equal(a_  ,  a_		)




@require_tf
class  __snake_case      (   unittest.TestCase			,SCREAMING_SNAKE_CASE):







							'''simple docstring'''






							if is_tf_available():
														UpperCamelCase__				:      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 _a      (							self		):
											# TF-only test: tf.saved_model export
											a__				      =		TFAutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""		)
											a__				      =		2
											a__				      =		2

											class  __snake_case      (   tf.Module):







																		'''simple docstring'''






																		def __init__(							self  ,  a_		):
																						super(a_  ,  self		).__init__()
																						a__				      =		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=a_  ,  )
																		def _a      (							self  ,  a_  ,  a_		):
																						a__				      =		self.model.generate(
																						    input_ids=a_  ,  attention_mask=a_  ,  max_new_tokens=a_  ,  return_dict_in_generate=a_  ,  )
																						return {"sequences": outputs["sequences"]}




											a__				      =		[[2, 0], [102, 103]]
											a__				      =		[[1, 0], [1, 1]]
											a__				      =		DummyModel(model=a_		)
											with tempfile.TemporaryDirectory() as tmp_dir:
															tf.saved_model.save(a_  ,  a_  ,  signatures={"""serving_default""": dummy_model.serving}		)
															a__				      =		tf.saved_model.load(a_		).signatures["""serving_default"""]
															for batch_size in range(1  ,  len(a_		) + 1		):
																			a__				      =		{
																			    """input_ids""": tf.constant(dummy_input_ids[:batch_size]		),
																			    """attention_mask""": tf.constant(dummy_attention_masks[:batch_size]		),
																			}
																			a__				      =		serving_func(**a_		)["""sequences"""]
																			a__				      =		test_model.generate(**a_  ,  max_new_tokens=a_		)
																			tf.debugging.assert_equal(a_  ,  a_		)


							@slow
							def _a      (							self		):
											# TF-only test: tf.saved_model export
											a__				      =		TFAutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""		)
											a__				      =		1
											a__				      =		2

											class  __snake_case      (   tf.Module):







																		'''simple docstring'''






																		def __init__(							self  ,  a_		):
																						super(a_  ,  self		).__init__()
																						a__				      =		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=a_  ,  )
																		def _a      (							self  ,  a_  ,  a_		):
																						a__				      =		self.model.generate(
																						    input_ids=a_  ,  attention_mask=a_  ,  max_new_tokens=a_  ,  return_dict_in_generate=a_  ,  )
																						return {"sequences": outputs["sequences"]}




											a__				      =		[[2], [102, 103]]
											a__				      =		[[1], [1, 1]]
											a__				      =		DummyModel(model=a_		)
											with tempfile.TemporaryDirectory() as tmp_dir:
															tf.saved_model.save(a_  ,  a_  ,  signatures={"""serving_default""": dummy_model.serving}		)
															a__				      =		tf.saved_model.load(a_		).signatures["""serving_default"""]
															for input_row in range(len(a_		)		):
																			a__				      =		{
																			    """input_ids""": tf.constant([dummy_input_ids[input_row]]		),
																			    """attention_mask""": tf.constant([dummy_attention_masks[input_row]]		),
																			}
																			a__				      =		serving_func(**a_		)["""sequences"""]
																			a__				      =		test_model.generate(**a_  ,  max_new_tokens=a_		)
																			tf.debugging.assert_equal(a_  ,  a_		)


							@slow
							@require_tensorflow_text
							def _a      (							self		):
											# TF-only test: tf.saved_model export
											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=a_		)

															class  __snake_case      (   tf.keras.layers.Layer):







																						'''simple docstring'''






																						def __init__(							self		):
																										super().__init__()
																										a__				      =		text.SentencepieceTokenizer(
																										    model=tf.io.gfile.GFile(os.path.join(a_  ,  """spiece.model"""		)  ,  """rb"""		).read()		)
																										a__				      =		TFAutoModelForSeqaSeqLM.from_pretrained("""hf-internal-testing/tiny-random-t5"""		)







																						def _a      (							self  ,  a_  ,  *a_  ,  **a_		):
																										a__				      =		self.tokenizer.tokenize(a_		)
																										a__      ,			a__				      =		text.pad_model_inputs(
																										    a_  ,  max_seq_length=64  ,  pad_value=self.model.config.pad_token_id		)
																										a__				      =		self.model.generate(input_ids=a_  ,  attention_mask=a_		)
																										return self.tokenizer.detokenize(a_		)




															a__				      =		CompleteSentenceTransformer()
															a__				      =		tf.keras.layers.Input(shape=(1,)  ,  dtype=tf.string  ,  name="""inputs"""		)
															a__				      =		complete_model(a_		)
															a__				      =		tf.keras.Model(a_  ,  a_		)
															keras_model.save(a_		)


							def _a      (							self		):
											# Has PT equivalent: this test relies on random sampling
											a__				      =		{
											    """do_sample""": True,
											    """num_beams""": 1,
											    """top_p""": 0.7,
											    """top_k""": 10,
											    """temperature""": 0.7,
											}
											a__				      =		14

											a__				      =		AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""		)
											a__				      =		"""Hello, my dog is cute and"""
											a__				      =		tokenizer(a_  ,  return_tensors="""tf"""		)
											a__				      =		TFAutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""		)

											a__				      =		638
											# forces the generation to happen on CPU, to avoid GPU-related quirks
											with tf.device(""":/CPU:0"""		):
															tf.random.set_seed(0		)
															a__				      =		model.generate(**a_  ,  eos_token_id=a_  ,  **a_		)
											self.assertTrue(expectation == len(generated_tokens[0]		)		)

											a__				      =		[638, 198]
											with tf.device(""":/CPU:0"""		):
															tf.random.set_seed(0		)
															a__				      =		model.generate(**a_  ,  eos_token_id=a_  ,  **a_		)
											self.assertTrue(expectation == len(generated_tokens[0]		)		)







							def _a      (							self		):
											# Has PT equivalent: ample use of framework-specific code
											a__				      =		AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-bart"""		)
											a__				      =		"""Hugging Face is a technology company based in New York and Paris."""
											a__				      =		bart_tokenizer(a_  ,  return_tensors="""tf"""		).input_ids
											a__				      =		TFBartForConditionalGeneration.from_pretrained("""hf-internal-testing/tiny-random-bart"""		)
											a__				      =		bart_model.generate(a_		).numpy()

											class  __snake_case      (   SCREAMING_SNAKE_CASE):







																		'''simple docstring'''






																		def _a      (							self  ,  a_  ,  a_=None  ,  **a_		):
																						return super().call(a_  ,  **a_		)




											a__				      =		FakeBart.from_pretrained("""hf-internal-testing/tiny-random-bart"""		)
											a__				      =		bart_model.generate(a_  ,  foo="""bar"""		).numpy()
											self.assertTrue(np.array_equal(a_  ,  a_		)		)

											class  __snake_case      (   bart_model.model.encoder.__class__):







																		'''simple docstring'''






																		def _a      (							self  ,  a_  ,  **a_		):
																						return super().call(a_  ,  **a_		)




											a__				      =		FakeEncoder(bart_model.config  ,  bart_model.model.shared		)
											a__				      =		fake_encoder

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






import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
	from .features import FeatureType


UpperCAmelCase ,       UpperCAmelCase ,       UpperCAmelCase							 =					False, False, False

@dataclass
class  __snake_case      :







							'''simple docstring'''






							UpperCamelCase__				:      Optional[int]         =					None
							UpperCamelCase__				:      bool         =					True
							UpperCamelCase__				:      bool         =					True
							UpperCamelCase__				:      Optional[str]         =					None
							# Automatically constructed
							UpperCamelCase__				:      ClassVar[str]         =					"dict"
							UpperCamelCase__				:      ClassVar[Any]         =					pa.struct({"""bytes""": pa.binary(), """path""": pa.string()})
							UpperCamelCase__				:      str         =					field(default="""Audio"""			,init=SCREAMING_SNAKE_CASE			,repr=SCREAMING_SNAKE_CASE)


							def __call__(							self		):
											return self.pa_type


							def _a      (							self  ,  a_		):
											try:
															import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
											except ImportError as err:
															raise ImportError("""To support encoding audio data, please install 'soundfile'."""		) from err
											if isinstance(a_  ,  a_		):
															return {"bytes": None, "path": value}
											elif isinstance(a_  ,  a_		):
															return {"bytes": value, "path": None}
											elif "array" in value:
															# convert the audio array to wav bytes
															a__				      =		BytesIO()
															sf.write(a_  ,  value["""array"""]  ,  value["""sampling_rate"""]  ,  format="""wav"""		)
															return {"bytes": buffer.getvalue(), "path": None}
											elif value.get("""path"""		) is not None and os.path.isfile(value["""path"""]		):
															# we set "bytes": None to not duplicate the data if they're already available locally
															if value["path"].endswith("""pcm"""		):
																			# "PCM" only has raw audio bytes
																			if value.get("""sampling_rate"""		) is None:
																							# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
																							raise KeyError("""To use PCM files, please specify a 'sampling_rate' in Audio object"""		)
																			if value.get("""bytes"""		):
																							# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
																							a__				      =		np.frombuffer(value["""bytes"""]  ,  dtype=np.intaa		).astype(np.floataa		) / 32_767
																			else:
																							a__				      =		np.memmap(value["""path"""]  ,  dtype="""h"""  ,  mode="""r"""		).astype(np.floataa		) / 32_767

																			a__				      =		BytesIO(bytes()		)
																			sf.write(a_  ,  a_  ,  value["""sampling_rate"""]  ,  format="""wav"""		)
																			return {"bytes": buffer.getvalue(), "path": None}
															else:
																			return {"bytes": None, "path": value.get("""path"""		)}
											elif value.get("""bytes"""		) is not None or value.get("""path"""		) is not None:
															# store the audio bytes, and path is used to infer the audio format using the file extension
															return {"bytes": value.get("""bytes"""		), "path": value.get("""path"""		)}
											else:
															raise ValueError(
															    F'''An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.'''		)


							def _a      (							self  ,  a_  ,  a_ = None		):
											if not self.decode:
															raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""		)

											a__      ,			a__				      =		(value["""path"""], BytesIO(value["""bytes"""]		)) if value["""bytes"""] is not None else (value["""path"""], None)
											if path is None and file is None:
															raise ValueError(F'''An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.'''		)

											try:
															import librosa
															import soundfile as sf
											except ImportError as err:
															raise ImportError("""To support decoding audio files, please install 'librosa' and 'soundfile'."""		) from err

											a__				      =		xsplitext(a_		)[1][1:].lower() if path is not None else None
											if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
															raise RuntimeError(
															    """Decoding 'opus' files requires system library 'libsndfile'>=1.0.31, """
															    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """		)
											elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
															raise RuntimeError(
															    """Decoding 'mp3' files requires system library 'libsndfile'>=1.1.0, """
															    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """		)

											if file is None:
															a__				      =		token_per_repo_id or {}
															a__				      =		path.split("""::"""		)[-1]
															try:
																			a__				      =		string_to_dict(a_  ,  config.HUB_DATASETS_URL		)["""repo_id"""]
																			a__				      =		token_per_repo_id[repo_id]
															except (ValueError, KeyError):
																			a__				      =		None

															with xopen(a_  ,  """rb"""  ,  use_auth_token=a_		) as f:
																			a__      ,			a__				      =		sf.read(a_		)

											else:
															a__      ,			a__				      =		sf.read(a_		)

											a__				      =		array.T
											if self.mono:
															a__				      =		librosa.to_mono(a_		)
											if self.sampling_rate and self.sampling_rate != sampling_rate:
															a__				      =		librosa.resample(a_  ,  orig_sr=a_  ,  target_sr=self.sampling_rate		)
															a__				      =		self.sampling_rate

											return {"path": path, "array": array, "sampling_rate": sampling_rate}


							def _a      (							self		):
											from .features import Value

											if self.decode:
															raise ValueError("""Cannot flatten a decoded Audio feature."""		)
											return {
											    "bytes": Value("""binary"""		),
											    "path": Value("""string"""		),
											}


							def _a      (							self  ,  a_		):
											if pa.types.is_string(storage.type		):
															a__				      =		pa.array([None] * len(a_		)  ,  type=pa.binary()		)
															a__				      =		pa.StructArray.from_arrays([bytes_array, storage]  ,  ["""bytes""", """path"""]  ,  mask=storage.is_null()		)
											elif pa.types.is_binary(storage.type		):
															a__				      =		pa.array([None] * len(a_		)  ,  type=pa.string()		)
															a__				      =		pa.StructArray.from_arrays([storage, path_array]  ,  ["""bytes""", """path"""]  ,  mask=storage.is_null()		)
											elif pa.types.is_struct(storage.type		) and storage.type.get_all_field_indices("""array"""		):
															a__				      =		pa.array([Audio().encode_example(a_		) if x is not None else None for x in storage.to_pylist()]		)
											elif pa.types.is_struct(storage.type		):
															if storage.type.get_field_index("""bytes"""		) >= 0:
																			a__				      =		storage.field("""bytes"""		)
															else:
																			a__				      =		pa.array([None] * len(a_		)  ,  type=pa.binary()		)
															if storage.type.get_field_index("""path"""		) >= 0:
																			a__				      =		storage.field("""path"""		)
															else:
																			a__				      =		pa.array([None] * len(a_		)  ,  type=pa.string()		)
															a__				      =		pa.StructArray.from_arrays([bytes_array, path_array]  ,  ["""bytes""", """path"""]  ,  mask=storage.is_null()		)
											return array_cast(a_  ,  self.pa_type		)







							def _a      (							self  ,  a_		):

											@no_op_if_value_is_null
											def path_to_bytes(a_		):
															with xopen(a_  ,  """rb"""		) as f:
																			a__				      =		f.read()
															return bytes_

											a__				      =		pa.array(
											    [
											        (path_to_bytes(x["""path"""]		) if x["""bytes"""] is None else x["""bytes"""]) if x is not None else None
											        for x in storage.to_pylist()
											    ]  ,  type=pa.binary()  ,  )
											a__				      =		pa.array(
											    [os.path.basename(a_		) if path is not None else None for path in storage.field("""path"""		).to_pylist()]  ,  type=pa.string()  ,  )
											a__				      =		pa.StructArray.from_arrays([bytes_array, path_array]  ,  ["""bytes""", """path"""]  ,  mask=bytes_array.is_null()		)
											return array_cast(a_  ,  self.pa_type		)

