
import gc
import unittest

from transformers import CTRLConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
			import torch

			from transformers import (
			    CTRL_PRETRAINED_MODEL_ARCHIVE_LIST,
			    CTRLForSequenceClassification,
			    CTRLLMHeadModel,
			    CTRLModel,
			)




class    SCREAMING_SNAKE_CASE__     :





	'''simple docstring'''



	def __init__(  self	:    List[Any]			,  lowercase	:    Optional[Any]			,  lowercase	:    int=14			,  lowercase	:    Tuple=7			,  lowercase	:    str=True			,  lowercase	:    Tuple=True			,  lowercase	:    Optional[int]=True			,  lowercase	:    Tuple=True			,  lowercase	:    List[str]=True			,  lowercase	:    List[str]=99			,  lowercase	:    Optional[Any]=32			,  lowercase	:    str=5			,  lowercase	:    Tuple=4			,  lowercase	:    str=37			,  lowercase	:    Any="gelu"			,  lowercase	:    Any=0.1			,  lowercase	:    str=0.1			,  lowercase	:    int=512			,  lowercase	:    List[str]=16			,  lowercase	:    List[Any]=2			,  lowercase	:    List[str]=0.02			,  lowercase	:    List[Any]=3			,  lowercase	:    Any=4			,  lowercase	:    int=None			,  ):


				'''simple docstring'''


				_snake_case		=			parent
				_snake_case		=			batch_size
				_snake_case		=			seq_length
				_snake_case		=			is_training
				_snake_case		=			use_token_type_ids
				_snake_case		=			use_input_mask
				_snake_case		=			use_labels
				_snake_case		=			use_mc_token_ids
				_snake_case		=			vocab_size
				_snake_case		=			hidden_size
				_snake_case		=			num_hidden_layers
				_snake_case		=			num_attention_heads
				_snake_case		=			intermediate_size
				_snake_case		=			hidden_act
				_snake_case		=			hidden_dropout_prob
				_snake_case		=			attention_probs_dropout_prob
				_snake_case		=			max_position_embeddings
				_snake_case		=			type_vocab_size
				_snake_case		=			type_sequence_label_size
				_snake_case		=			initializer_range
				_snake_case		=			num_labels
				_snake_case		=			num_choices
				_snake_case		=			scope
				_snake_case		=			self.vocab_size - 1



	def  A							(  self	:    List[str]	):


				'''simple docstring'''


				_snake_case		=			ids_tensor([self.batch_size, self.seq_length]			,  self.vocab_size	)

				_snake_case		=			None
				if self.use_input_mask:
							_snake_case		=			random_attention_mask([self.batch_size, self.seq_length]	)

				_snake_case		=			None
				if self.use_token_type_ids:
							_snake_case		=			ids_tensor([self.batch_size, self.seq_length]			,  self.type_vocab_size	)

				_snake_case		=			None
				if self.use_mc_token_ids:
							_snake_case		=			ids_tensor([self.batch_size, self.num_choices]			,  self.seq_length	)

				_snake_case		=			None
				_snake_case		=			None
				_snake_case		=			None
				if self.use_labels:
							_snake_case		=			ids_tensor([self.batch_size]			,  self.type_sequence_label_size	)
							_snake_case		=			ids_tensor([self.batch_size, self.seq_length]			,  self.num_labels	)
							_snake_case		=			ids_tensor([self.batch_size]			,  self.num_choices	)

				_snake_case		=			self.get_config()

				_snake_case		=			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	:    Optional[Any]	):


				'''simple docstring'''


				return 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			,  )



	def  A							(  self	:    int			,  lowercase	:    Dict			,  lowercase	:    Optional[int]			,  lowercase	:    int			,  lowercase	:    Optional[int]			,  lowercase	:    str			,  *lowercase	:    Any	):


				'''simple docstring'''


				_snake_case		=			CTRLModel(config=lowercase	)
				model.to(lowercase	)
				model.eval()

				model(lowercase			,  token_type_ids=lowercase			,  head_mask=lowercase	)
				model(lowercase			,  token_type_ids=lowercase	)
				_snake_case		=			model(lowercase	)
				self.parent.assertEqual(result.last_hidden_state.shape			,  (self.batch_size, self.seq_length, self.hidden_size)	)
				self.parent.assertEqual(len(result.past_key_values	)			,  config.n_layer	)



	def  A							(  self	:    Dict			,  lowercase	:    Any			,  lowercase	:    int			,  lowercase	:    List[str]			,  lowercase	:    Dict			,  lowercase	:    int			,  *lowercase	:    List[Any]	):


				'''simple docstring'''


				_snake_case		=			CTRLLMHeadModel(lowercase	)
				model.to(lowercase	)
				model.eval()

				_snake_case		=			model(lowercase			,  token_type_ids=lowercase			,  labels=lowercase	)
				self.parent.assertEqual(result.loss.shape			,  ()	)
				self.parent.assertEqual(result.logits.shape			,  (self.batch_size, self.seq_length, self.vocab_size)	)



	def  A							(  self	:    Any	):


				'''simple docstring'''


				_snake_case		=			self.prepare_config_and_inputs()

				(
				    (
				    _snake_case
				)	,					(
				    _snake_case
				)	,					(
				    _snake_case
				)	,					(
				    _snake_case
				)	,					(
				    _snake_case
				)	,					(
				    _snake_case
				)	,					(
				    _snake_case
				)	,					(
				    _snake_case
				)	,					(
				    _snake_case
				)	,					
				)		=			config_and_inputs

				_snake_case		=			{'input_ids': input_ids, 'token_type_ids': token_type_ids, 'head_mask': head_mask}

				return config, inputs_dict





	def  A							(  self	:    Optional[Any]			,  lowercase	:    List[str]			,  lowercase	:    List[Any]			,  lowercase	:    List[Any]			,  lowercase	:    Union[str, Any]			,  *lowercase	:    Optional[int]	):


				'''simple docstring'''


				_snake_case		=			self.num_labels
				_snake_case		=			CTRLForSequenceClassification(lowercase	)
				model.to(lowercase	)
				model.eval()
				_snake_case		=			ids_tensor([self.batch_size]			,  self.type_sequence_label_size	)
				_snake_case		=			model(lowercase			,  token_type_ids=lowercase			,  labels=lowercase	)
				self.parent.assertEqual(result.logits.shape			,  (self.batch_size, self.num_labels)	)




@require_torch
class    SCREAMING_SNAKE_CASE__     (	UpperCAmelCase							,UpperCAmelCase							,UpperCAmelCase							,unittest.TestCase					):





	'''simple docstring'''



	_UpperCAmelCase       :		Dict       =					(CTRLModel, CTRLLMHeadModel, CTRLForSequenceClassification) if is_torch_available() else ()
	_UpperCAmelCase       :		Optional[int]       =					(CTRLLMHeadModel,) if is_torch_available() else ()
	_UpperCAmelCase       :		Dict       =					(
	    {
	        "feature-extraction": CTRLModel,
	        "text-classification": CTRLForSequenceClassification,
	        "text-generation": CTRLLMHeadModel,
	        "zero-shot": CTRLForSequenceClassification,
	    }
	    if is_torch_available()
	    else {}
	)
	_UpperCAmelCase       :		str       =					True
	_UpperCAmelCase       :		int       =					False
	_UpperCAmelCase       :		Any       =					False



	def  A							(  self	:    List[str]			,  lowercase	:    int			,  lowercase	:    Union[str, Any]			,  lowercase	:    Tuple			,  lowercase	:    int			,  lowercase	:    Tuple	):


				'''simple docstring'''


				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	:    List[str]	):


				'''simple docstring'''


				_snake_case		=			CTRLModelTester(self	)
				_snake_case		=			ConfigTester(self			,  config_class=lowercase			,  n_embd=37	)



	def  A							(  self	:    Union[str, Any]	):


				'''simple docstring'''


				super().tearDown()
				# clean-up as much as possible GPU memory occupied by PyTorch
				gc.collect()
				torch.cuda.empty_cache()



	def  A							(  self	:    Optional[int]	):


				'''simple docstring'''


				self.config_tester.run_common_tests()



	def  A							(  self	:    Dict	):


				'''simple docstring'''


				_snake_case		=			self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_ctrl_model(*lowercase	)



	def  A							(  self	:    Tuple	):


				'''simple docstring'''


				_snake_case		=			self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_lm_head_model(*lowercase	)



	@unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'	)
	def  A							(  self	:    Optional[Any]	):


				'''simple docstring'''


				pass



	@slow
	def  A							(  self	:    List[Any]	):


				'''simple docstring'''


				for model_name in CTRL_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
							_snake_case		=			CTRLModel.from_pretrained(lowercase	)
							self.assertIsNotNone(lowercase	)





	@unittest.skip('The model doesn\'t support left padding'	)  # and it's not used enough to be worth fixing :)
	def  A							(  self	:    Tuple	):


				'''simple docstring'''


				pass




@require_torch
class    SCREAMING_SNAKE_CASE__     (	unittest.TestCase					):





	'''simple docstring'''



	def  A							(  self	:    Union[str, Any]	):


				'''simple docstring'''


				super().tearDown()
				# clean-up as much as possible GPU memory occupied by PyTorch
				gc.collect()
				torch.cuda.empty_cache()





	@slow
	def  A							(  self	:    str	):


				'''simple docstring'''


				_snake_case		=			CTRLLMHeadModel.from_pretrained('ctrl'	)
				model.to(lowercase	)
				_snake_case		=			torch.tensor(
				    [[11_859, 0, 1_611, 8]]			,  dtype=torch.long			,  device=lowercase	)  # Legal the president is
				_snake_case		=			[
				    11_859,
				    0,
				    1_611,
				    8,
				    5,
				    150,
				    26_449,
				    2,
				    19,
				    348,
				    469,
				    3,
				    2_595,
				    48,
				    20_740,
				    246_533,
				    246_533,
				    19,
				    30,
				    5,
				]  # Legal the president is a good guy and I don't want to lose my job. \n \n I have a

				_snake_case		=			model.generate(lowercase			,  do_sample=lowercase	)
				self.assertListEqual(output_ids[0].tolist()			,  lowercase	)

import os
import tempfile
import unittest
from pathlib import Path

from transformers import AutoConfig, is_torch_available
from transformers.testing_utils import require_torch, torch_device


if is_torch_available():
			from transformers import PyTorchBenchmark, PyTorchBenchmarkArguments




@require_torch
class    SCREAMING_SNAKE_CASE__     (	unittest.TestCase					):





	'''simple docstring'''



	def  A							(  self	:    List[Any]			,  lowercase	:    Dict	):


				'''simple docstring'''


				for model_result in results.values():
							for batch_size, sequence_length in zip(model_result['bs']			,  model_result['ss']	):
										_snake_case		=			model_result['result'][batch_size][sequence_length]
										self.assertIsNotNone(lowercase	)



	def  A							(  self	:    str	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_inference_result	)
				self.check_results_dict_not_empty(results.memory_inference_result	)



	def  A							(  self	:    Any	):


				'''simple docstring'''


				_snake_case		=			'sgugger/tiny-distilbert-classification'
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  only_pretrain_model=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_inference_result	)
				self.check_results_dict_not_empty(results.memory_inference_result	)



	def  A							(  self	:    Optional[int]	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  torchscript=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_inference_result	)
				self.check_results_dict_not_empty(results.memory_inference_result	)



	@unittest.skipIf(torch_device == 'cpu'			,  'Cant do half precision'	)
	def  A							(  self	:    Optional[Any]	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  fpaa=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_inference_result	)
				self.check_results_dict_not_empty(results.memory_inference_result	)



	def  A							(  self	:    str	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'
				_snake_case		=			AutoConfig.from_pretrained(lowercase	)
				# set architectures equal to `None`
				_snake_case		=			None
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase			,  configs=[config]	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_inference_result	)
				self.check_results_dict_not_empty(results.memory_inference_result	)



	def  A							(  self	:    Optional[Any]	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_train_result	)
				self.check_results_dict_not_empty(results.memory_train_result	)



	@unittest.skipIf(torch_device == 'cpu'			,  'Can\'t do half precision'	)
	def  A							(  self	:    str	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  fpaa=lowercase			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_train_result	)
				self.check_results_dict_not_empty(results.memory_train_result	)



	def  A							(  self	:    Tuple	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'
				_snake_case		=			AutoConfig.from_pretrained(lowercase	)
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase			,  configs=[config]	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_inference_result	)
				self.check_results_dict_not_empty(results.memory_inference_result	)



	def  A							(  self	:    Union[str, Any]	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tinier_bart'
				_snake_case		=			AutoConfig.from_pretrained(lowercase	)
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase			,  configs=[config]	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_inference_result	)
				self.check_results_dict_not_empty(results.memory_inference_result	)



	def  A							(  self	:    Dict	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'
				_snake_case		=			AutoConfig.from_pretrained(lowercase	)
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase			,  configs=[config]	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_train_result	)
				self.check_results_dict_not_empty(results.memory_train_result	)



	def  A							(  self	:    Dict	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tinier_bart'
				_snake_case		=			AutoConfig.from_pretrained(lowercase	)
				_snake_case		=			PyTorchBenchmarkArguments(
				    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  multi_process=lowercase			,  )
				_snake_case		=			PyTorchBenchmark(lowercase			,  configs=[config]	)
				_snake_case		=			benchmark.run()
				self.check_results_dict_not_empty(results.time_train_result	)
				self.check_results_dict_not_empty(results.memory_train_result	)



	def  A							(  self	:    Optional[Any]	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'
				with tempfile.TemporaryDirectory() as tmp_dir:
							_snake_case		=			PyTorchBenchmarkArguments(
							    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  save_to_csv=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  inference_time_csv_file=os.path.join(lowercase			,  'inf_time.csv'	)			,  train_memory_csv_file=os.path.join(lowercase			,  'train_mem.csv'	)			,  inference_memory_csv_file=os.path.join(lowercase			,  'inf_mem.csv'	)			,  train_time_csv_file=os.path.join(lowercase			,  'train_time.csv'	)			,  env_info_csv_file=os.path.join(lowercase			,  'env.csv'	)			,  multi_process=lowercase			,  )
							_snake_case		=			PyTorchBenchmark(lowercase	)
							benchmark.run()
							self.assertTrue(Path(os.path.join(lowercase			,  'inf_time.csv'	)	).exists()	)
							self.assertTrue(Path(os.path.join(lowercase			,  'train_time.csv'	)	).exists()	)
							self.assertTrue(Path(os.path.join(lowercase			,  'inf_mem.csv'	)	).exists()	)
							self.assertTrue(Path(os.path.join(lowercase			,  'train_mem.csv'	)	).exists()	)
							self.assertTrue(Path(os.path.join(lowercase			,  'env.csv'	)	).exists()	)





	def  A							(  self	:    Union[str, Any]	):


				'''simple docstring'''


				_snake_case		=			'sshleifer/tiny-gpt2'

				def _check_summary_is_not_empty(lowercase	:    Optional[Any]	):
							self.assertTrue(hasattr(lowercase			,  'sequential'	)	)
							self.assertTrue(hasattr(lowercase			,  'cumulative'	)	)
							self.assertTrue(hasattr(lowercase			,  'current'	)	)
							self.assertTrue(hasattr(lowercase			,  'total'	)	)

				with tempfile.TemporaryDirectory() as tmp_dir:
							_snake_case		=			PyTorchBenchmarkArguments(
							    models=[MODEL_ID]			,  training=lowercase			,  inference=lowercase			,  sequence_lengths=[8]			,  batch_sizes=[1]			,  log_filename=os.path.join(lowercase			,  'log.txt'	)			,  log_print=lowercase			,  trace_memory_line_by_line=lowercase			,  multi_process=lowercase			,  )
							_snake_case		=			PyTorchBenchmark(lowercase	)
							_snake_case		=			benchmark.run()
							_check_summary_is_not_empty(result.inference_summary	)
							_check_summary_is_not_empty(result.train_summary	)
							self.assertTrue(Path(os.path.join(lowercase			,  'log.txt'	)	).exists()	)