

'''simple docstring'''






import argparse
from copy import deepcopy

import numpy as np
from datasets import ClassLabel, DatasetDict, load_dataset
from evaluate import load

from transformers import (
    AutoModelForSequenceClassification,
    AutoTokenizer,
    DataCollatorWithPadding,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    set_seed,
)


def 		_A     (    ):






 """simple docstring"""



 __lowercase							  = argparse.ArgumentParser()
 parser.add_argument('''--model_ckpt'''     ,				type=A__     ,				default='''microsoft/unixcoder-base-nine'''      )
 parser.add_argument('''--num_epochs'''     ,				type=A__     ,				default=5      )
 parser.add_argument('''--batch_size'''     ,				type=A__     ,				default=6      )
 parser.add_argument('''--gradient_accumulation_steps'''     ,				type=A__     ,				default=1      )
 parser.add_argument('''--freeze'''     ,				type=A__     ,				default=A__      )
 parser.add_argument('''--learning_rate'''     ,				type=A__     ,				default=5e-4      )
 parser.add_argument('''--seed'''     ,				type=A__     ,				default=0      )
 parser.add_argument('''--lr_scheduler_type'''     ,				type=A__     ,				default='''cosine'''      )
 parser.add_argument('''--num_warmup_steps'''     ,				type=A__     ,				default=10      )
 parser.add_argument('''--weight_decay'''     ,				type=A__     ,				default=0.0_1      )
 parser.add_argument('''--output_dir'''     ,				type=A__     ,				default='''./results'''      )
 return parser.parse_args()


lowerCAmelCase__											=	load('''accuracy''')


def 		_A     (    A__      ):






 """simple docstring"""



 __lowercase  ,	__lowercase							  = eval_pred
 __lowercase							  = np.argmax(A__     ,				axis=1      )
 return metric.compute(predictions=A__     ,				references=A__      )






class 			lowercase_	(lowerCamelCase__      ):




  """simple docstring"""




  def __init__(      self   :     Union[str, Any]					,lowercase__   :     Optional[Any]			):
   super().__init__()
   __lowercase							  = trainer






  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[Any]					,lowercase__   :     List[str]					,lowercase__   :     Optional[int]					,lowercase__   :     Union[str, Any]					,**lowercase__   :     List[str]			):
   if control.should_evaluate:
    __lowercase							  = deepcopy(lowercase__			)
    self._trainer.evaluate(eval_dataset=self._trainer.train_dataset					,metric_key_prefix='''train'''			)
    return control_copy


def 		_A     (    ):






 """simple docstring"""



 __lowercase							  = get_args()
 set_seed(args.seed      )

 __lowercase							  = load_dataset('''codeparrot/codecomplex'''     ,				split='''train'''      )
 __lowercase							  = dataset.train_test_split(test_size=0.2      )
 __lowercase							  = train_test['''test'''].train_test_split(test_size=0.5      )
 __lowercase							  = DatasetDict(
     {
         '''train''': train_test['''train'''],
         '''test''': test_validation['''train'''],
         '''valid''': test_validation['''test'''],
     }      )

 print('''Loading tokenizer and model'''      )
 __lowercase							  = AutoTokenizer.from_pretrained(args.model_ckpt      )
 __lowercase							  = tokenizer.eos_token
 __lowercase							  = AutoModelForSequenceClassification.from_pretrained(args.model_ckpt     ,				num_labels=7      )
 __lowercase							  = model.config.eos_token_id

 if args.freeze:
  for param in model.roberta.parameters():
   __lowercase							  = False

 __lowercase							  = ClassLabel(num_classes=7     ,				names=list(set(train_test_validation['''train''']['''complexity''']      )      )      )

 def tokenize(A__      ):
  __lowercase							  = tokenizer(example['''src''']     ,				truncation=A__     ,				max_length=1024      )
  __lowercase							  = labels.straint(example['''complexity''']      )
  return {
      "input_ids": inputs["input_ids"],
      "attention_mask": inputs["attention_mask"],
      "label": label,
  }

 __lowercase							  = train_test_validation.map(
     A__     ,				batched=A__     ,				remove_columns=train_test_validation['''train'''].column_names     ,				)
 __lowercase							  = DataCollatorWithPadding(tokenizer=A__      )

 __lowercase							  = TrainingArguments(
     output_dir=args.output_dir     ,				learning_rate=args.learning_rate     ,				lr_scheduler_type=args.lr_scheduler_type     ,				evaluation_strategy='''epoch'''     ,				save_strategy='''epoch'''     ,				logging_strategy='''epoch'''     ,				per_device_train_batch_size=args.batch_size     ,				per_device_eval_batch_size=args.batch_size     ,				num_train_epochs=args.num_epochs     ,				gradient_accumulation_steps=args.gradient_accumulation_steps     ,				weight_decay=0.0_1     ,				metric_for_best_model='''accuracy'''     ,				run_name='''complexity-java'''     ,				report_to='''wandb'''     ,				)

 __lowercase							  = Trainer(
     model=A__     ,				args=A__     ,				train_dataset=tokenized_datasets['''train''']     ,				eval_dataset=tokenized_datasets['''valid''']     ,				tokenizer=A__     ,				data_collator=A__     ,				compute_metrics=A__     ,				)

 print('''Training...'''      )
 trainer.add_callback(CustomCallback(A__      )      )
 trainer.train()


if __name__ == "__main__":
 main()




'''simple docstring'''






import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase__											=	logging.get_logger(__name__)

lowerCAmelCase__											=	{'''vocab_file''': '''spiece.model'''}

lowerCAmelCase__											=	{
    '''vocab_file''': {
        '''bert_for_seq_generation''': (
            '''https://huggingface.co/google/bert_for_seq_generation_L-24_bbc_encoder/resolve/main/spiece.model'''
        ),
    }
}

lowerCAmelCase__											=	{'''bert_for_seq_generation''': 512}






class 			lowercase_	(lowerCamelCase__      ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE    :  List[Any]							=			VOCAB_FILES_NAMES
  SCREAMING_SNAKE_CASE    :  Dict							=			PRETRAINED_VOCAB_FILES_MAP
  SCREAMING_SNAKE_CASE    :  Optional[int]							=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  SCREAMING_SNAKE_CASE    :  List[int]							=			[]
  SCREAMING_SNAKE_CASE    :  Any							=			['input_ids', 'attention_mask']






  def __init__(      self   :     Optional[Any]					,lowercase__   :     Tuple					,lowercase__   :     Tuple="<s>"					,lowercase__   :     Union[str, Any]="</s>"					,lowercase__   :     str="<unk>"					,lowercase__   :     Tuple="<pad>"					,lowercase__   :     Union[str, Any]="<::::>"					,lowercase__   :     Optional[Dict[str, Any]] = None					,**lowercase__   :     Any					,):
   __lowercase							  = {} if sp_model_kwargs is None else sp_model_kwargs

   # Add extra_ids to the special token list
   super().__init__(
       bos_token=lowercase__					,eos_token=lowercase__					,unk_token=lowercase__					,pad_token=lowercase__					,sep_token=lowercase__					,sp_model_kwargs=self.sp_model_kwargs					,**lowercase__					,)

   __lowercase							  = vocab_file

   __lowercase							  = spm.SentencePieceProcessor(**self.sp_model_kwargs			)
   self.sp_model.Load(lowercase__			)






  @property
  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[int]			):
   return self.sp_model.get_piece_size()






  def 					SCREAMING_SNAKE_CASE      (      self   :     Union[str, Any]			):
   __lowercase							  = {self.convert_ids_to_tokens(lowercase__			): i for i in range(self.vocab_size			)}
   vocab.update(self.added_tokens_encoder			)
   return vocab






  def __getstate__(      self   :     str			):
   __lowercase							  = self.__dict__.copy()
   __lowercase							  = None
   return state






  def __setstate__(      self   :     Optional[int]					,lowercase__   :     Optional[Any]			):
   __lowercase							  = d

   # for backward compatibility
   if not hasattr(self					,'''sp_model_kwargs'''			):
    __lowercase							  = {}

   __lowercase							  = spm.SentencePieceProcessor(**self.sp_model_kwargs			)
   self.sp_model.Load(self.vocab_file			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     str					,lowercase__   :     str			):
   return self.sp_model.encode(lowercase__					,out_type=lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Tuple					,lowercase__   :     Union[str, Any]			):
   return self.sp_model.piece_to_id(lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Any					,lowercase__   :     Tuple			):
   __lowercase							  = self.sp_model.IdToPiece(lowercase__			)
   return token






  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[int]					,lowercase__   :     int			):
   __lowercase							  = []
   __lowercase							  = ''''''
   for token in tokens:
    # make sure that special tokens are not decoded using sentencepiece model
    if token in self.all_special_tokens:
     out_string += self.sp_model.decode(lowercase__			) + token
     __lowercase							  = []
    else:
     current_sub_tokens.append(lowercase__			)
   out_string += self.sp_model.decode(lowercase__			)
   return out_string.strip()






  def 					SCREAMING_SNAKE_CASE      (      self   :     int					,lowercase__   :     str					,lowercase__   :     Optional[str] = None			):
   if not os.path.isdir(lowercase__			):
    logger.error(F"Vocabulary path ({save_directory}) should be a directory"			)
    return
   __lowercase							  = os.path.join(
       lowercase__					,(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']			)

   if os.path.abspath(self.vocab_file			) != os.path.abspath(lowercase__			) and os.path.isfile(self.vocab_file			):
    copyfile(self.vocab_file					,lowercase__			)
   elif not os.path.isfile(self.vocab_file			):
    with open(lowercase__					,'''wb'''			) as fi:
     __lowercase							  = self.sp_model.serialized_model_proto()
     fi.write(lowercase__			)

   return (out_vocab_file,)

