"""simple docstring"""






from __future__ import annotations

import unittest

from transformers import CTRLConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
				import tensorflow as tf

				from transformers.models.ctrl.modeling_tf_ctrl import (
				    TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST,
				    TFCTRLForSequenceClassification,
				    TFCTRLLMHeadModel,
				    TFCTRLModel,
				)
class   UpperCAmelCase__				(    __lowerCamelCase			):


	"""simple docstring"""







	def __init__(			self       ,     _SCREAMING_SNAKE_CASE       ,     )    ->	Optional[Any]:
							a_							:			str     		=					parent
							a_							:			List[Any]     		=					1_3
							a_							:			Any     		=					7
							a_							:			Dict     		=					True
							a_							:			str     		=					True
							a_							:			Any     		=					True
							a_							:			List[Any]     		=					True
							a_							:			Any     		=					True
							a_							:			Dict     		=					9_9
							a_							:			str     		=					3_2
							a_							:			List[Any]     		=					2
							a_							:			Dict     		=					4
							a_							:			str     		=					3_7
							a_							:			Any     		=					"gelu"
							a_							:			Union[str, Any]     		=					0.1
							a_							:			Union[str, Any]     		=					0.1
							a_							:			Optional[int]     		=					5_1_2
							a_							:			Tuple     		=					1_6
							a_							:			Union[str, Any]     		=					2
							a_							:			Tuple     		=					0.0_2
							a_							:			int     		=					3
							a_							:			int     		=					4
							a_							:			List[Any]     		=					None
							a_							:			Union[str, Any]     		=					self.vocab_size - 1







	def       A    (			self )    ->	Tuple:
							a_							:			str     		=					ids_tensor([self.batch_size, self.seq_length]       ,     self.vocab_size )

							a_							:			Union[str, Any]     		=					None
							if self.use_input_mask:
													a_							:			List[str]     		=					random_attention_mask([self.batch_size, self.seq_length] )

							a_							:			List[Any]     		=					None
							if self.use_token_type_ids:
													a_							:			Union[str, Any]     		=					ids_tensor([self.batch_size, self.seq_length]       ,     self.type_vocab_size )

							a_							:			Tuple     		=					None
							if self.use_mc_token_ids:
													a_							:			Union[str, Any]     		=					ids_tensor([self.batch_size, self.num_choices]       ,     self.seq_length )

							a_							:			Optional[int]     		=					None
							a_							:			Dict     		=					None
							a_							:			Optional[int]     		=					None
							if self.use_labels:
													a_							:			Union[str, Any]     		=					ids_tensor([self.batch_size]       ,     self.type_sequence_label_size )
													a_							:			Tuple     		=					ids_tensor([self.batch_size, self.seq_length]       ,     self.num_labels )
													a_							:			Dict     		=					ids_tensor([self.batch_size]       ,     self.num_choices )

							a_							:			List[str]     		=					CTRLConfig(
							    vocab_size=self.vocab_size       ,     n_embd=self.hidden_size       ,     n_layer=self.num_hidden_layers       ,     n_head=self.num_attention_heads       ,     n_positions=self.max_position_embeddings       ,     pad_token_id=self.pad_token_id       ,     )

							a_							:			Tuple     		=					ids_tensor([self.num_hidden_layers, self.num_attention_heads]       ,     2 )

							return (
							    config,
							    input_ids,
							    input_mask,
							    head_mask,
							    token_type_ids,
							    mc_token_ids,
							    sequence_labels,
							    token_labels,
							    choice_labels,
							)







	def       A    (			self       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     *_SCREAMING_SNAKE_CASE )    ->	Optional[Any]:
							a_							:			List[Any]     		=					TFCTRLModel(config=_SCREAMING_SNAKE_CASE )
							a_							:			str     		=					{"input_ids": input_ids, "attention_mask": input_mask, "token_type_ids": token_type_ids}
							a_							:			Optional[Any]     		=					model(_SCREAMING_SNAKE_CASE )

							a_							:			str     		=					[input_ids, None, input_mask]  # None is the input for 'past'
							a_							:			Optional[int]     		=					model(_SCREAMING_SNAKE_CASE )

							a_							:			str     		=					model(_SCREAMING_SNAKE_CASE )

							self.parent.assertEqual(result.last_hidden_state.shape       ,     (self.batch_size, self.seq_length, self.hidden_size) )







	def       A    (			self       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     *_SCREAMING_SNAKE_CASE )    ->	List[str]:
							a_							:			Dict     		=					TFCTRLLMHeadModel(config=_SCREAMING_SNAKE_CASE )
							a_							:			List[Any]     		=					{"input_ids": input_ids, "attention_mask": input_mask, "token_type_ids": token_type_ids}
							a_							:			Union[str, Any]     		=					model(_SCREAMING_SNAKE_CASE )
							self.parent.assertEqual(result.logits.shape       ,     (self.batch_size, self.seq_length, self.vocab_size) )







	def       A    (			self       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     *_SCREAMING_SNAKE_CASE )    ->	int:
							a_							:			Optional[Any]     		=					self.num_labels
							a_							:			str     		=					ids_tensor([self.batch_size]       ,     self.type_sequence_label_size )
							a_							:			Any     		=					{
							    "input_ids": input_ids,
							    "token_type_ids": token_type_ids,
							    "labels": sequence_labels,
							}
							a_							:			List[Any]     		=					TFCTRLForSequenceClassification(_SCREAMING_SNAKE_CASE )
							a_							:			Optional[int]     		=					model(_SCREAMING_SNAKE_CASE )
							self.parent.assertEqual(result.logits.shape       ,     (self.batch_size, self.num_labels) )







	def       A    (			self )    ->	Tuple:
							a_							:			Optional[int]     		=					self.prepare_config_and_inputs()

							(
							    (
							    a_
							)		,					(
							    a_
							)		,					(
							    a_
							)		,					(
							    a_
							)		,					(
							    a_
							)		,					(
							    a_
							)		,					(
							    a_
							)		,					(
							    a_
							)		,					(
							    a_
							)		,					
							)							:			int     		=					config_and_inputs

							a_							:			Dict     		=					{"input_ids": input_ids, "token_type_ids": token_type_ids, "attention_mask": input_mask}
							return config, inputs_dict




@require_tf
class   UpperCAmelCase__				(    __lowerCamelCase,      __lowerCamelCase,      unittest.TestCase			):


	"""simple docstring"""
	lowerCAmelCase__    :					Union[str, Any]						      =							(TFCTRLModel, TFCTRLLMHeadModel, TFCTRLForSequenceClassification) if is_tf_available() else ()
	lowerCAmelCase__    :					Optional[Any]						      =							(TFCTRLLMHeadModel,) if is_tf_available() else ()
	lowerCAmelCase__    :					List[str]						      =							(
	    {
	        """feature-extraction""": TFCTRLModel,
	        """text-classification""": TFCTRLForSequenceClassification,
	        """text-generation""": TFCTRLLMHeadModel,
	        """zero-shot""": TFCTRLForSequenceClassification,
	    }
	    if is_tf_available()
	    else {}
	)
	lowerCAmelCase__    :					Optional[int]						      =							False
	lowerCAmelCase__    :					str						      =							False







	def       A    (			self       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE )    ->	Tuple:
							if pipeline_test_casse_name == "ZeroShotClassificationPipelineTests":
													# Get `tokenizer does not have a padding token` error for both fast/slow tokenizers.
													# `CTRLConfig` was never used in pipeline tests, either because of a missing checkpoint or because a tiny
													# config could not be created.
													return True

							return False







	def       A    (			self )    ->	Tuple:
							a_							:			Optional[Any]     		=					TFCTRLModelTester(self )
							a_							:			Optional[int]     		=					ConfigTester(self       ,     config_class=_SCREAMING_SNAKE_CASE       ,     n_embd=3_7 )







	def       A    (			self )    ->	Optional[int]:
							self.config_tester.run_common_tests()







	def       A    (			self )    ->	int:
							a_							:			List[str]     		=					self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_ctrl_model(*_SCREAMING_SNAKE_CASE )







	def       A    (			self )    ->	Optional[int]:
							a_							:			Union[str, Any]     		=					self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_ctrl_lm_head(*_SCREAMING_SNAKE_CASE )







	def       A    (			self )    ->	Any:
							a_							:			Dict     		=					self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_ctrl_for_sequence_classification(*_SCREAMING_SNAKE_CASE )







	def       A    (			self )    ->	Optional[int]:
							a_		,					a_							:			Union[str, Any]     		=					self.model_tester.prepare_config_and_inputs_for_common()
							a_							:			str     		=					[TFCTRLLMHeadModel]
							a_							:			str     		=					[TFCTRLForSequenceClassification]

							for model_class in self.all_model_classes:
													a_							:			Tuple     		=					model_class(_SCREAMING_SNAKE_CASE )
													model.build()  # may be needed for the get_bias() call below
													assert isinstance(model.get_input_embeddings()       ,     tf.keras.layers.Layer )

													if model_class in list_lm_models:
																			a_							:			List[Any]     		=					model.get_output_embeddings()
																			assert isinstance(_SCREAMING_SNAKE_CASE       ,     tf.keras.layers.Layer )
																			a_							:			Dict     		=					model.get_bias()
																			assert isinstance(_SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE )
																			for k, v in name.items():
																									assert isinstance(_SCREAMING_SNAKE_CASE       ,     tf.Variable )
													elif model_class in list_other_models_with_output_ebd:
																			a_							:			str     		=					model.get_output_embeddings()
																			assert isinstance(_SCREAMING_SNAKE_CASE       ,     tf.keras.layers.Layer )
																			a_							:			Any     		=					model.get_bias()
																			assert name is None
													else:
																			a_							:			Tuple     		=					model.get_output_embeddings()
																			assert x is None
																			a_							:			Dict     		=					model.get_bias()
																			assert name is None







	@slow
	def       A    (			self )    ->	Tuple:
							for model_name in TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
													a_							:			Optional[int]     		=					TFCTRLModel.from_pretrained(_SCREAMING_SNAKE_CASE )
													self.assertIsNotNone(_SCREAMING_SNAKE_CASE )




@require_tf
class   UpperCAmelCase__				(    unittest.TestCase			):


	"""simple docstring"""







	@slow
	def       A    (			self )    ->	int:
							a_							:			Any     		=					TFCTRLLMHeadModel.from_pretrained("ctrl" )
							a_							:			Tuple     		=					tf.convert_to_tensor([[1_1_8_5_9, 0, 1_6_1_1, 8]]       ,     dtype=tf.intaa )  # Legal the president is
							a_							:			str     		=					[
							    1_1_8_5_9,
							    0,
							    1_6_1_1,
							    8,
							    5,
							    1_5_0,
							    2_6_4_4_9,
							    2,
							    1_9,
							    3_4_8,
							    4_6_9,
							    3,
							    2_5_9_5,
							    4_8,
							    2_0_7_4_0,
							    2_4_6_5_3_3,
							    2_4_6_5_3_3,
							    1_9,
							    3_0,
							    5,
							]  # Legal the president is a good guy and I don't want to lose my job. \n \n I have a

							a_							:			List[str]     		=					model.generate(_SCREAMING_SNAKE_CASE       ,     do_sample=_SCREAMING_SNAKE_CASE )
							self.assertListEqual(output_ids[0].numpy().tolist()       ,     _SCREAMING_SNAKE_CASE )





"""simple docstring"""






import json
import os
import tempfile

import datasets
from datasets.arrow_writer import ArrowWriter
from datasets.features import ArrayaD
from utils import generate_examples, get_duration


UpperCamelCase     					=     (30, 4_87)
UpperCamelCase     					=     (36, 10_24)
UpperCamelCase     					=     (1_00, 1_00)
UpperCamelCase     					=     1_00

UpperCamelCase     					=     datasets.Features(
    {'text': ArrayaD(SHAPE_TEST_1, dtype='float32'), 'image': ArrayaD(SHAPE_TEST_2, dtype='float32')}
)

UpperCamelCase					,      UpperCamelCase     					=     os.path.split(__file__)
UpperCamelCase     					=     os.path.join(RESULTS_BASEPATH, 'results', RESULTS_FILENAME.replace('.py', '.json'))

@get_duration
def  lowerCAmelCase_      (_SCREAMING_SNAKE_CASE	:Tuple		,			_SCREAMING_SNAKE_CASE	:str		,			_SCREAMING_SNAKE_CASE	:Tuple  )				-> Any:
						with ArrowWriter(features=_SCREAMING_SNAKE_CASE		,			path=os.path.join(_SCREAMING_SNAKE_CASE		,			"beta.arrow"  )  ) as writer:
												for key, record in dummy_data:
																		a_							:			int     		=					my_features.encode_example(_SCREAMING_SNAKE_CASE  )
																		writer.write(_SCREAMING_SNAKE_CASE  )
												a_		,					a_							:			str     		=					writer.finalize()

@get_duration
def  lowerCAmelCase_      (_SCREAMING_SNAKE_CASE	:List[str]		,			_SCREAMING_SNAKE_CASE	:Any  )				-> Tuple:
						a_							:			Dict     		=					datasets.Dataset.from_file(
						    filename=os.path.join(_SCREAMING_SNAKE_CASE		,			"beta.arrow"  )		,			info=datasets.DatasetInfo(features=_SCREAMING_SNAKE_CASE  )  )
						for _ in dataset:
												pass

@get_duration
def  lowerCAmelCase_      (_SCREAMING_SNAKE_CASE	:str		,			_SCREAMING_SNAKE_CASE	:Dict  )				-> Optional[Any]:
						a_							:			List[str]     		=					datasets.Dataset.from_file(
						    filename=os.path.join(_SCREAMING_SNAKE_CASE		,			"beta.arrow"  )		,			info=datasets.DatasetInfo(features=_SCREAMING_SNAKE_CASE  )  )
						dataset.set_format("numpy"  )
						for _ in dataset:
												pass

@get_duration
def  lowerCAmelCase_      (_SCREAMING_SNAKE_CASE	:List[Any]		,			_SCREAMING_SNAKE_CASE	:List[str]  )				-> List[str]:
						a_							:			Optional[Any]     		=					10
						a_							:			Optional[int]     		=					datasets.Dataset.from_file(
						    filename=os.path.join(_SCREAMING_SNAKE_CASE		,			"beta.arrow"  )		,			info=datasets.DatasetInfo(features=_SCREAMING_SNAKE_CASE  )  )
						for i in range(0		,			len(_SCREAMING_SNAKE_CASE  )		,			_SCREAMING_SNAKE_CASE  ):
												a_							:			List[str]     		=					dataset[i : i + batch_size]

@get_duration
def  lowerCAmelCase_      (_SCREAMING_SNAKE_CASE	:Optional[int]		,			_SCREAMING_SNAKE_CASE	:List[str]  )				-> Optional[int]:
						a_							:			List[str]     		=					10
						a_							:			int     		=					datasets.Dataset.from_file(
						    filename=os.path.join(_SCREAMING_SNAKE_CASE		,			"beta.arrow"  )		,			info=datasets.DatasetInfo(features=_SCREAMING_SNAKE_CASE  )  )
						dataset.set_format("numpy"  )
						for i in range(0		,			len(_SCREAMING_SNAKE_CASE  )		,			_SCREAMING_SNAKE_CASE  ):
												a_							:			str     		=					dataset[i : i + batch_size]

@get_duration
def  lowerCAmelCase_      (_SCREAMING_SNAKE_CASE	:List[str]		,			_SCREAMING_SNAKE_CASE	:Optional[Any]  )				-> Dict:
						a_							:			List[str]     		=					datasets.Dataset.from_file(
						    filename=os.path.join(_SCREAMING_SNAKE_CASE		,			"beta.arrow"  )		,			info=datasets.DatasetInfo(features=_SCREAMING_SNAKE_CASE  )  )
						for col in feats:
												a_							:			List[Any]     		=					dataset[col]

@get_duration
def  lowerCAmelCase_      (_SCREAMING_SNAKE_CASE	:str		,			_SCREAMING_SNAKE_CASE	:Optional[Any]  )				-> Dict:
						a_							:			Any     		=					datasets.Dataset.from_file(
						    filename=os.path.join(_SCREAMING_SNAKE_CASE		,			"beta.arrow"  )		,			info=datasets.DatasetInfo(features=_SCREAMING_SNAKE_CASE  )  )
						dataset.set_format("numpy"  )
						for col in feats:
												a_							:			Any     		=					dataset[col]

def  lowerCAmelCase_      ()				-> Optional[Any]:
						a_							:			List[Any]     		=					{}
						a_							:			str     		=					(
						    read_unformated,
						    read_formatted_as_numpy,
						    read_batch_unformated,
						    read_batch_formatted_as_numpy,
						    read_col_unformated,
						    read_col_formatted_as_numpy,
						)
						with tempfile.TemporaryDirectory() as tmp_dir:
												a_							:			Optional[int]     		=					datasets.Features({"image": ArrayaD(_SCREAMING_SNAKE_CASE		,			dtype="float32"  )}  )
												a_							:			Optional[Any]     		=					generate_examples(features=_SCREAMING_SNAKE_CASE		,			num_examples=_SCREAMING_SNAKE_CASE  )
												a_							:			List[str]     		=					write(_SCREAMING_SNAKE_CASE		,			_SCREAMING_SNAKE_CASE		,			_SCREAMING_SNAKE_CASE  )
												for read_func in read_functions:
																		a_							:			Optional[int]     		=					read_func(_SCREAMING_SNAKE_CASE		,			_SCREAMING_SNAKE_CASE  )

						with tempfile.TemporaryDirectory() as tmp_dir:
												# don't use fixed length for fair comparison
												# feats = datasets.Features(
												#     {"image": datasets.Sequence(datasets.Sequence(datasets.Value("float32"), SPEED_TEST_SHAPE[1]), SPEED_TEST_SHAPE[0])}
												# )
												a_							:			Union[str, Any]     		=					datasets.Features({"image": datasets.Sequence(datasets.Sequence(datasets.Value("float32"  )  )  )}  )
												a_							:			List[str]     		=					generate_examples(
												    features=_SCREAMING_SNAKE_CASE		,			num_examples=_SCREAMING_SNAKE_CASE		,			seq_shapes={"image": SPEED_TEST_SHAPE}  )
												a_							:			str     		=					write(_SCREAMING_SNAKE_CASE		,			_SCREAMING_SNAKE_CASE		,			_SCREAMING_SNAKE_CASE  )
												for read_func in read_functions:
																		a_							:			List[str]     		=					read_func(_SCREAMING_SNAKE_CASE		,			_SCREAMING_SNAKE_CASE  )

						with tempfile.TemporaryDirectory() as tmp_dir:
												# don't use fixed length for fair comparison
												# feats = datasets.Features(
												#     {"image": datasets.Sequence(datasets.Value("float32"), SPEED_TEST_SHAPE[0] * SPEED_TEST_SHAPE[1])}
												# )
												a_							:			int     		=					datasets.Features({"image": datasets.Sequence(datasets.Value("float32"  )  )}  )
												a_							:			str     		=					generate_examples(
												    features=_SCREAMING_SNAKE_CASE		,			num_examples=_SCREAMING_SNAKE_CASE		,			seq_shapes={"image": [SPEED_TEST_SHAPE[0] * SPEED_TEST_SHAPE[1]]}		,			)
												a_							:			List[Any]     		=					write(_SCREAMING_SNAKE_CASE		,			_SCREAMING_SNAKE_CASE		,			_SCREAMING_SNAKE_CASE  )
												for read_func in read_functions:
																		a_							:			str     		=					read_func(_SCREAMING_SNAKE_CASE		,			_SCREAMING_SNAKE_CASE  )

						with open(_SCREAMING_SNAKE_CASE		,			"wb"  ) as f:
												f.write(json.dumps(_SCREAMING_SNAKE_CASE  ).encode("utf-8"  )  )


if __name__ == "__main__":  # useful to run the profiler
				benchmark_array_xd()




