
'''simple docstring'''





import logging
import os
from dataclasses import dataclass, field
from typing import Dict, List, Optional

import numpy as np
import torch
from utils_hans import HansDataset, InputFeatures, hans_processors, hans_tasks_num_labels

import transformers
from transformers import (
    AutoConfig,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    default_data_collator,
    set_seed,
)
from transformers.trainer_utils import is_main_process


SCREAMING_SNAKE_CASE    =					logging.getLogger(__name__)






@dataclass
class        UpperCAmelCase_       :

 '''simple docstring'''
 lowercase_				:						str				=       field(
     metadata={"help": "Path to pretrained model or model identifier from huggingface.co/models"}							)
 lowercase_				:						Optional[str]				=       field(
     default=lowerCAmelCase_				,							metadata={"help": "Pretrained config name or path if not the same as model_name"}							)
 lowercase_				:						Optional[str]				=       field(
     default=lowerCAmelCase_				,							metadata={"help": "Pretrained tokenizer name or path if not the same as model_name"}							)
 lowercase_				:						Optional[str]				=       field(
     default=lowerCAmelCase_				,							metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"}				,							)






@dataclass
class        UpperCAmelCase_       :

 '''simple docstring'''
 lowercase_				:						str				=       field(
     metadata={"help": "The name of the task to train selected in the list: " + ", ".join(hans_processors.keys()							)}							)
 lowercase_				:						str				=       field(
     metadata={"help": "The input data dir. Should contain the .tsv files (or other data files) for the task."}							)
 lowercase_				:						int				=       field(
     default=128				,							metadata={
         "help": (
             "The maximum total input sequence length after tokenization. Sequences longer "
             "than this will be truncated, sequences shorter will be padded."
         )
     }				,							)
 lowercase_				:						bool				=       field(
     default=lowerCAmelCase_				,							metadata={"help": "Overwrite the cached training and evaluation sets"}							)







def snake_case_    (				lowercase__		):
  UpperCAmelCase__			:				Optional[Any]												=       default_data_collator(a_		)
  UpperCAmelCase__			:				List[Any]												=       batch.pop("pairID"      ,					a_		)
  return batch







def snake_case_    (				):
  # See all possible arguments in src/transformers/training_args.py
  # or by passing the --help flag to this script.
  # We now keep distinct sets of args, for a cleaner separation of concerns.

  UpperCAmelCase__			:				int												=       HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)		)
  UpperCAmelCase__					, UpperCAmelCase__					, UpperCAmelCase__			:				List[Any]												=       parser.parse_args_into_dataclasses()

  if (
      os.path.exists(training_args.output_dir		)
      and os.listdir(training_args.output_dir		)
      and training_args.do_train
      and not training_args.overwrite_output_dir
  ):
    raise ValueError(
        F"""Output directory ({training_args.output_dir}) already exists and is not empty. Use"""
        " --overwrite_output_dir to overcome."		)

  # Setup logging
  logging.basicConfig(
      format="%(asctime)s - %(levelname)s - %(name)s - %(message)s"      ,					datefmt="%m/%d/%Y %H:%M:%S"      ,					level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN      ,					)
  logger.warning(
      "Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s"      ,					training_args.local_rank      ,					training_args.device      ,					training_args.n_gpu      ,					bool(training_args.local_rank != -1		)      ,					training_args.fpaa      ,					)
  # Set the verbosity to info of the Transformers logger (on main process only):
  if is_main_process(training_args.local_rank		):
    transformers.utils.logging.set_verbosity_info()
    transformers.utils.logging.enable_default_handler()
    transformers.utils.logging.enable_explicit_format()
  logger.info("Training/evaluation parameters %s"      ,					a_		)

  # Set seed
  set_seed(training_args.seed		)

  try:
    UpperCAmelCase__			:				Optional[int]												=       hans_tasks_num_labels[data_args.task_name]
  except KeyError:
    raise ValueError("Task not found: %s" % (data_args.task_name)		)

  # Load pretrained model and tokenizer
  #
  # Distributed training:
  # The .from_pretrained methods guarantee that only one local process can concurrently
  # download model & vocab.

  UpperCAmelCase__			:				Tuple												=       AutoConfig.from_pretrained(
      model_args.config_name if model_args.config_name else model_args.model_name_or_path      ,					num_labels=a_      ,					finetuning_task=data_args.task_name      ,					cache_dir=model_args.cache_dir      ,					)
  UpperCAmelCase__			:				List[Any]												=       AutoTokenizer.from_pretrained(
      model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path      ,					cache_dir=model_args.cache_dir      ,					)
  UpperCAmelCase__			:				List[str]												=       AutoModelForSequenceClassification.from_pretrained(
      model_args.model_name_or_path      ,					from_tf=bool(".ckpt" in model_args.model_name_or_path		)      ,					config=a_      ,					cache_dir=model_args.cache_dir      ,					)

  # Get datasets
  UpperCAmelCase__			:				Any												=       (
      HansDataset(
          data_dir=data_args.data_dir      ,					tokenizer=a_      ,					task=data_args.task_name      ,					max_seq_length=data_args.max_seq_length      ,					overwrite_cache=data_args.overwrite_cache      ,					)
      if training_args.do_train
      else None
  )
  UpperCAmelCase__			:				Dict												=       (
      HansDataset(
          data_dir=data_args.data_dir      ,					tokenizer=a_      ,					task=data_args.task_name      ,					max_seq_length=data_args.max_seq_length      ,					overwrite_cache=data_args.overwrite_cache      ,					evaluate=a_      ,					)
      if training_args.do_eval
      else None
  )

  # Initialize our Trainer
  UpperCAmelCase__			:				Any												=       Trainer(
      model=a_      ,					args=a_      ,					train_dataset=a_      ,					eval_dataset=a_      ,					data_collator=a_      ,					)

  # Training
  if training_args.do_train:
    trainer.train(
        model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path		) else None		)
    trainer.save_model()
    # For convenience, we also re-save the tokenizer to the same directory,
    # so that you can share your model easily on huggingface.co/models =)
    if trainer.is_world_master():
      tokenizer.save_pretrained(training_args.output_dir		)

    # Evaluation
  if training_args.do_eval:
    logger.info("*** Evaluate ***"		)

    UpperCAmelCase__			:				List[str]												=       trainer.predict(a_		)
    UpperCAmelCase__			:				Dict												=       output.predictions
    UpperCAmelCase__			:				Optional[Any]												=       np.argmax(a_      ,					axis=1		)

    UpperCAmelCase__			:				Union[str, Any]												=       [ex.pairID for ex in eval_dataset]
    UpperCAmelCase__			:				Optional[Any]												=       os.path.join(training_args.output_dir      ,					"hans_predictions.txt"		)
    UpperCAmelCase__			:				Union[str, Any]												=       eval_dataset.get_labels()
    if trainer.is_world_master():
      with open(a_      ,					"w"		) as writer:
        writer.write("pairID,gold_label\n"		)
        for pid, pred in zip(a_      ,					a_		):
          writer.write("ex" + str(a_		) + "," + label_list[int(a_		)] + "\n"		)

    trainer._log(output.metrics		)







def snake_case_    (				lowercase__		):
  # For xla_spawn (TPUs)
  main()


if __name__ == "__main__":
   main()





'''simple docstring'''





import random
import shutil
import tempfile
import unittest

import numpy as np

from transformers import TaTokenizer, TaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torch
from transformers.utils.import_utils import is_speech_available, is_torch_available


if is_torch_available():
      pass

if is_speech_available():
      from transformers import EncodecFeatureExtractor, MusicgenProcessor


SCREAMING_SNAKE_CASE    =					random.Random()







def snake_case_    (				lowercase__      ,					lowercase__=1.0      ,					lowercase__=None      ,					lowercase__=None		):
  if rng is None:
    UpperCAmelCase__			:				Optional[int]												=       global_rng

  UpperCAmelCase__			:				List[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_sentencepiece
class        UpperCAmelCase_       (						unittest.TestCase							):

 '''simple docstring'''







 def        UpperCamelCase     (				self		:     Optional[int]   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Tuple												=       "facebook/musicgen-small"
   UpperCAmelCase__			:				Union[str, Any]												=       tempfile.mkdtemp()







 def        UpperCamelCase     (				self		:     Union[str, Any] ,							**snake_case__		:     Dict   ):



   '''simple docstring'''




   return TaTokenizer.from_pretrained(self.checkpoint ,							**snake_case__   )







 def        UpperCamelCase     (				self		:     Optional[int] ,							**snake_case__		:     str   ):



   '''simple docstring'''




   return EncodecFeatureExtractor.from_pretrained(self.checkpoint ,							**snake_case__   )







 def        UpperCamelCase     (				self		:     List[str]   ):



   '''simple docstring'''




   shutil.rmtree(self.tmpdirname   )







 def        UpperCamelCase     (				self		:     Any   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Any												=       self.get_tokenizer()
   UpperCAmelCase__			:				Tuple												=       self.get_feature_extractor()

   UpperCAmelCase__			:				int												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   processor.save_pretrained(self.tmpdirname   )
   UpperCAmelCase__			:				Dict												=       MusicgenProcessor.from_pretrained(self.tmpdirname   )

   self.assertEqual(processor.tokenizer.get_vocab() ,							tokenizer.get_vocab()   )
   self.assertIsInstance(processor.tokenizer ,							snake_case__   )

   self.assertEqual(processor.feature_extractor.to_json_string() ,							feature_extractor.to_json_string()   )
   self.assertIsInstance(processor.feature_extractor ,							snake_case__   )







 def        UpperCamelCase     (				self		:     Tuple   ):



   '''simple docstring'''




   UpperCAmelCase__			:				int												=       MusicgenProcessor(tokenizer=self.get_tokenizer() ,							feature_extractor=self.get_feature_extractor()   )
   processor.save_pretrained(self.tmpdirname   )

   UpperCAmelCase__			:				List[Any]												=       self.get_tokenizer(bos_token="(BOS)" ,							eos_token="(EOS)"   )
   UpperCAmelCase__			:				int												=       self.get_feature_extractor(do_normalize=snake_case__ ,							padding_value=1.0   )

   UpperCAmelCase__			:				Optional[Any]												=       MusicgenProcessor.from_pretrained(
       self.tmpdirname ,							bos_token="(BOS)" ,							eos_token="(EOS)" ,							do_normalize=snake_case__ ,							padding_value=1.0   )

   self.assertEqual(processor.tokenizer.get_vocab() ,							tokenizer_add_kwargs.get_vocab()   )
   self.assertIsInstance(processor.tokenizer ,							snake_case__   )

   self.assertEqual(processor.feature_extractor.to_json_string() ,							feature_extractor_add_kwargs.to_json_string()   )
   self.assertIsInstance(processor.feature_extractor ,							snake_case__   )







 def        UpperCamelCase     (				self		:     List[Any]   ):



   '''simple docstring'''




   UpperCAmelCase__			:				int												=       self.get_feature_extractor()
   UpperCAmelCase__			:				int												=       self.get_tokenizer()

   UpperCAmelCase__			:				List[str]												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   UpperCAmelCase__			:				Optional[Any]												=       floats_list((3, 10_00)   )

   UpperCAmelCase__			:				List[Any]												=       feature_extractor(snake_case__ ,							return_tensors="np"   )
   UpperCAmelCase__			:				List[Any]												=       processor(snake_case__ ,							return_tensors="np"   )

   for key in input_feat_extract.keys():
     self.assertAlmostEqual(input_feat_extract[key].sum() ,							input_processor[key].sum() ,							delta=1e-2   )







 def        UpperCamelCase     (				self		:     List[str]   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Union[str, Any]												=       self.get_feature_extractor()
   UpperCAmelCase__			:				Union[str, Any]												=       self.get_tokenizer()

   UpperCAmelCase__			:				Optional[int]												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   UpperCAmelCase__			:				List[Any]												=       "This is a test string"

   UpperCAmelCase__			:				Dict												=       processor(text=snake_case__   )

   UpperCAmelCase__			:				Any												=       tokenizer(snake_case__   )

   for key in encoded_tok.keys():
     self.assertListEqual(encoded_tok[key] ,							encoded_processor[key]   )







 def        UpperCamelCase     (				self		:     int   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Union[str, Any]												=       self.get_feature_extractor()
   UpperCAmelCase__			:				str												=       self.get_tokenizer()

   UpperCAmelCase__			:				List[Any]												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   UpperCAmelCase__			:				Optional[int]												=       [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

   UpperCAmelCase__			:				Any												=       processor.batch_decode(sequences=snake_case__   )
   UpperCAmelCase__			:				Optional[Any]												=       tokenizer.batch_decode(snake_case__   )

   self.assertListEqual(snake_case__ ,							snake_case__   )







 def        UpperCamelCase     (				self		:     Tuple   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Any												=       self.get_feature_extractor()
   UpperCAmelCase__			:				List[Any]												=       self.get_tokenizer()

   UpperCAmelCase__			:				Union[str, Any]												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   self.assertListEqual(
       processor.model_input_names ,							feature_extractor.model_input_names ,							msg="`processor` and `feature_extractor` model input names do not match" ,							)







 def        UpperCamelCase     (				self		:     Dict   ):



   '''simple docstring'''




   UpperCAmelCase__			:				int												=       self.get_feature_extractor(padding_side="left"   )
   UpperCAmelCase__			:				str												=       self.get_tokenizer()

   UpperCAmelCase__			:				int												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   UpperCAmelCase__			:				List[Any]												=       [floats_list((1, x)   )[0] for x in range(5 ,							20 ,							5   )]
   UpperCAmelCase__			:				List[Any]												=       processor(snake_case__   ).padding_mask

   UpperCAmelCase__			:				Dict												=       np.asarray(floats_list((3, 20)   )   )[:, None, :]
   UpperCAmelCase__			:				Optional[Any]												=       processor.batch_decode(snake_case__ ,							padding_mask=snake_case__   )

   self.assertIsInstance(snake_case__ ,							snake_case__   )

   for audio in decoded_audios:
     self.assertIsInstance(snake_case__ ,							np.ndarray   )

   self.assertTrue(decoded_audios[0].shape == (1, 10)   )
   self.assertTrue(decoded_audios[1].shape == (1, 15)   )
   self.assertTrue(decoded_audios[2].shape == (1, 20)   )



