
'''simple docstring'''



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

import numpy as np
from utils_multiple_choice import MultipleChoiceDataset, Split, processors

import transformers
from transformers import (
    AutoConfig,
    AutoModelForMultipleChoice,
    AutoTokenizer,
    DataCollatorWithPadding,
    EvalPrediction,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    set_seed,
)
from transformers.trainer_utils import is_main_process


_lowercase          =						logging.getLogger(__name__)


def A (__lowerCamelCase :Tuple					,      __lowerCamelCase :Dict						):
						return (preds == labels).mean()




@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=_SCREAMING_SNAKE_CASE			,				metadata={'''help''': '''Pretrained config name or path if not the same as model_name'''}					)
						_lowercase						: Optional[str]										=  field(
						    default=_SCREAMING_SNAKE_CASE			,				metadata={'''help''': '''Pretrained tokenizer name or path if not the same as model_name'''}					)
						_lowercase						: Optional[str]										=  field(
						    default=_SCREAMING_SNAKE_CASE			,				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 on: ''' + ''', '''.join(processors.keys()					)}					)
						_lowercase						: str										=  field(metadata={'''help''': '''Should contain the data files for the task.'''}					)
						_lowercase						: int										=  field(
						    default=1_2_8			,				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=_SCREAMING_SNAKE_CASE			,				metadata={'''help''': '''Overwrite the cached training and evaluation sets'''}					)






def A ():
						# 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.

						_lowerCAmelCase    		=					HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)						)
						_lowerCAmelCase		,       _lowerCAmelCase		,       _lowerCAmelCase    		=					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"""					,      __lowerCamelCase						)

						# Set seed
						set_seed(training_args.seed						)

						try:
												_lowerCAmelCase    		=					processors[data_args.task_name]()
												_lowerCAmelCase    		=					processor.get_labels()
												_lowerCAmelCase    		=					len(__lowerCamelCase						)
						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.

						_lowerCAmelCase    		=					AutoConfig.from_pretrained(
						    model_args.config_name if model_args.config_name else model_args.model_name_or_path					,      num_labels=__lowerCamelCase					,      finetuning_task=data_args.task_name					,      cache_dir=model_args.cache_dir					,      )
						_lowerCAmelCase    		=					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					,      )
						_lowerCAmelCase    		=					AutoModelForMultipleChoice.from_pretrained(
						    model_args.model_name_or_path					,      from_tf=bool(""".ckpt""" in model_args.model_name_or_path						)					,      config=__lowerCamelCase					,      cache_dir=model_args.cache_dir					,      )

						# Get datasets
						_lowerCAmelCase    		=					(
						    MultipleChoiceDataset(
						        data_dir=data_args.data_dir					,      tokenizer=__lowerCamelCase					,      task=data_args.task_name					,      max_seq_length=data_args.max_seq_length					,      overwrite_cache=data_args.overwrite_cache					,      mode=Split.train					,      )
						    if training_args.do_train
						    else None
						)
						_lowerCAmelCase    		=					(
						    MultipleChoiceDataset(
						        data_dir=data_args.data_dir					,      tokenizer=__lowerCamelCase					,      task=data_args.task_name					,      max_seq_length=data_args.max_seq_length					,      overwrite_cache=data_args.overwrite_cache					,      mode=Split.dev					,      )
						    if training_args.do_eval
						    else None
						)

						def compute_metrics(__lowerCamelCase :EvalPrediction						) -> Dict:
												_lowerCAmelCase    		=					np.argmax(p.predictions					,      axis=1						)
												return {"acc": simple_accuracy(__lowerCamelCase					,      p.label_ids						)}

						# Data collator
						_lowerCAmelCase    		=					DataCollatorWithPadding(__lowerCamelCase					,      pad_to_multiple_of=8						) if training_args.fpaa else None

						# Initialize our Trainer
						_lowerCAmelCase    		=					Trainer(
						    model=__lowerCamelCase					,      args=__lowerCamelCase					,      train_dataset=__lowerCamelCase					,      eval_dataset=__lowerCamelCase					,      compute_metrics=__lowerCamelCase					,      data_collator=__lowerCamelCase					,      )

						# 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
						_lowerCAmelCase    		=					{}
						if training_args.do_eval:
												logger.info("""*** Evaluate ***"""						)

												_lowerCAmelCase    		=					trainer.evaluate()

												_lowerCAmelCase    		=					os.path.join(training_args.output_dir					,      """eval_results.txt"""						)
												if trainer.is_world_master():
																		with open(__lowerCamelCase					,      """w"""						) as writer:
																								logger.info("""***** Eval results *****"""						)
																								for key, value in result.items():
																														logger.info("""  %s = %s"""					,      __lowerCamelCase					,      __lowerCamelCase						)
																														writer.write("""%s = %s\n""" % (key, value)						)

																								results.update(__lowerCamelCase						)

						return results


def A (__lowerCamelCase :int						):
						# For xla_spawn (TPUs)
						main()


if __name__ == "__main__":
						main()


'''simple docstring'''



from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


_lowercase          =						logging.get_logger(__name__)




@add_end_docstrings(_SCREAMING_SNAKE_CASE					)
class     UpperCAmelCase_	(						_SCREAMING_SNAKE_CASE					):






						'''simple docstring'''






						def __init__(   self       ,  **_lowercase ):

												"""simple docstring"""



												super().__init__(**_lowercase )

												if self.framework != "pt":
																		raise ValueError(F'The {self.__class__} is only available in PyTorch.' )
												# No specific FOR_XXX available yet




						def __call__(   self       ,  _lowercase       ,  **_lowercase ):

												"""simple docstring"""



												return super().__call__(_lowercase       ,  **_lowercase )




						def    _lowercase      (   self       ,  **_lowercase ):

												"""simple docstring"""



												_lowerCAmelCase    		=					{}
												if "candidate_labels" in kwargs:
																		_lowerCAmelCase    		=					kwargs["""candidate_labels"""]
												if "hypothesis_template" in kwargs:
																		_lowerCAmelCase    		=					kwargs["""hypothesis_template"""]

												return preprocess_params, {}, {}




						def    _lowercase      (   self       ,  _lowercase       ,  _lowercase=None       ,  _lowercase="This is a sound of {}." ):

												"""simple docstring"""



												if isinstance(_lowercase       ,  _lowercase ):
																		if audio.startswith("""http://""" ) or audio.startswith("""https://""" ):
																								# We need to actually check for a real protocol, otherwise it's impossible to use a local file
																								# like http_huggingface_co.png
																								_lowerCAmelCase    		=					requests.get(_lowercase ).content
																		else:
																								with open(_lowercase       ,  """rb""" ) as f:
																														_lowerCAmelCase    		=					f.read()

												if isinstance(_lowercase       ,  _lowercase ):
																		_lowerCAmelCase    		=					ffmpeg_read(_lowercase       ,  self.feature_extractor.sampling_rate )

												if not isinstance(_lowercase       ,  np.ndarray ):
																		raise ValueError("""We expect a numpy ndarray as input""" )
												if len(audio.shape ) != 1:
																		raise ValueError("""We expect a single channel audio input for ZeroShotAudioClassificationPipeline""" )

												_lowerCAmelCase    		=					self.feature_extractor(
												    [audio]       ,  sampling_rate=self.feature_extractor.sampling_rate       ,  return_tensors="""pt""" )
												_lowerCAmelCase    		=					candidate_labels
												_lowerCAmelCase    		=					[hypothesis_template.format(_lowercase ) for x in candidate_labels]
												_lowerCAmelCase    		=					self.tokenizer(_lowercase       ,  return_tensors=self.framework       ,  padding=_lowercase )
												_lowerCAmelCase    		=					[text_inputs]
												return inputs




						def    _lowercase      (   self       ,  _lowercase ):

												"""simple docstring"""



												_lowerCAmelCase    		=					model_inputs.pop("""candidate_labels""" )
												_lowerCAmelCase    		=					model_inputs.pop("""text_inputs""" )
												if isinstance(text_inputs[0]       ,  _lowercase ):
																		_lowerCAmelCase    		=					text_inputs[0]
												else:
																		# Batching case.
																		_lowerCAmelCase    		=					text_inputs[0][0]

												_lowerCAmelCase    		=					self.model(**_lowercase       ,  **_lowercase )

												_lowerCAmelCase    		=					{
												    """candidate_labels""": candidate_labels,
												    """logits""": outputs.logits_per_audio,
												}
												return model_outputs




						def    _lowercase      (   self       ,  _lowercase ):

												"""simple docstring"""



												_lowerCAmelCase    		=					model_outputs.pop("""candidate_labels""" )
												_lowerCAmelCase    		=					model_outputs["""logits"""][0]

												if self.framework == "pt":
																		_lowerCAmelCase    		=					logits.softmax(dim=0 )
																		_lowerCAmelCase    		=					probs.tolist()
												else:
																		raise ValueError("""`tf` framework not supported.""" )

												_lowerCAmelCase    		=					[
												    {"""score""": score, """label""": candidate_label}
												    for score, candidate_label in sorted(zip(_lowercase       ,  _lowercase )       ,  key=lambda _lowercase : -x[0] )
												]
												return result
