import math

import tensorflow as tf
from packaging import version
def 						__UpperCAmelCase       (					UpperCAmelCase_					:       Any						)							->    int:

	'''simple docstring'''





	__snake_case    :							str   	=				tf.convert_to_tensor(UpperCAmelCase_						)
	__snake_case    :							Optional[int]   	=				0.5 * (1.0 + tf.math.erf(x / tf.cast(tf.sqrt(2.0						)    ,							x.dtype						)						))

	return x * cdf
def 						__UpperCAmelCase       (					UpperCAmelCase_					:       Optional[Any]						)							->    Any:

	'''simple docstring'''





	__snake_case    :							Tuple   	=				tf.convert_to_tensor(UpperCAmelCase_						)
	__snake_case    :							Dict   	=				tf.cast(math.pi    ,							x.dtype						)
	__snake_case    :							Optional[Any]   	=				tf.cast(0.044_715    ,							x.dtype						)
	__snake_case    :							List[Any]   	=				0.5 * (1.0 + tf.tanh(tf.sqrt(2.0 / pi						) * (x + coeff * tf.pow(UpperCAmelCase_    ,							3						))						))

	return x * cdf
def 						__UpperCAmelCase       (					UpperCAmelCase_					:       Any						)							->    Optional[int]:

	'''simple docstring'''





	__snake_case    :							Optional[int]   	=				tf.convert_to_tensor(UpperCAmelCase_						)

	return x * tf.tanh(tf.math.softplus(UpperCAmelCase_						)						)
def 						__UpperCAmelCase       (					UpperCAmelCase_					:       Union[str, Any]						)							->    List[Any]:

	'''simple docstring'''





	__snake_case    :							str   	=				tf.convert_to_tensor(UpperCAmelCase_						)
	__snake_case    :							Tuple   	=				tf.cast(0.044_715    ,							x.dtype						)
	__snake_case    :							Union[str, Any]   	=				tf.cast(0.7_978_845_608    ,							x.dtype						)

	return 0.5 * x * (1.0 + tf.tanh(x * coeffa * (1.0 + coeffa * x * x)						))
def 						__UpperCAmelCase       (					UpperCAmelCase_					:       str						)							->    Dict:

	'''simple docstring'''





	__snake_case    :							Dict   	=				tf.convert_to_tensor(UpperCAmelCase_						)
	__snake_case    :							Dict   	=				tf.cast(1.702    ,							x.dtype						)
	return x * tf.math.sigmoid(coeff * x						)
def 						__UpperCAmelCase       (					UpperCAmelCase_					:       List[str]						)							->    Optional[Any]:

	'''simple docstring'''





	return tf.clip_by_value(_gelu(UpperCAmelCase_						)    ,							-10    ,							10						)
def 						__UpperCAmelCase       (					UpperCAmelCase_					:       List[str]    ,							UpperCAmelCase_					:       Optional[Any]=-1						)							->    List[str]:

	'''simple docstring'''





	__snake_case	,       __snake_case    :							Optional[int]   	=				tf.split(UpperCAmelCase_    ,							2    ,							axis=UpperCAmelCase_						)
	return a * tf.math.sigmoid(UpperCAmelCase_						)


if version.parse(tf.version.VERSION) >= version.parse("2.4"):
	def 						__UpperCAmelCase       (					UpperCAmelCase_					:       Tuple						)							->    int:

		'''simple docstring'''





		return tf.keras.activations.gelu(UpperCAmelCase_    ,							approximate=UpperCAmelCase_						)

	_a  :  Tuple=				tf.keras.activations.gelu

	_a  :  Dict=				approximate_gelu_wrap
else:
	_a  :  Optional[int]=				_gelu
	_a  :  Optional[int]=				_gelu_new


_a  :  Union[str, Any]=				{
    """gelu""": gelu,
    """gelu_10""": gelu_aa,
    """gelu_fast""": gelu_fast,
    """gelu_new""": gelu_new,
    """glu""": glu,
    """mish""": mish,
    """quick_gelu""": quick_gelu,
    """relu""": tf.keras.activations.relu,
    """sigmoid""": tf.keras.activations.sigmoid,
    """silu""": tf.keras.activations.swish,
    """swish""": tf.keras.activations.swish,
    """tanh""": tf.keras.activations.tanh,
}

def 						__UpperCAmelCase       (					UpperCAmelCase_					:       str						)							->    Any:

	'''simple docstring'''





	if activation_string in ACTaFN:
		return ACTaFN[activation_string]
	else:
		raise KeyError(F"function {activation_string} not found in ACT2FN mapping {list(ACTaFN.keys()						)}"						)






"""simple docstring"""







import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
	import torch

	from transformers import AutoModelForCausalLM

@require_torch
class 	UpperCamelCase    (					unittest.TestCase		):



				def       _lowercase  (self       :   List[Any])					->			Dict:
					__snake_case    :							int   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							Optional[Any]   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							Union[str, Any]   	=				-1

					__snake_case    :							Optional[Any]   	=				ids_tensor((1, 5)			,						vocab_size=model.config.vocab_size).to(_A)
					__snake_case    :							Any   	=				model.generate(_A			,						max_new_tokens=10			,						do_sample=_A)
					__snake_case    :							str   	=				tokenizer.decode(greedy_ids[0])

					with CaptureStdout() as cs:
						__snake_case    :							List[str]   	=				TextStreamer(_A)
						model.generate(_A			,						max_new_tokens=10			,						do_sample=_A			,						streamer=_A)
					# The greedy text should be printed to stdout, except for the final "\n" in the streamer
					__snake_case    :							Optional[Any]   	=				cs.out[:-1]

					self.assertEqual(_A			,						_A)



				def       _lowercase  (self       :   int)					->			Optional[Any]:
					__snake_case    :							Any   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							List[Any]   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							Optional[int]   	=				-1

					__snake_case    :							str   	=				ids_tensor((1, 5)			,						vocab_size=model.config.vocab_size).to(_A)
					__snake_case    :							int   	=				model.generate(_A			,						max_new_tokens=10			,						do_sample=_A)
					__snake_case    :							int   	=				tokenizer.decode(greedy_ids[0])

					__snake_case    :							int   	=				TextIteratorStreamer(_A)
					__snake_case    :							List[Any]   	=				{'input_ids': input_ids, 'max_new_tokens': 10, 'do_sample': False, 'streamer': streamer}
					__snake_case    :							Any   	=				Thread(target=model.generate			,						kwargs=_A)
					thread.start()
					__snake_case    :							Dict   	=				''
					for new_text in streamer:
						streamer_text += new_text

					self.assertEqual(_A			,						_A)



				def       _lowercase  (self       :   Any)					->			List[str]:
					__snake_case    :							List[str]   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							Any   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							Union[str, Any]   	=				-1

					__snake_case    :							List[Any]   	=				ids_tensor((1, 5)			,						vocab_size=model.config.vocab_size).to(_A)
					__snake_case    :							Optional[int]   	=				model.generate(_A			,						max_new_tokens=10			,						do_sample=_A)
					__snake_case    :							Union[str, Any]   	=				greedy_ids[:, input_ids.shape[1] :]
					__snake_case    :							Optional[Any]   	=				tokenizer.decode(new_greedy_ids[0])

					with CaptureStdout() as cs:
						__snake_case    :							str   	=				TextStreamer(_A			,						skip_prompt=_A)
						model.generate(_A			,						max_new_tokens=10			,						do_sample=_A			,						streamer=_A)
					# The greedy text should be printed to stdout, except for the final "\n" in the streamer
					__snake_case    :							List[str]   	=				cs.out[:-1]

					self.assertEqual(_A			,						_A)



				def       _lowercase  (self       :   List[str])					->			Dict:
					# Tests that we can pass `decode_kwargs` to the streamer to control how the tokens are decoded. Must be tested
					# with actual models -- the dummy models' tokenizers are not aligned with their models, and
					# `skip_special_tokens=True` has no effect on them
					__snake_case    :							int   	=				AutoTokenizer.from_pretrained('distilgpt2')
					__snake_case    :							Dict   	=				AutoModelForCausalLM.from_pretrained('distilgpt2').to(_A)
					__snake_case    :							Union[str, Any]   	=				-1

					__snake_case    :							List[str]   	=				torch.ones((1, 5)			,						device=_A).long() * model.config.bos_token_id
					with CaptureStdout() as cs:
						__snake_case    :							Tuple   	=				TextStreamer(_A			,						skip_special_tokens=_A)
						model.generate(_A			,						max_new_tokens=1			,						do_sample=_A			,						streamer=_A)

					# The prompt contains a special token, so the streamer should not print it. As such, the output text, when
					# re-tokenized, must only contain one token
					__snake_case    :							Tuple   	=				cs.out[:-1]  # Remove the final "\n"
					__snake_case    :							int   	=				tokenizer(_A			,						return_tensors='pt')
					self.assertEqual(streamer_text_tokenized.input_ids.shape			,						(1, 1))






				def       _lowercase  (self       :   Optional[int])					->			List[str]:
					__snake_case    :							Optional[int]   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							Optional[int]   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							int   	=				-1

					__snake_case    :							List[Any]   	=				ids_tensor((1, 5)			,						vocab_size=model.config.vocab_size).to(_A)
					__snake_case    :							List[Any]   	=				TextIteratorStreamer(_A			,						timeout=0.001)
					__snake_case    :							str   	=				{'input_ids': input_ids, 'max_new_tokens': 10, 'do_sample': False, 'streamer': streamer}
					__snake_case    :							str   	=				Thread(target=model.generate			,						kwargs=_A)
					thread.start()

					# The streamer will timeout after 0.001 seconds, so an exception will be raised
					with self.assertRaises(_A):
						__snake_case    :							Any   	=				''
						for new_text in streamer:
							streamer_text += new_text





