


'''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


a__  : List[Any]				=	logging.getLogger(__name__)
@dataclass
class  lowercase_					:
    __UpperCAmelCase             =     field(
        metadata={'help': 'Path to pretrained model or model identifier from huggingface.co/models'}       )
    __UpperCAmelCase             =     field(
        default=a__  ,						metadata={'help': 'Pretrained config name or path if not the same as model_name'}       )
    __UpperCAmelCase             =     field(
        default=a__  ,						metadata={'help': 'Pretrained tokenizer name or path if not the same as model_name'}       )
    __UpperCAmelCase             =     field(
        default=a__  ,						metadata={'help': 'Where do you want to store the pretrained models downloaded from huggingface.co'}  ,						)

@dataclass
class  lowercase_					:
    __UpperCAmelCase             =     field(
        metadata={'help': 'The name of the task to train selected in the list: ' + ', '.join(hans_processors.keys()       )}       )
    __UpperCAmelCase             =     field(
        metadata={'help': 'The input data dir. Should contain the .tsv files (or other data files) for the task.'}       )
    __UpperCAmelCase             =     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.'
            )
        }  ,						)
    __UpperCAmelCase             =     field(
        default=a__  ,						metadata={'help': 'Overwrite the cached training and evaluation sets'}       )




def _UpperCamelCase	(					__A   )							->		Dict[str, torch.Tensor]:


      '''simple docstring'''



      UpperCamelCase__  =    default_data_collator(__A   )
      UpperCamelCase__  =    batch.pop("pairID"				,      __A   )
      return batch




def _UpperCamelCase	(					)							->		Optional[Any]:


      '''simple docstring'''



      UpperCamelCase__  =    HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)   )
      UpperCamelCase__ ,     UpperCamelCase__ ,     UpperCamelCase__  =    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__  =    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__  =    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__  =    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__  =    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__  =    (
          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__  =    (
          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__  =    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__  =    trainer.predict(__A   )
            UpperCamelCase__  =    output.predictions
            UpperCamelCase__  =    np.argmax(__A				,      axis=1   )

            UpperCamelCase__  =    [ex.pairID for ex in eval_dataset]
            UpperCamelCase__  =    os.path.join(training_args.output_dir				,      "hans_predictions.txt"   )
            UpperCamelCase__  =    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 _UpperCamelCase	(					__A   )							->		Optional[Any]:


      '''simple docstring'''



      main()


if __name__ == "__main__":
  main()









'''simple docstring'''







import json
import os
import shutil
import tempfile
import unittest

from transformers import is_speech_available, is_torch_available
from transformers.models.speechta import SpeechTaTokenizer
from transformers.testing_utils import get_tests_dir, require_torch
from transformers.utils import FEATURE_EXTRACTOR_NAME


if is_speech_available() and is_torch_available():
  from transformers import SpeechTaFeatureExtractor, SpeechTaProcessor

  from .test_feature_extraction_speechta import floats_list


a__  : int				=	get_tests_dir('fixtures/test_sentencepiece_bpe_char.model')
@require_torch
class  lowercase_					(   unittest.TestCase       ):


    def        __a							(					self   ):
          UpperCamelCase__  =    tempfile.mkdtemp()

          UpperCamelCase__  =    SpeechTaTokenizer(a   )
          tokenizer.save_pretrained(self.tmpdirname   )

          UpperCamelCase__  =    {
              "feature_size": 1,
              "padding_value": 0.0,
              "sampling_rate": 1_60_00,
              "do_normalize": False,
              "num_mel_bins": 80,
              "hop_length": 16,
              "win_length": 64,
              "win_function": "hann_window",
              "fmin": 80,
              "fmax": 76_00,
              "mel_floor": 1e-10,
              "reduction_factor": 2,
              "return_attention_mask": True,
          }

          UpperCamelCase__  =    os.path.join(self.tmpdirname  , a   )
          with open(self.feature_extraction_file  , "w"  , encoding="utf-8"   ) as fp:
                fp.write(json.dumps(a   ) + "\n"   )


    def        __a							(					self  , **a   ):
          return SpeechTaTokenizer.from_pretrained(self.tmpdirname  , **a   )


    def        __a							(					self  , **a   ):
          return SpeechTaFeatureExtractor.from_pretrained(self.tmpdirname  , **a   )


    def        __a							(					self   ):
          shutil.rmtree(self.tmpdirname   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_tokenizer()
          UpperCamelCase__  =    self.get_feature_extractor()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          processor.save_pretrained(self.tmpdirname   )
          UpperCamelCase__  =    SpeechTaProcessor.from_pretrained(self.tmpdirname   )

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

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


    def        __a							(					self   ):
          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=self.get_tokenizer()  , feature_extractor=self.get_feature_extractor()   )
          processor.save_pretrained(self.tmpdirname   )

          UpperCamelCase__  =    self.get_tokenizer(bos_token="(BOS)"  , eos_token="(EOS)"   )
          UpperCamelCase__  =    self.get_feature_extractor(do_normalize=a  , padding_value=1.0   )

          UpperCamelCase__  =    SpeechTaProcessor.from_pretrained(
              self.tmpdirname  , bos_token="(BOS)"  , eos_token="(EOS)"  , do_normalize=a  , padding_value=1.0   )

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

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


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    floats_list((3, 10_00)   )

          UpperCamelCase__  =    feature_extractor(audio=a  , return_tensors="np"   )
          UpperCamelCase__  =    processor(audio=a  , 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        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    floats_list((3, 10_00)   )

          UpperCamelCase__  =    feature_extractor(audio_target=a  , return_tensors="np"   )
          UpperCamelCase__  =    processor(audio_target=a  , 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        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    "This is a test string"

          UpperCamelCase__  =    processor(text=a   )
          UpperCamelCase__  =    tokenizer(a   )

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


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    "This is a test string"

          UpperCamelCase__  =    processor(text_target=a   )
          UpperCamelCase__  =    tokenizer(a   )

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


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

          UpperCamelCase__  =    processor.batch_decode(a   )
          UpperCamelCase__  =    tokenizer.batch_decode(a   )

          self.assertListEqual(a  , a   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

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





