


import argparse
import logging
import pickle
import random
import time

import numpy as np

from transformers import BertTokenizer, GPTaTokenizer, RobertaTokenizer


logging.basicConfig(
    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s''', datefmt='''%m/%d/%Y %H:%M:%S''', level=logging.INFO
)
snake_case							:       int     			=      logging.getLogger(__name__)






def 							__lowercase   ( ):
							a__             =     argparse.ArgumentParser(
							    description='Preprocess the data to avoid re-doing it several times by (tokenization + token_to_ids).'   )
							parser.add_argument('--file_path'					,    type=__lowerCAmelCase					,    default='data/dump.txt'					,    help='The path to the data.'   )
							parser.add_argument('--tokenizer_type'					,    type=__lowerCAmelCase					,    default='bert'					,    choices=['bert', 'roberta', 'gpt2']   )
							parser.add_argument('--tokenizer_name'					,    type=__lowerCAmelCase					,    default='bert-base-uncased'					,    help='The tokenizer to use.'   )
							parser.add_argument('--dump_file'					,    type=__lowerCAmelCase					,    default='data/dump'					,    help='The dump file prefix.'   )
							a__             =     parser.parse_args()

							logger.info(F'Loading Tokenizer ({args.tokenizer_name})'   )
							if args.tokenizer_type == "bert":
														a__             =     BertTokenizer.from_pretrained(args.tokenizer_name   )
														a__             =     tokenizer.special_tokens_map['cls_token']  # `[CLS]`
														a__             =     tokenizer.special_tokens_map['sep_token']  # `[SEP]`
							elif args.tokenizer_type == "roberta":
														a__             =     RobertaTokenizer.from_pretrained(args.tokenizer_name   )
														a__             =     tokenizer.special_tokens_map['cls_token']  # `<s>`
														a__             =     tokenizer.special_tokens_map['sep_token']  # `</s>`
							elif args.tokenizer_type == "gpt2":
														a__             =     GPTaTokenizer.from_pretrained(args.tokenizer_name   )
														a__             =     tokenizer.special_tokens_map['bos_token']  # `<|endoftext|>`
														a__             =     tokenizer.special_tokens_map['eos_token']  # `<|endoftext|>`

							logger.info(F'Loading text from {args.file_path}'   )
							with open(args.file_path					,    'r'					,    encoding='utf8'   ) as fp:
														a__             =     fp.readlines()

							logger.info('Start encoding'   )
							logger.info(F'{len(__lowerCAmelCase   )} examples to process.'   )

							a__             =     []
							a__             =     0
							a__             =     1_0_0_0_0
							a__             =     time.time()
							for text in data:
														a__             =     F'{bos} {text.strip()} {sep}'
														a__             =     tokenizer.encode(__lowerCAmelCase					,    add_special_tokens=__lowerCAmelCase   )
														rslt.append(__lowerCAmelCase   )

														iter += 1
														if iter % interval == 0:
																					a__             =     time.time()
																					logger.info(F'{iter} examples processed. - {(end-start):.2f}s/{interval}expl'   )
																					a__             =     time.time()
							logger.info('Finished binarization'   )
							logger.info(F'{len(__lowerCAmelCase   )} examples processed.'   )

							a__             =     F'{args.dump_file}.{args.tokenizer_name}.pickle'
							a__             =     tokenizer.vocab_size
							if vocab_size < (1 << 1_6):
														a__             =     [np.uintaa(__lowerCAmelCase   ) for d in rslt]
							else:
														a__             =     [np.intaa(__lowerCAmelCase   ) for d in rslt]
							random.shuffle(rslt_   )
							logger.info(F'Dump to {dp_file}'   )
							with open(__lowerCAmelCase					,    'wb'   ) as handle:
														pickle.dump(rslt_					,    __lowerCAmelCase					,    protocol=pickle.HIGHEST_PROTOCOL   )


if __name__ == "__main__":
						main()








import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
						from sagemaker import Session, TrainingJobAnalytics
						from sagemaker.huggingface import HuggingFace







@pytest.mark.skipif(
    literal_eval(os.getenv('''TEST_SAGEMAKER'''	,						'''False'''      )      ) is not True	,						reason='''Skipping test because should only be run when releasing minor transformers version'''	,						)
@pytest.mark.usefixtures('''sm_env'''      )
@parameterized_class(
    [
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_glue.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.g4dn.xlarge''',
            '''results''': {'''train_runtime''': 6_5_0, '''eval_accuracy''': 0.6, '''eval_loss''': 0.9},
        },
        {
            '''framework''': '''tensorflow''',
            '''script''': '''run_tf.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.g4dn.xlarge''',
            '''results''': {'''train_runtime''': 6_0_0, '''eval_accuracy''': 0.3, '''eval_loss''': 0.9},
        },
    ]      )
class        snake_case_   (unittest.TestCase      ):
					def  lowerCamelCase__(      self   :Dict      )					->    Union[str, Any]:
												if self.framework == "pytorch":
																			subprocess.run(
																			    F'cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py'.split()			,encoding='utf-8'			,check=__snake_case			,)
												assert hasattr(self			,'env'      )
					def  lowerCamelCase__(      self   :List[Any]			,__snake_case   :Optional[Any]=1      )					->    int:
												# creates estimator
												return HuggingFace(
												    entry_point=self.script			,source_dir=self.env.test_path			,role=self.env.role			,image_uri=self.env.image_uri			,base_job_name=F'{self.env.base_job_name}-single'			,instance_count=__snake_case			,instance_type=self.instance_type			,debugger_hook_config=__snake_case			,hyperparameters={**self.env.hyperparameters, 'model_name_or_path': self.model_name_or_path}			,metric_definitions=self.env.metric_definitions			,py_version='py36'			,)
					def  lowerCamelCase__(      self   :Dict			,__snake_case   :str      )					->    Optional[int]:
												TrainingJobAnalytics(__snake_case      ).export_csv(F'{self.env.test_path}/{job_name}_metrics.csv'      )



					def  lowerCamelCase__(      self   :Tuple      )					->    Union[str, Any]:
												# create estimator
												a__             =     self.create_estimator()

												# run training
												estimator.fit()

												# result dataframe
												a__             =     TrainingJobAnalytics(estimator.latest_training_job.name      ).dataframe()

												# extract kpis
												a__             =     list(result_metrics_df[result_metrics_df.metric_name == 'eval_accuracy']['value']      )
												a__             =     list(result_metrics_df[result_metrics_df.metric_name == 'eval_loss']['value']      )
												# get train time from SageMaker job, this includes starting, preprocessing, stopping
												a__             =     (
												    Session().describe_training_job(estimator.latest_training_job.name      ).get('TrainingTimeInSeconds'			,99_99_99      )
												)

												# assert kpis
												assert train_runtime <= self.results["train_runtime"]
												assert all(t >= self.results['eval_accuracy'] for t in eval_accuracy      )
												assert all(t <= self.results['eval_loss'] for t in eval_loss      )

												# dump tests result into json file to share in PR
												with open(F'{estimator.latest_training_job.name}.json'			,'w'      ) as outfile:
																			json.dump({'train_time': train_runtime, 'eval_accuracy': eval_accuracy, 'eval_loss': eval_loss}			,__snake_case      )




