


import argparse
import dataclasses
import json
import logging
import os
import shutil
from typing import List, Optional

import datasets
from accelerate import Accelerator
from datasets import load_dataset
from finetuning import finetune
from tqdm.auto import tqdm

import transformers
from transformers import AutoConfig, set_seed
from transformers.trainer_utils import IntervalStrategy


snake_case							: str      =   logging.getLogger(__name__)

snake_case							: List[str]      =   "pytorch_model.bin"







@dataclasses.dataclass
class  snake_case_     :
					UpperCAmelCase__      :					str					   =    dataclasses.field(
					    metadata={'''help''': '''Path to pretrained model or model identifier from huggingface.co/models.'''}   )
					UpperCAmelCase__      :					Optional[str]					   =    dataclasses.field(
					    default=_SCREAMING_SNAKE_CASE			,		metadata={'''help''': '''Where do you want to store the pretrained models downloaded from huggingface.co.'''}			,		)







@dataclasses.dataclass
class  snake_case_     :
					UpperCAmelCase__      :					str					   =    dataclasses.field(metadata={'''help''': '''A csv or a json file containing the training data.'''}   )
					UpperCAmelCase__      :					str					   =    dataclasses.field(metadata={'''help''': '''A csv or a json file containing the data to predict on.'''}   )
					UpperCAmelCase__      :					Optional[str]					   =    dataclasses.field(
					    default=_SCREAMING_SNAKE_CASE			,		metadata={'''help''': '''A csv or a json file containing the validation data.'''}   )
					UpperCAmelCase__      :					Optional[str]					   =    dataclasses.field(
					    default=_SCREAMING_SNAKE_CASE			,		metadata={'''help''': '''The name of the task to train on.'''}			,		)
					UpperCAmelCase__      :					Optional[List[str]]					   =    dataclasses.field(
					    default=_SCREAMING_SNAKE_CASE			,		metadata={'''help''': '''The list of labels for the task.'''}   )







@dataclasses.dataclass
class  snake_case_     :
					UpperCAmelCase__      :					str					   =    dataclasses.field(
					    metadata={'''help''': '''The output directory where the model predictions and checkpoints will be written.'''}   )
					UpperCAmelCase__      :					Optional[str]					   =    dataclasses.field(
					    default='''accuracy'''			,		metadata={'''help''': '''The evaluation metric used for the task.'''}   )
					UpperCAmelCase__      :					Optional[str]					   =    dataclasses.field(
					    default='''no'''			,		metadata={
					        '''help''': '''The evaluation strategy to adopt during training. Possible values are: [\"no\", \"step\", \"epoch]'''
					    }			,		)
					UpperCAmelCase__      :					Optional[int]					   =    dataclasses.field(
					    default=1_0			,		metadata={'''help''': '''Number of evaluation calls with no improvement after which training will be stopped.'''}			,		)
					UpperCAmelCase__      :					Optional[float]					   =    dataclasses.field(
					    default=0.0			,		metadata={
					        '''help''': '''How much the specified evaluation metric must improve to satisfy early stopping conditions.'''
					    }			,		)
					UpperCAmelCase__      :					Optional[bool]					   =    dataclasses.field(
					    default=_SCREAMING_SNAKE_CASE			,		metadata={'''help''': '''Whether to filter the pseudo-labeled data based on the confidence score.'''}			,		)
					UpperCAmelCase__      :					Optional[bool]					   =    dataclasses.field(
					    default=_SCREAMING_SNAKE_CASE			,		metadata={'''help''': '''Whether to filter the pseudo-labeled data based on the validation performance.'''}			,		)
					UpperCAmelCase__      :					Optional[bool]					   =    dataclasses.field(
					    default=_SCREAMING_SNAKE_CASE			,		metadata={'''help''': '''Whether to fine-tune on labeled data after pseudo training.'''}			,		)
					UpperCAmelCase__      :					Optional[float]					   =    dataclasses.field(
					    default=0.0			,		metadata={'''help''': '''Confidence threshold for pseudo-labeled data filtering.'''}			,		)
					UpperCAmelCase__      :					Optional[int]					   =    dataclasses.field(
					    default=1_0_0			,		metadata={'''help''': '''Number of evaluation calls with no improvement after which training will be stopped.'''}			,		)
					UpperCAmelCase__      :					Optional[int]					   =    dataclasses.field(
					    default=_SCREAMING_SNAKE_CASE			,		metadata={'''help''': '''Random seed for initialization.'''}			,		)






def 							__lowercase     (				__lowerCAmelCase							:       Tuple      ,					__lowerCAmelCase							:       Optional[int]      ,					__lowerCAmelCase							:       Dict      ,					__lowerCAmelCase							:       List[str]      ,					__lowerCAmelCase							:       Union[str, Any]      ,					__lowerCAmelCase							:       str     ):
							a__    =  datasets.concatenate_datasets([infer_input, infer_output]      ,					axis=1     )

							if args.do_filter_by_confidence:
														a__    =  dataset.filter(lambda __lowerCAmelCase     : example["probability"] > args.confidence_threshold     )

							if args.do_filter_by_val_performance:
														assert eval_result >= 0.0 and eval_result <= 1.0
														a__    =  int(eval_result * len(__lowerCAmelCase     )     )
														print(__lowerCAmelCase     )
														a__    =  dataset.sort('probability'      ,					reverse=__lowerCAmelCase     )
														a__    =  dataset.select(range(__lowerCAmelCase     )     )

							a__    =  dataset.remove_columns(['label', 'probability']     )
							a__    =  dataset.rename_column('prediction'      ,					'label'     )
							a__    =  dataset.map(lambda __lowerCAmelCase     : {"label": idalabel[example["label"]]}     )
							a__    =  dataset.shuffle(seed=args.seed     )

							a__    =  os.path.join(__lowerCAmelCase      ,					F'train_pseudo.{args.data_file_extension}'     )
							if args.data_file_extension == "csv":
														dataset.to_csv(__lowerCAmelCase      ,					index=__lowerCAmelCase     )
							else:
														dataset.to_json(__lowerCAmelCase     )






def 							__lowercase     (				__lowerCAmelCase							:       List[str]      ,					__lowerCAmelCase							:       Tuple      ,					__lowerCAmelCase							:       Tuple      ,					__lowerCAmelCase							:       List[str]      ,					**__lowerCAmelCase							:       Optional[Any]     ):
							a__    =  Accelerator()
							# Make one log on every process with the configuration for debugging.
							logging.basicConfig(
							    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s'      ,					datefmt='%m/%d/%Y %H:%M:%S'      ,					level=logging.INFO      ,					)
							logger.info(accelerator.state     )

							# Setup logging, we only want one process per machine to log things on the
							# screen. accelerator.is_local_main_process is only True for one process per
							# machine.
							logger.setLevel(logging.INFO if accelerator.is_local_main_process else logging.ERROR     )

							if accelerator.is_local_main_process:
														datasets.utils.logging.set_verbosity_warning()
														transformers.utils.logging.set_verbosity_info()
							else:
														datasets.utils.logging.set_verbosity_error()
														transformers.utils.logging.set_verbosity_error()

							a__    =  STModelArguments(model_name_or_path=__lowerCAmelCase     )
							a__    =  STDataArguments(train_file=__lowerCAmelCase      ,					infer_file=__lowerCAmelCase     )
							a__    =  STTrainingArguments(output_dir=__lowerCAmelCase     )
							a__    =  argparse.Namespace()

							for arg_class in (model_args, data_args, training_args):
														for key, value in vars(__lowerCAmelCase     ).items():
																					setattr(__lowerCAmelCase      ,					__lowerCAmelCase      ,					__lowerCAmelCase     )

							for key, value in kwargs.items():
														if hasattr(__lowerCAmelCase      ,					__lowerCAmelCase     ):
																					setattr(__lowerCAmelCase      ,					__lowerCAmelCase      ,					__lowerCAmelCase     )

    # Sanity checks
							a__    =  {}
							a__    =  None

							# You need to provide the training data and the data to predict on
							assert args.train_file is not None
							assert args.infer_file is not None
							a__    =  args.train_file
							a__    =  args.infer_file

							if args.evaluation_strategy != IntervalStrategy.NO.value:
														assert args.eval_file is not None
														a__    =  args.eval_file

							for key in data_files:
														a__    =  data_files[key].split('.'     )[-1]
														assert extension in ["csv", "json"], F'`{key}_file` should be a csv or a json file.'
														if args.data_file_extension is None:
																					a__    =  extension
														else:
																					assert extension == args.data_file_extension, F'`{key}_file` should be a {args.data_file_extension} file`.'

							assert (
							    args.eval_metric in datasets.list_metrics()
							), F'{args.eval_metric} not in the list of supported metrics {datasets.list_metrics()}.'

							# If passed along, set the training seed now.
							if args.seed is not None:
														set_seed(args.seed     )

							logger.info('Creating the initial data directory for self-training...'     )
							a__    =  F'{args.output_dir}/self-train_iter-{{}}'.format
							a__    =  data_dir_format(0     )

							if accelerator.is_main_process:
														if args.output_dir is not None:
																					os.makedirs(args.output_dir      ,					exist_ok=__lowerCAmelCase     )
																					os.makedirs(__lowerCAmelCase      ,					exist_ok=__lowerCAmelCase     )
							accelerator.wait_for_everyone()

							a__    =  None
							a__    =  None
							a__    =  0
							a__    =  False
							# Show the progress bar
							a__    =  tqdm(range(args.max_selftrain_iterations     )      ,					disable=not accelerator.is_local_main_process     )

							# Self-train
							for iteration in range(0      ,					int(args.max_selftrain_iterations     )     ):
														a__    =  data_dir_format(__lowerCAmelCase     )
														assert os.path.exists(__lowerCAmelCase     )

														# Stage 1: initial fine-tuning for iteration = 0 or pseudo-training for
														# iteration > 0
														a__    =  os.path.join(__lowerCAmelCase      ,					'stage-1'     )
														a__    =  {
														    'accelerator': accelerator,
														    'model_name_or_path': args.model_name_or_path,
														    'cache_dir': args.cache_dir,
														    'do_train': True,
														    'train_file': data_files['train'] if iteration == 0 else data_files['train_pseudo'],
														    'do_eval': True if args.eval_file is not None else False,
														    'eval_file': data_files['eval'],
														    'do_predict': True,
														    'infer_file': data_files['infer'],
														    'task_name': args.task_name,
														    'label_list': args.label_list,
														    'output_dir': current_output_dir,
														    'eval_metric': args.eval_metric,
														    'evaluation_strategy': args.evaluation_strategy,
														    'early_stopping_patience': args.early_stopping_patience,
														    'early_stopping_threshold': args.early_stopping_threshold,
														    'seed': args.seed,
														}
														# Add additional training arguments
														for key, value in kwargs.items():
																					if key not in arguments_dict and not hasattr(__lowerCAmelCase      ,					__lowerCAmelCase     ):
																												arguments_dict.update({key: value}     )

														a__    =  os.path.join(__lowerCAmelCase      ,					'best-checkpoint'      ,					__lowerCAmelCase     )
														if os.path.exists(__lowerCAmelCase     ):
																					logger.info(
																					    'Found existing model checkpoint at %s. Skipping self-training: iteration: %d, stage: 1.'      ,					__lowerCAmelCase      ,					__lowerCAmelCase      ,					)
														else:
																					logger.info('***** Running self-training: iteration: %d, stage: 1 *****'      ,					__lowerCAmelCase     )
																					finetune(**__lowerCAmelCase     )
																					accelerator.wait_for_everyone()
																					assert os.path.exists(__lowerCAmelCase     )
																					logger.info('Self-training job completed: iteration: %d, stage: 1.'      ,					__lowerCAmelCase     )

														if iteration > 0 and args.finetune_on_labeled_data:
																					# Stage 2 (optional): fine-tuning on the original labeled data
																					a__    =  os.path.join(__lowerCAmelCase      ,					'best-checkpoint'     )
																					a__    =  os.path.join(__lowerCAmelCase      ,					'stage-2'     )
																					# Update arguments_dict
																					a__    =  model_path
																					a__    =  data_files['train']
																					a__    =  current_output_dir

																					a__    =  os.path.join(__lowerCAmelCase      ,					'best-checkpoint'      ,					__lowerCAmelCase     )
																					if os.path.exists(__lowerCAmelCase     ):
																												logger.info(
																												    'Found existing model checkpoint at %s. Skipping self-training: iteration: %d, stage: 2.'      ,					__lowerCAmelCase      ,					__lowerCAmelCase      ,					)
																					else:
																												logger.info('***** Running self-training: iteration: %d, stage: 2 *****'      ,					__lowerCAmelCase     )
																												finetune(**__lowerCAmelCase     )
																												accelerator.wait_for_everyone()
																												assert os.path.exists(__lowerCAmelCase     )
																												logger.info('Self-training job completed: iteration: %d, stage: 2.'      ,					__lowerCAmelCase     )

														a__    =  iteration
														a__    =  data_dir_format(iteration + 1     )

														a__    =  AutoConfig.from_pretrained(os.path.join(__lowerCAmelCase      ,					'best-checkpoint'     )     )
														a__    =  config.idalabel
														a__    =  os.path.join(__lowerCAmelCase      ,					'eval_results_best-checkpoint.json'     )
														a__    =  os.path.join(__lowerCAmelCase      ,					'test_results_best-checkpoint.json'     )
														assert os.path.exists(__lowerCAmelCase     )

														with open(__lowerCAmelCase      ,					'r'     ) as f:
																					a__    =  float(json.load(__lowerCAmelCase     )[args.eval_metric]     )
														a__    =  os.path.join(__lowerCAmelCase      ,					'infer_output_best-checkpoint.csv'     )
														assert os.path.exists(__lowerCAmelCase     )
														# Loading the dataset from local csv or json files.
														a__    =  load_dataset(args.data_file_extension      ,					data_files={'data': data_files['infer']}     )['data']
														a__    =  load_dataset('csv'      ,					data_files={'data': infer_output_file}     )['data']

														if accelerator.is_main_process:
																					os.makedirs(__lowerCAmelCase      ,					exist_ok=__lowerCAmelCase     )
																					shutil.copy(__lowerCAmelCase      ,					os.path.join(__lowerCAmelCase      ,					F'eval_results_iter-{iteration}.json'     )     )
																					if os.path.exists(__lowerCAmelCase     ):
																												shutil.copy(__lowerCAmelCase      ,					os.path.join(__lowerCAmelCase      ,					F'test_results_iter-{iteration}.json'     )     )
																					create_pseudo_labeled_data(__lowerCAmelCase      ,					__lowerCAmelCase      ,					__lowerCAmelCase      ,					__lowerCAmelCase      ,					__lowerCAmelCase      ,					__lowerCAmelCase     )
														accelerator.wait_for_everyone()

														a__    =  os.path.join(__lowerCAmelCase      ,					F'train_pseudo.{args.data_file_extension}'     )

														if args.evaluation_strategy != IntervalStrategy.NO.value:
																					a__    =  eval_result

																					if best_iteration is None:
																												a__    =  new_iteration
																												a__    =  new_eval_result
																					else:
																												if new_eval_result - best_eval_result > args.early_stopping_threshold:
																																			a__    =  new_iteration
																																			a__    =  new_eval_result
																																			a__    =  0
																												else:
																																			if new_eval_result == best_eval_result:
																																										a__    =  new_iteration
																																										a__    =  new_eval_result
																																			early_stopping_patience_counter += 1

																												if early_stopping_patience_counter >= args.early_stopping_patience:
																																			a__    =  True

														progress_bar.update(1     )

														if should_training_stop:
																					break

							if best_iteration is not None:
														# Save the best iteration
														logger.info('Best iteration: %d'      ,					__lowerCAmelCase     )
														logger.info('Best evaluation result: %s = %f'      ,					args.eval_metric      ,					__lowerCAmelCase     )
														accelerator.wait_for_everyone()
														if accelerator.is_main_process:
																					shutil.copy(
																					    os.path.join(__lowerCAmelCase      ,					F'eval_results_iter-{iteration}.json'     )      ,					os.path.join(__lowerCAmelCase      ,					'eval_results_best-iteration.json'     )      ,					)
							else:
														# Assume that the last iteration is the best
														logger.info('Best iteration: %d'      ,					args.max_selftrain_iterations - 1     )
														logger.info('Best evaluation result: %s = %f'      ,					args.eval_metric      ,					__lowerCAmelCase     )
														accelerator.wait_for_everyone()
														if accelerator.is_main_process:
																					shutil.copy(
																					    os.path.join(__lowerCAmelCase      ,					F'eval_results_iter-{args.max_selftrain_iterations - 1}.json'     )      ,					os.path.join(__lowerCAmelCase      ,					'eval_results_best-iteration.json'     )      ,					)








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  snake_case_     :
					def __init__(   self     :Optional[Any]    ,__snake_case     :str    ,__snake_case     :Optional[Any]=14    ,__snake_case     :Dict=7    ,__snake_case     :Optional[int]=True    ,__snake_case     :Optional[int]=True    ,__snake_case     :Dict=True    ,__snake_case     :List[Any]=True    ,__snake_case     :Optional[int]=True    ,__snake_case     :Any=99    ,__snake_case     :List[str]=32    ,__snake_case     :List[str]=5    ,__snake_case     :Tuple=4    ,__snake_case     :Optional[int]=37    ,__snake_case     :Optional[int]="gelu"    ,__snake_case     :Tuple=0.1    ,__snake_case     :Tuple=0.1    ,__snake_case     :Dict=5_12    ,__snake_case     :Union[str, Any]=16    ,__snake_case     :str=2    ,__snake_case     :Optional[Any]=0.02    ,__snake_case     :Dict=3    ,__snake_case     :Optional[Any]=4    ,__snake_case     :Optional[Any]=None    ,)      ->					Tuple:
												a__    =  parent
												a__    =  batch_size
												a__    =  seq_length
												a__    =  is_training
												a__    =  use_token_type_ids
												a__    =  use_input_mask
												a__    =  use_labels
												a__    =  use_mc_token_ids
												a__    =  vocab_size
												a__    =  hidden_size
												a__    =  num_hidden_layers
												a__    =  num_attention_heads
												a__    =  intermediate_size
												a__    =  hidden_act
												a__    =  hidden_dropout_prob
												a__    =  attention_probs_dropout_prob
												a__    =  max_position_embeddings
												a__    =  type_vocab_size
												a__    =  type_sequence_label_size
												a__    =  initializer_range
												a__    =  num_labels
												a__    =  num_choices
												a__    =  scope
												a__    =  self.vocab_size - 1
					def 				lowerCamelCase__(   self     :Optional[int]   )      ->					Union[str, Any]:
												a__    =  ids_tensor([self.batch_size, self.seq_length]    ,self.vocab_size   )

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

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

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

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

												a__    =  self.get_config()

												a__    =  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 				lowerCamelCase__(   self     :Optional[Any]   )      ->					Tuple:
												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 				lowerCamelCase__(   self     :str    ,__snake_case     :List[str]    ,__snake_case     :Any    ,__snake_case     :Dict    ,__snake_case     :int    ,__snake_case     :Optional[Any]    ,*__snake_case     :List[str]   )      ->					List[Any]:
												a__    =  CTRLModel(config=__snake_case   )
												model.to(__snake_case   )
												model.eval()

												model(__snake_case    ,token_type_ids=__snake_case    ,head_mask=__snake_case   )
												model(__snake_case    ,token_type_ids=__snake_case   )
												a__    =  model(__snake_case   )
												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 				lowerCamelCase__(   self     :Optional[int]    ,__snake_case     :List[str]    ,__snake_case     :Union[str, Any]    ,__snake_case     :str    ,__snake_case     :str    ,__snake_case     :Dict    ,*__snake_case     :Dict   )      ->					Dict:
												a__    =  CTRLLMHeadModel(__snake_case   )
												model.to(__snake_case   )
												model.eval()

												a__    =  model(__snake_case    ,token_type_ids=__snake_case    ,labels=__snake_case   )
												self.parent.assertEqual(result.loss.shape    ,()   )
												self.parent.assertEqual(result.logits.shape    ,(self.batch_size, self.seq_length, self.vocab_size)   )
					def 				lowerCamelCase__(   self     :Optional[Any]   )      ->					Optional[Any]:
												a__    =  self.prepare_config_and_inputs()

												(
												    (
												    a__
												)   ,      (
												    a__
												)   ,      (
												    a__
												)   ,      (
												    a__
												)   ,      (
												    a__
												)   ,      (
												    a__
												)   ,      (
												    a__
												)   ,      (
												    a__
												)   ,      (
												    a__
												)   ,      
												)    =  config_and_inputs

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

												return config, inputs_dict



					def 				lowerCamelCase__(   self     :Optional[int]    ,__snake_case     :Tuple    ,__snake_case     :str    ,__snake_case     :str    ,__snake_case     :List[str]    ,*__snake_case     :Optional[int]   )      ->					List[Any]:
												a__    =  self.num_labels
												a__    =  CTRLForSequenceClassification(__snake_case   )
												model.to(__snake_case   )
												model.eval()
												a__    =  ids_tensor([self.batch_size]    ,self.type_sequence_label_size   )
												a__    =  model(__snake_case    ,token_type_ids=__snake_case    ,labels=__snake_case   )
												self.parent.assertEqual(result.logits.shape    ,(self.batch_size, self.num_labels)   )







@require_torch
class  snake_case_     (lowerCamelCase_			,		lowerCamelCase_			,		lowerCamelCase_			,		unittest.TestCase   ):
					UpperCAmelCase__      :					Union[str, Any]					   =    (CTRLModel, CTRLLMHeadModel, CTRLForSequenceClassification) if is_torch_available() else ()
					UpperCAmelCase__      :					Any					   =    (CTRLLMHeadModel,) if is_torch_available() else ()
					UpperCAmelCase__      :					Any					   =    (
					    {
					        '''feature-extraction''': CTRLModel,
					        '''text-classification''': CTRLForSequenceClassification,
					        '''text-generation''': CTRLLMHeadModel,
					        '''zero-shot''': CTRLForSequenceClassification,
					    }
					    if is_torch_available()
					    else {}
					)
					UpperCAmelCase__      :					Tuple					   =    True
					UpperCAmelCase__      :					List[Any]					   =    False
					UpperCAmelCase__      :					List[str]					   =    False
					def 				lowerCamelCase__(   self     :Union[str, Any]    ,__snake_case     :Optional[int]    ,__snake_case     :int    ,__snake_case     :Any    ,__snake_case     :List[str]    ,__snake_case     :Dict   )      ->					Union[str, Any]:
												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 				lowerCamelCase__(   self     :int   )      ->					List[str]:
												a__    =  CTRLModelTester(self   )
												a__    =  ConfigTester(self    ,config_class=__snake_case    ,n_embd=37   )
					def 				lowerCamelCase__(   self     :str   )      ->					str:
												super().tearDown()
												# clean-up as much as possible GPU memory occupied by PyTorch
												gc.collect()
												torch.cuda.empty_cache()
					def 				lowerCamelCase__(   self     :Tuple   )      ->					List[Any]:
												self.config_tester.run_common_tests()
					def 				lowerCamelCase__(   self     :str   )      ->					str:
												a__    =  self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_ctrl_model(*__snake_case   )
					def 				lowerCamelCase__(   self     :List[Any]   )      ->					Any:
												a__    =  self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_lm_head_model(*__snake_case   )
					@unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'   )
					def 				lowerCamelCase__(   self     :Union[str, Any]   )      ->					Tuple:
												pass
					@slow
					def 				lowerCamelCase__(   self     :int   )      ->					List[Any]:
												for model_name in CTRL_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																			a__    =  CTRLModel.from_pretrained(__snake_case   )
																			self.assertIsNotNone(__snake_case   )



					@unittest.skip('The model doesn\'t support left padding'   )  # and it's not used enough to be worth fixing :)
					def 				lowerCamelCase__(   self     :Dict   )      ->					List[str]:
												pass







@require_torch
class  snake_case_     (unittest.TestCase   ):
					def 				lowerCamelCase__(   self     :Union[str, Any]   )      ->					str:
												super().tearDown()
												# clean-up as much as possible GPU memory occupied by PyTorch
												gc.collect()
												torch.cuda.empty_cache()



					@slow
					def 				lowerCamelCase__(   self     :Any   )      ->					Dict:
												a__    =  CTRLLMHeadModel.from_pretrained('ctrl'   )
												model.to(__snake_case   )
												a__    =  torch.tensor(
												    [[1_18_59, 0, 16_11, 8]]    ,dtype=torch.long    ,device=__snake_case   )  # Legal the president is
												a__    =  [
												    1_18_59,
												    0,
												    16_11,
												    8,
												    5,
												    1_50,
												    2_64_49,
												    2,
												    19,
												    3_48,
												    4_69,
												    3,
												    25_95,
												    48,
												    2_07_40,
												    24_65_33,
												    24_65_33,
												    19,
												    30,
												    5,
												]  # Legal the president is a good guy and I don't want to lose my job. \n \n I have a

												a__    =  model.generate(__snake_case    ,do_sample=__snake_case   )
												self.assertListEqual(output_ids[0].tolist()    ,__snake_case   )




