






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  _lowercase  (      self  :						Optional[int] ):
						snake_case__				:   Optional[int]     =						tf.convert_to_tensor(
						    [
						        [
						            8.2_2_2_0_9_9_1,  # 3rd highest value; idx. 0
						            -0.5_6_2_0_0_4_4,
						            5.2_3_2_2_9_7_5_2,
						            4.0_3_8_6_3_9_3,
						            -6.8_7_9_8_3_7_8,
						            -0.5_4_7_8_5_8_0_2,
						            -3.2_0_1_2_1_5_3,
						            2.9_2_7_7_7_1_7_6,
						            1.8_8_1_7_1_9_5_3,
						            7.3_5_3_4_1_2_7_6,  # 5th highest value; idx. 9
						            8.4_3_2_0_7_8_3_3,  # 2nd highest value; idx. 10
						            -9.8_5_7_1_1_8_3_6,
						            -5.9_6_2_0_9_2_3_6,
						            -1.1_3_0_3_9_1_6_1,
						            -7.1_1_1_5_2_9_4,
						            -0.8_3_6_9_6_3_3,
						            -5.3_1_8_6_4_0_8,
						            7.0_6_4_2_7_4_0_7,
						            0.8_1_3_6_9_3_4_4,
						            -0.8_2_0_2_3_8_1_7,
						            -5.9_1_7_9_7_9_6,
						            0.5_8_8_1_3_4_4_3,
						            -6.9_9_7_7_8_4_3_8,
						            4.7_1_5_5_1_1_8_9,
						            -0.1_8_7_7_1_6_3_7,
						            7.4_4_0_2_0_7_5_9,  # 4th highest value; idx. 25
						            9.3_8_4_5_0_9_8_7,  # 1st highest value; idx. 26
						            2.1_2_6_6_2_9_4_1,
						            -9.3_2_5_6_2_0_3_8,
						            2.3_5_6_5_2_5_2_2,
						        ],  # cummulative prob of 5 highest values <= 0.6
						        [
						            0.5_8_4_2_5_5_1_8,
						            4.5_3_1_3_9_2_3_8,
						            -5.5_7_5_1_0_4_6_4,
						            -6.2_8_0_3_0_6_9_9,
						            -7.1_9_5_2_9_5_0_3,
						            -4.0_2_1_2_2_5_5_1,
						            1.3_9_3_3_7_0_3_7,
						            -6.0_6_7_0_7_0_5_7,
						            1.5_9_4_8_0_5_1_7,
						            -9.6_4_3_1_1_9,
						            0.0_3_9_0_7_7_9_9,
						            0.6_7_2_3_1_7_6_2,
						            -8.8_8_2_0_6_7_2_6,
						            6.2_7_1_1_5_9_2_2,  # 4th highest value; idx. 13
						            2.2_8_5_2_0_7_2_3,
						            4.8_2_7_6_7_5_0_6,
						            4.3_0_4_2_1_3_6_8,
						            8.8_2_7_5_3_1_3,  # 2nd highest value; idx. 17
						            5.4_4_0_2_9_9_5_8,  # 5th highest value; idx. 18
						            -4.4_7_3_5_7_9_4,
						            7.3_8_5_7_9_5_3_6,  # 3rd highest value; idx. 20
						            -2.9_1_0_5_1_6_6_3,
						            2.6_1_9_4_6_0_7_7,
						            -2.5_6_7_4_7_6_2,
						            -9.4_8_9_5_9_3_0_2,
						            -4.0_2_9_2_2_6_4_5,
						            -1.3_5_4_1_6_9_1_8,
						            9.6_7_7_0_2_3_2_3,  # 1st highest value; idx. 27
						            -5.8_9_4_7_8_5_5_3,
						            1.8_5_3_7_0_4_6_7,
						        ],  # cummulative prob of 5 highest values <= 0.6
						    ]							,	dtype=tf.floataa							,	)

						snake_case__				:   Optional[Any]     =						tf.convert_to_tensor(
						    [[0, 0], [0, 9], [0, 1_0], [0, 2_5], [0, 2_6], [1, 1_3], [1, 1_7], [1, 1_8], [1, 2_0], [1, 2_7]]							,	dtype=tf.intaa							,	)  # expected non filtered idx as noted above

						snake_case__				:   List[Any]     =						tf.convert_to_tensor(
						    [8.2_2_2_0_9_9, 7.3_5_3_4_1_2_6, 8.4_3_2_0_7_8, 7.4_4_0_2_0_7_5, 9.3_8_4_5_1, 6.2_7_1_1_5_9, 8.8_2_7_5_3_1, 5.4_4_0_2_9_9_5, 7.3_8_5_7_9_5_6, 9.6_7_7_0_2_3]							,	dtype=tf.floataa							,	)  # expected non filtered values as noted above

						snake_case__				:   Any     =						tf_top_k_top_p_filtering(__A							,	top_k=1_0							,	top_p=0.6							,	min_tokens_to_keep=4 )

						snake_case__				:   Optional[Any]     =						output[output != -float("inf" )]
						snake_case__				:   Optional[Any]     =						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 							SCREAMING_SNAKE_CASE__		(       unittest.TestCase				,     UpperCamelCase_		):





					"""simple docstring"""






					if is_tf_available():
										a_			=	{
										    "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  _lowercase  (      self  :						Tuple ):
						# TF-only test: tf.saved_model export
						snake_case__				:   Union[str, Any]     =						TFAutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2" )
						snake_case__				:   Any     =						2
						snake_case__				:   List[Any]     =						2



						class 							SCREAMING_SNAKE_CASE__		(       tf.Module		):





											"""simple docstring"""






											def __init__(      self  :						Optional[Any]							,	__A  :						List[str] ):
												super(__A							,	self ).__init__()
												snake_case__				:   Union[str, Any]     =						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  _lowercase  (      self  :						Optional[Any]							,	__A  :						List[str]							,	__A  :						str ):
												snake_case__				:   List[str]     =						self.model.generate(
												    input_ids=__A							,	attention_mask=__A							,	max_new_tokens=__A							,	return_dict_in_generate=__A							,	)
												return {"sequences": outputs["sequences"]}
						snake_case__				:   int     =						[[2, 0], [1_0_2, 1_0_3]]
						snake_case__				:   Tuple     =						[[1, 0], [1, 1]]
						snake_case__				:   Optional[Any]     =						DummyModel(model=__A )
						with tempfile.TemporaryDirectory() as tmp_dir:
							tf.saved_model.save(__A							,	__A							,	signatures={"serving_default": dummy_model.serving} )
							snake_case__				:   Tuple     =						tf.saved_model.load(__A ).signatures["serving_default"]
							for batch_size in range(1							,	len(__A ) + 1 ):
								snake_case__				:   Any     =						{
								    "input_ids": tf.constant(dummy_input_ids[:batch_size] ),
								    "attention_mask": tf.constant(dummy_attention_masks[:batch_size] ),
								}
								snake_case__				:   Optional[Any]     =						serving_func(**__A )["sequences"]
								snake_case__				:   Optional[int]     =						test_model.generate(**__A							,	max_new_tokens=__A )
								tf.debugging.assert_equal(__A							,	__A )



					@slow
					def  _lowercase  (      self  :						Dict ):
						# TF-only test: tf.saved_model export
						snake_case__				:   Optional[int]     =						TFAutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2" )
						snake_case__				:   Any     =						1
						snake_case__				:   Optional[int]     =						2



						class 							SCREAMING_SNAKE_CASE__		(       tf.Module		):





											"""simple docstring"""






											def __init__(      self  :						Any							,	__A  :						str ):
												super(__A							,	self ).__init__()
												snake_case__				:   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=__A							,	)
											def  _lowercase  (      self  :						int							,	__A  :						Tuple							,	__A  :						str ):
												snake_case__				:   int     =						self.model.generate(
												    input_ids=__A							,	attention_mask=__A							,	max_new_tokens=__A							,	return_dict_in_generate=__A							,	)
												return {"sequences": outputs["sequences"]}
						snake_case__				:   Tuple     =						[[2], [1_0_2, 1_0_3]]
						snake_case__				:   Optional[Any]     =						[[1], [1, 1]]
						snake_case__				:   Tuple     =						DummyModel(model=__A )
						with tempfile.TemporaryDirectory() as tmp_dir:
							tf.saved_model.save(__A							,	__A							,	signatures={"serving_default": dummy_model.serving} )
							snake_case__				:   Union[str, Any]     =						tf.saved_model.load(__A ).signatures["serving_default"]
							for input_row in range(len(__A ) ):
								snake_case__				:   Dict     =						{
								    "input_ids": tf.constant([dummy_input_ids[input_row]] ),
								    "attention_mask": tf.constant([dummy_attention_masks[input_row]] ),
								}
								snake_case__				:   int     =						serving_func(**__A )["sequences"]
								snake_case__				:   Union[str, Any]     =						test_model.generate(**__A							,	max_new_tokens=__A )
								tf.debugging.assert_equal(__A							,	__A )



					@slow
					@require_tensorflow_text
					def  _lowercase  (      self  :						int ):
						# 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 							SCREAMING_SNAKE_CASE__		(       tf.keras.layers.Layer		):





												"""simple docstring"""






												def __init__(      self  :						int ):
													super().__init__()
													snake_case__				:   Any     =						text.SentencepieceTokenizer(
													    model=tf.io.gfile.GFile(os.path.join(__A							,	"spiece.model" )							,	"rb" ).read() )
													snake_case__				:   Optional[int]     =						TFAutoModelForSeqaSeqLM.from_pretrained("hf-internal-testing/tiny-random-t5" )






												def  _lowercase  (      self  :						Union[str, Any]							,	__A  :						Any							,	*__A  :						Dict							,	**__A  :						str ):
													snake_case__				:   Dict     =						self.tokenizer.tokenize(__A )
													snake_case__,		snake_case__				:   List[Any]     =						text.pad_model_inputs(
													    __A							,	max_seq_length=6_4							,	pad_value=self.model.config.pad_token_id )
													snake_case__				:   str     =						self.model.generate(input_ids=__A							,	attention_mask=__A )
													return self.tokenizer.detokenize(__A )
							snake_case__				:   Any     =						CompleteSentenceTransformer()
							snake_case__				:   str     =						tf.keras.layers.Input(shape=(1,)							,	dtype=tf.string							,	name="inputs" )
							snake_case__				:   str     =						complete_model(__A )
							snake_case__				:   Optional[int]     =						tf.keras.Model(__A							,	__A )
							keras_model.save(__A )



					def  _lowercase  (      self  :						Tuple ):
						# Has PT equivalent: this test relies on random sampling
						snake_case__				:   Any     =						{
						    "do_sample": True,
						    "num_beams": 1,
						    "top_p": 0.7,
						    "top_k": 1_0,
						    "temperature": 0.7,
						}
						snake_case__				:   Union[str, Any]     =						1_4

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

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

						snake_case__				:   Optional[int]     =						[6_3_8, 1_9_8]
						with tf.device(":/CPU:0" ):
							tf.random.set_seed(0 )
							snake_case__				:   Union[str, Any]     =						model.generate(**__A							,	eos_token_id=__A							,	**__A )
						self.assertTrue(expectation == len(generated_tokens[0] ) )






					def  _lowercase  (      self  :						str ):
						# Has PT equivalent: ample use of framework-specific code
						snake_case__				:   Any     =						AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-bart" )
						snake_case__				:   Optional[int]     =						"Hugging Face is a technology company based in New York and Paris."
						snake_case__				:   List[str]     =						bart_tokenizer(__A							,	return_tensors="tf" ).input_ids
						snake_case__				:   Any     =						TFBartForConditionalGeneration.from_pretrained("hf-internal-testing/tiny-random-bart" )
						snake_case__				:   str     =						bart_model.generate(__A ).numpy()



						class 							SCREAMING_SNAKE_CASE__		(       UpperCamelCase_		):





											"""simple docstring"""






											def  _lowercase  (      self  :						Tuple							,	__A  :						str							,	__A  :						Optional[Any]=None							,	**__A  :						Optional[int] ):
												return super().call(__A							,	**__A )
						snake_case__				:   Union[str, Any]     =						FakeBart.from_pretrained("hf-internal-testing/tiny-random-bart" )
						snake_case__				:   Optional[Any]     =						bart_model.generate(__A							,	foo="bar" ).numpy()
						self.assertTrue(np.array_equal(__A							,	__A ) )



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





											"""simple docstring"""






											def  _lowercase  (      self  :						int							,	__A  :						Union[str, Any]							,	**__A  :						Optional[int] ):
												return super().call(__A							,	**__A )
						snake_case__				:   str     =						FakeEncoder(bart_model.config							,	bart_model.model.shared )
						snake_case__				:   List[Any]     =						fake_encoder

						# Normal generation still works (the output will be different because the encoder weights are different)
						snake_case__				:   str     =						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" )











from __future__ import annotations

import time

__lowerCamelCase      :    str   		=   list[tuple[int, int]]

__lowerCamelCase      :    Optional[int]   		=   [
    [0, 0, 0, 0, 0, 0, 0],
    [0, 1, 0, 0, 0, 0, 0],  # 0 are free path whereas 1's are obstacles
    [0, 0, 0, 0, 0, 0, 0],
    [0, 0, 1, 0, 0, 0, 0],
    [1, 0, 1, 0, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0],
    [0, 0, 0, 0, 1, 0, 0],
]

__lowerCamelCase      :    Tuple   		=   [[-1, 0], [0, -1], [1, 0], [0, 1]]  # up, left, down, right



class 							SCREAMING_SNAKE_CASE__		:





					"""simple docstring"""






					def __init__(      self  :						Union[str, Any]							,	__A  :						int							,	__A  :						int							,	__A  :						int							,	__A  :						int							,	__A  :						Node | None ):
						snake_case__				:   Optional[int]     =						pos_x
						snake_case__				:   Dict     =						pos_y
						snake_case__				:   int     =						(pos_y, pos_x)
						snake_case__				:   Optional[int]     =						goal_x
						snake_case__				:   Tuple     =						goal_y
						snake_case__				:   str     =						parent



class 							SCREAMING_SNAKE_CASE__		:





					"""simple docstring"""






					def __init__(      self  :						List[Any]							,	__A  :						tuple[int, int]							,	__A  :						tuple[int, int] ):
						snake_case__				:   Tuple     =						Node(start[1]							,	start[0]							,	goal[1]							,	goal[0]							,	__A )
						snake_case__				:   Tuple     =						Node(goal[1]							,	goal[0]							,	goal[1]							,	goal[0]							,	__A )

						snake_case__				:   int     =						[self.start]
						snake_case__				:   Union[str, Any]     =						False



					def  _lowercase  (      self  :						Dict ):
						while self.node_queue:
							snake_case__				:   Optional[Any]     =						self.node_queue.pop(0 )

							if current_node.pos == self.target.pos:
								snake_case__				:   Optional[Any]     =						True
								return self.retrace_path(__A )

							snake_case__				:   int     =						self.get_successors(__A )

							for node in successors:
								self.node_queue.append(__A )

						if not self.reached:
							return [self.start.pos]
						return None



					def  _lowercase  (      self  :						Union[str, Any]							,	__A  :						Node ):
						snake_case__				:   str     =						[]
						for action in delta:
							snake_case__				:   str     =						parent.pos_x + action[1]
							snake_case__				:   Union[str, Any]     =						parent.pos_y + action[0]
							if not (0 <= pos_x <= len(grid[0] ) - 1 and 0 <= pos_y <= len(__A ) - 1):
								continue

							if grid[pos_y][pos_x] != 0:
								continue

							successors.append(
							    Node(__A							,	__A							,	self.target.pos_y							,	self.target.pos_x							,	__A ) )
						return successors






					def  _lowercase  (      self  :						Optional[Any]							,	__A  :						Node | None ):
						snake_case__				:   Tuple     =						node
						snake_case__				:   Any     =						[]
						while current_node is not None:
							path.append((current_node.pos_y, current_node.pos_x) )
							snake_case__				:   Tuple     =						current_node.parent
						path.reverse()
						return path



class 							SCREAMING_SNAKE_CASE__		:





					"""simple docstring"""






					def __init__(      self  :						Dict							,	__A  :						str							,	__A  :						int ):
						snake_case__				:   str     =						BreadthFirstSearch(__A							,	__A )
						snake_case__				:   int     =						BreadthFirstSearch(__A							,	__A )
						snake_case__				:   Tuple     =						False



					def  _lowercase  (      self  :						Optional[Any] ):
						while self.fwd_bfs.node_queue or self.bwd_bfs.node_queue:
							snake_case__				:   Any     =						self.fwd_bfs.node_queue.pop(0 )
							snake_case__				:   List[str]     =						self.bwd_bfs.node_queue.pop(0 )

							if current_bwd_node.pos == current_fwd_node.pos:
								snake_case__				:   List[str]     =						True
								return self.retrace_bidirectional_path(
								    __A							,	__A )

							snake_case__				:   Union[str, Any]     =						current_bwd_node
							snake_case__				:   Dict     =						current_fwd_node

							snake_case__				:   List[Any]     =						{
							    self.fwd_bfs: self.fwd_bfs.get_successors(__A ),
							    self.bwd_bfs: self.bwd_bfs.get_successors(__A ),
							}

							for bfs in [self.fwd_bfs, self.bwd_bfs]:
								for node in successors[bfs]:
									bfs.node_queue.append(__A )

						if not self.reached:
							return [self.fwd_bfs.start.pos]
						return None






					def  _lowercase  (      self  :						Any							,	__A  :						Node							,	__A  :						Node ):
						snake_case__				:   List[str]     =						self.fwd_bfs.retrace_path(__A )
						snake_case__				:   Optional[Any]     =						self.bwd_bfs.retrace_path(__A )
						bwd_path.pop()
						bwd_path.reverse()
						snake_case__				:   List[Any]     =						fwd_path + bwd_path
						return path


if __name__ == "__main__":
	# all coordinates are given in format [y,x]
	import doctest

	doctest.testmod()
	__lowerCamelCase      :    str   		=   (0, 0)
	__lowerCamelCase      :    List[str]   		=   (len(grid) - 1, len(grid[0]) - 1)
	for elem in grid:
		print(elem)

	__lowerCamelCase      :    Any   		=   time.time()
	__lowerCamelCase      :    Optional[Any]   		=   BreadthFirstSearch(init, goal)
	__lowerCamelCase      :    str   		=   bfs.search()
	__lowerCamelCase      :    Optional[Any]   		=   time.time() - start_bfs_time

	print("""Unidirectional BFS computation time : """, bfs_time)

	__lowerCamelCase      :    Optional[Any]   		=   time.time()
	__lowerCamelCase      :    Optional[int]   		=   BidirectionalBreadthFirstSearch(init, goal)
	__lowerCamelCase      :    str   		=   bd_bfs.search()
	__lowerCamelCase      :    Optional[Any]   		=   time.time() - start_bd_bfs_time

	print("""Bidirectional BFS computation time : """, bd_bfs_time)



