




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 				lowerCAmelCase_	(     ):
      _A							:     Optional[int]			=						argparse.ArgumentParser()
      parser.add_argument("""--model_ckpt""",type=snake_case_,default="""microsoft/unixcoder-base-nine"""     )
      parser.add_argument("""--num_epochs""",type=snake_case_,default=5     )
      parser.add_argument("""--batch_size""",type=snake_case_,default=6     )
      parser.add_argument("""--gradient_accumulation_steps""",type=snake_case_,default=1     )
      parser.add_argument("""--freeze""",type=snake_case_,default=snake_case_     )
      parser.add_argument("""--learning_rate""",type=snake_case_,default=5e-4     )
      parser.add_argument("""--seed""",type=snake_case_,default=0     )
      parser.add_argument("""--lr_scheduler_type""",type=snake_case_,default="""cosine"""     )
      parser.add_argument("""--num_warmup_steps""",type=snake_case_,default=10     )
      parser.add_argument("""--weight_decay""",type=snake_case_,default=0.01     )
      parser.add_argument("""--output_dir""",type=snake_case_,default="""./results"""     )
      return parser.parse_args()


_snake_case										=       load("accuracy")







def 				lowerCAmelCase_	(     snake_case_     ):
      _A			,   _A							:     str			=						eval_pred
      _A							:     Optional[Any]			=						np.argmax(snake_case_,axis=1     )
      return metric.compute(predictions=snake_case_,references=snake_case_     )







class 						lowercase							(   UpperCamelCase__		):

     def __init__(  self		,     _a				)						->							None:
           super().__init__()
           _A							:     Dict			=						trainer



     def 					a__	(  self		,     _a		,     _a		,     _a		,     **_a				)						->							str:
           if control.should_evaluate:
                 _A							:     Tuple			=						deepcopy(_a				)
                 self._trainer.evaluate(eval_dataset=self._trainer.train_dataset		,     metric_key_prefix="""train"""				)
                 return control_copy







def 				lowerCAmelCase_	(     ):
      _A							:     List[Any]			=						get_args()
      set_seed(args.seed     )

      _A							:     Optional[int]			=						load_dataset("""codeparrot/codecomplex""",split="""train"""     )
      _A							:     Optional[int]			=						dataset.train_test_split(test_size=0.2     )
      _A							:     str			=						train_test["""test"""].train_test_split(test_size=0.5     )
      _A							:     Optional[int]			=						DatasetDict(
          {
              """train""": train_test["""train"""],
              """test""": test_validation["""train"""],
              """valid""": test_validation["""test"""],
          }     )

      print("""Loading tokenizer and model"""     )
      _A							:     List[Any]			=						AutoTokenizer.from_pretrained(args.model_ckpt     )
      _A							:     List[str]			=						tokenizer.eos_token
      _A							:     int			=						AutoModelForSequenceClassification.from_pretrained(args.model_ckpt,num_labels=7     )
      _A							:     Union[str, Any]			=						model.config.eos_token_id

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

      _A							:     int			=						ClassLabel(num_classes=7,names=list(set(train_test_validation["""train"""]["""complexity"""]     )     )     )

      def tokenize(snake_case_     ):
            _A							:     List[str]			=						tokenizer(example["""src"""],truncation=snake_case_,max_length=1024     )
            _A							:     List[Any]			=						labels.straint(example["""complexity"""]     )
            return {
                "input_ids": inputs["input_ids"],
                "attention_mask": inputs["attention_mask"],
                "label": label,
            }

      _A							:     Optional[Any]			=						train_test_validation.map(
          snake_case_,batched=snake_case_,remove_columns=train_test_validation["""train"""].column_names,)
      _A							:     List[str]			=						DataCollatorWithPadding(tokenizer=snake_case_     )

      _A							:     int			=						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.01,metric_for_best_model="""accuracy""",run_name="""complexity-java""",report_to="""wandb""",)

      _A							:     Dict			=						Trainer(
          model=snake_case_,args=snake_case_,train_dataset=tokenized_datasets["""train"""],eval_dataset=tokenized_datasets["""valid"""],tokenizer=snake_case_,data_collator=snake_case_,compute_metrics=snake_case_,)

      print("""Training..."""     )
      trainer.add_callback(CustomCallback(snake_case_     )     )
      trainer.train()


if __name__ == "__main__":
  main()




import itertools
import os
import random
import tempfile
import unittest

import numpy as np
from datasets import load_dataset

from transformers import is_speech_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
 from transformers import WhisperFeatureExtractor

if is_torch_available():
 import torch

_UpperCamelCase 			=		random.Random()







def 		_lowercase (     lowercase__   ,	lowercase__=1.0   ,	lowercase__=None   ,	lowercase__=None						):
      if rng is None:
            __lowerCAmelCase					:						Any					=					global_rng

      __lowerCAmelCase					:						str					=					[]
      for batch_idx in range(shape[0]						):
            values.append([]						)
            for _ in range(shape[1]						):
                  values[-1].append(rng.random() * scale						)

      return values







@require_torch
@require_torchaudio
class   __lowercase      (unittest.TestCase ):
 def __init__(							self      ,							A_      ,							A_=7      ,							A_=400      ,							A_=2000      ,							A_=10      ,							A_=160      ,							A_=8      ,							A_=0.0      ,							A_=4000      ,							A_=False      ,							A_=True      ,							)					->List[str]:



       '''simple docstring'''





       __lowerCAmelCase					:						Optional[int]					=					parent
       __lowerCAmelCase					:						Dict					=					batch_size
       __lowerCAmelCase					:						str					=					min_seq_length
       __lowerCAmelCase					:						int					=					max_seq_length
       __lowerCAmelCase					:						Any					=					(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
       __lowerCAmelCase					:						Any					=					padding_value
       __lowerCAmelCase					:						str					=					sampling_rate
       __lowerCAmelCase					:						Optional[Any]					=					return_attention_mask
       __lowerCAmelCase					:						Optional[Any]					=					do_normalize
       __lowerCAmelCase					:						Optional[Any]					=					feature_size
       __lowerCAmelCase					:						Optional[int]					=					chunk_length
       __lowerCAmelCase					:						Optional[Any]					=					hop_length
 def 	UpperCamelCase__  (							self					)					->Optional[Any]:



       '''simple docstring'''





       return {
           "feature_size": self.feature_size,
           "hop_length": self.hop_length,
           "chunk_length": self.chunk_length,
           "padding_value": self.padding_value,
           "sampling_rate": self.sampling_rate,
           "return_attention_mask": self.return_attention_mask,
           "do_normalize": self.do_normalize,
       }





 def 	UpperCamelCase__  (							self      ,							A_=False      ,							A_=False					)					->Optional[Any]:



       '''simple docstring'''





       def _flatten(A_					):
             return list(itertools.chain(*A_					)					)

       if equal_length:
             __lowerCAmelCase					:						str					=					[floats_list((self.max_seq_length, self.feature_size)					) for _ in range(self.batch_size					)]
       else:
             # make sure that inputs increase in size
             __lowerCAmelCase					:						Any					=					[
                 floats_list((x, self.feature_size)					)
                 for x in range(self.min_seq_length      ,							self.max_seq_length      ,							self.seq_length_diff					)
             ]
       if numpify:
             __lowerCAmelCase					:						Optional[Any]					=					[np.asarray(A_					) for x in speech_inputs]
       return speech_inputs







@require_torch
@require_torchaudio
class   __lowercase      (_UpperCAmelCase		,   unittest.TestCase ):
 _UpperCamelCase			  =  WhisperFeatureExtractor if is_speech_available() else None
 def 	UpperCamelCase__  (							self					)					->Optional[int]:



       '''simple docstring'''





       __lowerCAmelCase					:						Tuple					=					WhisperFeatureExtractionTester(self					)
 def 	UpperCamelCase__  (							self					)					->int:



       '''simple docstring'''





       __lowerCAmelCase					:						str					=					self.feature_extraction_class(**self.feat_extract_dict					)

       with tempfile.TemporaryDirectory() as tmpdirname:
             __lowerCAmelCase					:						List[str]					=					feat_extract_first.save_pretrained(A_					)[0]
             check_json_file_has_correct_format(A_					)
             __lowerCAmelCase					:						int					=					self.feature_extraction_class.from_pretrained(A_					)

       __lowerCAmelCase					:						Dict					=					feat_extract_first.to_dict()
       __lowerCAmelCase					:						Union[str, Any]					=					feat_extract_second.to_dict()
       __lowerCAmelCase					:						Union[str, Any]					=					feat_extract_first.mel_filters
       __lowerCAmelCase					:						Dict					=					feat_extract_second.mel_filters
       self.assertTrue(np.allclose(A_      ,							A_					)					)
       self.assertEqual(A_      ,							A_					)
 def 	UpperCamelCase__  (							self					)					->str:



       '''simple docstring'''





       __lowerCAmelCase					:						str					=					self.feature_extraction_class(**self.feat_extract_dict					)

       with tempfile.TemporaryDirectory() as tmpdirname:
             __lowerCAmelCase					:						Union[str, Any]					=					os.path.join(A_      ,							'''feat_extract.json'''					)
             feat_extract_first.to_json_file(A_					)
             __lowerCAmelCase					:						List[str]					=					self.feature_extraction_class.from_json_file(A_					)

       __lowerCAmelCase					:						List[str]					=					feat_extract_first.to_dict()
       __lowerCAmelCase					:						Tuple					=					feat_extract_second.to_dict()
       __lowerCAmelCase					:						Any					=					feat_extract_first.mel_filters
       __lowerCAmelCase					:						List[str]					=					feat_extract_second.mel_filters
       self.assertTrue(np.allclose(A_      ,							A_					)					)
       self.assertEqual(A_      ,							A_					)
 def 	UpperCamelCase__  (							self					)					->List[Any]:



       '''simple docstring'''





       __lowerCAmelCase					:						List[str]					=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
       # create three inputs of length 800, 1000, and 1200
       __lowerCAmelCase					:						Dict					=					[floats_list((1, x)					)[0] for x in range(800      ,							1400      ,							200					)]
       __lowerCAmelCase					:						Dict					=					[np.asarray(A_					) for speech_input in speech_inputs]

       # Test feature size
       __lowerCAmelCase					:						Tuple					=					feature_extractor(A_      ,							padding='''max_length'''      ,							return_tensors='''np'''					).input_features
       self.assertTrue(input_features.ndim == 3					)
       self.assertTrue(input_features.shape[-1] == feature_extractor.nb_max_frames					)
       self.assertTrue(input_features.shape[-2] == feature_extractor.feature_size					)

       # Test not batched input
       __lowerCAmelCase					:						Dict					=					feature_extractor(speech_inputs[0]      ,							return_tensors='''np'''					).input_features
       __lowerCAmelCase					:						List[str]					=					feature_extractor(np_speech_inputs[0]      ,							return_tensors='''np'''					).input_features
       self.assertTrue(np.allclose(A_      ,							A_      ,							atol=1e-3					)					)

       # Test batched
       __lowerCAmelCase					:						Union[str, Any]					=					feature_extractor(A_      ,							return_tensors='''np'''					).input_features
       __lowerCAmelCase					:						List[Any]					=					feature_extractor(A_      ,							return_tensors='''np'''					).input_features
       for enc_seq_a, enc_seq_a in zip(A_      ,							A_					):
             self.assertTrue(np.allclose(A_      ,							A_      ,							atol=1e-3					)					)

       # Test 2-D numpy arrays are batched.
       __lowerCAmelCase					:						int					=					[floats_list((1, x)					)[0] for x in (800, 800, 800)]
       __lowerCAmelCase					:						Optional[int]					=					np.asarray(A_					)
       __lowerCAmelCase					:						Dict					=					feature_extractor(A_      ,							return_tensors='''np'''					).input_features
       __lowerCAmelCase					:						Any					=					feature_extractor(A_      ,							return_tensors='''np'''					).input_features
       for enc_seq_a, enc_seq_a in zip(A_      ,							A_					):
             self.assertTrue(np.allclose(A_      ,							A_      ,							atol=1e-3					)					)

       # Test truncation required
       __lowerCAmelCase					:						Optional[int]					=					[floats_list((1, x)					)[0] for x in range(200      ,							(feature_extractor.n_samples + 500)      ,							200					)]
       __lowerCAmelCase					:						Dict					=					[np.asarray(A_					) for speech_input in speech_inputs]

       __lowerCAmelCase					:						Tuple					=					[x[: feature_extractor.n_samples] for x in speech_inputs]
       __lowerCAmelCase					:						Optional[int]					=					[np.asarray(A_					) for speech_input in speech_inputs_truncated]

       __lowerCAmelCase					:						Any					=					feature_extractor(A_      ,							return_tensors='''np'''					).input_features
       __lowerCAmelCase					:						List[str]					=					feature_extractor(A_      ,							return_tensors='''np'''					).input_features
       for enc_seq_a, enc_seq_a in zip(A_      ,							A_					):
             self.assertTrue(np.allclose(A_      ,							A_      ,							atol=1e-3					)					)
 def 	UpperCamelCase__  (							self					)					->Dict:



       '''simple docstring'''





       import torch

       __lowerCAmelCase					:						Tuple					=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
       __lowerCAmelCase					:						List[Any]					=					np.random.rand(100      ,							32					).astype(np.floataa					)
       __lowerCAmelCase					:						Any					=					np_speech_inputs.tolist()

       for inputs in [py_speech_inputs, np_speech_inputs]:
             __lowerCAmelCase					:						Tuple					=					feature_extractor.pad([{'''input_features''': inputs}]      ,							return_tensors='''np'''					)
             self.assertTrue(np_processed.input_features.dtype == np.floataa					)
             __lowerCAmelCase					:						int					=					feature_extractor.pad([{'''input_features''': inputs}]      ,							return_tensors='''pt'''					)
             self.assertTrue(pt_processed.input_features.dtype == torch.floataa					)
 def 	UpperCamelCase__  (							self      ,							A_					)					->str:



       '''simple docstring'''





       __lowerCAmelCase					:						Any					=					load_dataset('''hf-internal-testing/librispeech_asr_dummy'''      ,							'''clean'''      ,							split='''validation'''					)
       # automatic decoding with librispeech
       __lowerCAmelCase					:						Union[str, Any]					=					ds.sort('''id'''					).select(range(A_					)					)[:num_samples]['''audio''']

       return [x["array"] for x in speech_samples]
 def 	UpperCamelCase__  (							self					)					->int:



       '''simple docstring'''





       __lowerCAmelCase					:						Optional[int]					=					torch.tensor(
           [
               0.1_193, -0.0_946, -0.1_098, -0.0_196, 0.0_225, -0.0_690, -0.1_736, 0.0_951,
               0.0_971, -0.0_817, -0.0_702, 0.0_162, 0.0_260, 0.0_017, -0.0_192, -0.1_678,
               0.0_709, -0.1_867, -0.0_655, -0.0_274, -0.0_234, -0.1_884, -0.0_516, -0.0_554,
               -0.0_274, -0.1_425, -0.1_423, 0.0_837, 0.0_377, -0.0_854
           ]					)
       # fmt: on

       __lowerCAmelCase					:						int					=					self._load_datasamples(1					)
       __lowerCAmelCase					:						Any					=					WhisperFeatureExtractor()
       __lowerCAmelCase					:						Optional[Any]					=					feature_extractor(A_      ,							return_tensors='''pt'''					).input_features
       self.assertEqual(input_features.shape      ,							(1, 80, 3000)					)
       self.assertTrue(torch.allclose(input_features[0, 0, :30]      ,							A_      ,							atol=1e-4					)					)





 def 	UpperCamelCase__  (							self					)					->Any:



       '''simple docstring'''





       __lowerCAmelCase					:						Tuple					=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()					)
       __lowerCAmelCase					:						str					=					self._load_datasamples(1					)[0]
       __lowerCAmelCase					:						Optional[Any]					=					((audio - audio.min()) / (audio.max() - audio.min())) * 6_5535  # Rescale to [0, 65535] to show issue
       __lowerCAmelCase					:						Union[str, Any]					=					feat_extract.zero_mean_unit_var_norm([audio]      ,							attention_mask=A_					)[0]

       self.assertTrue(np.all(np.mean(A_					) < 1e-3					)					)
       self.assertTrue(np.all(np.abs(np.var(A_					) - 1					) < 1e-3					)					)


