






"""simple docstring"""
import json
import logging
import os
import re
import sys
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union

import datasets
import numpy as np
import torch
import torchaudio
from packaging import version
from torch import nn

import transformers
from transformers import (
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaForCTC,
    WavaVecaProcessor,
    is_apex_available,
    set_seed,
)
from transformers.trainer_utils import get_last_checkpoint, is_main_process


if is_apex_available():
							from apex import amp


if version.parse(version.parse(torch.__version__).base_version) >= version.parse("1.6"):
							_lowerCAmelCase       =				True
							from torch.cuda.amp import autocast

_lowerCAmelCase       =				logging.getLogger(__name__)






def       __UpperCamelCase						(		snake_case__=None , snake_case__=None       ):
	return field(default_factory=lambda: default , metadata=snake_case__       )



@dataclass
class   SCREAMING_SNAKE_CASE	:





	"""simple docstring"""




	_A							:					str 			=					field(
	    metadata={"""help""": """Path to pretrained model or model identifier from huggingface.co/models"""}    )
	_A							:					Optional[str] 			=					field(
	    default=_SCREAMING_SNAKE_CASE							,							metadata={"""help""": """Where do you want to store the pretrained models downloaded from huggingface.co"""}							,							)
	_A							:					Optional[bool] 			=					field(
	    default=_SCREAMING_SNAKE_CASE							,							metadata={"""help""": """Whether to freeze the feature extractor layers of the model."""}    )
	_A							:					Optional[float] 			=					field(
	    default=0.1							,							metadata={"""help""": """The dropout ratio for the attention probabilities."""}    )
	_A							:					Optional[float] 			=					field(
	    default=0.1							,							metadata={"""help""": """The dropout ratio for activations inside the fully connected layer."""}    )
	_A							:					Optional[float] 			=					field(
	    default=0.1							,							metadata={
	        """help""": """The dropout probabilitiy for all fully connected layers in the embeddings, encoder, and pooler."""
	    }							,							)
	_A							:					Optional[float] 			=					field(
	    default=0.1							,							metadata={"""help""": """The dropout probabilitiy for all 1D convolutional layers in feature extractor."""}							,							)
	_A							:					Optional[float] 			=					field(
	    default=0.05							,							metadata={
	        """help""": (
	            """Propability of each feature vector along the time axis to be chosen as the start of the vector"""
	            """span to be masked. Approximately ``mask_time_prob * sequence_length // mask_time_length`` feature"""
	            """vectors will be masked along the time axis. This is only relevant if ``apply_spec_augment is True``."""
	        )
	    }							,							)
	_A							:					Optional[float] 			=					field(default=0.0							,							metadata={"""help""": """The LayerDrop probability."""}    )






@dataclass
class   SCREAMING_SNAKE_CASE	:





	"""simple docstring"""




	_A							:					Optional[str] 			=					field(
	    default=_SCREAMING_SNAKE_CASE							,							metadata={"""help""": """The configuration name of the dataset to use (via the datasets library)."""}    )
	_A							:					Optional[str] 			=					field(
	    default="""train+validation"""							,							metadata={
	        """help""": """The name of the training data set split to use (via the datasets library). Defaults to 'train'"""
	    }							,							)
	_A							:					bool 			=					field(
	    default=_SCREAMING_SNAKE_CASE							,							metadata={"""help""": """Overwrite the cached preprocessed datasets or not."""}    )
	_A							:					Optional[int] 			=					field(
	    default=_SCREAMING_SNAKE_CASE							,							metadata={"""help""": """The number of processes to use for the preprocessing."""}							,							)
	_A							:					Optional[int] 			=					field(
	    default=_SCREAMING_SNAKE_CASE							,							metadata={
	        """help""": (
	            """For debugging purposes or quicker training, truncate the number of training examples to this """
	            """value if set."""
	        )
	    }							,							)
	_A							:					Optional[int] 			=					field(
	    default=_SCREAMING_SNAKE_CASE							,							metadata={
	        """help""": (
	            """For debugging purposes or quicker training, truncate the number of validation examples to this """
	            """value if set."""
	        )
	    }							,							)
	_A							:					List[str] 			=					list_field(
	    default=[""",""", """?""", """.""", """!""", """-""", """;""", """:""", """\"\"""", """%""", """'""", """\"""", """�"""]							,							metadata={"""help""": """A list of characters to remove from the transcripts."""}							,							)






@dataclass
class   SCREAMING_SNAKE_CASE	:





	"""simple docstring"""




	_A							:					WavaVecaProcessor
	_A							:					Union[bool, str] 			=					True
	_A							:					Optional[int] 			=					None
	_A							:					Optional[int] 			=					None
	_A							:					Optional[int] 			=					None
	_A							:					Optional[int] 			=					None





	def __call__(self  ,  lowerCAmelCase_				):
		# split inputs and labels since they have to be of different lenghts and need
		# different padding methods
		A_		:    str						     = [{"""input_values""": feature["""input_values"""]} for feature in features]
		A_		:    int						     = [{"""input_ids""": feature["""labels"""]} for feature in features]

		A_		:    Union[str, Any]						     = self.processor.pad(
		    lowerCAmelCase_  ,  padding=self.padding  ,  max_length=self.max_length  ,  pad_to_multiple_of=self.pad_to_multiple_of  ,  return_tensors="""pt"""  ,  )
		A_		:    Optional[int]						     = self.processor.pad(
		    labels=lowerCAmelCase_  ,  padding=self.padding  ,  max_length=self.max_length_labels  ,  pad_to_multiple_of=self.pad_to_multiple_of_labels  ,  return_tensors="""pt"""  ,  )

		# replace padding with -100 to ignore loss correctly
		A_		:    Optional[int]						     = labels_batch["""input_ids"""].masked_fill(labels_batch.attention_mask.ne(1				)  ,  -100				)

		A_		:    Optional[int]						     = labels

		return batch






class   SCREAMING_SNAKE_CASE	(					_SCREAMING_SNAKE_CASE    ):





	"""simple docstring"""




	def  lowerCamelCase(self  ,  lowerCAmelCase_  ,  lowerCAmelCase_				):
		model.train()
		A_		:    Tuple						     = self._prepare_inputs(lowerCAmelCase_				)

		if self.use_amp:
			with autocast():
				A_		:    str						     = self.compute_loss(lowerCAmelCase_  ,  lowerCAmelCase_				)
		else:
			A_		:    int						     = self.compute_loss(lowerCAmelCase_  ,  lowerCAmelCase_				)

		if self.args.n_gpu > 1:
			if model.module.config.ctc_loss_reduction == "mean":
				A_		:    Optional[int]						     = loss.mean()
			elif model.module.config.ctc_loss_reduction == "sum":
				A_		:    Union[str, Any]						     = loss.sum() / (inputs["""labels"""] >= 0).sum()
			else:
				raise ValueError(f"""{model.config.ctc_loss_reduction} is not valid. Choose one of ['mean', 'sum']"""				)

		if self.args.gradient_accumulation_steps > 1:
			A_		:    Any						     = loss / self.args.gradient_accumulation_steps

		if self.use_amp:
			self.scaler.scale(lowerCAmelCase_				).backward()
		elif self.use_apex:
			with amp.scale_loss(lowerCAmelCase_  ,  self.optimizer				) as scaled_loss:
				scaled_loss.backward()
		elif self.deepspeed:
			self.deepspeed.backward(lowerCAmelCase_				)
		else:
			loss.backward()

		return loss.detach()






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

	A_		:    List[str]						     = HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)       )
	if len(sys.argv       ) == 2 and sys.argv[1].endswith(""".json"""       ):
		# If we pass only one argument to the script and it's the path to a json file,
		# let's parse it to get our arguments.
		A_     ,  A_     ,  A_		:    Any						     = parser.parse_json_file(json_file=os.path.abspath(sys.argv[1]       )       )
	else:
		A_     ,  A_     ,  A_		:    List[str]						     = parser.parse_args_into_dataclasses()

	# Detecting last checkpoint.
	A_		:    List[str]						     = None
	if os.path.isdir(training_args.output_dir       ) and training_args.do_train and not training_args.overwrite_output_dir:
		A_		:    Union[str, Any]						     = get_last_checkpoint(training_args.output_dir       )
		if last_checkpoint is None and len(os.listdir(training_args.output_dir       )       ) > 0:
			raise ValueError(
			    F"""Output directory ({training_args.output_dir}) already exists and is not empty. """
			    """Use --overwrite_output_dir to overcome."""       )
		elif last_checkpoint is not None:
			logger.info(
			    F"""Checkpoint detected, resuming training at {last_checkpoint}. To avoid this behavior, change """
			    """the `--output_dir` or add `--overwrite_output_dir` to train from scratch."""       )

    # Setup logging
	logging.basicConfig(
	    format="""%(asctime)s - %(levelname)s - %(name)s - %(message)s""" , datefmt="""%m/%d/%Y %H:%M:%S""" , handlers=[logging.StreamHandler(sys.stdout       )] , )
	logger.setLevel(logging.INFO if is_main_process(training_args.local_rank       ) else logging.WARN       )

	# Log on each process the small summary:
	logger.warning(
	    F"""Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}"""
	    + F"""distributed training: {bool(training_args.local_rank != -1       )}, 16-bits training: {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()
	logger.info("""Training/evaluation parameters %s""" , snake_case__       )

	# Set seed before initializing model.
	set_seed(training_args.seed       )

	# Get the datasets:
	A_		:    str						     = datasets.load_dataset(
	    """common_voice""" , data_args.dataset_config_name , split=data_args.train_split_name       )
	A_		:    Dict						     = datasets.load_dataset("""common_voice""" , data_args.dataset_config_name , split="""test"""       )

	# Create and save tokenizer
	A_		:    Optional[Any]						     = F"""[{"".join(data_args.chars_to_ignore       )}]"""

	def remove_special_characters(snake_case__       ):
		A_		:    Tuple						     = re.sub(snake_case__ , """""" , batch["""sentence"""]       ).lower() + """ """
		return batch

	A_		:    Dict						     = train_dataset.map(snake_case__ , remove_columns=["""sentence"""]       )
	A_		:    Optional[int]						     = eval_dataset.map(snake_case__ , remove_columns=["""sentence"""]       )

	def extract_all_chars(snake_case__       ):
		A_		:    List[str]						     = """ """.join(batch["""text"""]       )
		A_		:    Tuple						     = list(set(snake_case__       )       )
		return {"vocab": [vocab], "all_text": [all_text]}

	A_		:    Optional[Any]						     = train_dataset.map(
	    snake_case__ , batched=snake_case__ , batch_size=-1 , keep_in_memory=snake_case__ , remove_columns=train_dataset.column_names , )
	A_		:    List[Any]						     = train_dataset.map(
	    snake_case__ , batched=snake_case__ , batch_size=-1 , keep_in_memory=snake_case__ , remove_columns=eval_dataset.column_names , )

	A_		:    Any						     = list(set(vocab_train["""vocab"""][0]       ) | set(vocab_test["""vocab"""][0]       )       )
	A_		:    List[str]						     = {v: k for k, v in enumerate(snake_case__       )}
	A_		:    List[Any]						     = vocab_dict[""" """]
	del vocab_dict[" "]
	A_		:    Optional[Any]						     = len(snake_case__       )
	A_		:    Optional[int]						     = len(snake_case__       )

	with open("""vocab.json""" , """w"""       ) as vocab_file:
		json.dump(snake_case__ , snake_case__       )

	# Load pretrained model and tokenizer
	#
	# Distributed training:
	# The .from_pretrained methods guarantee that only one local process can concurrently
	# download model & vocab.
	A_		:    List[str]						     = WavaVecaCTCTokenizer(
	    """vocab.json""" , unk_token="""[UNK]""" , pad_token="""[PAD]""" , word_delimiter_token="""|""" , )
	A_		:    Union[str, Any]						     = WavaVecaFeatureExtractor(
	    feature_size=1 , sampling_rate=16_000 , padding_value=0.0 , do_normalize=snake_case__ , return_attention_mask=snake_case__       )
	A_		:    Union[str, Any]						     = WavaVecaProcessor(feature_extractor=snake_case__ , tokenizer=snake_case__       )
	A_		:    Union[str, Any]						     = WavaVecaForCTC.from_pretrained(
	    model_args.model_name_or_path , cache_dir=model_args.cache_dir , activation_dropout=model_args.activation_dropout , attention_dropout=model_args.attention_dropout , hidden_dropout=model_args.hidden_dropout , feat_proj_dropout=model_args.feat_proj_dropout , mask_time_prob=model_args.mask_time_prob , gradient_checkpointing=training_args.gradient_checkpointing , layerdrop=model_args.layerdrop , ctc_loss_reduction="""mean""" , pad_token_id=processor.tokenizer.pad_token_id , vocab_size=len(processor.tokenizer       ) , )

	if data_args.max_train_samples is not None:
		A_		:    List[Any]						     = min(len(snake_case__       ) , data_args.max_train_samples       )
		A_		:    Union[str, Any]						     = train_dataset.select(range(snake_case__       )       )

	if data_args.max_val_samples is not None:
		A_		:    List[str]						     = eval_dataset.select(range(data_args.max_val_samples       )       )

	A_		:    Union[str, Any]						     = torchaudio.transforms.Resample(48_000 , 16_000       )

	# Preprocessing the datasets.
	# We need to read the aduio files as arrays and tokenize the targets.
	def speech_file_to_array_fn(snake_case__       ):
		A_     ,  A_		:    Optional[int]						     = torchaudio.load(batch["""path"""]       )
		A_		:    List[Any]						     = resampler(snake_case__       ).squeeze().numpy()
		A_		:    Union[str, Any]						     = 16_000
		A_		:    Any						     = batch["""text"""]
		return batch

	A_		:    Tuple						     = train_dataset.map(
	    snake_case__ , remove_columns=train_dataset.column_names , num_proc=data_args.preprocessing_num_workers , )
	A_		:    List[str]						     = eval_dataset.map(
	    snake_case__ , remove_columns=eval_dataset.column_names , num_proc=data_args.preprocessing_num_workers , )

	def prepare_dataset(snake_case__       ):
		# check that all files have the correct sampling rate
		assert (
		    len(set(batch["""sampling_rate"""]       )       ) == 1
		), F"""Make sure all inputs have the same sampling rate of {processor.feature_extractor.sampling_rate}."""

		A_		:    int						     = processor(
		    audio=batch["""speech"""] , text=batch["""target_text"""] , sampling_rate=batch["""sampling_rate"""][0]       )
		batch.update(snake_case__       )
		return batch

	A_		:    Dict						     = train_dataset.map(
	    snake_case__ , remove_columns=train_dataset.column_names , batch_size=training_args.per_device_train_batch_size , batched=snake_case__ , num_proc=data_args.preprocessing_num_workers , )
	A_		:    int						     = eval_dataset.map(
	    snake_case__ , remove_columns=eval_dataset.column_names , batch_size=training_args.per_device_train_batch_size , batched=snake_case__ , num_proc=data_args.preprocessing_num_workers , )

	# Metric
	A_		:    List[str]						     = datasets.load_metric("""wer"""       )

	def compute_metrics(snake_case__       ):
		A_		:    str						     = pred.predictions
		A_		:    Any						     = np.argmax(snake_case__ , axis=-1       )

		A_		:    str						     = processor.tokenizer.pad_token_id

		A_		:    Tuple						     = processor.batch_decode(snake_case__       )
		# we do not want to group tokens when computing the metrics
		A_		:    Optional[int]						     = processor.batch_decode(pred.label_ids , group_tokens=snake_case__       )

		A_		:    List[Any]						     = wer_metric.compute(predictions=snake_case__ , references=snake_case__       )

		return {"wer": wer}

	if model_args.freeze_feature_extractor:
		model.freeze_feature_extractor()

	# Data collator
	A_		:    Optional[Any]						     = DataCollatorCTCWithPadding(processor=snake_case__ , padding=snake_case__       )

	# Initialize our Trainer
	A_		:    Optional[int]						     = CTCTrainer(
	    model=snake_case__ , data_collator=snake_case__ , args=snake_case__ , compute_metrics=snake_case__ , train_dataset=train_dataset if training_args.do_train else None , eval_dataset=eval_dataset if training_args.do_eval else None , tokenizer=processor.feature_extractor , )

	# Training
	if training_args.do_train:
		if last_checkpoint is not None:
			A_		:    Optional[int]						     = last_checkpoint
		elif os.path.isdir(model_args.model_name_or_path       ):
			A_		:    Dict						     = model_args.model_name_or_path
		else:
			A_		:    Optional[int]						     = None

		# Save the feature_extractor and the tokenizer
		if is_main_process(training_args.local_rank       ):
			processor.save_pretrained(training_args.output_dir       )

		A_		:    Tuple						     = trainer.train(resume_from_checkpoint=snake_case__       )
		trainer.save_model()

		A_		:    List[str]						     = train_result.metrics
		A_		:    str						     = (
		    data_args.max_train_samples if data_args.max_train_samples is not None else len(snake_case__       )
		)
		A_		:    Dict						     = min(snake_case__ , len(snake_case__       )       )

		trainer.log_metrics("""train""" , snake_case__       )
		trainer.save_metrics("""train""" , snake_case__       )
		trainer.save_state()

	# Evaluation
	A_		:    int						     = {}
	if training_args.do_eval:
		logger.info("""*** Evaluate ***"""       )
		A_		:    Tuple						     = trainer.evaluate()
		A_		:    Optional[int]						     = data_args.max_val_samples if data_args.max_val_samples is not None else len(snake_case__       )
		A_		:    List[Any]						     = min(snake_case__ , len(snake_case__       )       )

		trainer.log_metrics("""eval""" , snake_case__       )
		trainer.save_metrics("""eval""" , snake_case__       )

	return results


if __name__ == "__main__":
							main()









"""simple docstring"""
import warnings
from typing import Dict

import numpy as np

from ..utils import ExplicitEnum, add_end_docstrings, is_tf_available, is_torch_available
from .base import PIPELINE_INIT_ARGS, GenericTensor, Pipeline


if is_tf_available():
							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING

if is_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING






def       __UpperCamelCase						(		snake_case__       ):
	return 1.0 / (1.0 + np.exp(-_outputs       ))






def       __UpperCamelCase						(		snake_case__       ):
	A_		:    Union[str, Any]						     = np.max(_outputs , axis=-1 , keepdims=snake_case__       )
	A_		:    str						     = np.exp(_outputs - maxes       )
	return shifted_exp / shifted_exp.sum(axis=-1 , keepdims=snake_case__       )



class   SCREAMING_SNAKE_CASE	(					_SCREAMING_SNAKE_CASE    ):





	"""simple docstring"""




	_A							:					Any 			=					"""sigmoid"""
	_A							:					Any 			=					"""softmax"""
	_A							:					Union[str, Any] 			=					"""none"""






@add_end_docstrings(
    _SCREAMING_SNAKE_CASE							,							R"""
        return_all_scores (`bool`, *optional*, defaults to `False`):
            Whether to return all prediction scores or just the one of the predicted class.
        function_to_apply (`str`, *optional*, defaults to `\"default\"`):
            The function to apply to the model outputs in order to retrieve the scores. Accepts four different values:

            - `\"default\"`: if the model has a single label, will apply the sigmoid function on the output. If the model
              has several labels, will apply the softmax function on the output.
            - `\"sigmoid\"`: Applies the sigmoid function on the output.
            - `\"softmax\"`: Applies the softmax function on the output.
            - `\"none\"`: Does not apply any function on the output.
    """							,							)
class   SCREAMING_SNAKE_CASE	(					_SCREAMING_SNAKE_CASE    ):





	"""simple docstring"""




	_A							:					Optional[int] 			=					False
	_A							:					Dict 			=					ClassificationFunction.NONE





	def __init__(self  ,  **lowerCAmelCase_				):
		super().__init__(**lowerCAmelCase_				)

		self.check_model_type(
		    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING
		    if self.framework == """tf"""
		    else MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING				)





	def  lowerCamelCase(self  ,  lowerCAmelCase_=None  ,  lowerCAmelCase_=None  ,  lowerCAmelCase_=""  ,  **lowerCAmelCase_				):
		# Using "" as default argument because we're going to use `top_k=None` in user code to declare
		# "No top_k"
		A_		:    Union[str, Any]						     = tokenizer_kwargs

		A_		:    List[str]						     = {}
		if hasattr(self.model.config  ,  """return_all_scores"""				) and return_all_scores is None:
			A_		:    Optional[int]						     = self.model.config.return_all_scores

		if isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				) or top_k is None:
			A_		:    Dict						     = top_k
			A_		:    Any						     = False
		elif return_all_scores is not None:
			warnings.warn(
			    """`return_all_scores` is now deprecated,  if want a similar functionality use `top_k=None` instead of"""
			    """ `return_all_scores=True` or `top_k=1` instead of `return_all_scores=False`."""  ,  lowerCAmelCase_  ,  )
			if return_all_scores:
				A_		:    List[Any]						     = None
			else:
				A_		:    List[str]						     = 1

		if isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				):
			A_		:    str						     = ClassificationFunction[function_to_apply.upper()]

		if function_to_apply is not None:
			A_		:    List[str]						     = function_to_apply
		return preprocess_params, {}, postprocess_params





	def __call__(self  ,  *lowerCAmelCase_  ,  **lowerCAmelCase_				):
		A_		:    List[str]						     = super().__call__(*lowerCAmelCase_  ,  **lowerCAmelCase_				)
		# TODO try and retrieve it in a nicer way from _sanitize_parameters.
		A_		:    Union[str, Any]						     = """top_k""" not in kwargs
		if isinstance(args[0]  ,  lowerCAmelCase_				) and _legacy:
			# This pipeline is odd, and return a list when single item is run
			return [result]
		else:
			return result





	def  lowerCamelCase(self  ,  lowerCAmelCase_  ,  **lowerCAmelCase_				):
		A_		:    Union[str, Any]						     = self.framework
		if isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				):
			return self.tokenizer(**lowerCAmelCase_  ,  return_tensors=lowerCAmelCase_  ,  **lowerCAmelCase_				)
		elif isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				) and len(lowerCAmelCase_				) == 1 and isinstance(inputs[0]  ,  lowerCAmelCase_				) and len(inputs[0]				) == 2:
			# It used to be valid to use a list of list of list for text pairs, keeping this path for BC
			return self.tokenizer(
			    text=inputs[0][0]  ,  text_pair=inputs[0][1]  ,  return_tensors=lowerCAmelCase_  ,  **lowerCAmelCase_				)
		elif isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				):
			# This is likely an invalid usage of the pipeline attempting to pass text pairs.
			raise ValueError(
			    """The pipeline received invalid inputs, if you are trying to send text pairs, you can try to send a"""
			    """ dictionary `{\"text\": \"My text\", \"text_pair\": \"My pair\"}` in order to send a text pair."""				)
		return self.tokenizer(lowerCAmelCase_  ,  return_tensors=lowerCAmelCase_  ,  **lowerCAmelCase_				)





	def  lowerCamelCase(self  ,  lowerCAmelCase_				):
		return self.model(**lowerCAmelCase_				)





	def  lowerCamelCase(self  ,  lowerCAmelCase_  ,  lowerCAmelCase_=None  ,  lowerCAmelCase_=1  ,  lowerCAmelCase_=True				):
		# `_legacy` is used to determine if we're running the naked pipeline and in backward
		# compatibility mode, or if running the pipeline with `pipeline(..., top_k=1)` we're running
		# the more natural result containing the list.
		# Default value before `set_parameters`
		if function_to_apply is None:
			if self.model.config.problem_type == "multi_label_classification" or self.model.config.num_labels == 1:
				A_		:    Optional[Any]						     = ClassificationFunction.SIGMOID
			elif self.model.config.problem_type == "single_label_classification" or self.model.config.num_labels > 1:
				A_		:    Tuple						     = ClassificationFunction.SOFTMAX
			elif hasattr(self.model.config  ,  """function_to_apply"""				) and function_to_apply is None:
				A_		:    Any						     = self.model.config.function_to_apply
			else:
				A_		:    Dict						     = ClassificationFunction.NONE

		A_		:    Optional[Any]						     = model_outputs["""logits"""][0]
		A_		:    Tuple						     = outputs.numpy()

		if function_to_apply == ClassificationFunction.SIGMOID:
			A_		:    str						     = sigmoid(lowerCAmelCase_				)
		elif function_to_apply == ClassificationFunction.SOFTMAX:
			A_		:    Dict						     = softmax(lowerCAmelCase_				)
		elif function_to_apply == ClassificationFunction.NONE:
			A_		:    Optional[int]						     = outputs
		else:
			raise ValueError(f"""Unrecognized `function_to_apply` argument: {function_to_apply}"""				)

		if top_k == 1 and _legacy:
			return {"label": self.model.config.idalabel[scores.argmax().item()], "score": scores.max().item()}

		A_		:    Optional[Any]						     = [
		    {"""label""": self.model.config.idalabel[i], """score""": score.item()} for i, score in enumerate(lowerCAmelCase_				)
		]
		if not _legacy:
			dict_scores.sort(key=lambda lowerCAmelCase_				: x["score"]  ,  reverse=lowerCAmelCase_				)
			if top_k is not None:
				A_		:    str						     = dict_scores[:top_k]
		return dict_scores

